Skip to content

Create PEA from scratch

The standalone PEA Engineering approach sets code first. You implement the PEA with the MTPPy2.0 API and the MTP-File will be generated based on your implementation.

The tutorial firstly defines a simplified workflow and then walks you through this workflow step by step, explaining the concepts, methods and generated artefacts and then applies it to a minimal example.

Development workflow

Developing a standalone PEA with MTPPy2.0 splits naturally into two parts: deciding what the PEA should do, and then expressing that design with the MTPPy2.0 API.

1. Design the PEA

Before writing code, decide on the asset’s capabilities:

  • Which services does the PEA offer?
  • For each service, which configuration parameters, procedures, procedure parameters, and report values are required?
  • Which DataAssemblies (indicator elements, active elements, etc.) live on the PEA level and are shared by services?
  • Which state transitions does each service need? Should a procedure be self-completing or wait for an external command?

Application to the RandomNumberGenerator Example

The example PEA offers a single service, rand_num_gen, that generates a random number inside a configured range on every execution cycle. Working the design through the questions above gives:

  • Service: one Service, rand_num_gen.
  • Procedure: one procedure, cont, that is not self-completing — it keeps executing until the POL sends a COMPLETE command.
  • Procedure parameters: lower_bound and upper_bound (the range), both DIntServParam.
  • Report value: generated_value, an AnaView, that the POL reads to obtain the result.
  • PEA-level DataAssemblies: one active element, a PIDCtrl, to exercise a non-service DataAssembly in the generated MTP.

2. Implement the service logic

Use the MTPPy2.0 API to define configuration parameters, procedures, procedure parameters, and report values.

  1. Create your own Service class

    1. Configure the class

      • Add service-level configuration parameters with add_configuration_parameter.
      • Create one or more Procedure instances and add them with add_procedure.
      • Add procedure parameters with add_procedure_parameter and report values with add_report_value.

    2. Model the service logic by overriding the state methods.

      • Service declares all of idle, starting, execute, completing, completed, pausing, paused, resuming, holding, held, unholding, stopping, stopped, aborting, aborted, and resetting as async abstract methods, so every one of them must be overridden in your subclass (an empty pass body is fine for states you do not use).
      • Cyclic states (idle, execute, …) are implemented as a loop that runs while the service is in that state and has not been stopped, with an await asyncio.sleep(...) once per cycle. Transient states (starting, completing, resetting, …) do their one-shot work and then call self.state_change() to advance the state machine.
      • Inside the state methods, read parameter values and write report values through the procedure object (get_base_interface(...), procedure_parameters[...], report_values[...]).

For the concrete API usage, see pea_minimal.py. The class RandomNumberGenerator defines the service, adds a cont procedure with three DIntServParam parameters and an AnaView report value, and writes the generated random number into generated_value in execute.

3. Assemble and run the PEA

  • Instantiate PEA with the desired OPC UA endpoint.
  • Register your services with add_service.
  • Register PEA-level DataAssemblies with add_data_assembly (and input/output process values with add_process_value_in / add_process_value_out, and alarms with add_alarm, if any).
  • Start the PEA with pea.start(). This initializes the embedded OPC UA server and starts each service's single long-lived run task, which drives the state machine.

4. Generate the MTP file

The MTP file is a CAEX archive that describes the PEA (its data assemblies, services, process values and alarms). It is generated from the same in-memory model used to run the PEA, by MTPGenerator:

from mtppy.pea.mtp_generator.mtp_generator import MTPGenerator

MTPGenerator().generate(pea, out_path="output")
  • pea is the assembled PEA model and out_path is the directory the archive is written to. The optional source_document_information dict overrides the SourceDocumentInformation fields in the CAEX manifest.
  • generate() writes the archive to <out_path>/MyPrettyMTP-<uuid>.mtp and returns. It builds the CAEX XML and writes the file synchronously, so it does not require a running OPC UA server: you can generate the MTP before pea.start(), after it, or on a model you never start at all.
  • A data item that is exposed over OPC UA is emitted as an IDReferenceType link to its OPC UA node; a data item with no communication object is emitted as a StaticValueAttributeType instead.

