Skip to content

PyAutomation Core API

The PyAutomation class is the central entry point for the framework. It implements the Singleton pattern to ensure a unified control point for all services (Tags, Alarms, Database, Workers).

Usage

from automation import PyAutomation

app = PyAutomation()
app.run(debug=True)

Class Documentation

class automation.PyAutomation(*args, **kwargs)

Automation is a singleton <https://en.wikipedia.org/wiki/Singleton_pattern>_ class designed to develop multi-threaded web applications for Industrial Applications.

You can initialize and run PyAutomation Framework in different ways depending on your requirements.

Example 1: Using only PyAutomation Framework

from automation import PyAutomation
app = PyAutomation(certfile=certfile, keyfile=keyfile)
app.run(debug=True, create_tables=True)

Example 2: Extending PyAutomation Framework with Flask Application

from automation import PyAutomation
from app import CreateApp
application = CreateApp()
server = application()  # Flask App
app = PyAutomation(certfile=certfile, keyfile=keyfile)
app.run(create_tables=True)
PORTS

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal.

int('0b100', base=0) 4

add_db_table(self, table)

Registers a new table model in the database manager.

Parameters:

  • table (BaseModel): The Peewee model class to register.

Usage:

>>> from automation import PyAutomation
>>> from peewee import CharField
>>> from automation.dbmodels.core import BaseModel
>>> class MyTable(BaseModel):
...     name = CharField()
>>> app = PyAutomation()
>>> app.add_db_table(MyTable)
>>> app.get_db_table("MyTable") is not None
True
add_opcua_client(*args, **kwargs)
change_password(*args, **kwargs)
connect_to_db(*args, **kwargs)
create_alarm(*args, **kwargs)
create_linear_referencing_geospatial(*args, **kwargs)
create_log(self, message, user, description=None, classification=None, alarm_summary_id=None, event_id=None, timestamp=None)

Creates a new log entry.

Parameters:

  • message (str): Log message.
  • user (User): User associated with the log.
  • description (str, optional): Detailed description.
  • classification (str, optional): Log category.

Returns:

  • tuple: Created log object and status message.

Usage:

>>> from automation import PyAutomation
>>> from automation.modules.users.users import User
>>> from automation.modules.users.roles import Role
>>> app = PyAutomation()
>>> # Mock user for logging
>>> role = Role(name="logger_role", level=1)
>>> user = User(username="logger_test", email="log@test.com", role=role, password="password")
>>> # Example of log creation (requires active DB connection)
>>> # if app.is_db_connected():
>>> #     log, msg = app.create_log("System started", user)
>>> #     print(log.message)
>>> # 'System started'
create_opcua_server_record(self, name, namespace, access_type='Read')

Creates a record for an OPC UA server node in the database.

Parameters:

  • name (str): Name of the node.
  • namespace (str): Namespace URI or index.
  • access_type (str): Access level (Read, Write, ReadWrite).

Usage:

>>> from automation import PyAutomation
>>> from unittest.mock import MagicMock
>>> app = PyAutomation()
>>> app.opcua_server_engine.create = MagicMock(return_value=True)
>>> app.create_opcua_server_record("Node1", "ns=1;i=1", "Read")
True
create_system_user(self)

Ensures a 'system' user exists with 'sudo' role. Used for automated internal actions.

create_tag(*args, **kwargs)
create_token(*args, **kwargs)
define_socketio(self, server, certfile=None, keyfile=None)

Initializes and configures the Socket.IO server integrated within PyAutomation.

Parameters:

  • certfile (str, optional): Path to the SSL certificate file.
  • keyfile (str, optional): Path to the SSL key file.

This method sets up the Socket.IO server, configures callbacks, and initializes the Socket.IO server for real-time communication.

delete_alarm(*args, **kwargs)
delete_linear_referencing_geospatial_point(*args, **kwargs)
delete_tag(*args, **kwargs)
delete_tag_by_name(*args, **kwargs)
disconnect_to_db(*args, **kwargs)
ensure_db_config_from_env(self)

Bootstrap database configuration from environment variables if no db/db_config.json is present yet.

