openpectus.engine.uod_builder_api

Defines the standard imports for uod modules.

Classes

`HardwareLayerBase`	Base class for hardware layer implementations.
`Register`	Represents a hardware register or named value.
`RegisterDirection`	Support for flags
`UnitOperationDefinitionBase`	Represets the Unit Operation Definition interface used by the OpenPectus engine.
`UodCommand`	Represents a command that targets hardware, such as setting a valve state.
`UodBuilder`	Provides a builder api to define a Unit Operation Definition
`PlotConfiguration`	!!! abstract "Usage Documentation"
`SubPlot`	!!! abstract "Usage Documentation"
`PlotAxis`	!!! abstract "Usage Documentation"
`PlotColorRegion`	!!! abstract "Usage Documentation"

Functions

`RegexNumber`(units[, non_negative, int_only])	Create a regex that parses a number with optional unit to arguments number and optionally number_unit.
`RegexText`([allow_empty])	Parses text into an argument named text.
`RegexCategorical`([exclusive_options, additive_options])	Create a regex that parses categorical text into an argument named option.
`as_float`(value)	Parse string value as a float and return it. If the value is not a number, return None.
`as_int`(value)	Parse string value as int and return it. If the value is not an int, return None.

Module Contents

class openpectus.engine.uod_builder_api.HardwareLayerBase

Base class for hardware layer implementations.

_registers: dict[str, Register]

_is_connected: bool = False

__enter__(): Allow use as context manager.

__exit__(type, value, traceback)

tick(): Invoked on each tick by engine.

__str__()

Return type:: str

property registers: dict[str, Register]

Return type:: dict[str, Register]

property is_connected: bool

Returns a value indicating whether there is an active connection to the hardware. Virtual property.

Return type:: bool

read(r)

Read single register value. Abstract method.

Parameters:: r (Register)
Return type:: Any

read_batch(registers)

Read batch of register values. Override to provide efficient implementation. Virtual method.

Parameters:: registers (Sequence[Register])
Return type:: list[Any]

write(value, r)

Write single register value. Abstract method.

Parameters:

value (Any)
r (Register)

Return type:

None

write_batch(values, registers)

Write batch of register values. Override to provide efficient implementation. Virtual method.

Parameters:

values (Sequence[Any])
registers (Sequence[Register])

connect(): Connect to hardware. Throw HardwareLayerException on error. Virtual method. Implementations must call base method on completed connect.

disconnect(): Disconnect from hardware. Throw HardwareLayerException on error. Virtual method. Implementations must call base method on completed disconnect.

reconnect(): Perform a reconnect. The default implementation just disconnects (ignoring any error) and connects. Override to perform a more advanced reconnect, like using a thread.

validate_offline(): Perform checks that verify that the registers definition is valid. Raise on validation error. Virtual method.

validate_online()

Perform checks that verify online hardware access. Raise on validation error. Virtual method.

The default implementation checks that all Read registers can be read. Override to add additional checks.

show_online_register_details(): Displays online register details to use in uod definition.

class openpectus.engine.uod_builder_api.Register(name, direction, **options)

Represents a hardware register or named value.

Parameters:

name (str)
direction (RegisterDirection)

_name: str

_direction: RegisterDirection

_options: dict[str, Any]

property name

property direction: RegisterDirection

Return type:: RegisterDirection

property options: dict[str, Any]

Return type:: dict[str, Any]

__str__()

Return type:: str

__repr__()

class openpectus.engine.uod_builder_api.RegisterDirection(*args, **kwds)

Bases: enum.Flag

Support for flags

Read

Write

Both

openpectus.engine.uod_builder_api.RegexNumber(units, non_negative=False, int_only=False)

Create a regex that parses a number with optional unit to arguments number and optionally number_unit.

number_unit is only matched if one or more units are given.

Parameters:

units (list[str] | None)
non_negative (bool)
int_only (bool)

Return type:

str

openpectus.engine.uod_builder_api.RegexText(allow_empty=False)

Parses text into an argument named text.

Parameters:: allow_empty (bool)
Return type:: str

openpectus.engine.uod_builder_api.RegexCategorical(exclusive_options=None, additive_options=None)

Create a regex that parses categorical text into an argument named option.

Examples - exclusive:

regex = RegexCategorical(exclusive_options=["Open", "Closed"])

