init added link tag functionality

rlabbeptc · rlabbeptc · commit 0ff30ff107fe · 2025-06-23T18:08:06.000-04:00
diff --git a/kepconfig/adv_tags/__init__.py b/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, min_tags, max_tags
+from . import adv_tag_group, average_tags, derived_tags, complex_tags, cumulative_tags, min_tags, max_tags, link_tags
 ADV_TAGS_ROOT = '/project/_advancedtags'
 
 def adv_tag_path_split(path: str, *, isItem=False) -> dict:
diff --git a/kepconfig/adv_tags/link_tags.py b/kepconfig/adv_tags/link_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"""`link_tags` exposes an API to allow modifications (add, delete, modify) to 
+link 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
+
+LINK_TAGS_ROOT = '/link_tags'
+
+def _get_link_tags_url(tag: str = None) -> str:
+    '''Creates url object for the "link_tags" branch of Kepware's project tree.
+    
+    Returns the link tag specific url when a value is passed as the tag name.
+    '''
+    if tag is None:
+        return LINK_TAGS_ROOT
+    else:
+        return f'{LINK_TAGS_ROOT}/{_url_parse_object(tag)}'
+
+def add_link_tag(server: server, adv_tag_group_path: str, DATA: Union[dict, list]) -> Union[bool, list]:
+    '''Add `"link_tag"` or multiple `"link_tag"` objects to a specific path in Kepware.
+    Can be used to pass a list of link 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 link 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 link 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 
+    link 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_link_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_link_tag(server: server, link_tag_path: str, DATA: dict, force: bool = False) -> bool:
+    '''Modify a `"link_tag"` object and its properties in Kepware.
+
+    :param server: instance of the `server` class
+    :param link_tag_path: path identifying location and link tag to modify. Standard Kepware address decimal 
+    notation string including the link tag such as "_advancedtags.AdvTagGroup1.LinkTag1"
+    :param DATA: Dict of the `link_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
+    '''
+    link_tag_data = server._force_update_check(force, DATA)
+    path_obj = adv_tags.adv_tag_path_split(link_tag_path, isItem=True)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_link_tags_url(path_obj['item'])
+
+    r = server._config_update(url, link_tag_data)
+    if r.code == 200:
+        return True
+    else:
+        raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+def del_link_tag(server: server, link_tag_path: str) -> bool:
+    '''Delete `"link_tag"` object at a specific path in Kepware.
+
+    :param server: instance of the `server` class
+    :param link_tag_path: path identifying location and link tag to delete. Standard Kepware address decimal 
+    notation string including the link tag such as "_advancedtags.AdvTagGroup1.LinkTag1"
+
+    :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(link_tag_path, isItem=True)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_link_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_link_tag(server: server, link_tag_path: str) -> dict:
+    '''Returns the properties of the `"link_tag"` object at a specific path in Kepware.
+
+    :param server: instance of the `server` class
+    :param link_tag_path: path identifying location and link tag to retrieve. Standard Kepware address decimal 
+    notation string including the link tag such as "_advancedtags.AdvTagGroup1.LinkTag1"
+
+    :return: Dict of data for the link tag requested
+
+    :raises KepHTTPError: If urllib provides an HTTPError
+    :raises KepURLError: If urllib provides an URLError
+    '''
+    path_obj = adv_tags.adv_tag_path_split(link_tag_path, isItem=True)
+    url = adv_tags._create_adv_tags_base_url(server.url, path_obj) + _get_link_tags_url(path_obj['item'])
+
+    r = server._config_get(url)
+    return r.payload
+
+def get_all_link_tags(server: server, adv_tag_group_path: str, *, options: dict = None) -> list:
+    '''Returns the properties of all `"link_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 link 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 link tags. Options are `filter`,
+        `sortOrder`, `sortProperty`, `pageNumber`, and `pageSize`
+
+    :return: List of data for all link 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_link_tags_url()
+
+    r = server._config_get(url, params=options)
+    return r.payload
diff --git a/tests/adv_tags_test.py b/tests/adv_tags_test.py
@@ -9,7 +9,7 @@
 
 import os, sys
 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-from kepconfig import adv_tags, error, connection
+from kepconfig import adv_tags, error, connection, connectivity
 import pytest
 
 adv_tag_list_avail  = []
@@ -19,6 +19,10 @@
 adv_tag_group_child = 'AdvTagGroupChild'
 adv_tag_child_group_path = f'_advancedtags.{adv_tag_group_name}.{adv_tag_group_child}'
 
+sim_ch_name = 'SimulatedDevice'
+sim_dev_name = 'SimulatedDevice'
+sim_tag_name = 'tag1'
+
 avg_tag_name = 'AvgTag1'
 avg_tag_data = [
         {
@@ -83,6 +87,17 @@
     }
 ]
 
+link_tag_name = 'LinkTag1'
+link_tag_data = [
+    {
+        "common.ALLTYPES_NAME": link_tag_name,
+        "common.ALLTYPES_DESCRIPTION": "",
+        "advanced_tags.ENABLED": True,
+        "advanced_tags.LINK_INPUT_TAG": "_System._Time_Hour",
+        "advanced_tags.LINK_OUTPUT_TAG": f"{sim_ch_name}.{sim_dev_name}.{sim_tag_name}",
+    }
+]
+
 def HTTPErrorHandler(err):
     if err.__class__ is error.KepHTTPError:
         print(err.code)
@@ -101,6 +116,31 @@ def initialize(server: connection.server):
         server._config_get(server.url +"/project/_advancedtags")
     except Exception as err:
         pytest.skip("Advanced Tags is not installed", allow_module_level=True)
+    
+    # Create simulated device for testing
+    sim_device = {
+        "common.ALLTYPES_NAME": sim_ch_name,
+        "servermain.MULTIPLE_TYPES_DEVICE_DRIVER": "Simulator",
+        "devices": [
+            {
+                "common.ALLTYPES_NAME": sim_dev_name,
+                "servermain.MULTIPLE_TYPES_DEVICE_DRIVER": "Simulator",
+                "tags": [
+                    {
+                        "common.ALLTYPES_NAME": sim_tag_name,
+                        "servermain.TAG_ADDRESS": "K0"
+                    }
+                ],
+            }
+        ]
+    }
+    try:
+        connectivity.channel.add_channel(server, sim_device)
+    except Exception as err:
+        HTTPErrorHandler(err)
+        return False
+    return True
+
 
 def complete(server: connection.server):
     # Remove advanced tag group and all tags created during the test
@@ -126,8 +166,22 @@ def complete(server: connection.server):
             for obj in obj_list:
                 minimum_tag_path = f'_advancedtags.{obj["name"]}'
                 adv_tags.min_tags.del_minimum_tag(server, minimum_tag_path)
-        
-    pass
+        elif key == 'maximum_tags':
+            for obj in obj_list:
+                maximum_tag_path = f'_advancedtags.{obj["name"]}'
+                adv_tags.max_tags.del_maximum_tag(server, maximum_tag_path)
+        elif key == 'link_tags':
+            for obj in obj_list:
+                link_tag_path = f'_advancedtags.{obj["name"]}'
+                adv_tags.link_tags.del_link_tag(server, link_tag_path)
+    
+    # Delete all Channels
+    try:
+        ch_left = connectivity.channel.get_all_channels(server)
+        for x in ch_left:
+            print(connectivity.channel.del_channel(server,x['common.ALLTYPES_NAME']))
+    except Exception as err:
+        HTTPErrorHandler(err)
 
 @pytest.fixture(scope="module")
 def server(kepware_server):
@@ -136,7 +190,8 @@ def server(kepware_server):
     server_type = kepware_server[1]
     
     # Initialize any configuration before testing in module
-    initialize(server)
+    if not initialize(server):
+        pytest.fail("Simulation Device creation failed")
 
     # Everything below yield is run after module tests are completed
     yield server
@@ -420,6 +475,47 @@ def test_maximum_tag_del(server):
     maximum_tag_path = f'_advancedtags.{adv_tag_group_name}.{maximum_tag_name}'
     assert adv_tags.max_tags.del_maximum_tag(server, maximum_tag_path)
 
+def test_link_tag_add(server):
+    # Add a link tag to the root advanced tag plug-in
+    assert adv_tags.link_tags.add_link_tag(server, f'_advancedtags', link_tag_data)
+
+    testTag = {
+        "common.ALLTYPES_NAME": "newLinkTag",
+        "common.ALLTYPES_DESCRIPTION": "",
+        "advanced_tags.ENABLED": True,
+        "advanced_tags.LINK_INPUT_TAG": "_System._Time_Hour",
+        "advanced_tags.LINK_OUTPUT_TAG": f"{sim_ch_name}.{sim_dev_name}.{sim_tag_name}",
+    }
+    link_tag_data.append(testTag)
+    # Add a link tag to the advanced tag group
+    assert adv_tags.link_tags.add_link_tag(server, f'_advancedtags.{adv_tag_group_name}', link_tag_data)
+
+def test_link_tag_get(server):
+    # Get the link tag
+    link_tag_path = f'_advancedtags.{adv_tag_group_name}.{link_tag_name}'
+    result = adv_tags.link_tags.get_link_tag(server, link_tag_path)
+    assert type(result) == dict
+    assert result.get("common.ALLTYPES_NAME") == link_tag_name
+
+def test_link_tag_modify(server):
+    # Modify the link tag
+    link_tag_path = f'_advancedtags.{adv_tag_group_name}.{link_tag_name}'
+    tag_data = {
+        "common.ALLTYPES_DESCRIPTION": "Modified link tag"
+    }
+    assert adv_tags.link_tags.modify_link_tag(server, link_tag_path, tag_data, force=True)
+
+def test_link_tag_get_all(server):
+    # Get all link tags under the group
+    result = adv_tags.link_tags.get_all_link_tags(server, f'_advancedtags.{adv_tag_group_name}')
+    assert type(result) == list
+    assert any(tag.get("common.ALLTYPES_NAME") == link_tag_name for tag in result)
+
+def test_link_tag_del(server):
+    # Delete the link tag
+    link_tag_path = f'_advancedtags.{adv_tag_group_name}.{link_tag_name}'
+    assert adv_tags.link_tags.del_link_tag(server, link_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}')
