CVT

class pyhades.tags.CVTEngine(*args, **kwargs)

Current Value Table Engine class for Tag thread-safe based repository.

This class is intended hold in memory tag based values and observers for those required tags, it is implemented as a singleton so each sub-thread within the PyHades application can access tags in a thread-safe mechanism.

Usage:

>>> from pyhades.tags import CVTEngine
>>> tag_egine = CVTEngine()
set_tag(self, name, unit, data_type, description, min_value=None, max_value=None, tcp_source_address='', node_namespace='')

Defines a new tag.

Parameters:

  • name (str): Tag name.
  • unit (str): Engineering units.
  • data_type (float, int, bool): Tag value ("int", "float", "bool")
  • description (str): Tag description
  • min_value (int - float)[Optional]: Field instrument lower value
  • max_value (int - float)[Optional]: Field instrument higher value
  • tcp_source_address (str)[Optional]: Url for tcp communication with a server.
  • node_namespace (str)[Optional]: Node ID or Namespace (OPC UA) to get element value from server.

Usage:

>>> tag_engine.set_tag("speed", "float", "km/h", "Speed of car", 0.0, 240.0)
set_tags(self, tags)

Sets new values for a defined list of tags, in thread-safe mechanism.

Parameters:

  • tags (list): List of tag name, unit, data type, description, min and max value
>>> tags = [
        ("TAG1", 'ºC', 'float', 'Inlet temperature', 0.0, 100.0),
        ("TAG2", 'kPa', 'float', 'Inlet pressure', 100.0),
        ("TAG3", 'm3/s', 'float', 'Inlet flow')
    ]
>>> tag_engine.set_tags(tags)
set_config(self, config_file)

Allows to define tags using a YaML file.

Parameters:

  • config_file (str): Url where the configuration file is, the configuratio file must have .yml extension.

Returns

  • None
>>> tag_egine.set_config('base_url/config.yml')

Configuration File Structure

version: '3'

modules:
    tags:

        groups:

            cvt:

                PT-01:
                    name: 'PT-01'
                    unit: 'Pa'
                    data_type: 'float'
                    description: 'Inlet Pressure'
                    min_value: 0.00
                    max_value: 100.00
                    tcp_source_address: ''
                    node_namespace: ''
set_data_type(self, data_type)

Sets a new data_type as string format.

Parameters: * data_type (str): Data type.

Returns

  • None
set_group(self, group, *tags)

Sets new tags group, which can be retrieved by group name.

Parameters:

  • group (str): Group name.
  • tags (list): List of defined tag names.
>>> temp_tags = [
        ("TAG1", 'ºC', 'float', 'Inlet temperature', 0.0, 100.0),
        ("TAG2", 'ºC', 'float', 'Outlet temperature', 0.0, 100.0)
    ]
>>> pressure_tags = [
        ("TAG3", 'kPa', 'float', 'Inlet pressure', 100.0, 200.0),
        ("TAG4", 'kPa', 'float', 'Outlet pressure', 0.0, 100.0)
    ]
>>> tag_engine.set_group('Temperatures', temp_tags)
>>> tag_engine.set_group('Pressures', pressure_tags)
get_group(self, group)

Returns the tag list of the a defined group.

Parameters:

  • group (str): Group name.

Returns

  • tag_names (list) Tag names binded to a group
>>> tag_engine.get_group('Temperatures')
['TAG1', 'TAG2']
>>> tag_engine.get_group('Pressures')
['TAG3', 'TAG4']
get_tags(self)

Returns a list of the tag names defined.

get_groups(self)

Returns a list of the group names defined.

get_data_type(self, name)

Gets a tag data type as string format.

Parameters:

  • name (str): Tag name.

Returns

  • data_type (str): Tag's data type
>>> tag_egine.get_data_type('TAG1')
get_unit(self, name)

Gets tag's unit.

Parameters:

  • name (str): Tag name.

Returns

  • unit (str): Tag's unit
get_description(self, name)

Gets tag's description.

Parameters:

  • name (str): Tag's name.

Returns

  • description (str): Tag's description
get_min_value(self, name)

Gets tag's min value defined.

Parameters:

  • name (str): Tag name.

Returns

  • min_value (float) Tag's min value
get_max_value(self, name)

Gets tag's max value defined.

Parameters:

  • name (str): Tag's name.

Returns

  • max_value (float): Tag's max value
get_tagname_by_node_namespace(self, node_namespace)

Gets tag's name binded to a tcp node namespace

Parameters

  • node_namespace (str): TCP node namespace

Returns

  • tag_name (str): Tag's name binded to a tcp node namespace
get_node_namespace_by_tag_name(self, name)

Gets tcp node namespace binded to a tag name

Parameters

  • name (str): tag name binded to a tcp node namespace

Returns

  • node_namespace (str): TCP node namespace
tag_defined(self, name)

Checks if a tag name is already defined into database and tags repository.

Parameters:

  • name (str): Tag name.

Returns

  • flag (bool): True if tag is already defined
update_tag(self, id, **kwargs)

Updates tag's definition attributes

Parameters

  • id (int): Tag ID into database
  • name (str)[Optional]: New tag name
  • unit (str)[Optional]: New tag unit
  • data_type (str)[Optional]: New tag data type
  • description (str)[Optional]: New tag description
  • min_value (float)[Optional]: New tag min value
  • max_value (float)[Optional]: New tag max value
  • tcp_source_address (str)[Optional]: New tcp source address to tag binding
  • node_namespace (str)[Optional]: New tcp node namespace to tag binding