self.assertEqual(re.search(regex, "Closed").groupdict(), dict(option="Closed"))

self.assertEqual(re.search(regex, "VA01"), None)

self.assertEqual(re.search(regex, "Open").groupdict(), dict(option="Open"))

self.assertEqual(re.search(regex, "Open+Closed"), None)

Examples - exclusive and additive:

regex = RegexCategorical(exclusive_options=['Closed'], additive_options=['VA01', 'VA02', 'VA03'])

self.assertEqual(re.search(regex, "Closed").groupdict(), dict(option="Closed"))

self.assertEqual(re.search(regex, "Closed+VA01"), None)

self.assertEqual(re.search(regex, "VA01").groupdict(), dict(option="VA01"))

self.assertEqual(re.search(regex, "VA01+VA02").groupdict(), dict(option="VA01+VA02"))

self.assertEqual(re.search(regex, "Open+Closed"), None)

Parameters:

exclusive_options (list[str] | None)
additive_options (list[str] | None)

Return type:

str

class openpectus.engine.uod_builder_api.UnitOperationDefinitionBase(instrument_name, hwl, author_name, author_email, filename, location, tags, readings, command_factories, command_descriptions, overlapping_command_names_lists, plot_configuration, required_roles, data_log_interval_seconds, base_unit_provider)

Represets the Unit Operation Definition interface used by the OpenPectus engine.

Parameters:

instrument_name (str)
hwl (openpectus.engine.hardware.HardwareLayerBase)
author_name (str)
author_email (str)
filename (str)
location (str)
tags (openpectus.lang.exec.tags.TagCollection)
readings (openpectus.lang.exec.readings.ReadingCollection)
command_factories (dict[str, UodCommandBuilder])
command_descriptions (dict[str, openpectus.lang.exec.readings.UodCommandDescription])
overlapping_command_names_lists (list[list[str]])
plot_configuration (openpectus.protocol.models.PlotConfiguration)
required_roles (set[str])
data_log_interval_seconds (float)
base_unit_provider (openpectus.lang.exec.base_unit.BaseUnitProvider)

instrument

hwl

author_name

author_email

filename

location

tags

system_tags: openpectus.lang.exec.tags.TagCollection | None = None

readings

command_factories

command_descriptions

command_instances: dict[str, UodCommand]

overlapping_command_names_lists: list[list[str]]

plot_configuration

required_roles

data_log_interval_seconds

base_unit_provider: openpectus.lang.exec.base_unit.BaseUnitProvider

__str__()

Return type:: str

property options: dict[str, str]

Return type:: dict[str, str]

build_commands(): Complete configuration of a validated uod. This builds the commands.

validate_configuration()

Validates these areas: - Each Reading matches a defined tag - Each Reading Command is verified

Must have an exec function with an appropriate signature

If the command uses a regular expression as parser function, it is verified that the exec function arguments match the regular expression.

Each process value is verified
- Checks that entry process values have matching tag value type and process value entry_data_type.
Plot configuration
- Process value names exist
- Process values of a single axis share unit

validate_command_signatures(): Validate the signatures of command exec functions

validate_plot_configuration(): Validate plot configuration

validate_read_registers(): Verify that all read registers have a matching tag.

has_command_name(name)

Check whether the command name is defined in the Uod

Parameters:: name (str)
Return type:: bool

has_command_instance(name)

Check whether the Uod currently has an active instance of the named command

Parameters:: name (str)
Return type:: bool

has_any_command_instances()

Return type:: bool

create_command(name)

Create a new command instance. Only one command instance with a given name can exist at a time.

Parameters:: name (str)
Return type:: UodCommand

dispose_command(cmd)

Remove command from list of instances

Parameters:: cmd (UodCommand)

get_command(name)

Get existing command instance by name

Parameters:: name (str)
Return type:: UodCommand | None

get_command_names()

Return type:: list[str]

validate_tag_name(tag_name)

Parameters:: tag_name (str)
Return type:: bool

validate_command_name(command_name)

Parameters:: command_name (str)
Return type:: bool

generate_pcode_examples()

Generate example code as tuples of (description, example_pcode).

Return type:: list[Tuple[str, str]]

create_lsp_definition()

Return type:: openpectus.protocol.models.UodDefinition

class openpectus.engine.uod_builder_api.UodCommand(context, name)

