utils : Utility functions¶

MTP File¶

`mtppy.utils.mtp_file_handling.mtp_reader.mtp_reader.MtpReader(mtp_path=None)` ¶

A class to read and parse MTP files.

This class provides functionality to read MTP files, extract their manifest, and create mappings between aliases and OPC UA node IDs.

Initialize the MtpReader with an optional MTP file path.

Parameters:

Name	Type	Description	Default
`mtp_path`	`str \| Path \| None`	Path to the MTP file to read. If None, no file is read initially but all the public data structured are initialized	`None`

`create_alias_opcua_node_map(data_assemblies, node_ids)` ¶

Create a mapping between MTP aliases and OPC UA node IDs.

Returns:

Type	Description
`dict[str, str]`	Dictionary mapping aliases to OPC UA node IDs.

`read_mtp(mtp_path)` ¶

Read and parse an MTP file.

Parameters:

Name	Type	Description	Default
`mtp_path`	`str \| Path`	Path to the MTP file to read.	required

Returns:

Type	Description
`ElementTree`	Parsed XML tree of the manifest

Raises:

Type	Description
`FileNotFoundError`	If file does not exist
`ValueError`	If file is not readable

`mtppy.utils.mtp_file_handling.mtp_writer.mtp_writer.MtpWriter()` ¶

Writer for MTP files.

`mtppy.utils.mtp_file_handling.mtp_reader.mtp_xml` ¶

Utilities for XML file handling.

`get_all_child_attribute_text(elem, node_ids=None)` ¶

Get all child attribute texts from an element.

Parameters:

Name	Type	Description	Default
`elem`	`Element`	The XML element to process.	required
`node_ids`	`dict[str, str] \| None`	Optional mapping of node IDs to their replacement values.	`None`

Returns:

Name	Type	Description
`dict`	`dict[str, dict[str, str]]`	Dictionary containing attribute names and values.

`get_child_default_value_text(elem)` ¶

Get the first content of an element.

Parameters:

Name	Type	Description	Default
`elem`	`Element`	The XML element to search.	required

Returns:

Name	Type	Description
`str`	`str`	The default value text if found, empty string otherwise.

`get_child_value_text(elem)` ¶

Get the first content child of an element.

Parameters:

Name	Type	Description	Default
`elem`	`Element`	The XML element to search.	required

Returns:

Name	Type	Description
`str`	`str`	The value text if found, empty string otherwise.

`get_data_assemblies(root)` ¶

Extract data assemblies from the XML root element.

Parameters:

Name	Type	Description	Default
`root`	`Element \| None`	The root ElementTree to search.	required

Returns:

Name	Type	Description
`dict`	`dict[str, DataAssemblyConfig]`	A dictionary mapping data assembly names to their configurations.

`get_named_childs_recursive(parent, xpath)` ¶

Recursively get named children from an XML element.

Parameters:

Name	Type	Description	Default
`parent`	`Element`	The parent XML element.	required
`xpath`	`str`	The XPath expression to find child elements.	required

Returns:

Name	Type	Description
`dict`	`dict[str, Any]`	A dictionary of named children and their attributes.

`get_opcua_node_id(root, identifier)` ¶

Get OPC UA node ID from an ExternalInterface element.

Parameters:

Name	Type	Description	Default
`root`	`Element`	The root ElementTree to search.	required
`identifier`	`str`	The ID of the ExternalInterface element.	required

Returns:

Name	Type	Description
`str`	`str`	The OPC UA node ID in the format "ns=;s=", or an empty string if not found.

`get_opcua_node_ids(root)` ¶

Get all OPC UA node IDs from ExternalInterface elements.

Parameters:

Name	Type	Description	Default
`root`	`Element \| None`	The root ElementTree to search.	required

Returns:

Name	Type	Description
`dict`	`dict[str, str]`	A dictionary mapping ExternalInterface IDs to their node information.

`get_services(root)` ¶

Extract services from the XML root element.

Parameters:

Name	Type	Description	Default
`root`	`Element \| None`	The root ElementTree to search.	required

