PyHades

class pyhades.PyHades(*args, **kwargs)

PyHades is a singleton class to develop multi threads web application for general purposes .

Usage:

>>> from pyhades import PyHades
>>> app = PyHades()
run(self)

Runs main app thread and all defined threads by decorators and State Machines besides this method starts app logger

Returns: None

Usage

>>> app.run()
safe_start(self)

Run the app without a main thread, only run the app with the threads and state machines define

safe_stop(self)

Stops the app in safe way with the threads

info(self)

Shows a summary threads on execution

get_start_up_datetime(self)

Gets the start up datetime of the application

Returns

  • date_time (datetime): Start up datetime of the app

Usage

>>> start_up = app.get_start_up_datetime()
set_log(self, level=20, file='app.log')

Sets the log file and level.

Parameters:

  • level (str): logging.LEVEL (default: logging.INFO).
  • file (str): log filename (default: 'app.log').

Returns: None

Usage:

>>> app.set_log(file="app.log")
thread(self, function=None, **kwargs)

Decorator method to register functions plugins.

This method will register into the PyHades application a new function to be executed by the Thread Pool Executor

Parameters:

  • period (float): Value of the default loop execution time.

Returns: None

Usage:

@app.thread(period=0.5)
def hello():
    print("Hello!!!")

@app.thread
def hello_world():
    print("Hello World!!!")
set_mode(self, mode)

Allows to you define "Development" or "Production" mode.

For Development mode you use SQLite database by default when you define it.

For Production mode you can use Sqlite - Postgres - MySQL

Parameters

  • mode (str): App mode ('Development' or 'Production')
>>> app = PyHades()
>>> app.set_mode('Development')
get_mode(self)

Gets app mode

Returns

  • mode (str) Application mode.
>>> app.get_mode()
'Development'
set_socketio(self, sio)

Sets a SocketIO Client Object defined from any web server

get_socketio(self)

Gets SocketIO Client

set_max_threads(self, max_threads)

Sets maximum numbers of threads

Parameters

  • max_threads (int): Max numbers of threads

Returns None

Usage

>>> app.set_max_threads(20)
threads_running(self)

Gets thread numbers defined in the app

Returns: (int)

Usage

>>> threads_running = app.threads_running()
threads_info(self)

Gets information of all defined threads

Returns: (str)

Usage

>>> app.threads_info()
get_manager(self, name='state')

Returns a specified application manager.

Available Managers

  • Alarm Manager: name='alarm'
  • State Machine Manager: name='state'
  • Database Manager: name='db'

Parameters:

  • name (str): a manager name.
get_state_machine_manager(self)

Gets state machine Manager

Returns: StateMachineManager instance

>>> state_manager = app.get_state_machine_manager()
define_machine(self, name='', interval=1, mode='sync', **kwargs)

Append an state machine to the state machine manager by decoration.

Parameters:

  • name (str): State machine name
  • interval (int): Interval execution time in seconds.
  • mode (str): Syncronic or Asyncronic thread mode - allowed values ['sync', 'async']

Returns Class (cls)

Usage

>>> from pyhades import PyHades, PyHadesStateMachine
>>> app = PyHades()
>>> @app.define_machine(name='state_machine_name', interval=1, mode='async')
    class StateMachine(PyHadesStateMachine):
        ...
get_machine(self, name)

Returns a PyHades State Machine defined by its name.

Parameters:

  • name (str): a pyhades state machine name.

Usage

>>> state_machine = app.get_machine('state_machine_name')
get_machines(self)

Returns all defined PyHades state machines.

Returns (list)

Usage

>>> state_machines = app.get_machines()
state_machines_running(self)

Gets state machines numbers defined in the app

Returns: (int)

Usage

>>> state_machines_running = app.state_machines_running()
state_machines_info(self)

Gets information of all defined state machines

Returns: (str)

Usage

>>> app.state_machines_info()
get_db_manager(self)

Gets Database Manager

Returns: DBManager instance

>>> db_manager = app.db_manager()
set_db_from_config_file(self, config_file)

Defines the database configuration from a .yml configuration file

Parameters

  • config_file (str) Url where the .yml configuration file is