Bases: openpectus.engine.commands.ContextEngineCommand[UnitOperationDefinitionBase]

Represents a command that targets hardware, such as setting a valve state.

Uod commands are specified/implemented by using the UodCommandBuilder class.

Parameters:

context (UnitOperationDefinitionBase)
name (str)

init_fn: INIT_FN | None = None

exec_fn: EXEC_FN | None = None

finalize_fn: FINAL_FN | None = None

arg_parse_fn: PARSE_FN | None = None

_performance_warned = False

cmd_state: dict[str, Any]

static builder()

Helper method that provides a command builder object that must be used to specify the command.

Return type:: UodCommandBuilder

initialize()

execute(args)

Parameters:: args (openpectus.engine.commands.CommandArgs)
Return type:: None

cancel()

finalize(): Overrides must call super().finalize() and must dispose the command.

parse_args(args)

Parse argument input to concrete values. Return None to indicate invalid arguments.

Parameters:: args (str)
Return type:: openpectus.engine.commands.CommandArgs | None

class openpectus.engine.uod_builder_api.UodBuilder

Provides a builder api to define a Unit Operation Definition

instrument: str = ''

hwl: openpectus.engine.hardware.HardwareLayerBase | None = None

tags

command_factories: dict[str, UodCommandBuilder]

command_descriptions: dict[str, openpectus.lang.exec.readings.UodCommandDescription]

overlapping_command_names_lists: list[list[str]] = []

readings

author_name: str = ''

author_email: str = ''

filename: str = ''

location: str = ''

plot_configuration: openpectus.protocol.models.PlotConfiguration

required_roles: set[str]

data_log_interval_seconds = 5

base_unit_provider: openpectus.lang.exec.base_unit.BaseUnitProvider

__str__()

Return type:: str

get_logger()

validate()

with_instrument(instrument_name)

Parameters:: instrument_name (str)
Return type:: UodBuilder

with_hardware_none()

Use no hardware. Mainly used for testing.

Return type:: UodBuilder

with_hardware_opcua(host)

Use OPCUA hardware

Parameters:: host (str)
Return type:: UodBuilder

with_hardware_labjack(serial_number=None)

Use LabJack hardware

Parameters:: serial_number (str | None)
Return type:: UodBuilder

with_hardware(hwl)

Parameters:: hwl (openpectus.engine.hardware.HardwareLayerBase)
Return type:: UodBuilder

with_author(name, email)

Parameters:

name (str)
email (str)

Return type:

UodBuilder

with_filename(filename)

Parameters:: filename (str)
Return type:: UodBuilder

with_location(location)

Parameters:: location (str)
Return type:: UodBuilder

with_hardware_register(name, direction=RegisterDirection.Read, **options)

Define a hardware register.

Parameters:

name (str):

The register name. Used to match the register to a tag.

direction:

Specify whether the register is read, written or both

options:

Specify additional named options. Specific hardwares may require certain options, e.g. the opcua hardware requires a ‘path’ option.

Two special options are callbacks named ‘to_tag’ and ‘from_tag’. If specified, these callbacks convert between hardware values and tag values.

Parameters:

name (str)
direction (openpectus.engine.hardware.RegisterDirection)

Return type:

UodBuilder

with_tag(tag)

Parameters:: tag (openpectus.lang.exec.tags.Tag)
Return type:: UodBuilder

_with_tag(tag)

Parameters:: tag (openpectus.lang.exec.tags.Tag)
Return type:: UodBuilder

with_accumulated_volume(totalizer_tag_name)

Parameters:: totalizer_tag_name (str)
Return type:: UodBuilder

with_accumulated_cv(cv_tag_name, totalizer_tag_name)

Parameters:

cv_tag_name (str)
totalizer_tag_name (str)

Return type:

UodBuilder

with_base_unit_provider(units, provider_tag_name, provider_block_tag_name)

Parameters:

units (list[str])
provider_tag_name (str)
provider_block_tag_name (str)

with_command(name, exec_fn, init_fn=None, finalize_fn=None, arg_parse_fn='default')

Define a custom unit operation command that can be executed from pcode.

Parameters:

name (str):: The command name. Used in pcode to refer to the command
exec_fn:: The function to execute when the command is run. The function will be called once in each engine tick until it is completed. To signal completion, use cmd.set_complete().
init_fn, optional:: An optional initialization function that is invoked once before the command runs
finalize_fn, optional:: An optional function that is run once when the command is completed
arg_parse_fn, optional:: An optional function that is used to parse the command argument in pcode to **kvargs that are given as named arguments to the exec function. A number of cases are possible: The default parsing function (arg_parse_fn=”default”) passes the argument from the pcode line verbatim to the exec function using the argument name ‘value’. If no parse function is set (arg_parse_fn=None), no arguments are passed to the exec function. To use a regular expression as argument parser, use the method with_command_regex_arguments() instead.

Parameters:

name (str)
exec_fn (Callable[Ellipsis, None])
init_fn (Callable[[UodCommand], None] | None)
finalize_fn (Callable[[UodCommand], None] | None)
arg_parse_fn (Callable[[str], openpectus.engine.commands.CommandArgs | None] | Literal['default'] | None)

Return type:

UodBuilder

with_command_regex_arguments(name, arg_parse_regex, exec_fn, init_fn=None, finalize_fn=None)

Define a custom unit operation command with a regular expression as argument parser.

Parameters:

name (str):: The command name. Used in pcode to refer to the command
arg_parse_regex (str):: The regular expression to use for parsing. The expression must use named groups to specify the argument names and values.
exec_fn:: The function to execute when the command is run. The function will be called once in each engine tick until it is completed. To signal completion, use cmd.set_complete().
init_fn, optional:: An optional initialization function that is invoked once before the command runs
finalize_fn, optional:: An optional function that is run once when the command is completed

Parameters:

name (str)
arg_parse_regex (str)
exec_fn (Callable[Ellipsis, None])
init_fn (Callable[[UodCommand], None] | None)
finalize_fn (Callable[[UodCommand], None] | None)

Return type:

UodBuilder

with_command_overlap(command_names)

Specify that the given commands names overlap, i.e. at most one can run at a time.

Parameters:: command_names (list[str])
Return type:: UodBuilder

with_process_value(tag_name)

Add process value that displays the given tag in the UI.

Parameters:: tag_name (str)
Return type:: UodBuilder

with_process_value_entry(tag_name, entry_data_type='auto', execute_command_name=None)

Add process value for the given tag and enable setting its value by typing in a value in the UI.

Parameters:

tag_name: str

The tag to display

entry_data_type: “str” | “int” | “float” | “auto”, optional, default=”auto”

The data type to display and validate in the UI. Default is “auto” which uses the type of the tag’s value. The “auto” value requires that the tag has a default value with the correct type.

execute_command_name: str: optional

The command to execute to ‘set the process value’. In principle, this can be any pcode command that takes a single argument as given by the user. However, the intension is that the command ‘sets the process value’, probably by writing the value to the register backing the tag.

If not specified, it defaults to the tag name, assuming that there is a command with the same name as the tag that takes a single argument.

Parameters:

tag_name (str)
entry_data_type (openpectus.protocol.models.EntryDataType)
execute_command_name (str | None)

Return type:

UodBuilder

with_process_value_choice(tag_name, command_options)

Add process value for the given tag and enable command choices.

Parameters:

tag_name: str: The tag to display. The tag’s choices are also used as commands, unless command_options are specified.
command_options: optional: Dictionary that maps command names to their pcode implementation.

Parameters:

tag_name (str)
command_options (dict[str, str])

Return type:

UodBuilder

_with_process_value(reading)

Parameters:: reading (openpectus.lang.exec.readings.Reading)
Return type:: UodBuilder

with_plot_configuration(plot_configuration)

Parameters:: plot_configuration (openpectus.protocol.models.PlotConfiguration)

with_required_roles(required_roles)

Parameters:: required_roles (set[str])
Return type:: UodBuilder

with_data_log_interval_seconds(data_log_interval_seconds)

Parameters:: data_log_interval_seconds (float)
Return type:: UodBuilder

with_measurement_unit(unit, unit_relation=None, quantity_relation=None, quantity=None)

Parameters:

unit (str)
unit_relation (None | str)
quantity_relation (None | dict[str, str])
quantity (None | str)

Return type:

UodBuilder

build()

Return type:: UnitOperationDefinitionBase

class openpectus.engine.uod_builder_api.PlotConfiguration(/, **data)

Bases: ProtocolModel

!!! abstract “Usage Documentation”: [Models](../concepts/models.md)

