11. Python API reference

11.1. High-level API

11.1.1. Device management

class model_runtime.OptionFlags

__init__(*args, **kwargs)

Overloaded function.

1. __init__(self: model_runtime.OptionFlags) -> None

2. __init__(self: model_runtime.OptionFlags, arg0: List[Tuple[str, str]]) -> None

at(self: model_runtime.OptionFlags, option: str) → str

clear(self: model_runtime.OptionFlags) → None

set(*args, **kwargs)

Overloaded function.

1. set(self: model_runtime.OptionFlags, list: List[Tuple[str, str]]) -> None

2. set(self: model_runtime.OptionFlags, option: str, value: str) -> None

class model_runtime.DeviceWaitStrategy

Members:

NO_WAIT

WAIT_WITH_TIMEOUT

WAIT_FOREVER

property name

class model_runtime.DeviceWaitConfig

__init__(self: model_runtime.DeviceWaitConfig, strategy: model_runtime.DeviceWaitStrategy = NO_WAIT, timeout: datetime.timedelta = 0s, sleepTime: datetime.timedelta = 15s) → None

__repr__(self: model_runtime.DeviceWaitConfig) → str

property sleepTime

property strategy

property timeout

class model_runtime.DeviceConstraints

__init__(self: model_runtime.DeviceConstraints, requiresRemoteBuffersSupport: bool = False) → None

__repr__(self: model_runtime.DeviceConstraints) → str

property requiresRemoteBuffersSupport

class model_runtime.Device

__repr__(self: model_runtime.Device) → str

attach(self: model_runtime.Device) → None

detach(self: model_runtime.Device) → None

getId(self: model_runtime.Device) → int

getNumIPUs(self: model_runtime.Device) → int

ipuVersion(self: model_runtime.Device) → int

class model_runtime.DeviceManager

__init__(self: model_runtime.DeviceManager) → None

createIpuModelDevice(*args, **kwargs)

Overloaded function.

1. createIpuModelDevice(self: model_runtime.DeviceManager, num_ipus: int, ipu_version: int = 2, tiles_per_ipu: int = 0) -> model_runtime.Device

2. createIpuModelDevice(self: model_runtime.DeviceManager, model: popef::Model, tiles_per_ipu: int = 0) -> model_runtime.Device

createSimulatorDevice(*args, **kwargs)

Overloaded function.

1. createSimulatorDevice(self: model_runtime.DeviceManager, num_ipus: int, tiles_per_ipu: int, ipu_ver: int) -> model_runtime.Device

2. createSimulatorDevice(self: model_runtime.DeviceManager, model: popef::Model) -> model_runtime.Device

createSmallIpuModelDevice(*args, **kwargs)

Overloaded function.

1. createSmallIpuModelDevice(self: model_runtime.DeviceManager, num_ipus: int, ipu_version: int = 2) -> model_runtime.Device

2. createSmallIpuModelDevice(self: model_runtime.DeviceManager, model: popef::Model) -> model_runtime.Device

getDevice(*args, **kwargs)

Overloaded function.

1. getDevice(self: model_runtime.DeviceManager, model: popef::Model, wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig()) -> model_runtime.Device

2. getDevice(self: model_runtime.DeviceManager, num_ipus: int, device_options: model_runtime.OptionFlags = model_runtime.OptionFlags(), wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig(), constrains: model_runtime.DeviceConstraints = model_runtime.DeviceConstraints()) -> model_runtime.Device

getSpecificDevice(*args, **kwargs)

Overloaded function.

1. getSpecificDevice(self: model_runtime.DeviceManager, device_id: int, model: popef::Model, wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig()) -> model_runtime.Device

2. getSpecificDevice(self: model_runtime.DeviceManager, dev_id: int, device_options: model_runtime.OptionFlags = model_runtime.OptionFlags(), wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig()) -> model_runtime.Device

ipuHardwareVersion(self: model_runtime.DeviceManager) → int

tryGetDevice(*args, **kwargs)

Overloaded function.

1. tryGetDevice(self: model_runtime.DeviceManager, model: popef::Model, wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig()) -> model_runtime.Device

2. tryGetDevice(self: model_runtime.DeviceManager, num_ipus: int, device_options: model_runtime.OptionFlags = model_runtime.OptionFlags(), wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig(), constrains: model_runtime.DeviceConstraints = model_runtime.DeviceConstraints()) -> model_runtime.Device

