double threaded auxiliary

Double Threaded based Auxiliary Interface

module: dt_auxiliary
synopsis: common double threaded based auxiliary interface

class pykiso.interfaces.dt_auxiliary.AuxCommand(value)[source]

Contain all available auxiliary’s commands.

CREATE_AUXILIARY = 1: create auxiliary command id

DELETE_AUXILIARY = 2: delete auxiliary command id

class pykiso.interfaces.dt_auxiliary.DTAuxiliaryInterface(name=None, is_proxy_capable=False, activate_log=None, tx_task_on=True, rx_task_on=True, auto_start=True)[source]

Common interface for all double threaded auxiliary. A so called << double threaded >> auxiliary, simply encapsulate two threads one for the reception and one for the transmmission.

Initialize auxiliary attributes

Parameters

name (Optional[str]) – alias of the auxiliary instance
is_proxy_capable (bool) – notify if the current auxiliary could be (or not) associated to a proxy-auxiliary.
activate_log (Optional[List[str]]) – loggers to deactivate
tx_task_on – enable or not the tx thread
rx_task_on – enable or not the rx thread
auto_start (bool) – determine if the auxiliayry is automatically started (magic import) or manually (by user)

create_instance()[source]

Start auxiliary’s running tasks and activities.

Return type: bool
Returns: True if the auxiliary is created otherwise False
Raises: AuxiliaryCreationError – if instance creation failed

delete_instance()[source]

Stop auxiliary’s running tasks and activities.

Return type: bool
Returns: True if the auxiliary is deleted otherwise False

resume()[source]

Resume current auxiliary’s run.

Warning

due to the usage of create_instance if an issue occurred the exception AuxiliaryCreationError is raised.

Return type: bool
Returns: True if the auxiliary is resumed otherwise False

run_command(cmd_message, cmd_data=None, blocking=True, timeout_in_s=5)[source]

Send a request by transmitting it through queue_in and waiting for a response using queue_out.

Parameters

cmd_message (Any) – command request to the auxiliary
cmd_data (Optional[Any]) – data you would like to populate the command with
blocking (bool) – If you want the command request to be blocking or not
timeout_in_s (int) – Number of time (in s) you want to wait for an answer

Return type

Any

Returns

True if the request is correctly executed otherwise False

start()[source]

Force the auxiliary to start all running tasks and activities.

Warning

due to the usage of create_instance if an issue occurred the exception AuxiliaryCreationError is raised.

Return type: bool
Returns: True if the auxiliary is started otherwise False

stop()[source]

Force the auxiliary to stop all running tasks and activities.

Return type: bool
Returns: True if the auxiliary is stopped otherwise False

suspend()[source]

Supend current auxiliary’s run.

Return type: bool
Returns: True if the auxiliary is suspend otherwise False

wait_for_queue_out(blocking=False, timeout_in_s=0)[source]

Wait for data from the queue out.

Parameters

blocking (bool) – True: wait for timeout to expire, False: return immediately
timeout_in_s (int) – if blocking, wait the defined time in seconds

Return type

Optional[Any]

Returns

data contained in the auxiliary’s queue_out

pykiso.interfaces.dt_auxiliary.close_connector(func)[source]

Close current associated auxiliary’s channel.

Parameters: func (Callable) – decorated method
Return type: Callable
Returns: inner decorated function

pykiso.interfaces.dt_auxiliary.flash_target(func)[source]

Flash firmware on the target, using associated auxiliary’s flasher channel.

Parameters: func (Callable) – decorated method
Return type: Callable
Returns: inner decorated function

pykiso.interfaces.dt_auxiliary.open_connector(func)[source]

Open current associated auxiliary’s channel.

Parameters: func (Callable) – decorated method
Return type: Callable
Returns: inner decorated function