Returns:

Name	Type	Description
`dict`	`dict[str, dict[str, str \| dict[str, str]]]`	A dictionary mapping service names to their configurations.

`is_valid_guid(guid_string)` ¶

Check if the given string is a valid GUID/UUID.

Parameters:

Name	Type	Description	Default
`guid_string`	`str`	The string to validate.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if the string is a valid GUID, False otherwise.

DataItem and DataItemCollection: Data Storage, Access and Callbacks¶

`mtppy.utils.data_item.DataItem(name, w3c_xml_data_type, access, init_value=0, description='', external_write_cb=None, internal_write_cb=None)` ¶

Represent a scalar typed value hosted by a data-item collection.

A DataItem validates and stores its value according to a W3C XML data type. It can notify an internal callback and any number of external communication observers when its value is written.

The class is hosted by a DataItemCollection, the base class for DataAssembly.

A DataItem is associated with an Attribute of a DataAssembly definition within MTP 2.0.

Attributes:

Name	Type	Description
`name`	`str`	DataItem name.
`w3c_xml_data_type`	`W3cXmlDataTypes`	DataItem value type.
`access`	`MTPAccess`	DataItem access mode.
`init_value`	`ANY_TYPE`	Initial, type-corrected value.
`description`	`str`	Human-readable description.
`ext_wrt_cb`	`Callable[[ANY_TYPE], None] \| None`	Callback invoked for external writes.
`int_wrt_cb`	`Callable[[ANY_TYPE], None] \| None`	Callback invoked for internal writes.
`communication_objects`	`list[CommunicationObject]`	External communication observers notified on value writes.
`comm_obj`	`CommunicationObject \| None`	Backwards-compatible alias for the first communication observer.
`_value`	`ANY_TYPE`	Current type-corrected value.
`_is_notifying`	`bool`	Whether an internal notification is in progress.

Initialize a DataItem.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the data item.	required
`w3c_xml_data_type`	`W3cXmlDataTypes`	W3C XML type used to convert assigned values.	required
`access`	`MTPAccess`	Access mode for the data item.	required
`init_value`	`ANY_TYPE \| None`	Initial value. `None` is treated as `0`.	`0`
`description`	`str`	Human-readable description.	`''`
`external_write_cb`	`Callable[[ANY_TYPE], None] \| None`	Callback for external writes received through the data item.	`None`
`internal_write_cb`	`Callable[[ANY_TYPE], None] \| None`	Callback for internal writes to the data item.	`None`

Raises:

Type	Description
`TypeError`	If `w3c_xml_data_type` or `access` is not the expected enumeration type.

`comm_obj` `property` `writable` ¶

Return the first communication object, if one is attached.

Returns:

Type	Description
`CommunicationObject \| None`	The first attached observer, or `None` when no observer exists.

`attach_communication_object(communication_object)` ¶

Attach an external communication observer.

Duplicate observers are ignored. Observers are notified in attachment order whenever the DataItem is written.

Parameters:

Name	Type	Description	Default
`communication_object`	`CommunicationObject`	Communication object that exposes this DataItem to an external client.	required

`get_value()` ¶

Return the current value of the DataItem.

Returns:

Type	Description
`ANY_TYPE`	The current value, already converted to the DataItem's
`ANY_TYPE`	pythonic `w3c_xml_data_type`

`get_value_str()` ¶

Return a human-readable string representation of the DataItem.

Returns:

Type	Description
`str`	A string of the form `"<name>=<value>"`, e.g. `"WQC=128"`.

`set_value(value, notify_external=True, source_observer=None)` ¶

Set the value and notify external subscribers except its source.

Parameters:

Name	Type	Description	Default
`value`	`ANY_TYPE`	Value to convert, store, and publish.	required
`notify_external`	`bool`	Whether to notify attached communication observers.	`True`
`source_observer`	`CommunicationObject \| None`	Observer that supplied `value`. This observer is excluded from notification to prevent echoing a polled value back to its source.	`None`

`write_value(value)` ¶

