ih_muse.muse.muse

Core functionality for Muse.

Module Contents

Classes

Muse	The main class for interacting with the Muse system in Python.

API

class ih_muse.muse.muse.Muse(config: ih_muse.config.Config)

The main class for interacting with the Muse system in Python.

Parameters:

config (Config) – The configuration object for initializing Muse.

# Example usage:
from ih_muse import Muse, Config

config = Config(...)
muse = Muse(config)
await muse.initialize(timeout=5.0)

Initialization

Initialize the Muse instance.

_muse: ih_muse.ih_muse.PyMuse

None

async initialize(timeout: Optional[float] = None) → None

Initialize the Muse client and starts background tasks.

Parameters:

timeout (Optional[float]) – Optional timeout in seconds for the initialization process.

Raises:

MuseInitializationTimeoutError – If initialization times out.

async classmethod create(config: ih_muse.config.Config, timeout: Optional[float] = None) → ih_muse.muse.muse.Muse

Create and initialize a Muse instance.

Parameters:

• config (Config) – The configuration object.

• timeout (Optional[float]) – Optional timeout in seconds for initialization.

Returns:

An initialized Muse instance.

property is_initialized: bool

Check whether the Muse client is initialized.

Returns:

True if initialized, False otherwise.

property finest_resolution: ih_muse.ih_muse.TimestampResolution

Get the finest resolution stored by the Poet.

Returns:

The current TimestampResolution stored in the poet.

get_remote_element_id(local_elem_id: str) → Optional[int]

Retrieve the remote Element ID associated with a local Element ID.

Parameters:

local_elem_id (str) – The local Element ID to query.

Returns:

The corresponding remote Element ID, or None if not registered.

async register_element(kind_code: str, name: str, metadata: dict[str, str], parent_id: Optional[str] = None) → str

Register a new element with the Muse system.

Parameters:

• kind_code (str) – The kind code of the element.

• name (str) – The name of the element.

• metadata (dict[str, str]) – Metadata associated with the element.

• parent_id (Optional[int]) – The parent element ID, if any.

Returns:

The local element ID assigned to the registered element.

Raises:

MuseError – If registration fails.

async send_metric(local_elem_id: str, metric_code: str, value: float) → None

Send a metric value associated with an element.

Parameters:

• local_elem_id (int) – The local ID of the element.

• metric_code (str) – The code identifying the metric.

• value (float) – The value of the metric to send.

Raises:

MuseError – If sending the metric fails.

async get_metrics(query: ih_muse.proto.MetricQuery) → list[ih_muse.proto.MetricPayload]

Retrieve metrics based on a query.

Parameters:

query (MetricQuery) – The query specifying the criteria for retrieving metrics.

Returns:

A list of MetricPayload matching the query.

Raises:

MuseError – If retrieving the metrics fails.

async replay(replay_path: str) → None

Replays events from a recording file.

Parameters:

replay_path (str) – The file path to the recording.

Raises:

MuseError – If replaying the events fails.