This is intended for the very first startup of the application:

  • If db/db_config.json already exists, it is considered the single source of truth and environment variables are ignored.
  • If it does not exist and the appropriate AUTOMATION_DB_* env variables are defined, a new configuration file is written and will be used for all subsequent connections.

Supported environment variables:

  • AUTOMATION_DB_TYPE: sqlite (default), postgresql or mysql.
  • For SQLite:
  • AUTOMATION_DB_FILE: database filename (default: app.db).
  • For PostgreSQL/MySQL:
  • AUTOMATION_DB_HOST (default: 127.0.0.1)
  • AUTOMATION_DB_PORT (default: 5432 for PostgreSQL, 3306 for MySQL)
  • AUTOMATION_DB_USER (required)
  • AUTOMATION_DB_PASSWORD (required)
  • AUTOMATION_DB_NAME (required)
export_configuration(*args, **kwargs)
filter_alarms_by(self, **fields)

Filters alarm history based on provided fields.

Parameters:

  • fields: Keyword arguments for filtering (e.g., name, tag, state).

Returns:

  • list: Filtered list of alarms.

Usage:

>>> from automation import PyAutomation
>>> app = PyAutomation()
>>> # Assuming database is already connected
>>> # Filter by alarm names (returns dict with data and pagination)
>>> alarms = app.filter_alarms_by(names=["NonExistentAlarm"])
>>> isinstance(alarms, dict)
True
>>> isinstance(alarms['data'], list)
True
filter_events_by(self, usernames=None, priorities=None, criticities=None, message='', classification='', description='', greater_than_timestamp=None, less_than_timestamp=None, timezone='UTC', page=1, limit=20)

Filters system events based on multiple criteria.

Parameters:

  • usernames (list[str]): Filter by user.
  • priorities (list[int]): Filter by priority level.
  • criticities (list[int]): Filter by criticity level.
  • message (str): Text search in message.
  • classification (str): Filter by event classification.
  • timezone (str): Timezone for timestamp filtering.
  • page, limit: Pagination parameters.

Returns:

  • list: Filtered list of events.
filter_logs_by(self, usernames=None, alarm_names=None, event_ids=None, classification='', message='', description='', greater_than_timestamp=None, less_than_timestamp=None, timezone='UTC', page=1, limit=20)

Filters system logs based on criteria with pagination.

Parameters:

  • usernames (list[str]): Filter by user.
  • alarm_names (list[str]): Filter by linked alarm names.
  • event_ids (list[int]): Filter by linked event IDs.
  • classification (str): Filter by category.
  • message (str): Partial match message.
  • description (str): Partial match description.
  • greater_than_timestamp (datetime): Start time.
  • less_than_timestamp (datetime): End time.
  • timezone (str): Timezone.
  • page (int): Page number for pagination.
  • limit (int): Items per page.

Returns:

  • dict: {data: list, pagination: dict}
find_opcua_servers(*args, **kwargs)
get_alarm(*args, **kwargs)
get_alarm_by_name(*args, **kwargs)
get_alarm_manager(*args, **kwargs)
get_alarms(*args, **kwargs)
get_alarms_by_kp_range(*args, **kwargs)
get_alarms_by_tag(*args, **kwargs)
get_app_config(self)

Retrieves the application configuration from app_config.json.

If the file doesn't exist, it creates one with default values.

Returns:

  • dict: Application configuration dictionary.

Usage:

>>> from automation import PyAutomation
>>> app = PyAutomation()
>>> config = app.get_app_config()
>>> isinstance(config, dict)
True
get_db_config(*args, **kwargs)
get_db_table(self, tablename)

Retrieves a registered database table model by its name.

Parameters:

  • tablename (str): The name of the table.

Returns:

  • Model: The Peewee model class.

Usage:

>>> from automation import PyAutomation
>>> app = PyAutomation()
>>> # Assuming 'Users' table is registered by default
>>> table = app.get_db_table("Users")
>>> table is not None
True
get_geospatial_by_segment_and_kp(*args, **kwargs)
get_lasts_active_alarms(*args, **kwargs)
get_lasts_alarms(*args, **kwargs)
get_lasts_events(*args, **kwargs)
get_lasts_logs(*args, **kwargs)
get_linear_referencing_geospatial_point(*args, **kwargs)
get_linear_referencing_geospatial_points(*args, **kwargs)
get_linear_referencing_geospatial_points_by_segment(*args, **kwargs)
get_machine(*args, **kwargs)
get_machines(self)