Returns

  • tag (dict): Tag definition updated
write_tag(self, name, value)

Writes a new value for a defined tag, in thread-safe mechanism.

Parameters:

  • name (str): Tag's name.
  • value (float, int, bool, str): Tag's value ("int", "float", "bool", "str")

Returns

  • msg (dict): Message for write request
>>> tag_engine.write_tag('TAG1', 50.53)
read_tag(self, name, unit=None)

Returns a tag value defined by name, in thread-safe mechanism.

Parameters:

  • name (str): Tag name.

Returns

  • value (float) Tag's value
>>> tag_engine.read_tag('TAG1')
50.53
read_unit(self, name)

Returns the tag's unit, in thread-safe mechanism.

Parameters:

  • name (str): Tag name.

Returns

  • unit (str): Tag's unit.
read_data_type(self, name)

Returns the tag's data type, in thread-safe mechanism.

Parameters:

  • name (str): Tag's name.

Returns

  • data_type (str) Tag's data type
read_description(self, name)

Returns the tag's description, in thread-safe mechanism.

Parameters:

  • name (str): Tag name.

Returns

  • description (str): Tag's description.
read_min_value(self, name)

Returns the tag's min value, in thread-safe mechanism.

Parameters:

  • name (str): Tag name.

Returns

  • min_value (float): Tag's min value.
read_max_value(self, name)

Returns the tag's max value, in thread-safe mechanism.

Parameters:

  • name (str): Tag name.

Returns

  • max_value (float): Tag's max value.
read_attributes(self, name)

Returns all tag's attributes, in thread-safe mechanism.

Parameters:

  • name (str): Tag name.

Returns

  • attrs (dict) Tag's attributes
>>> tag_engine.read_attributes('TAG1')
{
    "value":{
            "status_code":{
            "name": 'GOOD',
            "value": '0x000000000',
            "description":  'Operation succeeded'
        },
        "source_timestamp": '03/25/2022, 14:39:29.189422',
        "value": 50.53,
    },
    'name': 'TAG1',
    'unit': 'ºC',
    'data_type': 'float',
    'description': 'Inlet temperature',
    'min_value': 0.0,
    'max_value': 100.0,
    'tcp_source_address': '',
    'node_namespace': ''
}
load_tag_from_db_to_cvt(self)

It's necessary when initialize your app and already exist a database defined with this app, you must load all tag's definition in your database to your Current Value Table (CVT) or tags repository

delete_tag(self, name)

Deletes tag from database and tags repository by tag name

Parameters

  • name (str): Tag name to delete

Returns

  • msg (dict) Request message

Message structure:

msg = {
    'message': 'Request message'
}
attach(self, name, observer)

Attaches an observer object to a Tag, observer gets notified when the Tag value changes.

Parameters:

  • name (str): Tag name.
  • observer (str): TagObserver instance.
detach(self, name, observer)

Detaches an observer object from a Tag, observer no longer gets notified when the Tag value changes.

Parameters:

  • name (str): Tag name.
  • observer (str): TagObserver instance.
request(self, query)

It does the request to the tags repository according query's structure, in a thread-safe mechanism

Parameters

  • query (dict): Query to tags repository

Query Structure

query = {
    "action": (str)
    "parameters": (dict)
}

Valid actions in query

  • set_tag
  • get_tags
  • get_value
  • get_data_type
  • get_unit
  • get_description
  • get_min_value
  • get_max_value
  • get_attributes
  • set_value
  • attach
  • detach

Parameters strcuture in query

parameters = {
    "name": (str) tag name to do request
    "unit": (str)[Optional] Unit to get value
    "value": (float)[Optional] If you use set_value function, you must pass this parameter
    "observer": (TagObserver)[Optional] If you use attach and detach function, you must pass this parameter
}
response(self)

Handles the python GIL to emit the request's response in a thread-safe mechanism.

serialize_tag(self, id)

Serialize a Tag Object in a jsonable object.

Parameters

  • id (id) Tag name to serialize Tag Object

Returns

  • tag (dict): Tag attributes in a jsonable object
serialize(self)

Serializes all tag's repository in a jsonable object.

Returns

  • cvt (list): Tag's repository in a jsonable object.
serialize_group(self, name)

Serializes all groups of the tag's repository in a jsonable object.

Returns

  • groups (list): Groups of the tag's repository in a jsonable object.
add_conversions(self, conversions_path)

Add new custom conversion factores between tag's units.

Parameters

  • conversions_path (str) Url where the json file configuration for conversion factors is.
add_variables(self, variables_path)

Add new variables and units.

Parameters

  • variables_path (str) Url where the json file configuration for variables and units is.
class pyhades.tags.TagBinding(tag, direction='read')

Class used within PyHades State Machine.

This class is used to bind tag values with an instance of a PyHades State Machine object, in the machine loop, before executing current state, tag bindings of an object are updated with last values from the CVT Engine, after execution, the CVT Engine is updated, the direction of the binding must be provided, otherwise read direction is used.

Usage:

>>> time_left = TagBinding('time_left')
>>> time_left_write = TagBinding('time_left', direction='write')
class pyhades.tags.GroupBinding(group, direction='read')

Class used within PyHades State Machine.

This class is used to bind a tag group values with an instance of a PyHades State Machine object, in the machine loop, before executing current state, group bindings of an object are updated with last values of all tags in that group from the Tag Engine, after execution, the Tag Engine is updated, the direction of the binding must be provided, otherwise read direction is used.

Usage:

>>> g1 = GroupBinding("G1")
>>> g2 = GroupBinding("G2", direction="write")