Skip to content

PEA : Bringing the things together

PEA Base Class

mtppy.pea.pea.PEA(endpoint='opc.tcp://localhost:4840/')

Main Process Equipment Assembly (PEA) class.

The PEA class serves as the central coordinator for managing all aspects of a Process Element Assembly, including data assemblies, process values, services, and OPC UA server functionality. It provides methods to add components and control the overall operation of the PEA system.

Attributes:

Name Type Description
service_set DataAssemblyDict[Service]

Dictionary of service instances managed by this PEA
data_assemblies DataAssemblyDict[DataAssembly]

Dictionary of data assemblies managed by this PEA
process_value_ins DataAssemblyDict[InputElement]

Dictionary of input process values managed by this PEA
process_value_outs DataAssemblyDict[IndicatorElement]

Dictionary of output process values managed by this PEA
alarms DataAssemblyDict[Alarm]

Dictionary of alarm instances managed by this PEA
endpoint

OPC UA server endpoint address
opcua_options_provider

Provider for OPC UA server options
pea_opcua_server

OPC UA server instance for this PEA
subscription_list

List of active subscriptions
web_server_url

URL for web server access
pea_information_label

Information label containing PEA metadata

Initializes the PEA with default or specified OPC UA endpoint.

Parameters:

Name Type Description Default
endpoint str

Endpoint of the OPC UA server. If not specified, opc.tcp://localhost:4840/ is used.

 'opc.tcp://localhost:4840/'

activate_and_start(service, proc_id, timeout)

Activate a service and start the given procedure.

Convenience method that brings a service from its initial undefined/Off state into execute for the given procedure:

  1. Sets the service's operation mode to Aut.
  2. Sets the internal procedure request (PROCEDURE_REQ) to proc_id.
  3. Polls (up to timeout seconds) until the start bit in the service's CommandEn mask is set (i.e. the service has reached idle with a valid procedure latched).
  4. If start is enabled within the timeout, issues the start command via the internal channel and returns True; otherwise returns False.

Parameters:

Name Type Description Default
service Service

The Service to activate and start.

 required
proc_id int

The procedure id to select and start.

 required
timeout float

Maximum time in seconds to wait for START to be enabled.

 required

Returns:

Type Description
bool

True if the start command was issued, False on timeout.

add_alarm(alarm)

Add a static AlarmSet.Base alarm to the PEA.

Parameters:

Name Type Description Default
alarm Alarm

Alarm description to register.

 required

Raises:

Type Description
TypeError

If alarm is not an Alarm instance.
ValueError

If an alarm with the same tag name is already registered.

add_data_assembly(data_assembly)

Add a DataAssembly to the PEA.

Parameters:

Name Type Description Default
data_assembly DataAssembly | list[DataAssembly] | dict[str, DataAssembly]

The DataAssembly instance to add

 required

add_process_value_in(process_value_in)

Add an input ProcessValue to the PEA.

Parameters:

Name Type Description Default
process_value_in InputElement

The InputElement instance to add as input process value

 required

add_process_value_out(process_value_out)

Add an output ProcessValue to the PEA.

Parameters:

Name Type Description Default
process_value_out IndicatorElement

The output process value to add.

 required

add_service(service)

Add a Service to the PEA.

Parameters:

Name Type Description Default
service Service

The Service instance to add

 required

start()

Start the OPC UA server instance and begin PEA operation.

Initializes and starts the OPC UA server with the configured data assemblies and services, making the PEA available for communication and operation. Also attaches the PEA event loop to every service and starts each service's single long-lived run task.

stop()

Stop the OPC UA server instance and shut down PEA operation.

Stops the services' run tasks, the OPC UA server, and (if the PEA created its own loop) that loop.

User Logic

mtppy.pea.rmtp_pea.RMTPPea(endpoint='opc.tcp://localhost:4840/', mtp='', ssd='', code='', cycle_time=1.0)

Bases: PEA

PEA assembled from an MTP package and an application logic module.

Create an RMTP PEA, optionally loading an MTP package.

activate_and_start(service, proc_id, timeout)

Activate a service and start the given procedure.

Convenience method that brings a service from its initial undefined/Off state into execute for the given procedure:

  1. Sets the service's operation mode to Aut.
  2. Sets the internal procedure request (PROCEDURE_REQ) to proc_id.
  3. Polls (up to timeout seconds) until the start bit in the service's CommandEn mask is set (i.e. the service has reached idle with a valid procedure latched).
  4. If start is enabled within the timeout, issues the start command via the internal channel and returns True; otherwise returns False.

