Skip to content

Automation State Machine API

The AutomationStateMachine is the base class for implementing custom automation logic. It extends the statemachine library with PyAutomation-specific features like data buffering, tag subscription, and database logging.

class automation.state_machine.StateMachineCore(name, description='', classification='', interval=1.0, identifier=None, buffer_size=10)

Base class for all state machines in PyAutomation.

It defines the standard lifecycle states: * Start: Initialization state. * Wait: Waiting for data or conditions. * Run: Main execution logic. * Reset: Resetting variables or states. * Restart: Restarting the machine cycle.

while_starting(self)

Executed every machine loop while in Start state.

Default behavior: 1. Initialize buffer size. 2. Transition to Wait state.

while_waiting(self)

Executed every machine loop while in Wait state.

Default behavior: 1. Checks if internal buffers for subscribed tags are full. 2. If full, transitions to Run state.

while_running(self)

Executed every machine loop while in Run state.

Note: This method should be overridden by child classes to implement custom logic.

while_resetting(self)

Executed every machine loop while in Reset state.

Default behavior: Transition to Start state.

while_restarting(self)

Executed every machine loop while in Restart state.

Default behavior: 1. Clear buffers. 2. Transition to Wait state.

set_socketio(self, sio)
put_attr(self, attr_name, value, user=None)

Updates an attribute of the state machine and persists it to the database.

Parameters:

  • attr_name (str): Attribute name.
  • value: New value.
  • user (User): User performing the action.
add_process_variable(self, name, tag, read_only=False)

Registers a process variable (ProcessType) dynamically.

Parameters:

  • name (str): Variable name.
  • tag (Tag): Associated tag.
  • read_only (bool): If True, the variable cannot be modified by the machine logic (input only).
get_process_variables(self)

Retrieves all ProcessType variables defined in the machine.

Returns:

  • dict: Serialized process variables.
get_process_variable(self, name)

Retrieves a specific ProcessType variable by name.

Parameters:

  • name (str): Variable name.

Returns:

  • dict: Serialized process variable.
set_buffer_size(*args, **kwargs)
restart_buffer(self)

Clears and reinitializes data buffers for all subscribed tags.

get_subscribed_tags(*args, **kwargs)
get_not_subscribed_tags(*args, **kwargs)
subscribe_to(self, tag, default_tag_name=None)

Subscribes the machine to a tag.

If default_tag_name is provided, it attempts to map the tag to that specific internal variable. Otherwise, it automatically maps to a matching variable or creates a new one.

Parameters:

  • tag (Tag): The tag to subscribe.
  • default_tag_name (str, optional): Internal variable name target.

Returns:

  • tuple: (Success bool, Message str) or True/False.
unsubscribe_to(*args, **kwargs)
process_type_exists(*args, **kwargs)
get_internal_process_type_variables(*args, **kwargs)
get_read_only_process_type_variables(self)

Returns ProcessType variables that ARE read-only (inputs).

notify(*args, **kwargs)
attach(self, machine, tag)

Attaches the machine as an observer to a tag in the CVT.

transition(*args, **kwargs)
get_interval(*args, **kwargs)
set_interval(*args, **kwargs)
get_allowed_actions(self)

Returns a list of allowed target states for transitions from the current state.

Used for UI controls.

loop(self)

Main execution loop called by the worker thread.

It dynamically calls the while_<state> method corresponding to the current state.

get_states(*args, **kwargs)
get_serialized_models(*args, **kwargs)
serialize(*args, **kwargs)
on_start_to_wait(self)

Transition action: Start -> Wait.

on_wait_to_run(self)

Transition action: Wait -> Run.

on_wait_to_restart(self)

Transition action: Wait -> Restart.

on_wait_to_reset(self)

Transition action: Wait -> Reset.

on_run_to_restart(self)

Transition action: Run -> Restart.

on_run_to_reset(self)

Transition action: Run -> Reset.

on_reset_to_start(self)

Transition action: Reset -> Start.

on_restart_to_wait(self)

Transition action: Restart -> Wait.

on_enter_starting(self)
on_enter_waiting(self)
on_enter_running(self)
on_enter_restarting(self)
on_enter_resetting(self)
class automation.state_machine.DAQ(name='DAQ', description='', classification='Data Acquisition System')

Data Acquisition (DAQ) State Machine.

Specialized state machine for polling OPC UA tags at a specific interval.

while_waiting(self)

Executed in Wait state. DAQ immediately transitions to Run.

while_running(self)

Executed in Run state.

Reads values from OPC UA using the client manager and updates the CVT and DAS buffers.

set_opcua_client_manager(self, manager)

Sets the OPC UA Client Manager reference.

class automation.state_machine.OPCUAServer(name='OPCUAServer', description='', classification='OPC UA Server')

OPC UA Server State Machine.

Manages the lifecycle of an embedded OPC UA Server, exposing CVT tags, alarms, and machine states.

while_starting(self)

Executed in Start state.

Initializes the OPC UA Server, configures endpoints, creates namespaces, and populates the address space with CVT tags, Alarms, and Engines folders.

while_waiting(self)

Executed in Wait state. Transitions to Run.

while_running(self)

Executed in Run state.

Continuously updates the values of tags, alarms, and engines in the OPC UA address space.

while_resetting(self)

Executed in Reset state. Transitions back to Starting to restart the server.

class automation.state_machine.AutomationStateMachine(name, description='', classification='', interval=1.0, identifier=None, buffer_size=10)

Extended State Machine with additional states for testing and sleeping.

States: * Test: Testing or simulation mode. * Sleep: Idle or low-power mode.

while_testing(self)

Executed in Test state.

while_sleeping(self)

Executed in Sleep state.

on_test_to_restart(self)
on_test_to_reset(self)
on_sleep_to_restart(self)
on_sleep_to_reset(self)
on_enter_sleeping(self)
on_enter_testing(self)