A base class for creating Pydantic models.

Attributes:

__class_vars__: The names of the class variables defined on the model. __private_attributes__: Metadata about the private attributes of the model. __signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__: Whether model building is completed, or if there are still undefined fields. __pydantic_core_schema__: The core schema of the model. __pydantic_custom_init__: Whether the model has a custom __init__ function. __pydantic_decorators__: Metadata containing the decorators defined on the model.

This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to: __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.

__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models. __pydantic_post_init__: The name of the post-init method for the model, if defined. __pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel]. __pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model. __pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects. __pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra]: is set to ‘allow’.

__pydantic_fields_set__: The names of fields explicitly set during instantiation. __pydantic_private__: Values of private attributes set on the model instance.

Parameters:: data (Any)

process_value_names_to_annotate: list[str]

color_regions: list[PlotColorRegion]

sub_plots: list[SubPlot]

x_axis_process_value_names: list[str]

static empty()

Return type:: PlotConfiguration

class openpectus.engine.uod_builder_api.SubPlot(/, **data)

Bases: ProtocolModel

!!! abstract “Usage Documentation”: [Models](../concepts/models.md)

A base class for creating Pydantic models.

Attributes:

__class_vars__: The names of the class variables defined on the model. __private_attributes__: Metadata about the private attributes of the model. __signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__: Whether model building is completed, or if there are still undefined fields. __pydantic_core_schema__: The core schema of the model. __pydantic_custom_init__: Whether the model has a custom __init__ function. __pydantic_decorators__: Metadata containing the decorators defined on the model.

This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to: __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.

__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models. __pydantic_post_init__: The name of the post-init method for the model, if defined. __pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel]. __pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model. __pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects. __pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra]: is set to ‘allow’.

__pydantic_fields_set__: The names of fields explicitly set during instantiation. __pydantic_private__: Values of private attributes set on the model instance.

Parameters:: data (Any)

axes: list[PlotAxis]

ratio: int | float

class openpectus.engine.uod_builder_api.PlotAxis(/, **data)

Bases: ProtocolModel

!!! abstract “Usage Documentation”: [Models](../concepts/models.md)

A base class for creating Pydantic models.

Attributes:

__class_vars__: The names of the class variables defined on the model. __private_attributes__: Metadata about the private attributes of the model. __signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__: Whether model building is completed, or if there are still undefined fields. __pydantic_core_schema__: The core schema of the model. __pydantic_custom_init__: Whether the model has a custom __init__ function. __pydantic_decorators__: Metadata containing the decorators defined on the model.

This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to: __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.

__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models. __pydantic_post_init__: The name of the post-init method for the model, if defined. __pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel]. __pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model. __pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects. __pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra]: is set to ‘allow’.

__pydantic_fields_set__: The names of fields explicitly set during instantiation. __pydantic_private__: Values of private attributes set on the model instance.

Parameters:: data (Any)

label: str

process_value_names: list[str]

y_max: int | float

y_min: int | float

color: str

class openpectus.engine.uod_builder_api.PlotColorRegion(/, **data)

Bases: ProtocolModel

!!! abstract “Usage Documentation”: [Models](../concepts/models.md)

A base class for creating Pydantic models.

Attributes:

__class_vars__: The names of the class variables defined on the model. __private_attributes__: Metadata about the private attributes of the model. __signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__: Whether model building is completed, or if there are still undefined fields. __pydantic_core_schema__: The core schema of the model. __pydantic_custom_init__: Whether the model has a custom __init__ function. __pydantic_decorators__: Metadata containing the decorators defined on the model.

This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to: __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.

__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models. __pydantic_post_init__: The name of the post-init method for the model, if defined. __pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel]. __pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model. __pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects. __pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra]: is set to ‘allow’.

__pydantic_fields_set__: The names of fields explicitly set during instantiation. __pydantic_private__: Values of private attributes set on the model instance.

Parameters:: data (Any)

process_value_name: str

value_color_map: dict[str | int | float, str]

openpectus.engine.uod_builder_api.as_float(value)

Parse string value as a float and return it. If the value is not a number, return None.

Parameters:: value (str)
Return type:: float | None

openpectus.engine.uod_builder_api.as_int(value)

Parse string value as int and return it. If the value is not an int, return None.

Parameters:: value (str)
Return type:: int | None