Parameters:

Name Type Description Default
service Service

The Service to activate and start.

 required
proc_id int

The procedure id to select and start.

 required
timeout float

Maximum time in seconds to wait for START to be enabled.

 required

Returns:

Type Description
bool

True if the start command was issued, False on timeout.

add_alarm(alarm)

Add a static AlarmSet.Base alarm to the PEA.

Parameters:

Name Type Description Default
alarm Alarm

Alarm description to register.

 required

Raises:

Type Description
TypeError

If alarm is not an Alarm instance.
ValueError

If an alarm with the same tag name is already registered.

add_data_assembly(data_assembly)

Add a DataAssembly to the PEA.

Parameters:

Name Type Description Default
data_assembly DataAssembly | list[DataAssembly] | dict[str, DataAssembly]

The DataAssembly instance to add

 required

add_process_value_in(process_value_in)

Add an input ProcessValue to the PEA.

Parameters:

Name Type Description Default
process_value_in InputElement

The InputElement instance to add as input process value

 required

add_process_value_out(process_value_out)

Add an output ProcessValue to the PEA.

Parameters:

Name Type Description Default
process_value_out IndicatorElement

The output process value to add.

 required

add_service(service)

Add a Service to the PEA.

Parameters:

Name Type Description Default
service Service

The Service instance to add

 required

load_requirements(mtp, ssd='', code='')

Load application logic, data assemblies, and services from an MTP package.

start()

Start the OPC UA server instance and begin PEA operation.

Initializes and starts the OPC UA server with the configured data assemblies and services, making the PEA available for communication and operation. Also attaches the PEA event loop to every service and starts each service's single long-lived run task.

stop()

Stop the OPC UA server instance and shut down PEA operation.

Stops the services' run tasks, the OPC UA server, and (if the PEA created its own loop) that loop.

mtppy.pea.application_logic.application_logic_pea.ApplicationLogicPEA(endpoint='opc.tcp://localhost:4840/', mtp='', ssd='', code='')

Bases: PEA

PEA that loads an MTP package and a user application-logic module.

Initialize the PEA, optionally loading an MTP package.

activate_and_start(service, proc_id, timeout)

Activate a service and start the given procedure.

Convenience method that brings a service from its initial undefined/Off state into execute for the given procedure:

  1. Sets the service's operation mode to Aut.
  2. Sets the internal procedure request (PROCEDURE_REQ) to proc_id.
  3. Polls (up to timeout seconds) until the start bit in the service's CommandEn mask is set (i.e. the service has reached idle with a valid procedure latched).
  4. If start is enabled within the timeout, issues the start command via the internal channel and returns True; otherwise returns False.

Parameters:

Name Type Description Default
service Service

The Service to activate and start.

 required
proc_id int

The procedure id to select and start.

 required
timeout float

Maximum time in seconds to wait for START to be enabled.

 required

Returns:

Type Description
bool

True if the start command was issued, False on timeout.

add_alarm(alarm)

Add a static AlarmSet.Base alarm to the PEA.

Parameters:

Name Type Description Default
alarm Alarm

Alarm description to register.

 required

Raises:

Type Description
TypeError

If alarm is not an Alarm instance.
ValueError

If an alarm with the same tag name is already registered.

add_data_assembly(data_assembly)

Add a DataAssembly to the PEA.

Parameters:

Name Type Description Default
data_assembly DataAssembly | list[DataAssembly] | dict[str, DataAssembly]

The DataAssembly instance to add

 required

add_process_value_in(process_value_in)

Add an input ProcessValue to the PEA.

Parameters:

Name Type Description Default
process_value_in InputElement

The InputElement instance to add as input process value

 required

add_process_value_out(process_value_out)

Add an output ProcessValue to the PEA.

Parameters:

Name Type Description Default
process_value_out IndicatorElement

The output process value to add.

 required

add_service(service)

Add a Service to the PEA.

Parameters:

Name Type Description Default
service Service

The Service instance to add

 required

load_requirements(mtp, ssd, code)

Loads requirements mtp and application logic. ignores namespace for the moment.

start()

Start the OPC UA server instance and begin PEA operation.

Initializes and starts the OPC UA server with the configured data assemblies and services, making the PEA available for communication and operation. Also attaches the PEA event loop to every service and starts each service's single long-lived run task.

stop()

Stop the OPC UA server instance and shut down PEA operation.

Stops the services' run tasks, the OPC UA server, and (if the PEA created its own loop) that loop.