What the generated MTP looks like

The .mtp file is a zip archive holding the CAEX manifest and its schema:

MyPrettyMTP-<uuid>.mtp
├── manifest.aml                  # the CAEX model of your PEA
├── _rels/.rels
├── [Content_Types].xml
└── CAEX_ClassModel_V.3.0.xsd

manifest.aml is a CAEXFile whose InstanceHierarchy sections mirror the PEA model: ModuleTypePackage (PEA metadata and version), ServerAssemblies (the OPC UA server assembly and the external interface for each data item), DataAssemblies, Services (each service with its procedures, parameters and report values), ProcessValues, and Alarms. The same manifest can be inspected with the mtp_inspector tool, e.g.

mtp_inspector --f output/MyPrettyMTP-<uuid>.mtp -d -a Name -a RefB

{'PeaInformationLabel': {'Name': 'PeaInformationLabel',
                 'RefBaseSystemUnitPath': 'MTPDataAssemblySUCLib/DataAssembly/PeaElement/PeaInformationLabel'},
 'cont': {'Name': 'cont',
          'RefBaseSystemUnitPath': 'MTPDataAssemblySUCLib/DataAssembly/ServiceElement/ProcedureHealthView'},
 'generated_value': {'Name': 'generated_value',
                     'RefBaseSystemUnitPath': 'MTPDataAssemblySUCLib/DataAssembly/IndicatorElement/AnaView'},
 'lower_bound': {'Name': 'lower_bound',
                 'RefBaseSystemUnitPath': 'MTPDataAssemblySUCLib/DataAssembly/ServiceElement/ParameterElement/DIntServParam'},
 'pid_ctrl': {'Name': 'pid_ctrl',
              'RefBaseSystemUnitPath': 'MTPDataAssemblySUCLib/DataAssembly/ActiveElement/PIDCtrl'},
 'rand_num_gen': {'Name': 'rand_num_gen',
                  'RefBaseSystemUnitPath': 'MTPDataAssemblySUCLib/DataAssembly/ServiceElement/ServiceControl'},
 'upper_bound': {'Name': 'upper_bound',
                 'RefBaseSystemUnitPath': 'MTPDataAssemblySUCLib/DataAssembly/ServiceElement/ParameterElement/DIntServParam'}}
}

Example

The complete self-contained example is maintained in src/examples/scratch/pea_minimal.py. It defines a RandomNumberGenerator service, assembles a PEA, starts the OPC UA server, runs a small control sequence, and (optionally) generates the MTP file.

Service and PEA definition

"""**pea_minimal**, a minimal example of a self-contained PEA application."""

__copyright__ = (
    "Copyright (c) 2026 Dresden University of Technology, Process-to-Order Group"
)
__license__ = "MIT"

import argparse
import asyncio
import random
import sys
import time
from pathlib import Path

from mtppy.part3.data_assembly_set.active_elements.pid.pid_ctrl import PIDCtrl
from mtppy.part3.data_assembly_set.indicator_elements.views.ana_view import AnaView
from mtppy.part4.automation_services.procedure import Procedure
from mtppy.part4.automation_services.service import Service
from mtppy.part4.automation_services.service_elements.parameter_elements.dint_serv_param import (  # noqa: E501
    DIntServParam,
)
from mtppy.pea.mtp_generator.mtp_generator import MTPGenerator
from mtppy.pea.pea import PEA
from mtppy.utils.logging import CHOICES, logger, setLogLevel
from utils.resolve_unit.unit_lib import Units


