qmi.core.context

Implementation of the QMI_Context class.

Functions

ping_qmi_contexts(workgroup_name_filter[, ...])

Broadcast an info request message to discover all QMI contexts on the network.

Classes

QMI_Context(name[, config])

Represents an application or script within the QMI system.
qmi.core.context.ping_qmi_contexts(workgroup_name_filter: str, context_name_filter: str = '*', timeout: float = 0.1) → list[_UdpPingResponse]

Broadcast an info request message to discover all QMI contexts on the network.

Parameters:

  • workgroup_name_filter – Filter on workgroup name.

  • context_name_filter – Filter on context name (default: “*”).

  • timeout – Time to wait for answers (default: 0.1).

class qmi.core.context.QMI_Context(name: str, config: CfgQmi | None = None)

Represents an application or script within the QMI system.

The QMI_Context is the main entry point to QMI functionality. It keeps track of instruments, tasks, signals, RPC objects and connections to other contexts. It provides functionality to create and access these objects, and to interact with objects in other contexts.

Internally, the QMI context will create Python threads to perform certain tasks in the background.

Unless stated otherwise, methods of the QMI_Context instance may only be called from the thread that created the context. A few specific methods may safely be called from any thread; this is explicitly stated in the documentation of the method when it applies.

In most Python programs that use the QMI framework, there is exactly one instance of QMI_Context. This instance is created when the application calls qmi.start(...) and destroyed when the application calls qmi.stop().

DEFAULT_UDP_RESPONDER_PORT

Default port number for UDP responses.

property suppress_version_mismatch_warnings: bool

If set to True, no warnings will be issued when connecting to a peer that runs a different version of QMI.

property workgroup_name: str

Make ‘_workgroup_name’ as read-only property.

get_config() → CfgQmi

Return the QMI configuration.

The QMI configuration is a hierarchical data structure consisting of Python dataclass instances. The QMI configuration is typically initialized by reading the QMI configuration file during program startup.

It is not allowed to modify the configuration data structure returned by this function.

Returns:

The QMI configuration data.

Return type:

self._config

get_context_config() → CfgContext

Return the subset of the QMI configuration specific for this context.

The QMI configuration contains a mapping that applies to this context.

Returns:

The QMI configuration subset specific for this context.

Return type:

ctxcfg

get_qmi_home_dir() → str

Return the QMI home directory.

The QMI home directory is the location where QMI expects to find configuration files and data.

The QMI home directory may be specified in the QMI configuration file. Otherwise, if the environment variable QMI_HOME is set, it contains the path of the QMI home directory. Otherwise, the user home directory will be used as QMI home directory.

Returns:

The QMI home directory.

Return type:

qmi_home

get_log_dir() → str

Return the directory where log files should be written.

get_datastore_dir() → str

Return the base directory of the DataStore repository.

resolve_file_name(file_name: str) → str

Apply substitutions to a configured file name.

The following substitutions are supported:

$$            -> "$"
${context}    -> context name
${qmi_home}   -> QMI home directory
${datastore}  -> QMI datastore directory
${config_dir} -> directory of QMI configuration file
${date}       -> date of program start, UTC, formatted as YYYY-mm-dd
${datetime}   -> time of program start, UTC, formatted as YYYY-mm-ddTHH-MM-SS
Returns:

The substituted file name if any of the supported substitutions were present. Else, the file name.

run(_rpc_thread_run: Callable) → None

Run thread. From Python 3.14 onwards objects that are run within a context, need a run method with a single input where the callable to run is given. Addition of this function should have no effect when using earlier Python versions.

start() → None

Start the context.

A QMI_Context instance must be started before creating any tasks or sending any messages. This happens automatically when the application calls qmi.start(...).

stop() → None

Stop the context.

After the context is stopped, it must not be used anymore. A stopped context may not be started a second time.

shutdown_requested() → bool

Return True if the context has received a shutdown request via RPC.

A background program may call this function to check whether a shutdown request has been received. If this function returns True, the program should clean up and exit.

wait_until_shutdown(duration: float) → bool

Wait until a shutdown request is received.

Parameters:

duration – Maximum wait time in seconds.

Returns:

If a shutdown request is received. False: If the wait duration expires before a shutdown request is received.

Return type:

True

register_stop_handler(handler: Callable[[], None]) → None

Register a callback function to be invoked when this context stops.

Parameters:

handler – Function to be called (without arguments) when the context stops.

send_message(message: QMI_Message) → None

Send a message to its destination.

If the destination is within the local context, this function delivers the message to the local message handler. Otherwise, this function sends the message to the correct peer context.

This method is thread-safe. It can be called from any thread.

Raises:

QMI_MessageDeliveryException – If the message can not be routed.

connect_to_peer(peer_context_name: str, peer_address: str | None = None, ignore_duplicate: bool = False) → None

Connect to the specified peer context.

Parameters:

  • peer_context_name – Name of the peer context to connect to.

  • peer_address – IP address and TCP port of the peer context, formatted as "<addr>:<port>". When not specified, the peer address will be taken from the QMI configuration.

  • ignore_duplicate – When True, nothing will happen if a connection to the specified context already exists.

