openpectus.lang.exec.uod

Attributes

`logger`
`INIT_FN`	Command initialization method.
`EXEC_FN`	Command execution function. Must take UodCommand and **kvargs as inputs. May be invoked multiple times.
`FINAL_FN`	Command finalization method.
`PARSE_FN`	Command argument parse function. Parses a string into a dictionary. Returns None on invalid input.

Classes

`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.
`UodCommandBuilder`	Used to builds command specifications and as factory to instantiate commands from the specifications.
`UodBuilder`	Provides a builder api to define a Unit Operation Definition
`RegexNamedArgumentParser`

Functions

`defaultArgumentParser`(args)	The default command argument parser.
`unescape`(re_escaped_string)	re.escape is used to escape options supplied to

Module Contents

openpectus.lang.exec.uod.logger

class openpectus.lang.exec.uod.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

openpectus.lang.exec.uod.INIT_FN: Command initialization method.

openpectus.lang.exec.uod.EXEC_FN

Command execution function. Must take UodCommand and **kvargs as inputs. May be invoked multiple times.

NOTE: The type system does not seem to support describing the real signature (which should be something like Callable[[UodCommand, Unpack[CommandArgs]], None]).

For this reason we validate the exec function dynamically during uod validation. This validation check must be updated if EXEC_FN changes. It is implemented in UnitOperationDefinitionBase.verify_command_signatures().

openpectus.lang.exec.uod.FINAL_FN: Command finalization method.

openpectus.lang.exec.uod.PARSE_FN: Command argument parse function. Parses a string into a dictionary. Returns None on invalid input.

class openpectus.lang.exec.uod.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.lang.exec.uod.UodCommandBuilder

Used to builds command specifications and as factory to instantiate commands from the specifications.

name = ''

init_fn: Callable[[UodCommand], None] | None = None

exec_fn: Callable[Ellipsis, None] | None = None

finalize_fn: Callable[[UodCommand], None] | None = None

arg_parse_fn: Callable[[str], openpectus.engine.commands.CommandArgs | None] | None = None

__str__()

Return type:: str

with_name(name)

Define the name of the command. Required.

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

with_init_fn(init_fn)

Define an initialization function for the command. Optional.

Parameters:: init_fn (Callable[[UodCommand], None])
Return type:: UodCommandBuilder

with_exec_fn(exec_fn)

Define the command’s execution function. Required.

Example:

def my_command_exec(cmd: UodCommand, **kvargs):
    # kvargs contains the result of the arg_parse function.
    pass

Parameters:: exec_fn (Callable[Ellipsis, None])
Return type:: UodCommandBuilder

with_finalize_fn(finalize_fn)

Define a finalizer function for the command. Optional.

Parameters:: finalize_fn (Callable[[UodCommand], None])
Return type:: UodCommandBuilder

with_arg_parse_fn(arg_parse_fn)

Define an argument parser function. Optional.

The function is given the command argument string from the method and should parse that and return the result as a dictionary. The result is passed to the excution function automatically.

If not arg_parse function is defined, no arguments are provided to the execute function.

Example:

def my_command_parse_args(args: str) -> dict[str, Any] | None:
  # parse the argument string and return the values as a dictionary.
  return None # if args are invalid.

Parameters:: arg_parse_fn (Callable[[str], openpectus.engine.commands.CommandArgs | None])
Return type:: UodCommandBuilder

build(uod)

Construct the command

Parameters:: uod (UnitOperationDefinitionBase)
Return type:: UodCommand

class openpectus.lang.exec.uod.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

openpectus.lang.exec.uod.defaultArgumentParser(args)

The default command argument parser.

Passes the string argument verbatim to the command exec function as a single argument named value.

E.g. for the command pcode MyCommand: 87, value will have the value ‘87’.

When using this argument parser, the exec function looks like this:

def exec(cmd: UodCommand, value: str):
  # implementation here
  cmd.set_complete()

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

openpectus.lang.exec.uod.unescape(re_escaped_string)

re.escape is used to escape options supplied to RegexCategorical. This function can reverses operation.

assert unescape(re.escape(‘A B’)) == ‘A B’ assert unescape(re.escape(‘A/B’)) == ‘A/B’

Parameters:: re_escaped_string (str)
Return type:: str

class openpectus.lang.exec.uod.RegexNamedArgumentParser(regex, name=None)

Parameters:

regex (str)
name (str | None)

regex

name = None

__str__()

Return type:: str

parse(args)

Parameters:: args (str)
Return type:: dict[str, Any] | None

validate(args)

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

get_named_groups()

Return type:: list[str]

get_units()

Return type:: list[str]

get_exclusive_options()

Return type:: list[str]

get_additive_options()

Return type:: list[str]

static get_instance(parse_func)

Gets the RegexNamedArgumentParser associated with parse_func or None if parse_func is not a regex validator

Return type:: RegexNamedArgumentParser | None

serialize()

Return type:: str

static deserialize(serialized, name=None)

Parameters:

serialized (str)
name (str | None)

Return type:

RegexNamedArgumentParser | None