mtppy.pea.application_logic.uml_state_machine_service.UMLStateMachineService(tag_name, tag_description, registry=None, procedures=None, cycle_time=0.5)

Bases: Service

Implementation of Service that calls application logic, registered by the ApplicationLogicRegistry.

This service acts as a bridge between the service execution framework and user-defined application logic functions that are registered in an ApplicationLogicRegistry.

The state model follows the UML-State-Chart notation with entry, do, and exit actions. Each state in the service lifecycle has three phases: - entry: Executed once when entering the state - do: Executed repeatedly while in the state (for cyclic states) until user function returns False - exit: Executed once when leaving the state

Attributes:

Name Type Description
_registry

The ApplicationLogicRegistry containing user-defined functions.

Initialize the UMLStateMachineService.

Parameters:

Name Type Description Default
tag_name str

The name of the service instance.

 required
tag_description str

A description of the service.

 required
registry ApplicationLogicRegistry | None

Optional ApplicationLogicRegistry containing user functions.

 None
procedures list[Procedure] | None

Optional list of procedures for the service.

 None
cycle_time float

Time interval (in seconds) for cyclic state execution.

 0.5

name property writable

Return the service tag name.

Returns:

Type Description
str

The current service tag name.

aborted() async

Execute the entry, do, and exit functions bound to aborted.

aborting() async

Execute the entry, do, and exit functions bound to aborting.

add_configuration_parameter(configuration_parameter)

Add a configuration parameter to the service.

Parameters:

Name Type Description Default
configuration_parameter ParameterElement

Parameter to add, keyed by its tag name.

 required

add_procedure(procedure)

Add one or more procedures to the service.

Parameters:

Name Type Description Default
procedure Procedure | list[Procedure | None] | dict[str, Procedure | None] | None

A procedure, a list of procedures, a mapping of names to procedures, or None.

 required

Raises:

Type Description
ValueError

If any procedure has an ID of zero.

apply_configuration_parameters()

Apply all configuration parameters to the service.

Each parameter is updated through its update_v_out method.

apply_procedure_parameters()

Apply parameters for the currently selected procedure.

attach_loop(loop)

Attach the PEA's event loop to this service's run task.

Also binds the service's control-plane DataItem collections (service_control and its operation/source sub-assemblies) to the same loop so that every DataItem write and guarded transition funnels onto the single run task (the single-writer principle, see docs/architecture.md "Single-Writer Principle").

Parameters:

Name Type Description Default
loop AbstractEventLoop

The running event loop owned by the PEA.

 required

command_execution(value)

Capture a command for deferred, deterministic execution.

Unlike the base :class:Service, an UMLStateMachineService does NOT execute a command the instant it arrives. Instead the command is captured on the run task and only honoured at the next state's do-phase checkpoint, guaranteeing that the current state's entry has already run.

  • STOP and ABORT are always held (the most recent one wins) so an operator can always intervene, even mid-transition.
  • Any other command is retained only while a state's do-phase is accepting commands; otherwise it is discarded. Checking that a command is enabled before sending it is the external system's responsibility.

Parameters:

Name Type Description Default
value int

Command value defined by the service control interface.

 required

completed() async

Execute the entry, do, and exit functions bound to completed.

completing() async

Execute the entry, do, and exit functions bound to completing.

dispatch(name, *args, **kwargs)

Dispatch call to the registered function.

Parameters:

Name Type Description Default
name str

Name of the function to be called.

 required
*args object

Positional arguments to forward to the registered function.

 ()
**kwargs object

Keyword arguments to forward to the registered function.

 {}

Returns:

Type Description
bool

The result of the registered function, or the default answer if no
bool

function is registered under the given name.

Raises:

Type Description
RuntimeError

If no registry is attached to the service.

execute() async

Execute the entry, do, and exit functions bound to execute.

get_base_interface(tag_name)

Return the data assembly with tag_name from the current procedure.

Parameters:

Name Type Description Default
tag_name str

Tag name of the data assembly to retrieve.

 required

Returns:

Type Description
DataAssembly

The matching data assembly.

Raises:

Type Description
RuntimeError

If the service has no current procedure.
KeyError

If no data assembly with that tag name is allocated to the current procedure.

get_current_procedure()

Return the currently selected procedure.

Returns:

Type Description
Procedure

The procedure selected by PROCEDURE_CUR.

Raises:

Type Description
RuntimeError

If no current procedure is selected.

held() async

Execute the entry, do, and exit functions bound to held.

holding() async

Execute the entry, do, and exit functions bound to holding.

idle() async

Execute the entry, do, and exit functions bound to idle.