class RandomNumberGenerator(Service):
    """A minimal service that generates random numbers between two bounds."""

    def __init__(self, tag_name: str, tag_description: str) -> None:
        """Initialise the service with a single procedure and two DINT params."""
        super().__init__(tag_name, tag_description)
        self.set_cycle_time(0.2)

        # Define a single procedure with two parameters, one report value
        proc_1 = Procedure(1, "cont", is_self_completing=False)
        proc_1.add_procedure_parameter(
            DIntServParam(
                "lower_bound",
                v_op=10,
                v_min=0,
                v_max=100,
                v_scl_min=0,
                v_scl_max=100,
                v_unit=Units.UNSPECIFIED.code,
            ),
        )
        proc_1.add_procedure_parameter(
            DIntServParam(
                "upper_bound",
                v_op=90,
                v_min=0,
                v_max=100,
                v_scl_min=0,
                v_scl_max=100,
                v_unit=Units.UNSPECIFIED.code,
            ),
        )
        proc_1.add_report_value(
            AnaView(
                "generated_value",
                v_scl_min=0,
                v_scl_max=100,
                v_unit=Units.UNSPECIFIED.code,
            ),
        )
        self.add_procedure(proc_1)
        proc_1.add_required_equipment(proc_1.procedure_parameters)
        proc_1.add_required_equipment(proc_1.report_values)

    # Define the application logic
    async def idle(self) -> None:
        """Idle: log state each cycle."""
        logger.app("- Idle -")
        cycle = 0
        while self.is_state("idle") and not self.thread_ctrl.is_stopped():
            logger.app(
                f" Idle cycle={cycle}, sm.act_state={self.state_machine.act_state}",
            )
            cycle += 1
            await asyncio.sleep(0.2)

    async def starting(self) -> None:
        """Starting: apply parameters then transition to execute."""
        logger.app("- Starting -")
        logger.app("  Applying procedure parameters...")
        self.state_change()

    async def execute(self) -> None:
        """Execute: generate random numbers within the configured bounds."""
        logger.app("- Execute -")
        cycle = 0
        while self.is_state("execute") and not self.thread_ctrl.is_stopped():
            logger.debug(f"  About to execute cycle {cycle}")
            # Read procedure parameter data assemblies
            lower_bound: DIntServParam = self.get_base_interface("lower_bound")
            upper_bound: DIntServParam = self.get_base_interface("upper_bound")
            generated_value: AnaView = self.get_base_interface("generated_value")

            generated_value.V = random.randint(  # noqa: S311
                lower_bound.VOut,
                upper_bound.VOut,
            )

            logger.app(
                f" {cycle}, generated_number={generated_value.V}, "
                f"bounds={[lower_bound.VOut, upper_bound.VOut]}",
            )

            cycle += 1
            await asyncio.sleep(self.cycle_time)

    async def completing(self) -> None:
        """Completing: transition to completed."""
        self.state_change()

    async def completed(self) -> None:
        """Completed: no-op."""

    async def pausing(self) -> None:
        """Pausing: no-op."""

    async def paused(self) -> None:
        """Paused: no-op."""

    async def resuming(self) -> None:
        """Resuming: no-op."""

    async def holding(self) -> None:
        """Holding: no-op."""

    async def held(self) -> None:
        """Held: no-op."""

    async def unholding(self) -> None:
        """Unholding: no-op."""

    async def stopping(self) -> None:
        """Stopping: no-op."""

    async def stopped(self) -> None:
        """Stopped: no-op."""

    async def aborting(self) -> None:
        """Aborting: no-op."""

    async def aborted(self) -> None:
        """Aborted: no-op."""

    async def resetting(self) -> None:
        """Resetting: log and transition."""
        logger.app("- Resetting -")
        self.state_change()


