PyAutomation Modules¶
class
automation.core.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
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:
serialize_machines(*args, **kwargs)create_tag(*args, **kwargs)get_tags(*args, **kwargs)get_tag_by_name(*args, **kwargs)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.
delete_tag(*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:
delete_tag_by_name(*args, **kwargs)signup(*args, **kwargs)login(*args, **kwargs)create_token(*args, **kwargs)set_role(*args, **kwargs)change_password(*args, **kwargs)reset_password(*args, **kwargs)find_opcua_servers(*args, **kwargs)get_opcua_clients(*args, **kwargs)get_opcua_client(*args, **kwargs)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:
add_opcua_client(*args, **kwargs)subscribe_opcua(*args, **kwargs)subscribe_tag(*args, **kwargs)unsubscribe_opcua(*args, **kwargs)set_log(*args, **kwargs)set_db(*args, **kwargs)set_db_config(*args, **kwargs)get_db_config(*args, **kwargs)is_db_connected(*args, **kwargs)connect_to_db(*args, **kwargs)disconnect_to_db(*args, **kwargs)load_db_to_cvt(*args, **kwargs)load_db_to_alarm_manager(*args, **kwargs)load_db_to_roles(*args, **kwargs)load_db_to_users(*args, **kwargs)load_opcua_clients_from_db(*args, **kwargs)get_alarm_manager(*args, **kwargs)create_alarm(*args, **kwargs)get_lasts_alarms(*args, **kwargs)update_alarm(*args, **kwargs)get_alarm(*args, **kwargs)get_alarm_by_name(*args, **kwargs)get_alarms_by_tag(*args, **kwargs)delete_alarm(*args, **kwargs)get_lasts_events(*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'
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}
get_lasts_logs(*args, **kwargs)export_configuration(*args, **kwargs)import_configuration(*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:
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)