is_state(state_str)

Check whether the service is in a specified state.

Parameters:

Name Type Description Default
state_str str

State name to compare with the current state.

 required

Returns:

Type Description
bool

True if the service is in state_str; otherwise, False.

paused() async

Execute the entry, do, and exit functions bound to paused.

pausing() async

Execute the entry, do, and exit functions bound to pausing.

resetting() async

Execute the entry, do, and exit functions bound to resetting.

resuming() async

Execute the entry, do, and exit functions bound to resuming.

run() async

The single long-lived task driving the service's lifecycle.

Repeatedly executes the state machine's current state (its entry/do*/exit application logic). A state change requested elsewhere only updates the state machine; this loop observes it, lets the current state's exit finish, then moves on to the new state's entry.

While in undefined, the loop checks whether the op-mode is online and activates the service (undefined → idle) if so. This reconciles the case where a service is constructed while already in Aut/Op (no offline→online event fires).

run_on_task(fn)

Run fn on the service's run task, deferring from other threads.

This is the single-writer entry point for any operation that mutates service state and may be invoked from an arbitrary thread (e.g. PEA startup) rather than from the run task itself. When the caller is already on the run task's event loop, fn runs inline (serialized with the run loop); otherwise it is scheduled onto that loop via call_soon_threadsafe (see :meth:ThreadControl.run_or_post).

Parameters:

Name Type Description Default
fn Callable[[], None]

Zero-argument callable that mutates service state.

 required

set_cycle_time(cycle_time)

Set the cycle time for this service.

Parameters:

Name Type Description Default
cycle_time float

Time interval (in seconds) for cyclic state execution.

 required

set_registry(registry)

Set the application logic registry for this service.

Parameters:

Name Type Description Default
registry ApplicationLogicRegistry

The ApplicationLogicRegistry containing user functions to be used.

 required

start()

Start the service's single long-lived run task.

Idempotent: if the task is already running this is a no-op. Uses the attached event loop when available, otherwise falls back to a single daemon thread running a private loop (useful for synchronous tests).

starting() async

Execute the entry, do, and exit functions bound to starting.

state_change()

Request a state change from a cyclic state to the next one.

stop()

Stop the service's run task.

stopped() async

Execute the entry, do, and exit functions bound to stopped.

stopping() async

Execute the entry, do, and exit functions bound to stopping.

unholding() async

Execute the entry, do, and exit functions bound to unholding.

mtppy.pea.application_logic.application_logic_registry.ApplicationLogicRegistry()

Stores and dispatches user-defined logic functions by name.

A registry maps string names to callables that accept a Service instance and return a boolean. If a name is dispatched that has not been registered, the configured default answer is returned instead.

Attributes:

Name Type Description
_registry dict[str, UserFunc]

Dictionary to store registered functions.
_default_answer bool

Default return value when no function is registered.

Initialize an empty registry and set default answer to True.

dispatch(them, name, *args, **kwargs)

Call the function registered under name if it exists.

Parameters:

Name Type Description Default
them Service

The service instance passed as the first argument to the registered function.

 required
name str

The registered name of the function to call.

 required
*args object

Positional arguments forwarded to the registered function.

 ()
**kwargs object

Keyword arguments forwarded to the registered function.

 {}

Returns:

Type Description
bool

The result of the registered function, or _default_answer if no
bool

function is registered under name.

items()

Iterate over registered callback names and functions.

register(*names)

Register a function under several names.

Parameters:

Name Type Description Default
*names str

The keys used to store and later dispatch the function.

 ()

Returns:

Type Description
Callable[[UserFunc], UserFunc]

A decorator that records the decorated function and returns it
Callable[[UserFunc], UserFunc]

unchanged.

Raises:

Type Description
KeyError

If one of the names was already used or is not of str type.

registered_names()

Return registered callback names in registration order.

mtppy.pea.pea_opcua_server.pea_opcua_server.PeaOPCUAServer(opcua_options_provider, endpoint='opc.tcp://localhost:4840/', pea_name='Internal')

OPC UA server that exposes the PEA's DataAssemblies, Services and Procedures to external OPC UA clients (POLs).

Initializes an OPC UA server for PEA.

Parameters:

Name Type Description Default
opcua_options_provider OPCUAOptionsProvider

Instance of an OPCUAOptionsProvider.

 required
endpoint str

OPC UA server endpoint. If not specified, opc.tcp://localhost:4840/ is used.

 'opc.tcp://localhost:4840/'
pea_name str