def parse_args() -> argparse.Namespace:
    """Parse command-line arguments for the minimal PEA example."""
    p = argparse.ArgumentParser("minimal PEA example")
    p.add_argument(
        "-l",
        "--log-level",
        dest="log_level",
        default="APP",
        type=str,
        choices=["DEBUG", "INFO", *CHOICES, "WARNING", "ERROR", "CRITICAL"],
        help="Set the logging level (default: APP)",
    )
    p.add_argument(
        "--duration",
        type=float,
        default=10.0,
        help="Set the duration of the execute phase",
    )
    p.add_argument(
        "-e",
        "--endpoint",
        default="opc.tcp://localhost:4840",
        help="Set the OPCUA endpoint (default opc.tcp://localhost:4840).",
    )
    p.add_argument(
        "--daemon",
        action="store_true",
        default=False,
        help="don't stop, continue to run (default: exit ).",
    )
    p.add_argument(
        "--mtp",
        default=None,
        help="Directory to generate the MTP file into (default: no MTP is generated).",
    )
    return p.parse_args()


if __name__ == "__main__":
    args = parse_args()
    setLogLevel(args.log_level)

    # -------------------------- PEA setup --------------------------
    # create PEA instance with specified endpoint
    pea = PEA(endpoint=args.endpoint)

    # define service and add it to the PEA
    service_1 = RandomNumberGenerator(
        "rand_num_gen",
        "This service generates random numbers",
    )
    pea.add_service(service_1)

    # define an active element and add it to the PEA
    pid_ctrl = PIDCtrl("pid_ctrl")
    pea.add_data_assembly(pid_ctrl)

    # Setup done, now start the PEA
    logger.app("--- Start PEA ---")
    pea.start()
    time.sleep(1)

    # -------------------------- POL test script --------------------------
    from mtppy.part4.definitions.command_codes import (
        CommandCodes as CC,  # noqa: N817 - acronym for readability
    )

    opcua_server = pea.pea_opcua_server.opcua_server
    ns_index = pea.pea_opcua_server._get_opcua_ns_index()  # noqa: SLF001

    def nodeid(cmd: str) -> str:
        """Build an OPC UA node ID string from a command name."""
        return f"ns={ns_index};s={cmd}"

    def opcua_write(cmd: str, val: object) -> None:
        """Write a value to the OPC UA node identified by cmd."""
        opcua_server.get_node(nodeid(cmd)).set_value(val)

    logger.pol("--- Set procedure parameters to Operator mode ---")
    opcua_write("lower_bound.StateOpOp", True)
    opcua_write("upper_bound.StateOpOp", True)

    logger.pol("--- Set procedure parameter values ---")
    opcua_write("lower_bound.VOp", 40)
    opcua_write("upper_bound.VOp", 60)

    logger.pol("--- Set service to Operator mode ---")
    opcua_write("rand_num_gen.StateOpOp", True)
    time.sleep(0.1)

    logger.pol("--- Start service ---")
    opcua_write("rand_num_gen.ProcedureOp", 1)
    opcua_write("rand_num_gen.CommandOp", CC.START.code)
    time.sleep(args.duration)

    logger.pol("--- Complete service ---")
    opcua_write("rand_num_gen.CommandOp", CC.COMPLETE.code)
    time.sleep(1)

    logger.pol("--- Reset service ---")
    opcua_write("rand_num_gen.CommandOp", CC.RESET.code)
    time.sleep(1)

    # -------------------------- MTP generation --------------------------
    # The MTP file is generated from the same in-memory model that defines the
    # running PEA. MTPGenerator serializes the PEA's data assemblies, services,
    # process values and alarms into a CAEX .mtp archive.
    if args.mtp:
        out_dir = Path(args.mtp)
        out_dir.mkdir(parents=True, exist_ok=True)
        logger.app(f"--- Generate MTP into {out_dir} ---")
        MTPGenerator().generate(pea, out_path=str(out_dir))

    # -------------------------- PEA shutdown or detach --------------------------
    if not args.daemon:
        logger.app("--- Stop PEA ---")
        pea.stop()
        sys.exit(0)

Run the example

From the project root, run:

python -m examples.scratch.pea_minimal --endpoint opc.tcp://localhost:4840 --duration 10

To also generate the MTP file into the output directory, pass --mtp output:

python -m examples.scratch.pea_minimal --duration 10 --mtp output