tryGetSpecificDevice(*args, **kwargs)

Overloaded function.

1. tryGetSpecificDevice(self: model_runtime.DeviceManager, device_id: int, model: popef::Model, wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig()) -> model_runtime.Device

2. tryGetSpecificDevice(self: model_runtime.DeviceManager, dev_id: int, device_options: model_runtime.OptionFlags = model_runtime.OptionFlags(), wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig()) -> model_runtime.Device

11.1.2. Tensor memory representation

class model_runtime.TensorMemoryView

__init__(*args, **kwargs)

Overloaded function.

1. __init__(self: model_runtime.TensorMemoryView, data: capsule, data_size_bytes: int) -> None

2. __init__(self: model_runtime.TensorMemoryView, data: numpy.ndarray) -> None

__repr__(self: model_runtime.TensorMemoryView) → str

property data

property data_size_in_bytes

class model_runtime.ConstTensorMemoryView

__init__(*args, **kwargs)

Overloaded function.

1. __init__(self: model_runtime.ConstTensorMemoryView, data: capsule, data_size_bytes: int) -> None

2. __init__(self: model_runtime.ConstTensorMemoryView, data: numpy.ndarray) -> None

__repr__(self: model_runtime.ConstTensorMemoryView) → str

property data

property data_size_in_bytes

class model_runtime.TensorMemory

__init__(self: model_runtime.TensorMemory, data_size_bytes: int) → None

__repr__(self: model_runtime.TensorMemory) → str

property data

property data_size_in_bytes

getView(*args, **kwargs)

Overloaded function.

1. getView(self: model_runtime.TensorMemory) -> model_runtime.TensorMemoryView

2. getView(self: model_runtime.TensorMemory) -> model_runtime.ConstTensorMemoryView

11.1.3. ModelRunner

class model_runtime.FuturePyArrayMemory

get(self: model_runtime.FuturePyArrayMemory) → None

numpy(self: model_runtime.FuturePyArrayMemory) → numpy.ndarray

wait(self: model_runtime.FuturePyArrayMemory) → None

class model_runtime.FuturePyArrayMemoryView

get(self: model_runtime.FuturePyArrayMemoryView) → None

numpy(self: model_runtime.FuturePyArrayMemoryView) → numpy.ndarray

wait(self: model_runtime.FuturePyArrayMemoryView) → None

class model_runtime.FutureTensorMemoryView

get(self: model_runtime.FutureTensorMemoryView) → model_runtime.TensorMemoryView

wait(self: model_runtime.FutureTensorMemoryView) → None

class model_runtime.FutureTensorMemory

get(self: model_runtime.FutureTensorMemory) → model_runtime.TensorMemory

wait(self: model_runtime.FutureTensorMemory) → None

class model_runtime.InputMemoryView

items(self: model_runtime.InputMemoryView) → model_runtime.ItemsView[InputMemoryView]

keys(self: model_runtime.InputMemoryView) → model_runtime.KeysView[InputMemoryView]

values(self: model_runtime.InputMemoryView) → model_runtime.ValuesView[InputMemoryView]

class model_runtime.OutputMemoryView

items(self: model_runtime.OutputMemoryView) → model_runtime.ItemsView[OutputMemoryView]

keys(self: model_runtime.OutputMemoryView) → model_runtime.KeysView[OutputMemoryView]

values(self: model_runtime.OutputMemoryView) → model_runtime.ValuesView[OutputMemoryView]

class model_runtime.OutputMemory

items(self: model_runtime.OutputMemory) → model_runtime.ItemsView[OutputMemory]

keys(self: model_runtime.OutputMemory) → model_runtime.KeysView[OutputMemory]

values(self: model_runtime.OutputMemory) → model_runtime.ValuesView[OutputMemory]

class model_runtime.OutputFutureMemoryView

items(self: model_runtime.OutputFutureMemoryView) → model_runtime.ItemsView[OutputFutureMemoryView]

keys(self: model_runtime.OutputFutureMemoryView) → model_runtime.KeysView[OutputFutureMemoryView]

values(self: model_runtime.OutputFutureMemoryView) → model_runtime.ValuesView[OutputFutureMemoryView]

wait(self: model_runtime.OutputFutureMemoryView) → None

class model_runtime.OutputFutureMemory