Retrieves a list of all registered state machines along with their configuration.

Returns:

  • list[tuple[Machine, int, str]]: A list of tuples containing (Machine instance, interval, execution_mode).

Usage:

>>> from automation import PyAutomation
>>> app = PyAutomation()
>>> machines = app.get_machines()
>>> isinstance(machines, list)
True
get_node_attributes(*args, **kwargs)
get_node_values(*args, **kwargs)
get_opcua_client(*args, **kwargs)
get_opcua_client_by_address(*args, **kwargs)
get_opcua_clients(*args, **kwargs)
get_opcua_server_attrs(*args, **kwargs)
get_opcua_server_record_by_namespace(self, namespace)

Retrieves an OPC UA server node record by its namespace.

Parameters:

  • namespace (str): The namespace to search for.

Returns:

  • dict: Node record data.

Usage:

>>> from automation import PyAutomation
>>> from unittest.mock import MagicMock
>>> app = PyAutomation()
>>> app.opcua_server_engine.read_by_namespace = MagicMock(return_value={'name': 'Node1'})
>>> record = app.get_opcua_server_record_by_namespace("ns=1;i=1")
>>> record['name']
'Node1'
get_opcua_tree(self, client_name, *, mode='generic', max_depth=10, max_nodes=50000, include_properties=True, include_property_values=False)

Retrieves the hierarchical node tree structure from a connected OPC UA server.

Parameters:

  • client_name (str): The name of the client to use.

Returns:

  • dict: Nested dictionary representing the node tree.

Usage:

>>> from automation import PyAutomation
>>> from unittest.mock import MagicMock
>>> app = PyAutomation()
>>> app.opcua_client_manager.get_opcua_tree = MagicMock(return_value={'Root': {}})
>>> app.get_opcua_tree("PLC1")
{'Root': {}}
get_opcua_tree_children(self, client_name, node_id, *, mode='generic', max_nodes=5000, include_properties=True, include_property_values=False, fallback_to_legacy=True)

Devuelve los hijos directos de un NodeId (lazy-loading), para expandir carpetas profundas desde el HMI sin pedir todo el árbol.

get_opcua_variables(self, client_name, *, mode='generic', max_depth=20, max_nodes=50000, fallback_to_legacy=True)

Devuelve SOLO Variables del servidor OPC UA (para dropdowns de Tags).

get_segments(self)

Retrieves all unique segments (logical groupings of tags/machines) defined in the system.

Returns:

  • list: List of segment names.
get_tabular_data(self, start, stop, timezone, tags, sample_time, page=1, limit=20)

Retrieves historical data resampled to a specific time interval (tabular format).

Parameters:

  • start (str): Start datetime.
  • stop (str): Stop datetime.
  • timezone (str): Timezone.
  • tags (list): List of tag names.
  • sample_time (int): Resampling interval in seconds.
  • page (int): Page number.
  • limit (int): Records per page.

Returns:

  • dict: Resampled tabular data.
get_tag_by_name(*args, **kwargs)
get_tag_by_node_namespace(*args, **kwargs)
get_tags(*args, **kwargs)
get_tags_by_kp_range(*args, **kwargs)
get_tags_by_names(*args, **kwargs)
get_tags_list(*args, **kwargs)
get_tags_tables(self, start, stop, timezone, tags, page=1, limit=20)

Retrieves historical data in a paginated table format.

Parameters:

  • start (str): Start datetime.
  • stop (str): Stop datetime.
  • timezone (str): Timezone.
  • tags (list): List of tag names.
  • page (int): Page number for pagination.
  • limit (int): Number of records per page.

Returns:

  • dict: Paginated historical data.
get_trends(self, start, stop, timezone, *tags)

Retrieves historical trend data for specified tags within a time range.

Parameters:

  • start (str): Start datetime string.
  • stop (str): Stop datetime string.
  • timezone (str): Timezone for the query.
  • tags (tuple): One or more tag names to query.