disconnect_from_peer(peer_context_name: str) → None

Disconnect from the specified remote QMI context.

Raises:

QMI_UnknownNameException – If the specified context is not connected.

has_peer_context(peer_context_name: str) → bool

Return True if a context with specified name is currently connected as a peer.

Note that the result of this method may become invalid at any time as the set of connected contexts may change asynchronously.

class PeerDescriptor(name: str, address_port: str)

Descriptor of a peer returned by discover_peer_contexts.

name

The name of the peer context.

Type:

str

address_port

The location to find the peer, in the format of “address:port”.

Type:

str

name: str

Alias for field number 0

address_port: str

Alias for field number 1

count(value, /)

Return number of occurrences of value.

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

discover_peer_contexts(workgroup_name_filter: str | None = None, context_name_filter: str = '*') → list[PeerDescriptor]

Discover QMI contexts on the network.

You can filter on workgroup name and/or context name via the optional arguments. Use a “*” to match any sequence of characters (e.g. “ba*” matches “bar”, “baz” and “ball”) and a “?” to match a single character (e.g. “ba?” matches “bar” and “baz”, but not “ball”). Filters are case-sensitive.

Parameters:

  • workgroup_name_filter – Filter on workgroup name (None: use the workgroup name of this context’s config).

  • context_name_filter – Filter on context name (default: “*”).

Returns:

A list of PeerDescriptors that can be passed to connect_to_peer.

Return type:

contexts

register_message_handler(message_handler: QMI_MessageHandler) → None

Register a new message handler.

The message handler must be a local object (in the same context) and it must have a unique object ID within the context.

This function is intended for internal use within QMI. Application programs should not need to call this function.

Parameters:

message_handler – New message handler to be registered.

unregister_message_handler(message_handler: QMI_MessageHandler) → None

Unregister an existing message handler.

This function removes a message handler from the context. After this function returns, no further messages will be delivered to the specified handler. The specified object must currently be registered as a message handler.

This function is intended for internal use within QMI. Application programs should not need to call this function.

Parameters:

message_handler – The message handler to be unregistered.

info() → str

Return information about the context.

get_version() → str

Return the QMI version used by this context.

get_tcp_server_port() → int

Return the TCP server port for this context, or return 0 if this context does not have a TCP server.

get_rpc_object_descriptors() → list[RpcObjectDescriptor]

Return a list of descriptors for all local RPC objects.

get_rpc_object_descriptor(rpc_object_name: str) → RpcObjectDescriptor | None

Return a descriptor for the specified local RPC object if object is managed, else None.

make_unique_address(prefix: str) → QMI_MessageHandlerAddress

Generate a unique message handler address with a specified prefix.

make_unique_token(prefix: str = '$lock_') → QMI_LockTokenDescriptor

Generate and return a unique token descriptor.

make_rpc_object(rpc_object_name: str, rpc_object_class: type[QMI_RpcObject], *args: Any, **kwargs: Any) → Any

Create an instance of a QMI_RpcObject and return a proxy for the new object instance.

The actual object instance will be created in a separate background thread. You can call its methods via RPC, using the proxy object returned by this function.

The return type of the make_XXX() methods is not annotated. It is not possible to provide a precise annotation, because the actual return type is “rpc_object_class.Proxy” which is a programmatically constructed class and thus not available for static type checking.

Parameters:

  • rpc_object_name – Name for the new object instance, unique within the local context.

  • rpc_object_class – Class that implements this object (must be a subclass of QMI_RpcObject).

  • args – Optional arguments for the class constructor.

  • kwargs – Optional keyword arguments for the class constructor.

Returns:

An RPC proxy that provides access to the new object instance.

remove_rpc_object(proxy: QMI_RpcProxy) → None

Stop the specified RPC object and release its resources.

This can only be used for RPC objects running in the local context.

Parameters:

proxy – Proxy for the RPC object to be removed.

make_instrument(instrument_name: str, instrument_class: type[QMI_Instrument], *args: Any, **kwargs: Any) → Any

Create an instance of a QMI_Instrument subclass and make it accessible via RPC.

The actual instrument instance will be created in a separate background thread. To access the instrument, you can call its methods via RPC.

Parameters:

  • instrument_name – Unique name for the new instrument instance. This name will also be used to access the instrument via RPC.

  • instrument_class – Class that implements this instrument (must be a subclass of QMI_Instrument).

  • args – Optional arguments for the instrument class constructor.

  • kwargs – Optional keyword arguments for the instrument class constructor.

Returns:

An RPC proxy that provides access to the new instrument instance.

make_task(task_name: str, task_class: type[~qmi.core.task.QMI_Task], *args: ~typing.Any, task_runner: type[~qmi.core.task.QMI_TaskRunner] = <class 'qmi.core.task.QMI_TaskRunner'>, **kwargs: ~typing.Any) → Any

Create an instance of a QMI_Task subclass and make it accessible via RPC.

The task instance will be created in a separate thread. A QMI_TaskRunner will be created to manage the task thread. The task can be accessed by calling the methods of the task runner via RPC.