items(self: model_runtime.OutputFutureMemory) → model_runtime.ItemsView[OutputFutureMemory]

keys(self: model_runtime.OutputFutureMemory) → model_runtime.KeysView[OutputFutureMemory]

values(self: model_runtime.OutputFutureMemory) → model_runtime.ValuesView[OutputFutureMemory]

wait(self: model_runtime.OutputFutureMemory) → None

class model_runtime.OutputFuturePyArrayMemoryView

items(self: model_runtime.OutputFuturePyArrayMemoryView) → model_runtime.ItemsView[OutputFuturePyArrayMemoryView]

keys(self: model_runtime.OutputFuturePyArrayMemoryView) → model_runtime.KeysView[OutputFuturePyArrayMemoryView]

values(self: model_runtime.OutputFuturePyArrayMemoryView) → model_runtime.ValuesView[OutputFuturePyArrayMemoryView]

wait(self: model_runtime.OutputFuturePyArrayMemoryView) → None

class model_runtime.OutputFuturePyArrayMemory

items(self: model_runtime.OutputFuturePyArrayMemory) → model_runtime.ItemsView[OutputFuturePyArrayMemory]

keys(self: model_runtime.OutputFuturePyArrayMemory) → model_runtime.KeysView[OutputFuturePyArrayMemory]

values(self: model_runtime.OutputFuturePyArrayMemory) → model_runtime.ValuesView[OutputFuturePyArrayMemory]

wait(self: model_runtime.OutputFuturePyArrayMemory) → None

class model_runtime.ReplicaIdToOutputMemoryView

items(self: model_runtime.ReplicaIdToOutputMemoryView) → model_runtime.ItemsView[ReplicaIdToOutputMemoryView]

keys(self: model_runtime.ReplicaIdToOutputMemoryView) → model_runtime.KeysView[ReplicaIdToOutputMemoryView]

values(self: model_runtime.ReplicaIdToOutputMemoryView) → model_runtime.ValuesView[ReplicaIdToOutputMemoryView]

class model_runtime.ReplicaIdToDevice

items(self: model_runtime.ReplicaIdToDevice) → model_runtime.ItemsView[ReplicaIdToDevice]

keys(self: model_runtime.ReplicaIdToDevice) → model_runtime.KeysView[ReplicaIdToDevice]

values(self: model_runtime.ReplicaIdToDevice) → model_runtime.ValuesView[ReplicaIdToDevice]

class model_runtime.PopefPaths

append(self: model_runtime.PopefPaths, x: str) → None

Add an item to the end of the list

clear(self: model_runtime.PopefPaths) → None

Clear the contents

count(self: model_runtime.PopefPaths, x: str) → int

Return the number of times x appears in the list

extend(*args, **kwargs)

Overloaded function.

1. extend(self: model_runtime.PopefPaths, L: model_runtime.PopefPaths) -> None

Extend the list by appending all the items in the given list

2. extend(self: model_runtime.PopefPaths, L: Iterable) -> None

Extend the list by appending all the items in the given list

insert(self: model_runtime.PopefPaths, i: int, x: str) → None

Insert an item at a given position.

pop(*args, **kwargs)

Overloaded function.

1. pop(self: model_runtime.PopefPaths) -> str

Remove and return the last item

2. pop(self: model_runtime.PopefPaths, i: int) -> str

Remove and return the item at index i

remove(self: model_runtime.PopefPaths, x: str) → None

Remove the first item from the list whose value is x. It is an error if there is no such item.

class model_runtime.DataDesc

__init__(self: model_runtime.DataDesc, name: str, size_in_bytes: int, shape: List[int], data_type: popef::DataType, popef_contains_tensor_data: bool = False) → None

__repr__(self: model_runtime.DataDesc) → str

property data_type

property name

numpy_data_type(self: model_runtime.DataDesc) → dtype

property popef_contains_tensor_data

property shape

property size_in_bytes

class model_runtime.ModelRunnerConfig