Name used to label writes that originate inside the PEA itself (see _install_write_hooks). Typically the PEA's TagName read from its PeaInformationLabel data assembly; falls back to "Internal" when unknown.

 'Internal'

add_folder_if_not_exists(root_node, tag_name)

Return the child folder with the given name, creating it if absent.

If the folder already exists (as a child of root_node or anywhere in the namespace) it is returned and referenced; otherwise a new folder is created under root_node.

add_node_or_reference(data_item, parent_node, node_id, browse_name, init_value, variant_type, reference_type=ua.ObjectIds.Organizes)

Add a node to the namespace, or reference it if it already exists.

Parameters:

Name Type Description Default
data_item DataAssembly

DataAssembly that owns the DataItem node.

 required
parent_node NodeId

Parent node under which the node or reference is added.

 required
node_id str

The NodeId of the node to add or reference.

 required
browse_name str

The BrowseName of the node to add or reference.

 required
init_value object

Initial value for the node.

 required
variant_type VariantType

The OPC UA variant type of the variable node.

 required
reference_type int

Reference type created if the node already exists (default is Organizes).

 Organizes

Returns:

Type Description
Node

The created or referenced node.

get_node_part_ns_index()

Get OPC UA server namespace index of default namespace uri.

Returns:

Type Description
str

The OPC UA server namespace index as a string.

init_opcua_server()

Initializes an OPC UA server.

is_folder(nodeid)

Return True if the node with the given id is a FolderType object.

node_exists(nodeid)

Return True if a node with the given id exists in the namespace.

run_opcua_server(data_assemblies, service_set)

Start the OPC UA server instance.

%% ### ::: mtppy.pea.pea_opcua_server.opcua_options_provider.opcua_options_provider.OPCUAOptionsProvider %% ### ::: mtppy.pea.pea_opcua_server.opcua_options_provider.instances.basic_opcua_options_provider.BasicOPCUAOptionsProvider

MTP Generator

mtppy.pea.mtp_generator.mtp_generator.MTPGenerator()

Represents an MTP generator.

MTP Factories for DataAssemblies and Services

mtppy.pea.service_factory.ServiceFactory

Factory class for creating DataAssembly instances.

create_instances(service_dict, data_assemblies=None) classmethod

Recursively create instances from a configuration dictionary.

Parameters:

Name Type Description Default
service_dict dict[str, ServiceConfig]

Dictionary mapping names to ServiceConfig dictionaries.

 required
data_assemblies dict[str, DataAssembly] | None

Pre-resolved data assemblies keyed by name.

 None

Returns:

Type Description
dict[str, dict[str, Service | Procedure | DataAssembly]]

Dictionary mapping names to created instances.

create_procedure(config, data_assemblies=None) classmethod

Create a single Procedure instance from a configuration entry.

Parameters:

Name Type Description Default
config ServiceConfig

Configuration dictionary for the procedure.

 required
data_assemblies dict[str, DataAssembly] | None

Pre-resolved data assemblies keyed by name.

 None

Returns:

Type Description
Procedure

Created procedure instance.

create_service(config, data_assemblies=None) staticmethod

Create a single Service instance from a configuration entry.

Parameters:

Name Type Description Default
config ServiceConfig

Configuration dictionary for the service.

 required
data_assemblies dict[str, DataAssembly] | None

Pre-resolved data assemblies keyed by name.

 None

Returns:

Type Description
Service

Created service instance.

create_services(service_dict, data_assemblies=None) classmethod

Create service instances from a configuration dictionary.

get_data_assembly(config, data_assemblies=None) staticmethod

Look up a pre-resolved data assembly by its name.

initialize_factory() classmethod

Populate the class-level factory dispatch table.

mtppy.pea.data_assembly_factory.DataAssemblyFactory

Factory class for creating DataAssembly instances.

create_data_assemblies(da_dict) staticmethod

Create multiple DataAssembly instances from a dictionary of configurations.

Parameters:

Name Type Description Default
da_dict dict[str, DataAssemblyConfig]

Dictionary mapping names to DataAssemblyConfig dictionaries.

 required

Returns:

Type Description
dict[str, DataAssembly]

Dictionary mapping names to created DataAssembly instances.

create_data_assembly(config) staticmethod

Create a single DataAssembly instance from a configuration entry. If a default value is given in the config this value is set as init_value of the underlying DataItem and the actual value of the DataAssembly is initialized accordingly.

Parameters:

Name Type Description Default
config DataAssemblyConfig

Configuration dictionary for the data assembly.

 required

Returns:

Type Description
DataAssembly | None

Created DataAssembly instance.