init added min tags functionality

Author: rlabbeptc

kepconfig/adv_tags/__init__.py

@@ -8,7 +8,7 @@
 specific objects within the Kepware Configuration API
 """
 
-from . import adv_tag_group, average_tags, derived_tags, complex_tags, cumulative_tags
+from . import adv_tag_group, average_tags, derived_tags, complex_tags, cumulative_tags, min_tags
 ADV_TAGS_ROOT = '/project/_advancedtags'
 
 def adv_tag_path_split(path: str, *, isItem=False) -> dict:

kepconfig/adv_tags/min_tags.py

@@ -0,0 +1,138 @@
+# -------------------------------------------------------------------------
+# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
+# See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+r"""`minimum_tags` exposes an API to allow modifications (add, delete, modify) to 
+minimum tag objects within the Kepware Configuration API
+"""
+
+from ..connection import server
+from ..error import KepError, KepHTTPError
+from ..utils import _url_parse_object
+from typing import Union
+from .. import adv_tags
+
+MINIMUM_TAGS_ROOT = '/minimum_tags'
+
+def _get_minimum_tags_url(tag: str = None) -> str:
+    '''Creates url object for the "minimum_tags" branch of Kepware's project tree.
+    
+    Returns the minimum tag specific url when a value is passed as the tag name.
+    '''
+    if tag is None:
+        return MINIMUM_TAGS_ROOT
+    else:
+        return f'{MINIMUM_TAGS_ROOT}/{_url_parse_object(tag)}'
+
+def add_minimum_tag(server: server, adv_tag_group_path: str, DATA: Union[dict, list]) -> Union[bool, list]:
+    '''Add `"minimum_tag"` or multiple `"minimum_tag"` objects to a specific path in Kepware.
+    Can be used to pass a list of minimum tags to be added at one path location.
+
+    :param server: instance of the `server` class
+    :param adv_tag_group_path: path identifying where to add minimum tag(s). Standard Kepware address decimal 
+    notation string such as "_advancedtags.AdvTagGroup1" or "_advancedtags.AdvTagGroup1.AdvTagGroupChild"
+    :param DATA: Dict or List of Dicts of the minimum tag(s) to add
+
+    :return: True - If a "HTTP 201 - Created" is received from Kepware server
+    :return: If a "HTTP 207 - Multi-Status" is received from Kepware with a list of dict error responses for all 
+    minimum tags added that failed.
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    path_obj = adv_tags.adv_tag_path_split(adv_tag_group_path, isItem=False)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_minimum_tags_url()
+
+    r = server._config_add(url, DATA)
+    if r.code == 201:
+        return True
+    elif r.code == 207:
+        errors = [item for item in r.payload if item['code'] != 201]
+        return errors
+    else:
+        raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+def modify_minimum_tag(server: server, min_tag_path: str, DATA: dict, force: bool = False) -> bool:
+    '''Modify a `"minimum_tag"` object and its properties in Kepware.
+
+    :param server: instance of the `server` class
+    :param min_tag_path: path identifying location and minimum tag to modify. Standard Kepware address decimal 
+    notation string including the minimum tag such as "_advancedtags.AdvTagGroup1.MinTag1"
+    :param DATA: Dict of the `minimum_tag` properties to be modified
+    :param force: *(optional)* if True, will force the configuration update to the Kepware server
+
+    :return: True - If a "HTTP 200 - OK" is received from Kepware server
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    min_tag_data = server._force_update_check(force, DATA)
+    path_obj = adv_tags.adv_tag_path_split(min_tag_path, isItem=True)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_minimum_tags_url(path_obj['item'])
+
+    r = server._config_update(url, min_tag_data)
+    if r.code == 200:
+        return True
+    else:
+        raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+def del_minimum_tag(server: server, min_tag_path: str) -> bool:
+    '''Delete `"minimum_tag"` object at a specific path in Kepware.
+
+    :param server: instance of the `server` class
+    :param min_tag_path: path identifying location and minimum tag to delete. Standard Kepware address decimal 
+    notation string including the minimum tag such as "_advancedtags.AdvTagGroup1.MinTag1"
+
+    :return: True - If a "HTTP 200 - OK" is received from Kepware server
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    path_obj = adv_tags.adv_tag_path_split(min_tag_path, isItem=True)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_minimum_tags_url(path_obj['item'])
+
+    r = server._config_del(url)
+    if r.code == 200:
+        return True
+    else:
+        raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+def get_minimum_tag(server: server, min_tag_path: str) -> dict:
+    '''Returns the properties of the `"minimum_tag"` object at a specific path in Kepware.
+
+    :param server: instance of the `server` class
+    :param min_tag_path: path identifying location and minimum tag to retrieve. Standard Kepware address decimal 
+    notation string including the minimum tag such as "_advancedtags.AdvTagGroup1.MinTag1"
+
+    :return: Dict of data for the minimum tag requested
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    path_obj = adv_tags.adv_tag_path_split(min_tag_path, isItem=True)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_minimum_tags_url(path_obj['item'])
+
+    r = server._config_get(url)
+    return r.payload
+
+def get_all_minimum_tags(server: server, adv_tag_group_path: str, *, options: dict = None) -> list:
+    '''Returns the properties of all `"minimum_tag"` objects at a specific path in Kepware.
+
+    :param server: instance of the `server` class
+    :param adv_tag_group_path: path identifying location to retrieve minimum tag list. Standard Kepware address decimal
+    notation string such as "_advancedtags.AdvTagGroup1" or "_advancedtags.AdvTagGroup1.AdvTagGroupChild"
+    :param options: *(optional)* Dict of parameters to filter, sort or paginate the list of minimum tags. Options are `filter`,
+        `sortOrder`, `sortProperty`, `pageNumber`, and `pageSize`
+
+    :return: List of data for all minimum tags
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    path_obj = adv_tags.adv_tag_path_split(adv_tag_group_path, isItem=False)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_minimum_tags_url()
+
+    r = server._config_get(url, params=options)
+    return r.payload

tests/adv_tags_test.py

@@ -61,6 +61,17 @@
     }
 ]
 
+minimum_tag_name = 'MinimumTag1'
+minimum_tag_data = [
+    {
+        "common.ALLTYPES_NAME": minimum_tag_name,
+        "common.ALLTYPES_DESCRIPTION": "",
+        "advanced_tags.ENABLED": True,
+        "advanced_tags.MINIMUM_TAG": "_System._Time_Hour",
+        "advanced_tags.RUN_TAG": "_System._Time_Second"
+    }
+]
+
 def HTTPErrorHandler(err):
     if err.__class__ is error.KepHTTPError:
         print(err.code)
@@ -96,6 +107,15 @@ def complete(server: connection.server):
             for obj in obj_list:
                 complex_tag_path = f'_advancedtags.{obj["name"]}'
                 adv_tags.complex_tags.del_complex_tag(server, complex_tag_path)
+        elif key == 'cumulative_tags':
+            for obj in obj_list:
+                cumulative_tag_path = f'_advancedtags.{obj["name"]}'
+                adv_tags.cumulative_tags.del_cumulative_tag(server, cumulative_tag_path)
+        elif key == 'minimum_tags':
+            for obj in obj_list:
+                minimum_tag_path = f'_advancedtags.{obj["name"]}'
+                adv_tags.min_tags.del_minimum_tag(server, minimum_tag_path)
+        
     pass
 
 @pytest.fixture(scope="module")
@@ -307,6 +327,47 @@ def test_cumulative_tag_del(server):
     cumulative_tag_path = f'_advancedtags.{adv_tag_group_name}.{cumulative_tag_name}'
     assert adv_tags.cumulative_tags.del_cumulative_tag(server, cumulative_tag_path)
 
+def test_minimum_tag_add(server):
+    # Add a minimum tag to the root advanced tag plug-in
+    assert adv_tags.min_tags.add_minimum_tag(server, f'_advancedtags', minimum_tag_data)
+
+    testTag = {
+        "common.ALLTYPES_NAME": "newMinimumTag",
+        "common.ALLTYPES_DESCRIPTION": "",
+        "advanced_tags.ENABLED": True,
+        "advanced_tags.MINIMUM_TAG": "_System._Time_Hour",
+        "advanced_tags.RUN_TAG": "_System._Time_Second"
+    }
+    minimum_tag_data.append(testTag)
+    # Add a minimum tag to the advanced tag group
+    assert adv_tags.min_tags.add_minimum_tag(server, f'_advancedtags.{adv_tag_group_name}', minimum_tag_data)
+
+def test_minimum_tag_get(server):
+    # Get the minimum tag
+    minimum_tag_path = f'_advancedtags.{adv_tag_group_name}.{minimum_tag_name}'
+    result = adv_tags.min_tags.get_minimum_tag(server, minimum_tag_path)
+    assert type(result) == dict
+    assert result.get("common.ALLTYPES_NAME") == minimum_tag_name
+
+def test_minimum_tag_modify(server):
+    # Modify the minimum tag
+    minimum_tag_path = f'_advancedtags.{adv_tag_group_name}.{minimum_tag_name}'
+    tag_data = {
+        "common.ALLTYPES_DESCRIPTION": "Modified minimum tag"
+    }
+    assert adv_tags.min_tags.modify_minimum_tag(server, minimum_tag_path, tag_data, force=True)
+
+def test_minimum_tag_get_all(server):
+    # Get all minimum tags under the group
+    result = adv_tags.min_tags.get_all_minimum_tags(server, f'_advancedtags.{adv_tag_group_name}')
+    assert type(result) == list
+    assert any(tag.get("common.ALLTYPES_NAME") == minimum_tag_name for tag in result)
+
+def test_minimum_tag_del(server):
+    # Delete the minimum tag
+    minimum_tag_path = f'_advancedtags.{adv_tag_group_name}.{minimum_tag_name}'
+    assert adv_tags.min_tags.del_minimum_tag(server, minimum_tag_path)
+
 def test_adv_tag_group_del(server):
     # Delete parent advanced tag group
     assert adv_tags.adv_tag_group.del_tag_group(server, f'_advancedtags.{adv_tag_group_name}')