__init__(self: model_runtime.ModelRunnerConfig, replication_factor: int = 1, run_save_programs: bool = False, thread_safe: bool = False, frozen_inputs: model_runtime.InputMemoryView = model_runtime.InputMemoryView(), replica_to_save_programs_outputs: model_runtime.ReplicaIdToOutputMemoryView = model_runtime.ReplicaIdToOutputMemoryView(), replica_to_device: model_runtime.ReplicaIdToDevice = model_runtime.ReplicaIdToDevice(), device_wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig(), check_package_hash: bool = True, timeout_ns: datetime.timedelta = 5s, validate_io_params: bool = True, batching_dim: int = 0xFFFFFFFF, auto_reset: bool = False, max_look_ahead: str = 0, ring_buffer_size_multiplier: int = 2, flush_on_waiting_outputs: bool = False, batch_size_timeout_ns: datetime.timedelta = 9223372036854775807, data_parallel_timeout_ns: datetime.timedelta = 9223372036854775807, is_batch_size_timeout_enabled: bool = False, request_tracepoints_buffer_size: int = 1000) → None

__repr__(self: model_runtime.ModelRunnerConfig) → str

property auto_reset

property batch_size_timeout_ns

property batching_dim

property check_package_hash

property data_parallel_timeout_ns

property device_wait_config

property flush_on_waiting_outputs

property frozen_inputs

property is_batch_size_timeout_enabled

property max_look_ahead

property replica_to_device

property replica_to_save_programs_outputs

property replication_factor

property request_tracepoints_buffer_size

property ring_buffer_size_multiplier

property run_save_programs

property thread_safe

property timeout_ns

property validate_io_params

class model_runtime.ModelRunner

__init__(*args, **kwargs)

Overloaded function.

1. __init__(self: model_runtime.ModelRunner, model: popef::Model, config: model_runtime.ModelRunnerConfig = model_runtime.ModelRunnerConfig()) -> None

2. __init__(self: model_runtime.ModelRunner, popef_path: str, config: model_runtime.ModelRunnerConfig = model_runtime.ModelRunnerConfig()) -> None

3. __init__(self: model_runtime.ModelRunner, popef_paths: model_runtime.PopefPaths, config: model_runtime.ModelRunnerConfig = model_runtime.ModelRunnerConfig()) -> None

__repr__(self: model_runtime.ModelRunner) → str

execute(*args, **kwargs)

Overloaded function.

1. execute(self: model_runtime.ModelRunner, input_data: model_runtime.InputMemoryView, replica_id: int = 0) -> model_runtime.OutputMemory

2. execute(self: model_runtime.ModelRunner, input_data: model_runtime.InputMemoryView, output_data: model_runtime.OutputMemoryView, replica_id: int = 0) -> None

3. execute(self: model_runtime.ModelRunner, input_data: Dict[str, numpy.ndarray], replica_id: int = 0) -> Dict[str, numpy.ndarray]

4. execute(self: model_runtime.ModelRunner, input_data: Dict[str, numpy.ndarray], output_data: Dict[str, numpy.ndarray], replica_id: int = 0) -> None

executeAsync(*args, **kwargs)

Overloaded function.

1. executeAsync(self: model_runtime.ModelRunner, input_data: model_runtime.InputMemoryView, replica_id: int = 0) -> model_runtime.OutputFutureMemory

2. executeAsync(self: model_runtime.ModelRunner, input_data: model_runtime.InputMemoryView, output_data: model_runtime.OutputMemoryView, replica_id: int = 0) -> model_runtime.OutputFutureMemoryView

3. executeAsync(self: model_runtime.ModelRunner, input_data: Dict[str, numpy.ndarray], replica_id: int = 0) -> model_runtime.OutputFuturePyArrayMemory

4. executeAsync(self: model_runtime.ModelRunner, input_data: Dict[str, numpy.ndarray], output_data: Dict[str, numpy.ndarray], replica_id: int = 0) -> model_runtime.OutputFuturePyArrayMemoryView

getExecuteInputs(self: model_runtime.ModelRunner) → List[model_runtime.DataDesc]

getExecuteOutputs(self: model_runtime.ModelRunner) → List[model_runtime.DataDesc]

getModelInputs(self: model_runtime.ModelRunner) → List[model_runtime.DataDesc]

getModelOutputs(self: model_runtime.ModelRunner) → List[model_runtime.DataDesc]

11.2. Low-level API

11.2.1. Anchor callback management

class model_runtime.PopefDataUsagePolicy

Members:

USE_POPEF_DATA_IF_ANY

USE_USER_DATA

property name

model_runtime.null_popef_data_usage_predicate = None

