Programs¶

class pyniryo_nate.components.Programs(http_client: HttpClient, mqtt_client: MqttClient, correlation_id: str)[source]¶

Bases: BaseAPIComponent

Programs component for managing and executing robot programs.

This component provides methods for: - Creating, uploading, and managing programs - Executing programs with custom environment variables and arguments - Monitoring program execution status and output - Controlling program execution (pause, resume, stop)

Methods

`close`	Close the component and release any resources if needed.
`create`	Upload a new program.
`delete`	Delete a program by its ID.
`execute`	Execute a program.
`executor_status`	Get the current status of the program executor.
`get`	Get a program by its ID.
`get_all`	Get all the registered programs.
`get_execution`	Get a program execution by its ID.
`get_execution_output`	Download the full output of a program execution.
`get_executions`	Get all the executions of a program.
`on_execution_output`
`on_execution_status`
`pause`	Pause the current program execution.
`resume`	Resume a program execution paused by `pause()`.
`stop`	Stop the current program execution.
`update`	Update a program.

Parameters:

http_client (HttpClient)
mqtt_client (MqttClient)
correlation_id (str)

get_all() → list[Program][source]¶

Get all the registered programs.

Returns:: The list of all the programs.
Return type:: list[Program]

create(name: str | None, program: IO[bytes], program_type: ProgramType) → Program[source]¶

Upload a new program.

Parameters:

name (str | None) – The name of the program. If None, the robot uses the file name.
program (IO[bytes]) – A file-like object containing the program content.
program_type (ProgramType) – The type of the program.

Returns:

The created program.

Return type:

Program

get(program_id: str, dst: IO[bytes] | None = None) → Program[source]¶

Get a program by its ID.

Parameters:

program_id (str) – The ID of the program.
dst (IO[bytes] | None) – Optional file-like object to download the program content into. If None, the content is not downloaded.

Returns:

The corresponding program.

Return type:

Program

delete(program_id: str) → None[source]¶

Delete a program by its ID.

Parameters:: program_id (str) – The ID of the program.
Returns:: None
Return type:: None

update(program: Program, src: IO[bytes] | None = None) → Program[source]¶

Update a program. The metadata and the content are two requests: a content upload that fails leaves the metadata already changed.

Parameters:

program (Program) – The program to update.
src (IO[bytes] | None) – Optional file-like object containing the new program content. If None, the content is not updated.

Returns:

The updated program.

Return type:

Program

get_executions(program_id: str) → list[ProgramExecution][source]¶

Get all the executions of a program.

Parameters:: program_id (str) – The ID of the program.
Returns:: The list of all the executions of the program.
Return type:: list[ProgramExecution]

get_execution(program_id: str, execution_id: str) → ProgramExecution[source]¶

Get a program execution by its ID.

Parameters:

program_id (str) – The ID of the program.
execution_id (str) – The ID of the program execution.

Returns:

The corresponding program execution.

Return type:

ProgramExecution

get_execution_output(program_id: str, execution_id: str, dst: IO[bytes]) → None[source]¶

Download the full output of a program execution.

Parameters:

program_id (str) – The ID of the program.
execution_id (str) – The ID of the execution.
dst (IO[bytes]) – A file-like object to write the output content into.

Return type:

None

execute(program_id: str, environment: dict[str, str] | None = None, arguments: list[str] | None = None, timeout: float = -1, on_output: Callable[[str, bool], None] | None = None, on_status: Callable[[ExecutorStatus], None] | None = None) → ExecutionBaseCommand[source]¶

Execute a program.

Parameters:

program_id (str) – The ID of the program.
environment (dict[str, str] | None) – Optional dictionary of environment variables to pass to the program.
arguments (list[str] | None) – Optional list of command line arguments to pass to the program.
timeout (float) – Maximum time in seconds to wait before considering the program failed. A negative value means no timeout.
on_output (Callable[[str, bool], None] | None) – Optional callback to listen to the program outputs. The callback must take the output string and a boolean indicating if it’s the end of the output as parameters.
on_status (Callable[[ExecutorStatus], None] | None) – Optional callback to listen the program execution status changes. The callback must take the new status as a parameter.

Returns:

The created program execution command.

Raises:

ClientError – If another workstation holds the robot (code == 'robot_held') or an execution is already running.

Return type:

ExecutionBaseCommand

executor_status() → ProgramExecutorStatus[source]¶

Get the current status of the program executor.

Returns:: The current status of the program executor.
Return type:: ProgramExecutorStatus

pause() → None[source]¶

Pause the current program execution. A pause is taken from the moment the execution is requested, the program’s own startup included: a program paused before it runs a line carries on its startup once resumed. A pause asked by a client holds until resume() is called, a protective stop pauses and resumes on its own.

Raises:: ClientError – If nothing is running (code == 'no_execution') or the execution is in a state a pause does not follow from.
Return type:: None

stop(sigkill: bool = False) → None[source]¶

Stop the current program execution. A stop is taken from the moment the execution is requested, the program’s own startup included: a program stopped before it runs a line never runs one.

Parameters:: sigkill (bool) – If True, force kill the program. If False, send a graceful stop signal.
Raises:: ClientError – If nothing is running (code == 'no_execution').
Return type:: None

resume() → None[source]¶

Resume a program execution paused by pause().

Raises:: ClientError – If nothing is running (code == 'no_execution'), the execution is not paused, or the safety chain closes the resume.
Return type:: None

on_execution_output(program_id: str, execution_id: str, callback: Callable[[str, bool], None]) → Callable[[], None][source]¶

Parameters:

program_id (str)
execution_id (str)
callback (Callable[[str, bool], None])

Return type:

Callable[[], None]

on_execution_status(program_id: str, execution_id: str, callback: Callable[[ExecutorStatus], None]) → Callable[[], None][source]¶

Parameters:

program_id (str)
execution_id (str)
callback (Callable[[ExecutorStatus], None])

Return type:

Callable[[], None]