Note that the task is not yet started. To start the task, perform an explicit call to the start() method of the returned task.

Parameters:

  • task_name – Unique name for the new task instance. This name will also be used to access the task runner via RPC.

  • task_class – Class that implements this task (must be a subclass of QMI_Task).

  • args – Optional arguments for the task class constructor.

  • task_runner – Class that implements the managing of the task (must be a subclass of QMI_Taskrunner)

  • kwargs – Optional keyword arguments for the task class constructor.

Returns:

An RPC proxy that provides access to the new task.

make_proxy(rpc_object_descriptor: RpcObjectDescriptor) → Any

Return a proxy for the specified object descriptor.

This function is intended for internal use within QMI. Application programs should not call this function directly.

Parameters:

rpc_object_descriptor – RpcObjectDescriptor uniquely identifying an RPC object.

Returns:

Proxy for the specified RPC object.

make_peer_context_proxy(context_name: str) → Any

Return a proxy for the internal $context object of the specified peer context.

This function is intended for internal use within QMI. Application programs should not call this function directly.

Parameters:

context_name – Name of the peer context.

Returns:

A proxy for the internal context RPC object of the specified peer context.

show_network_contexts()

Show a list of contexts that are on the network.

list_rpc_objects(category: str | None = None) → list[tuple[str, str]]

Returns a list of tuple strings of RPC objects addresses in the local context and in peer context.

show_rpc_objects(category: str | None = None) → None

Show a list of RPC objects in the local context and peer contexts.

show_tasks() → None

Show a list of tasks in the local context and peer contexts.

show_instruments() → None

Show a list of instruments in the local context and peer contexts.

show_contexts() → None

Show a list of currently connected contexts (including the local context).

get_rpc_object_by_name(rpc_object_name: str, auto_connect: bool = False, host_port: str | None = None) → Any

Return a proxy for the specified RPC object.

The object may exist either in the local context, or in a peer context. Note that when using auto_connect keyword parameter, in Windows environment the peer might not be found without giving the exact host:port address for connect_to_peer command.

Parameters:

  • rpc_object_name – Object name, formatted as "<context_name>.<object_id>".

  • auto_connect – If True, connect automatically to the RPC object peer.

  • host_port – Optional host:port string pattern to guide the auto_connect.

Returns:

A proxy for the specified object.

Raises:

ValueError – If the given RPC object descriptor was not found.

get_instrument(instrument_name: str, auto_connect: bool = False, host_port: str | None = None) → Any

Return a proxy for the specified instrument.

The instrument may exist either in the local context, or in a peer context.

Parameters:

  • instrument_name – Instrument name, formatted as "<context_name>.<instrument_name>".

  • auto_connect – If True, connect automatically to the instrument peer.

  • host_port – Optional host:port string pattern to guide the auto_connect.

Returns:

A proxy for the specified instrument.

get_task(task_name: str, auto_connect: bool = False, host_port: str | None = None) → Any

Return a proxy for the specified task.

The task may exist either in the local context, or in a peer context.

Parameters:

  • task_name – Task name, formatted as "<context_name>.<task_name>".

  • auto_connect – If True, connect automatically to the task peer.

  • host_port – Optional host:port string pattern to guide the auto_connect.

Returns:

A proxy for the specified task.

get_configured_contexts() → dict[str, CfgContext]

Get active QMI contexts.

Returns:

An OrderedDict object of contexts in the config file.

subscribe_signal(publisher_context: str, publisher_name: str, signal_name: str, receiver: QMI_SignalReceiver) → None

Subscribe to a specified signal.

While subscribed, the SignalReceiver object will receive and queue all published signals of the specified type.

A SignalReceiver object can be simultaneously subscribed to multiple signals (from different publishers). Similarly, multiple receivers can be simultaneously subscribed to the same signal. However, it is an error to try to subscribe a receiver to a signal to which it is already subscribed.

This method may safely be called from any thread.

Parameters:

  • publisher_context – Name of the context that publishes the signal. An empty string may be used to refer to the local context.

  • publisher_name – Name of the publisher of the signal (e.g. instrument name).

  • signal_name – Name of the signal to subscribe to.

  • receiver – A SignalReceiver object which will receive the published signals.

unsubscribe_signal(publisher_context: str, publisher_name: str, signal_name: str, receiver: QMI_SignalReceiver) → None

Unsubscribe from a specified signal.

It is an error to unsubscribe a receiver from a signal to which is not currently subscribed.

This method may safely be called from any thread.

Parameters:

  • publisher_context – Name of the context that publishes the signal, or empty string for local context.

  • publisher_name – Name of the publisher of the signal.

  • signal_name – Name of the signal to unsubscribe from.

  • receiver – The SignalReceiver object to unsubscribe from the signal.

publish_signal(publisher_name: str, signal_name: str, *args: Any) → None

Publish the specified signal to the QMI network.

The published signal will be received by any SignalReceivers that are currently subscribed to the specified signal from the specified publisher.

Parameters:

  • publisher_name – Name of the publisher of the signal.

  • signal_name – Name of the signal.

  • args – Additional data to send along with the signal.