Returns:

  • dict: Historical data for the requested tags.
import_configuration(*args, **kwargs)
import_linear_referencing_profile(*args, **kwargs)
is_db_connected(*args, **kwargs)
load_db_tags_to_machine(self)

Loads tag subscriptions for state machines from the database.

load_db_to_alarm_manager(*args, **kwargs)
load_db_to_cvt(*args, **kwargs)
load_db_to_roles(*args, **kwargs)
load_db_to_users(*args, **kwargs)
load_opcua_clients_from_db(*args, **kwargs)
login(*args, **kwargs)
reconnect_to_db(*args, **kwargs)
remove_opcua_client(*args, **kwargs)
reset_password(*args, **kwargs)
run(self, server, debug=False, test=False, create_tables=False, machines=None, certfile=None, keyfile=None)

Starts the PyAutomation application.

Initializes the logger, database connections, and worker threads.

Parameters:

  • debug (bool): Enable debug mode for Dash server.
  • test (bool): Run in test mode (using test DB).
  • create_tables (bool): Create database tables on startup.
  • machines (tuple, optional): Initial state machines to run.

Returns: None

Usage:

# Standard execution
# from automation import PyAutomation
# app = PyAutomation()
# if name == "main":
#     app.run(debug=True, create_tables=True)
safe_start(self, test=False, create_tables=True, machines=None)

Initializes app components without blocking the main thread.

Used by run() and for testing.

safe_stop(*args, **kwargs)
serialize_alarms(*args, **kwargs)
serialize_machines(*args, **kwargs)
set_app_config(self, **kwargs)

Updates application configuration in app_config.json.

Parameters:

  • kwargs: Configuration keys and values to update (e.g., logger_period).

Usage:

>>> from automation import PyAutomation
>>> app = PyAutomation()
>>> app.set_app_config(logger_period=5.0)
>>> config = app.get_app_config()
>>> config['logger_period']
5.0
set_db(*args, **kwargs)
set_db_config(*args, **kwargs)
set_log(*args, **kwargs)
set_role(*args, **kwargs)
signup(*args, **kwargs)
state_machine_diagrams(self, folder_path)

Generates and saves state diagrams for all active state machines.

Parameters:

  • folder_path (str): Destination directory for PNG images.
subscribe_opcua(*args, **kwargs)
subscribe_tag(*args, **kwargs)
subscribe_tag_into_automation_machine(*args, **kwargs)
unsubscribe_opcua(*args, **kwargs)
update_alarm(*args, **kwargs)
update_linear_referencing_geospatial_point(self, id, **kwargs)

Updates a linear-referencing geospatial point by ID.

update_log_config(*args, **kwargs)
update_log_level(*args, **kwargs)
update_logger_period(*args, **kwargs)
update_opcua_client(*args, **kwargs)
update_opcua_server_access_type(self, namespace, access_type)

Updates the access type for a specific OPC UA server node.

Parameters:

  • namespace (str): The namespace of the node.
  • access_type (str): New access type (Read, Write, ReadWrite).

Usage:

>>> from automation import PyAutomation
>>> from unittest.mock import MagicMock
>>> app = PyAutomation()
>>> app.opcua_server_engine.put = MagicMock(return_value=True)
>>> app.update_opcua_server_access_type("ns=1;i=1", "Write")
True
update_opcua_server_node_access_type(*args, **kwargs)
update_tag(self, id, user=None, **kwargs)

Updates the configuration of an existing tag.

Parameters:

  • id (str): Tag ID.
  • user (User, optional): User performing the update.
  • kwargs: Tag attributes to update (e.g., name, unit, scan_time, alarm limits).

Returns:

  • tuple[Tag|None, str]: A tuple containing the updated Tag object (or None on failure) and a status message.

Usage:

>>> from automation import PyAutomation
>>> app = PyAutomation()
>>> tag, _ = app.create_tag("Tag_Update", "C", "Temperature")
>>> # Update scan time
>>> updated_tag, msg = app.update_tag(tag.id, scan_time=1000)
>>> updated_tag.scan_time
1000
update_user_role(*args, **kwargs)
write_opcua_value(*args, **kwargs)