model_runtime.predicate_factory.popef_data_usage.predAnchorsPolicies(accepted_anchors_policies: List[model_runtime::predicate_factory::AnchorWithPolicy<model_runtime::PopefDataUsagePolicy>], reject_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_POPEF_DATA_IF_ANY) → Callable[[popef::Anchor], model_runtime.PopefDataUsagePolicy]

model_runtime.predicate_factory.popef_data_usage.predProgramFlowLoad(flow: popef::ProgramFlow, accept_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_USER_DATA, reject_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_POPEF_DATA_IF_ANY) → Callable[[popef::Anchor], model_runtime.PopefDataUsagePolicy]

model_runtime.predicate_factory.popef_data_usage.predProgramFlowMain(flow: popef::ProgramFlow, accept_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_USER_DATA, reject_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_POPEF_DATA_IF_ANY) → Callable[[popef::Anchor], model_runtime.PopefDataUsagePolicy]

model_runtime.predicate_factory.popef_data_usage.predProgramFlowSave(flow: popef::ProgramFlow, accept_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.BIND_USER_CB, reject_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_POPEF_DATA_IF_ANY) → Callable[[popef::Anchor], model_runtime.PopefDataUsagePolicy]

model_runtime.predicate_factory.popef_data_usage.predProgramIndexes(*args, **kwargs)

Overloaded function.

1. predProgramIndexes(program_indexes: List[int], accept_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_USER_DATA, reject_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_POPEF_DATA_IF_ANY) -> Callable[[popef::Anchor], model_runtime.PopefDataUsagePolicy]

2. predProgramIndexes(accepted_programs_policies: List[model_runtime::predicate_factory::ProgramsWithPolicy<model_runtime::PopefDataUsagePolicy>], reject_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_POPEF_DATA_IF_ANY) -> Callable[[popef::Anchor], model_runtime.PopefDataUsagePolicy]

model_runtime.predicate_factory.popef_data_usage.predProgramNotAssigned(accept_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_USER_DATA, reject_policy: model_runtime.PopefDataUsagePolicy = model_runtime.PopefDataUsagePolicy.USE_POPEF_DATA_IF_ANY) → Callable[[popef::Anchor], model_runtime.PopefDataUsagePolicy]

class model_runtime.AnchorCallbackPolicy

Members:

BIND_USER_CB

BIND_EMPTY_CB

SKIP_CB

property name

model_runtime.null_anchor_callback_predicate = None

model_runtime.predicate_factory.anchor_callbacks.predAnchorsPolicies(accepted_anchors_policies: List[model_runtime::predicate_factory::AnchorWithPolicy<model_runtime::AnchorCallbackPolicy>], reject_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_EMPTY_CB) → Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy]

model_runtime.predicate_factory.anchor_callbacks.predProgramFlowLoad(flow: popef::ProgramFlow, accept_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_USER_CB, reject_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_EMPTY_CB) → Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy]

model_runtime.predicate_factory.anchor_callbacks.predProgramFlowMain(flow: popef::ProgramFlow, accept_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_USER_CB, reject_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_EMPTY_CB) → Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy]

model_runtime.predicate_factory.anchor_callbacks.predProgramFlowSave(flow: popef::ProgramFlow, accept_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_USER_CB, reject_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_EMPTY_CB) → Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy]

model_runtime.predicate_factory.anchor_callbacks.predProgramIndexes(*args, **kwargs)

Overloaded function.

1. predProgramIndexes(program_indexes: List[int], accept_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_USER_CB, reject_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_EMPTY_CB) -> Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy]

2. predProgramIndexes(accepted_programs_policies: List[model_runtime::predicate_factory::ProgramsWithPolicy<model_runtime::AnchorCallbackPolicy>], reject_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_EMPTY_CB) -> Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy]

model_runtime.predicate_factory.anchor_callbacks.predProgramNotAssigned(accept_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_USER_CB, reject_policy: model_runtime.AnchorCallbackPolicy = model_runtime.AnchorCallbackPolicy.BIND_EMPTY_CB) → Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy]

11.2.2. Queue management

class model_runtime.SpscRingBuffer

Ring Buffer for void* data type.

__init__(self: model_runtime.SpscRingBuffer, num_buffers: int, label: str, timeout_cb: Callable[[model_runtime.SpscRingBuffer], None] = None, timeout_ns: datetime.timedelta = 0ns) → None

__repr__(self: model_runtime.SpscRingBuffer) → str