Configuration File Structure For Database Definition

version: '3'

db:

    dev_mode:
        db_name: \({DB_NAME}</span><span class="w"></span>

<span class="w">    </span><span class="nt">prod_mode</span><span class="p">:</span><span class="w"></span>
<span class="w">        </span><span class="nt">db_type</span><span class="p">:</span><span class="w"> </span><span class="s">"POSTGRESQL"</span><span class="w"></span>
<span class="w">        </span><span class="nt">db_name</span><span class="p">:</span><span class="w"> </span><span class="l l-Scalar l-Scalar-Plain">\){DB_NAME}
        db_user: \({DB_USER}</span><span class="w"></span>
<span class="w">        </span><span class="nt">db_password</span><span class="p">:</span><span class="w"> </span><span class="l l-Scalar l-Scalar-Plain">\){DB_PASSWORD}
        db_host: \({DB_HOST}</span><span class="w"></span>
<span class="w">        </span><span class="nt">db_port</span><span class="p">:</span><span class="w"> </span><span class="l l-Scalar l-Scalar-Plain">\){DB_PORT}

    sample_time: ${DB_SAMPLE_TIME}

you can define your file based on environment variables or you can complete the file directly.

drop_db(dbfile)

Class method to remove SQLite database files

Parameters

  • dbfile (str): DB file name, default for hades app.db
append_table(self, table)

Append a database model class definition.

Parameters:

  • table (BaseModel): A Base Model Inheritance.
define_table(self, cls)

Appends a database model class definition using a class decoration.

Example code

from peewee import CharField
from pyhades.dbmodels import BaseModel
from pyhades import PyHades

app = PyHades()

@app.define_table
class NewTable(BaseModel):
    item1 = CharField()
    item2 = CharField()
set_db(self, dbtype='sqlite', drop_table=False, clear_default_tables=False, **kwargs)

Sets the database, it supports SQLite and Postgres, in case of SQLite, the filename must be provided.

if app mode is "Development" you must use SQLite Databse

Parameters:

  • dbfile (str): a path to database file.
  • drop_table* (bool): If you want to drop table.
  • cascade (bool): if there are some table dependency, drop it as well
  • kwargs: Same attributes to a postgres connection.

Returns: None

Usage:

>>> app.set_db(dbfile="app.db")
init_db(self)

Initialize Logger Worker

Returns

  • db_worker: (LoggerWorker Object)
stop_db(self, db_worker)

Stops Database Worker

set_dbtags(self, tags, period=0.5, delay=1.0)

Sets the database tags for logging.

If you want to log any tag defined in CVTengine, you must define it here

Parameters:

  • tags (list): A list of the tags.

Returns: None

Usage:

>>> tags = ["P1", "P2", "T1"]
>>> app.set_dbtags(tags, period=1.0)
get_dbtags(self)

Returns the database tags for logging.

get_alarm_manager(self)

Gets Alarm Manager

Returns: AlarmManager instance

>>> alarm_manager = app.alarm_manager()
define_alarm_from_config_file(self, config_file)

Defines alarms from a .yml configuration file

Parameters

  • config_file (str) Url where the .yml configuration file is

Configuration File Structure For Alarms Definition

modules:

    alarms:

        alarm1:
            name: "alarm1_name"
            tag: "tag1_name"
            description: "alarm description"
            type: "bool"
            trigger: True

        alarm2:
            name: "alarm2_name"
            tag: "tag2_name"
            description: "alarm description"
            type: "high"
            trigger: 100.0

You must have previously defined the tag to which you are going to bind the alarm, see (CVT set_config)[api_cvt.md] section or (CVT User Guide)[user_guide_cvt.md]

append_alarm(self, alarm)

Appends an alarm to the alarm manager.

Parameters:

  • alarm (Alarm Object)
get_alarm(self, name)

Returns an Alarm defined by its name.

Parameters:

  • name (str): an alarm name.

Returns

  • alarm (Alarm Object)
get_alarm_by_tag(self, tag)

Returns an Alarm binded to any tag name.

Parameters:

  • tag (str): Tag name binded to alarm.

Returns

  • alarm (Alarm Object)