Set the local value and publish it to the communication nodes, without invoking the external write callback.

This is the server-side half of an external handshake: it stores value locally (most often a reset to 0) and pushes it to the attached communication nodes so that external clients observe the change, but it does NOT call :attr:ext_wrt_cb. That callback is the entry point for writes coming from an external client, so invoking it here would make the PEA react to its own handshake. On OPC UA the communication object additionally suppresses the datachange event that the server's own self-subscription would otherwise fire, which would call :attr:ext_wrt_cb with the new value.

Parameters:

Name	Type	Description	Default
`value`	`ANY_TYPE`	Value to store locally and publish to the nodes.	required

`mtppy.utils.data_item.DataItemCollection()` ¶

Bases: MutableMapping[str, DataItem]

Key-immutable, dict-like container of DataItem instances.

The collection is populated at construction time from the ClassVar CLASS_DATA_ITEMS mapping declared statically on the owning class (and all of its bases in MRO order). Each entry is deep-copied so that two collections derived from the same class do not share DataItems.

The collection supports both mapping-style (da["WQC"]) and attribute-style (da.WQC) access to the current value of a DataItem.

Both the dict-keys and the attributes of the class are immutable, that is a lhs da["NEW"] does not add a new entry with key "NDW" to the DataItemCollection. The same is true for lhs da.new or self.new - this hardens the class against typos.

Note

In cases where the SNAKE_CASE string holding constant is not distinguishable from the CamelCase PNO MTP 2.0 name, our best practice is to have the constant name end with an , e.g. WQC, P_, PV_

Attention

The collection is key-immutable and attribute-immutable. Setting an unknown key or attribute raises KeyError, while reading an unknown attribute raises AttributeError. Consequently, hasattr(collection, name) returns False for missing attributes, as it does for normal Python objects.

Initialize the collection from the class-level CLASS_DATA_ITEMS.

Walks the class MRO in reverse order so that base-class entries are inserted first and subclass entries take precedence on duplicate keys. Each declared data item is deep-copied to avoid shared state between instances.

`attach_loop(loop)` ¶

Bind this collection to the run-task event loop of its service.

Once bound, every write made from a foreign thread is scheduled onto that loop so that all mutations happen on the single run task (the single-writer principle). Writes made from the run task itself execute inline. Collections that are never bound keep the legacy synchronous, inline behaviour.

Parameters:

Name	Type	Description	Default
`loop`	`AbstractEventLoop`	The running event loop owned by the service's run task.	required

`attach_wake(wake_fn)` ¶

Register the run-task wake callable for this collection.

When a guarded write (e.g. a service op-mode or command change) is scheduled onto the bound loop from a foreign thread, the owning service's run task may be parked in the offline undefined state. Registering its wake callable lets such a write release that wait immediately instead of waiting out the next poll interval.

Parameters:

Name	Type	Description	Default
`wake_fn`	`Callable[[], None]`	Zero-argument callable that signals the run task to wake.	required

`get_item(key)` ¶

Access the underlying DataItem for introspection.

Unlike __getitem__, this returns the full DataItem object instead of only its current value, allowing callers to inspect or modify attributes such as the data type, access mode or subscription callback.

Parameters:

Name	Type	Description	Default
`key`	`str`	The data item name.	required

Returns:

Type	Description
`DataItem`	The DataItem instance registered
`DataItem`	under `key`.

`items()` ¶

Return the current {name: value} dict of all data items.

Returns:

Type	Description
`list[tuple[str, ANY_TYPE]]`	A list of `(key, value)` tuples for every data item, where
`list[tuple[str, ANY_TYPE]]`	`value` is the current (already type-cast) value.

`keys()` ¶

Return a view of the data item names.

Returns:

Type	Description
`KeysView[str]`	A `KeysView` of the registered data item keys.

`run_on_loop(fn)` ¶

Run fn on the bound loop, deferring from foreign threads.

This is the single-writer primitive. Three cases:

No bound loop -> run fn inline (unbound / test / offline).
Caller already on the bound loop -> run fn inline (avoids a re-post loop for nested / cascade writes fired inside a posted operation).
Caller on a foreign thread -> schedule fn via loop.call_soon_threadsafe so it runs on the run task.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[[], None]`	Zero-argument callable that mutates this collection.	required

`values()` ¶

Return the current values of all data items.

Returns:

Type	Description
`list[ANY_TYPE]`	A list with the current value of every data item, in the same
`list[ANY_TYPE]`	order as `keys`.

logging¶

`mtppy.utils.logging` ¶

predefined logging levels for the MTPPy2.0 framework.

`setLogLevel(log_level)` ¶

Set the root logger level from a numeric value or level name.

testing¶

`mtppy.utils.testing` ¶

testing provides repeated pieces of codes used in testing.

`data_object_set_modes(instance, init_op_mode, init_src_mode)` ¶

Set initial mode of ActiveElement instance to a certain state.

General utility functions¶

The utils module delivers some general helper functions that have their own use independent of the MTP application.

Conversion of Automation Types¶

`utils.type_conversion` ¶

type_conversion collects utility functions that cast values to basic types used in automation.

OPC UA, MTP, IEC 61131 and other automation standards have quite specialized data types, which are not always directly supported by Python, for instance signed/unsigned integer types of different sizes (UNSIGNED SHORT != SHORT, INT != SHORT), or strings with a maximum length.

The collected functions help to convert/cast values to the appropriate python types, ensuring compatibility with these standards.

`cast_to_bool(value)` ¶

Casts canonical boolean values without treating arbitrary strings as true.

`cast_to_byte(value)` ¶

Casts value to byte, ensuring it falls within the range of INT8_MIN to INT8_MAX (-128 to 127).

Parameters:

Name	Type	Description	Default
`value`	`SupportsInt`	The value to be cast to a byte.	required

Returns:

Type	Description
`int`	The byte representation of the value, clamped to the valid range.

`cast_to_int(value)` ¶

Casts a value to an integer.

Parameters:

Name	Type	Description	Default
`value`	`SupportsInt`	The value to be cast to an integer.	required

Returns:

Type	Description
`int`	The integer representation of the value.

`cast_to_long(value)` ¶

Casts value to long integer, ensuring it falls within the range of INT64_MIN to INT64_MAX (-9223372036854775808 to 9223372036854775807).

Parameters:

Name	Type	Description	Default
`value`	`SupportsInt`	The value to be cast to a long integer.	required

Returns:

Type	Description
`int`	The signed integer representation of the value, clamped to the valid range.

`cast_to_short(value)` ¶

Casts a value to a short integer, ensuring it falls within the range of -32768 to 32767.

Parameters:

Name	Type	Description	Default
`value`	`SupportsInt`	The value to be cast to a short integer.	required

Returns:

Type	Description
`int`	The signed short integer representation of the value, clamped to the valid
`int`	range.

`cast_to_string(value, max_len=None)` ¶

Casts a value to a string, optionally truncating it to a maximum length.

Parameters:

Name	Type	Description	Default
`value`	`object`	The value to be cast to a string.	required
`max_len`	`SupportsIndex \| None`	The maximum length of the resulting string.	`None`

Returns:

Type	Description
`str`	The string representation of the value, optionally truncated.

`cast_to_ubyte(value)` ¶

Casts value to unsigned byte, ensuring it falls within the range of 0 to UINT8_MAX (255).

Parameters:

Name	Type	Description	Default
`value`	`SupportsInt`	The value to be cast to an unsigned byte.	required

Returns:

Type	Description
`int`	The unsigned byte representation of the value, clamped to the valid range.

`cast_to_uint(value)` ¶

Casts a value to an unsigned integer, ensuring it falls within the range of 0 to 4294967295.

Parameters:

Name	Type	Description	Default
`value`	`SupportsInt`	The value to be cast to an unsigned integer.	required

Returns:

Type	Description
`int`	The unsigned integer representation, clamped to the valid range.

`cast_to_ulong(value)` ¶

Casts value to unsigned long integer, ensuring it falls within the range of 0 to UINT64_MAX (18446744073709551615).

Parameters:

Name	Type	Description	Default
`value`	`SupportsInt`	The value to be cast to an unsigned long integer.	required

Returns:

Type	Description
`int`	The unsigned long integer representation, clamped to the valid range.

`cast_to_ushort(value)` ¶

Casts value to unsigned short integer, ensuring it falls within the range of 0 to 65535.

Parameters:

Name	Type	Description	Default
`value`	`SupportsInt`	The value to be cast to an unsigned short integer.	required

Returns:

Type	Description
`int`	The unsigned short integer representation, clamped to the valid range.

`clamp(value, limit, reset=None)` ¶

Clamp a value between a lower and an upper bound.

When value is below the lower bound or above the upper bound, it is replaced by the corresponding element of reset. If reset is not provided, the bounds themselves are used as the replacement values.

Parameters:

Name	Type	Description	Default
`value`	`float`	The value to clamp.	required
`limit`	`tuple[float, float]`	A `(min, max)` pair defining the inclusive bounds.	required
`reset`	`tuple[float, float] \| None`	An optional `(min, max)` pair returned when `value` is outside the bounds. Defaults to `limit`.	`None`

Returns:

Type	Description
`float`	The clamped value, or the matching fallback bound.

Example

clamp(value,[0,100],[0,0]) - generates a saw-tooth when fed with a ramp clamp(value,[0,0]) - clamps the value between upper and lower bound

logging¶

`utils.logging.CustomLogger(name, level=logging.NOTSET)` ¶

Bases: Logger

Custom logger class that allows adding custom logging levels.

The levels are between the standard logging levels INFO and WARNING. The logger is configured to display the timestamp, log level, thread, module, function name, and message in the log output.

Example

from utils.logging import CustomLogger
logging.setLoggerClass(CustomLogger)
logger = logging.getLogger("MTPPy2.0Logger")
logger.add_custom_levels({"opcua": 1, "mtp": 2, "app": 4, "pol": 6})

Initialize a logger with the standard logging constructor.

`add_custom_level(level_name, level_value)` ¶

Register one custom level name without modifying the logger API.

`add_custom_levels(custom_levels)` ¶

Register level names using offsets from logging.INFO.

The logger methods are explicit and are not generated dynamically. This keeps the framework API visible to static type checkers.

`app(msg, *args, **kwargs)` ¶

Log a message at the application level.

`mtp(msg, *args, **kwargs)` ¶

Log a message at the MTP level.

`opcua(msg, *args, **kwargs)` ¶

Log a message at the OPC UA level.

`pol(msg, *args, **kwargs)` ¶

Log a message at the POL level.

utils : Utility functions¶

MTP File¶

mtppy.utils.mtp_file_handling.mtp_reader.mtp_reader.MtpReader(mtp_path=None) ¶

create_alias_opcua_node_map(data_assemblies, node_ids) ¶

read_mtp(mtp_path) ¶

mtppy.utils.mtp_file_handling.mtp_writer.mtp_writer.MtpWriter() ¶

mtppy.utils.mtp_file_handling.mtp_reader.mtp_xml ¶

get_all_child_attribute_text(elem, node_ids=None) ¶

get_child_default_value_text(elem) ¶

get_child_value_text(elem) ¶

get_data_assemblies(root) ¶

get_named_childs_recursive(parent, xpath) ¶

get_opcua_node_id(root, identifier) ¶

get_opcua_node_ids(root) ¶

get_services(root) ¶

is_valid_guid(guid_string) ¶

DataItem and DataItemCollection: Data Storage, Access and Callbacks¶

mtppy.utils.data_item.DataItem(name, w3c_xml_data_type, access, init_value=0, description='', external_write_cb=None, internal_write_cb=None) ¶

comm_obj property writable ¶

attach_communication_object(communication_object) ¶

get_value() ¶

get_value_str() ¶

set_value(value, notify_external=True, source_observer=None) ¶

write_value(value) ¶

mtppy.utils.data_item.DataItemCollection() ¶

attach_loop(loop) ¶

attach_wake(wake_fn) ¶

get_item(key) ¶

items() ¶

keys() ¶

run_on_loop(fn) ¶

values() ¶

logging¶

mtppy.utils.logging ¶

setLogLevel(log_level) ¶

testing¶

mtppy.utils.testing ¶

data_object_set_modes(instance, init_op_mode, init_src_mode) ¶

General utility functions¶

Conversion of Automation Types¶

utils.type_conversion ¶

cast_to_bool(value) ¶

cast_to_byte(value) ¶

cast_to_int(value) ¶

cast_to_long(value) ¶

cast_to_short(value) ¶

cast_to_string(value, max_len=None) ¶

cast_to_ubyte(value) ¶

cast_to_uint(value) ¶

cast_to_ulong(value) ¶

cast_to_ushort(value) ¶

clamp(value, limit, reset=None) ¶

logging¶

utils.logging.CustomLogger(name, level=logging.NOTSET) ¶

add_custom_level(level_name, level_value) ¶

add_custom_levels(custom_levels) ¶

app(msg, *args, **kwargs) ¶

mtp(msg, *args, **kwargs) ¶

opcua(msg, *args, **kwargs) ¶

pol(msg, *args, **kwargs) ¶

`mtppy.utils.mtp_file_handling.mtp_reader.mtp_reader.MtpReader(mtp_path=None)` ¶

`create_alias_opcua_node_map(data_assemblies, node_ids)` ¶

`read_mtp(mtp_path)` ¶

`mtppy.utils.mtp_file_handling.mtp_writer.mtp_writer.MtpWriter()` ¶

`mtppy.utils.mtp_file_handling.mtp_reader.mtp_xml` ¶

`get_all_child_attribute_text(elem, node_ids=None)` ¶

`get_child_default_value_text(elem)` ¶

`get_child_value_text(elem)` ¶

`get_data_assemblies(root)` ¶

`get_named_childs_recursive(parent, xpath)` ¶

`get_opcua_node_id(root, identifier)` ¶

`get_opcua_node_ids(root)` ¶

`get_services(root)` ¶

`is_valid_guid(guid_string)` ¶

`mtppy.utils.data_item.DataItem(name, w3c_xml_data_type, access, init_value=0, description='', external_write_cb=None, internal_write_cb=None)` ¶

`comm_obj` `property` `writable` ¶

`attach_communication_object(communication_object)` ¶

`get_value()` ¶

`get_value_str()` ¶

`set_value(value, notify_external=True, source_observer=None)` ¶

`write_value(value)` ¶

`mtppy.utils.data_item.DataItemCollection()` ¶

`attach_loop(loop)` ¶

`attach_wake(wake_fn)` ¶

`get_item(key)` ¶

`items()` ¶

`keys()` ¶

`run_on_loop(fn)` ¶

`values()` ¶

`mtppy.utils.logging` ¶

`setLogLevel(log_level)` ¶

`mtppy.utils.testing` ¶

`data_object_set_modes(instance, init_op_mode, init_src_mode)` ¶

`utils.type_conversion` ¶

`cast_to_bool(value)` ¶

`cast_to_byte(value)` ¶

`cast_to_int(value)` ¶

`cast_to_long(value)` ¶

`cast_to_short(value)` ¶

`cast_to_string(value, max_len=None)` ¶

`cast_to_ubyte(value)` ¶

`cast_to_uint(value)` ¶

`cast_to_ulong(value)` ¶

`cast_to_ushort(value)` ¶

`clamp(value, limit, reset=None)` ¶

`utils.logging.CustomLogger(name, level=logging.NOTSET)` ¶

`add_custom_level(level_name, level_value)` ¶

`add_custom_levels(custom_levels)` ¶

`app(msg, *args, **kwargs)` ¶

`mtp(msg, *args, **kwargs)` ¶

`opcua(msg, *args, **kwargs)` ¶

`pol(msg, *args, **kwargs)` ¶