getState(self: model_runtime.SpscRingBuffer, msg: str) → str

invalidate(self: model_runtime.SpscRingBuffer) → None

isValid(self: model_runtime.SpscRingBuffer) → bool

label(self: model_runtime.SpscRingBuffer) → str

numBuffers(self: model_runtime.SpscRingBuffer) → int

readAvailable(self: model_runtime.SpscRingBuffer) → bool

readComplete(self: model_runtime.SpscRingBuffer) → None

readLock(self: model_runtime.SpscRingBuffer) → capsule

readReset(self: model_runtime.SpscRingBuffer) → None

reset(self: model_runtime.SpscRingBuffer) → None

write(self: model_runtime.SpscRingBuffer, obj: capsule) → None

writeComplete(self: model_runtime.SpscRingBuffer) → None

writeLock(self: model_runtime.SpscRingBuffer) → capsule

class model_runtime.InputData

__init__(self: model_runtime.InputData) → None

__repr__(self: model_runtime.InputData) → str

property data

property data_size

property readStartCallback

class model_runtime.OutputData

__init__(self: model_runtime.OutputData) → None

__repr__(self: model_runtime.OutputData) → str

property data

property data_size

property writeCompleteCallback

class model_runtime.InputRingBuffer

Ring Buffer for InputData data type.

__init__(self: model_runtime.InputRingBuffer, num_buffers: int, label: str, timeout_cb: Callable[[model_runtime.InputRingBuffer], None] = None, timeout_ns: datetime.timedelta = 0ns) → None

__repr__(self: model_runtime.InputRingBuffer) → str

getState(self: model_runtime.InputRingBuffer, msg: str) → str

invalidate(self: model_runtime.InputRingBuffer) → None

isValid(self: model_runtime.InputRingBuffer) → bool

label(self: model_runtime.InputRingBuffer) → str

numBuffers(self: model_runtime.InputRingBuffer) → int

readAvailable(self: model_runtime.InputRingBuffer) → bool

readComplete(self: model_runtime.InputRingBuffer) → None

readLock(self: model_runtime.InputRingBuffer) → model_runtime.InputData

readReset(self: model_runtime.InputRingBuffer) → None

reset(self: model_runtime.InputRingBuffer) → None

write(self: model_runtime.InputRingBuffer, obj: model_runtime.InputData) → None

writeComplete(self: model_runtime.InputRingBuffer) → None

writeLock(self: model_runtime.InputRingBuffer) → model_runtime.InputData

class model_runtime.OutputRingBuffer

Ring Buffer for OutputData data type.

__init__(self: model_runtime.OutputRingBuffer, num_buffers: int, label: str, timeout_cb: Callable[[model_runtime.OutputRingBuffer], None] = None, timeout_ns: datetime.timedelta = 0ns) → None

__repr__(self: model_runtime.OutputRingBuffer) → str

getState(self: model_runtime.OutputRingBuffer, msg: str) → str

invalidate(self: model_runtime.OutputRingBuffer) → None

isValid(self: model_runtime.OutputRingBuffer) → bool

label(self: model_runtime.OutputRingBuffer) → str

numBuffers(self: model_runtime.OutputRingBuffer) → int

readAvailable(self: model_runtime.OutputRingBuffer) → bool

readComplete(self: model_runtime.OutputRingBuffer) → None

readLock(self: model_runtime.OutputRingBuffer) → model_runtime.OutputData

readReset(self: model_runtime.OutputRingBuffer) → None

reset(self: model_runtime.OutputRingBuffer) → None

write(self: model_runtime.OutputRingBuffer, obj: model_runtime.OutputData) → None

writeComplete(self: model_runtime.OutputRingBuffer) → None

writeLock(self: model_runtime.OutputRingBuffer) → model_runtime.OutputData

class model_runtime.InputQueue

__repr__(self: model_runtime.InputQueue) → str

enqueue(self: model_runtime.InputQueue, input: numpy.ndarray, read_start_callback: Callable[[], None] = None, read_complete_callback: Callable[[], None] = None) → None

class model_runtime.OutputQueue

__repr__(self: model_runtime.OutputQueue) → str

enqueue(self: model_runtime.OutputQueue, output: numpy.ndarray, write_complete_callback: Callable[[], None] = None) → None

class model_runtime.QueueManager

__repr__(self: model_runtime.QueueManager) → str

