Utils

Async/sync bridging utilities (decorators and event-loop helpers).

pymirokai.utils.async_wrapper.make_async(func)

Convert a synchronous function into one that supports asynchronous execution.

The returned wrapper can be awaited in async contexts or called directly in sync contexts. If called from within a running event loop, the function is executed in a thread pool.

Parameters:

func (Callable[..., Any]) – The function to make asynchronous.

Return type:

Callable[..., Any]

Returns:

A callable that can be invoked both synchronously and asynchronously.

Enum/identifier converters for robot entities (arm/ear, etc.).

pymirokai.utils.converter.arm_to_arm_entity_identifier(arm)

Convert an Arm enum value into a standardized entity identifier.

Parameters:

arm (Arm | None) – The Arm enum member (Arm.LEFT or Arm.RIGHT), or None.

Return type:

str | None

Returns:

The corresponding string identifier (e.g., "left_arm") or None.

pymirokai.utils.converter.ear_to_ear_entity_identifier(ear)

Convert an Ear enum value into a standardized entity identifier.

Parameters:

ear (Ear) – The Ear enum member (Ear.LEFT or Ear.RIGHT).

Return type:

str | None

Returns:

The corresponding string identifier (e.g., "right_ear") or None if not recognized.

Get the local IP address of the machine.

pymirokai.utils.get_local_ip.get_local_ip()

Return the local IPv4 address of the current machine.

Uses a dummy UDP connection to determine the machine’s outward-facing IP address. If detection fails, defaults to 127.0.0.1.

Return type:

str

Returns:

The local IPv4 address as a string.

Recursive JSON parsing utilities that decode strings within nested structures.

pymirokai.utils.json_parser.parse_json_recursively(obj)

Parse a JSON string recursively.

Parameters:

obj (Union[str, Any]) – The JSON string to parse or an already parsed JSON object.

Return type:

Any

Returns:

Any – The parsed JSON object or the original object if it’s not a valid JSON string.

Run an async function until interrupted.

pymirokai.utils.run_until_interruption.run_until_interruption(callback)

Run an asynchronous callback until SIGINT or SIGTERM is received.

Gracefully handles shutdown by cancelling all running tasks and closing the loop once the program is interrupted.

Parameters:

callback (Callable[[], Awaitable[None]]) – The asynchronous function to execute until interrupted.

Return type:

None

Serialization helpers that coerce arbitrary data into JSON-safe structures.

pymirokai.utils.serialize_data.serialize_data(data)

Recursively serialize arbitrary data into JSON-safe structures.

Handles tuples, dictionaries, iterables, and primitive types. Non-serializable objects are converted to strings as a fallback.

Parameters:

data (Any) – The data to serialize.

Return type:

Any

Returns:

A JSON-serializable version of the input data.

CLI and runtime helpers for the skills manager (RPC handling, publishing).

async pymirokai.utils.skills_manager_util.handle_cancel(msg, robot, skills_manager)

Handle an incoming ‘cancel’ RPC command to stop a skill or mission.

Parameters:

  • msg (dict) – The RPC cancel message.

  • robot (Robot) – The connected Robot instance.

  • skills_manager (SkillsManager) – The SkillsManager managing running missions.

async pymirokai.utils.skills_manager_util.handle_run(msg, robot, skills_manager)

Handle an incoming ‘run’ RPC command for skill execution.

Parameters:

  • msg (dict) – The RPC command message.

  • robot (Robot) – The connected Robot instance.

  • skills_manager (SkillsManager) – The SkillsManager managing available skills.

async pymirokai.utils.skills_manager_util.main()

Command-line entry point for the skills manager.

Return type:

None

async pymirokai.utils.skills_manager_util.run(ip, api_key, upload_dir, shutdown_event)

Connect to the robot and start the skills manager.

Automatically handles reconnection attempts and error propagation.

Parameters:

  • ip (str) – Target robot IP address.

  • api_key (str) – Robot API key for authentication.

  • upload_dir (str) – Path to the directory of skill files.

  • shutdown_event (Event) – Event used to stop the main loop.

Return type:

None

async pymirokai.utils.skills_manager_util.run_skills_manager(robot, upload_dir, shutdown_event)

Manage the lifecycle of the SkillsManager, including subscriptions and error handling.

Establishes communication between the robot and SkillsManager, monitors connectivity, and republishes skill states over WebSocket channels.

Parameters:

  • robot (Robot) – Connected Robot instance.

  • upload_dir (Path) – Path to the directory containing uploaded skill files.

  • shutdown_event (Event) – Event used to signal shutdown.

Return type:

None

async pymirokai.utils.skills_manager_util.send_auto_start_skills_list(robot, skills_manager)

Publish the list of auto-start skills to subscribers.

Parameters:

  • robot (Robot) – The connected Robot instance.

  • skills_manager (SkillsManager) – The SkillsManager providing serialized skills.

async pymirokai.utils.skills_manager_util.send_skills_list(robot, skills_manager)

Publish the list of available (non-auto-start) skills to subscribers.

Parameters:

  • robot (Robot) – The connected Robot instance.

  • skills_manager (SkillsManager) – The SkillsManager providing serialized skills.

pymirokai.utils.skills_manager_util.start_skills_manager()

CLI-compatible synchronous entry point for running the skills manager.

Skills Wrapper provides the _mission helper to be able to call priority.

class pymirokai.utils.skills_wrapper.SkillWrapper

Bases: object

Provide the _mission helper for creating Mission objects.

This mixin does not perform any dynamic attribute interception. The priority-handling logic is defined in accepts_priority().