disconnectAll(self: model_runtime.QueueManager) → None

flushAll(self: model_runtime.QueueManager) → None

inputQueue(self: model_runtime.QueueManager, name: str) → model_runtime.InputQueue

property inputs

outputQueue(self: model_runtime.QueueManager, name: str) → model_runtime.OutputQueue

property outputs

resetAll(self: model_runtime.QueueManager) → None

model_runtime.ring_size_multiplier_factory.ringSizeMultForMainProgs(model: popef::Model, main_ring_size_multiplier: int, load_save_ring_size_multiplier: int = 1) → Callable[[popef::Anchor], int]

model_runtime.ring_size_multiplier_factory.ringSizeMultForProgs(programs: List[int], selected_ring_size_multiplier: int, others_ring_size_multiplier: int = 1) → Callable[[popef::Anchor], int]

11.2.3. Runtime management

class model_runtime.HandlerType

Members:

Input

Output

property name

class model_runtime.LaunchPolicy

Members:

Immediate

Deferred

property name

class model_runtime.SessionConfig

__init__(self: model_runtime.SessionConfig, policy: model_runtime.LaunchPolicy = model_runtime.LaunchPolicy.Deferred, pred_tensor_data: Callable[[popef::Anchor], model_runtime.PopefDataUsagePolicy] = None, check_package_hash: bool = True, wait_config: model_runtime.DeviceWaitConfig = model_runtime.DeviceWaitConfig(), max_look_ahead: str = unlimited) → None

__repr__(self: model_runtime.SessionConfig) → str

property check_package_hash

property max_look_ahead

property policy

property pred_tensor_data

property wait_config

class model_runtime.Session

__init__(*args, **kwargs)

Overloaded function.

1. __init__(self: model_runtime.Session, model: popef::Model, config: model_runtime.SessionConfig = model_runtime.SessionConfig()) -> None

2. __init__(self: model_runtime.Session, popef_paths: model_runtime.PopefPaths, config: model_runtime.SessionConfig = model_runtime.SessionConfig()) -> None

anchorsNotConnectedToCallbacks(self: model_runtime.Session, progs: List[int]) → Dict[int, List[popef::Anchor]]

bindToDevice(self: model_runtime.Session, device: model_runtime.Device) → None

createQueueManager(*args, **kwargs)

Overloaded function.

1. createQueueManager(self: model_runtime.Session, ring_size_multiplier: int = 2, timeout_cb: Callable[[model_runtime.InputRingBuffer], None] = None, timeout_ns: datetime.timedelta = 0ns, input_handle_predicate: Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy] = None, output_handle_predicate: Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy] = None) -> model_runtime.QueueManager

2. createQueueManager(self: model_runtime.Session, ring_size_multiplier_producer: Callable[[popef::Anchor], int], timeout_cb: Callable[[model_runtime.InputRingBuffer], None] = None, timeout_ns: datetime.timedelta = 0ns, input_handle_predicate: Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy] = None, output_handle_predicate: Callable[[popef::Anchor], model_runtime.AnchorCallbackPolicy] = None) -> model_runtime.QueueManager

errorIfAnchorsAreNotConnectedToCallbacks(self: model_runtime.Session, progs: List[int]) → None

getInputAnchors(self: model_runtime.Session) → List[popef::Anchor]

getUserInputAnchors(self: model_runtime.Session) → List[popef::Anchor]

getUserOutputAnchors(self: model_runtime.Session) → List[popef::Anchor]

model(self: model_runtime.Session) → popef::Model

reload(self: model_runtime.Session) → None

runLoadPrograms(self: model_runtime.Session) → None

runMainPrograms(self: model_runtime.Session) → None

runPrograms(self: model_runtime.Session, arg0: List[int]) → None

runSavePrograms(self: model_runtime.Session) → None

setCallbackForAnchor(*args, **kwargs)

Overloaded function.

1. setCallbackForAnchor(self: model_runtime.Session, arg0: str, arg1: Callable[[capsule], None]) -> None

2. setCallbackForAnchor(self: model_runtime.Session, data: numpy.ndarray, name: str, handler_type: model_runtime.HandlerType) -> None

   param numpy.ndarray data

   Numpy array with tensor data.

   param str name

   The name of the tensor to be connected. Connect each Anchor if empty.

stop(self: model_runtime.Session) → None

unloadFromDevice(self: model_runtime.Session) → None