kepconfig.connectivity.tag

  1# -------------------------------------------------------------------------
  2# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
  3# See License.txt in the project root for
  4# license information.
  5# --------------------------------------------------------------------------
  6
  7
  8r"""`tag` exposes an API to allow modifications (add, delete, modify) to 
  9tag and tag group objects within the Kepware Configuration API
 10"""
 11
 12from ..connection import server
 13from ..error import KepError, KepHTTPError
 14from ..utils import _url_parse_object, path_split
 15from typing import Union
 16from . import channel, device
 17import inspect
 18
 19TAGS_ROOT = '/tags'
 20TAG_GRP_ROOT = '/tag_groups'
 21
 22def _create_tags_url(tag = None):
 23    '''Creates url object for the "tags" branch of Kepware's project tree. Used 
 24    to build a part of Kepware Configuration API URL structure
 25    
 26    Returns the tag specific url when a value is passed as the tag name.
 27    '''
 28    if tag == None:
 29        return TAGS_ROOT
 30    else: 
 31        return '{}/{}'.format(TAGS_ROOT, _url_parse_object(tag))
 32
 33def _create_tag_groups_url(tag_group = None):
 34    '''Creates url object for the "tag_group" branch of Kepware's project tree. Used 
 35    to build a part of Kepware Configuration API URL structure
 36    
 37    Returns the tag group specific url when a value is passed as the tag group name.
 38    '''
 39    if tag_group == None:
 40        return TAG_GRP_ROOT
 41    else: 
 42        return '{}/{}'.format(TAG_GRP_ROOT,_url_parse_object(tag_group))
 43
 44def add_tag(server: server, tag_path: str, DATA: Union[dict, list]) -> Union[bool, list]:
 45    '''Add `"tag"` or multiple `"tag"` objects to a specific path in Kepware. 
 46    Can be used to pass a list of tags to be added at one path location.
 47
 48    :param server: instance of the `server` class
 49    :param device_path: path identifying where to add tag(s). Standard Kepware address decimal 
 50    notation string that tags exists such as "channel1.device1.tag_group1" or "channel1.device1"
 51    :param DATA: Dict or List of Dicts of the tag(s) to add
 52
 53    :return: True - If a "HTTP 201 - Created" is received from Kepware server
 54    :return: If a "HTTP 207 - Multi-Status" is received from Kepware with a list of dict error responses for all 
 55    tags added that failed.
 56
 57    :raises KepHTTPError: If urllib provides an HTTPError
 58    :raises KepURLError: If urllib provides an URLError
 59    '''
 60
 61    path_obj = path_split(tag_path)
 62    try:
 63        url = server.url+channel._create_url(path_obj['channel'])+device._create_url(path_obj['device'])
 64        if 'tag_path' in path_obj:
 65            for tg in path_obj['tag_path']:
 66                url += _create_tag_groups_url(tag_group=tg)
 67        url += _create_tags_url()
 68    except KeyError as err:
 69        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
 70        raise KepError(err_msg)
 71    except Exception as e:
 72        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
 73        raise KepError(err_msg)
 74    r = server._config_add(url, DATA)
 75    if r.code == 201: return True 
 76    elif r.code == 207:
 77        errors = [] 
 78        for item in r.payload:
 79            if item['code'] != 201:
 80                errors.append(item)
 81        return errors
 82    else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
 83
 84def add_tag_group(server: server, tag_group_path: str, DATA: Union[dict, list]) -> Union[bool, list]:
 85    '''Add `"tag_group"` or multiple `"tag_group"` objects to a specific path in Kepware. 
 86    Can be used to pass a list of tag_groups and children (tags or tag groups) to be added at one 
 87    path location.
 88
 89    :param server: instance of the `server` class
 90    :param tag_group_path: path identifying where to add tag group(s). Standard Kepware address decimal 
 91    notation string that tag groups exists such as "channel1.device1.tag_group1" or "channel1.device1"
 92    :param DATA: Dict or List of Dicts of the tag group(s) to add and it's children (tags or tag groups)
 93
 94    :return: True - If a "HTTP 201 - Created" is received from Kepware server
 95    :return: If a "HTTP 207 - Multi-Status" is received from Kepware with a list of dict error responses for all 
 96    tag groups added that failed.
 97
 98    :raises KepHTTPError: If urllib provides an HTTPError
 99    :raises KepURLError: If urllib provides an URLError
100    '''
101
102    path_obj = path_split(tag_group_path)
103    try:
104        url = server.url+channel._create_url(path_obj['channel'])+device._create_url(path_obj['device'])
105        if 'tag_path' in path_obj:
106            for tg in path_obj['tag_path']:
107                url += _create_tag_groups_url(tag_group=tg)
108        url += _create_tag_groups_url()
109    except KeyError as err:
110        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
111        raise KepError(err_msg)
112    except Exception as e:
113        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
114        raise KepError(err_msg)
115    r = server._config_add(url, DATA)
116    if r.code == 201: return True 
117    elif r.code == 207:
118        errors = [] 
119        for item in r.payload:
120            if item['code'] != 201:
121                errors.append(item)
122        return errors
123    else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
124
125def add_all_tags(server: server, ch_dev_path: str, DATA: dict) -> Union[bool, list]:
126    '''Add `"tag"` and `"tag group"` objects to a device in Kepware. To be used to 
127    pass a list of tags, tag groups and/or children of tag groups (tags and tag 
128    groups) to be added at once. See example below for required structure with 
129    "tags" and "tag_groups" as keys:
130
131    Example DATA:
132
133        {
134            'tags': [tag1_dict, tag2_dict,...],
135            'tag_groups':[
136                {
137                    tag_group1_properties,
138                    'tags': [tag1_dict, tag2_dict,...]
139                    'tag_groups':[sub_group1, subgroup2,...]
140                }, 
141                {
142                    tag_group2_properties,
143                    'tags': [tag1_dict, tag2_dict,...]
144                    'tag_groups':[sub_group1, subgroup2,...]
145                },...]
146        }
147    
148    :param server: instance of the `server` class
149    :param ch_dev_path: device path identifying where to add tags and tag groups. Standard Kepware address decimal 
150    notation string that tag groups exists such as "channel1.device1"
151    :param DATA: Dict of the tags and tag groups to add and it's children (tags or tag groups). 
152
153    
154    :return: True - If a "HTTP 201 - Created" is received from Kepware server
155    :return: List [tag failure list, tag group failure list] - If a "HTTP 207 - Multi-Status" is received from 
156    Kepware for either tags or tag groups, a list of dict error responses for all tags and/or tag groups added that failed.
157    :return: False - If tags or tag groups are not found in DATA
158
159    :raises KepHTTPError: If urllib provides an HTTPError
160    :raises KepURLError: If urllib provides an URLError
161    '''
162    
163    tags_result = False
164    tag_groups_result = False
165
166    # check to see if there are dict entries for tags or tag groups
167    if ('tag_groups' not in DATA) and ('tags' not in DATA):
168        return False
169
170    if 'tags' in DATA:
171        tags_result = add_tag(server, ch_dev_path, DATA['tags'])
172    if 'tag_groups' in DATA:
173        #Add all Tag Groups
174        tag_groups_result = add_tag_group(server, ch_dev_path, DATA['tag_groups'])
175    
176    # build results return from both calls
177    if tags_result == True and tag_groups_result == True:
178        return True
179    elif tags_result == True:
180        return [[], tag_groups_result]
181    elif tag_groups_result == True:
182        return [tags_result, []]
183    else:
184        # mixed results from both tags and tag groups
185        return [tags_result, tag_groups_result]
186
187def modify_tag(server: server, full_tag_path: str, DATA: dict, force: bool = False) -> bool:
188    '''Modify a `"tag"` object and it's properties in Kepware.
189
190    :param server: instance of the `server` class
191    :param full_tag_path: path identifying location and tag to modify. Standard Kepware address decimal 
192    notation string including the tag such as "channel1.device1.tag_group1.tag1"
193    :param DATA: Dict of the `tag` properties to be modified
194    :param force: *(optional)* if True, will force the configuration update to the Kepware server
195
196    :return: True - If a "HTTP 200 - OK" is received from Kepware server
197
198    :raises KepHTTPError: If urllib provides an HTTPError
199    :raises KepURLError: If urllib provides an URLError
200    '''
201
202    tag_data = server._force_update_check(force, DATA)
203
204    path_obj = path_split(full_tag_path)
205    try:
206        url = server.url+channel._create_url(path_obj['channel'])+device._create_url(path_obj['device'])
207        for x in range(0, len(path_obj['tag_path'])-1):
208            url += _create_tag_groups_url(tag_group=path_obj['tag_path'][x])
209        url += _create_tags_url(tag=path_obj['tag_path'][len(path_obj['tag_path'])-1])
210    except KeyError as err:
211        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
212        raise KepError(err_msg)
213    except Exception as e:
214        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
215        raise KepError(err_msg)
216    r = server._config_update(url, tag_data)
217    if r.code == 200: return True 
218    else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
219
220def modify_tag_group(server: server, tag_group_path: str, DATA: dict, force: bool = False) -> bool:
221    '''Modify a `"tag group"` object and it's properties in Kepware.
222
223    :param server: instance of the `server` class
224    :param tag_group_path: path identifying location and tag group to modify. Standard Kepware address decimal 
225    notation string that tag groups exists such as "channel1.device1.tag_group1"
226    :param DATA: Dict of the `tag group` properties to be modified
227    :param force: *(optional)* if True, will force the configuration update to the Kepware server
228
229    :return: True - If a "HTTP 200 - OK" is received from Kepware server
230
231    :raises KepHTTPError: If urllib provides an HTTPError
232    :raises KepURLError: If urllib provides an URLError
233    '''
234
235    tag_group_data = server._force_update_check(force, DATA)
236
237    path_obj = path_split(tag_group_path)
238    try:
239        url = server.url+channel._create_url(path_obj['channel'])+device._create_url(path_obj['device'])
240        for tg in path_obj['tag_path']:
241            url += _create_tag_groups_url(tag_group=tg)
242    except KeyError as err:
243        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
244        raise KepError(err_msg)
245    except Exception as e:
246        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
247        raise KepError(err_msg)
248    r = server._config_update(url, tag_group_data)
249    if r.code == 200: return True 
250    else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
251
252def del_tag(server: server, full_tag_path: str) -> bool:
253    '''Delete `"tag"` object at a specific path in Kepware.
254
255    :param server: instance of the `server` class
256    :param full_tag_path: path identifying location and tag to delete. Standard Kepware address decimal 
257    notation string including the tag such as "channel1.device1.tag_group1.tag1"
258
259    :return: True - If a "HTTP 200 - OK" is received from Kepware server
260
261    :raises KepHTTPError: If urllib provides an HTTPError
262    :raises KepURLError: If urllib provides an URLError
263    '''
264
265    path_obj = path_split(full_tag_path)
266    try:
267        url = server.url+channel._create_url(path_obj['channel'])+device._create_url(path_obj['device'])
268        for x in range(0, len(path_obj['tag_path'])-1):
269            url += _create_tag_groups_url(tag_group=path_obj['tag_path'][x])
270        url += _create_tags_url(tag=path_obj['tag_path'][len(path_obj['tag_path'])-1])
271    except KeyError as err:
272        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
273        raise KepError(err_msg)
274    except Exception as e:
275        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
276        raise KepError(err_msg)
277    r = server._config_del(url)
278    if r.code == 200: return True 
279    else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
280
281def del_tag_group(server: server, tag_group_path: str) -> bool:
282    '''Delete `"tag group"` object at a specific path in Kepware.
283
284    :param server: instance of the `server` class
285    :param tag_group_path: path identifying location and tag group to delete. Standard Kepware address decimal 
286    notation string that tag groups exists such as "channel1.device1.tag_group1"
287
288    :return: True - If a "HTTP 200 - OK" is received from Kepware server
289
290    :raises KepHTTPError: If urllib provides an HTTPError
291    :raises KepURLError: If urllib provides an URLError
292    '''
293
294    path_obj = path_split(tag_group_path)
295    try:
296        url = server.url+channel._create_url(path_obj['channel'])+device._create_url(path_obj['device'])
297        for tg in path_obj['tag_path']:
298            url += _create_tag_groups_url(tag_group=tg)
299    except KeyError as err:
300        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
301        raise KepError(err_msg)
302    except Exception as err:
303        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(err))
304        raise KepError(err_msg)
305    r = server._config_del(url)
306    if r.code == 200: return True 
307    else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
308
309def get_tag(server: server, full_tag_path: str) -> dict:
310    '''Returns the properties of the `"tag"` object at a specific path in Kepware. 
311
312    :param server: instance of the `server` class
313    :param full_tag_path: path identifying location and tag to delete. Standard Kepware address decimal 
314    notation string including the tag such as "channel1.device1.tag_group1.tag1"
315
316    :return: Dict of data for the tag requested
317
318    :raises KepHTTPError: If urllib provides an HTTPError
319    :raises KepURLError: If urllib provides an URLError
320    '''
321
322    path_obj = path_split(full_tag_path)
323    try:
324        url = server.url+channel._create_url(path_obj['channel'])+device._create_url(path_obj['device'])
325        for x in range(0, len(path_obj['tag_path'])-1):
326            url += _create_tag_groups_url(tag_group=path_obj['tag_path'][x])
327        url += _create_tags_url(tag=path_obj['tag_path'][len(path_obj['tag_path'])-1])
328    except KeyError as err:
329        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
330        raise KepError(err_msg)
331    except Exception as e:
332        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
333        raise KepError(err_msg)
334    r = server._config_get(url)
335    return r.payload
336
337def get_all_tags(server: server, full_tag_path: str, *, options: dict = None) -> list:
338    '''Returns the properties of all `"tag"` object at a specific path in Kepware. 
339
340    :param server: instance of the `server` class
341    :param full_tag_path: path identifying location to retreive tag list. Standard Kepware address decimal 
342    notation string including the tag such as "channel1.device1.tag_group1.tag1"
343    :param options: *(optional)* Dict of parameters to filter, sort or pagenate the list of tags. Options are `filter`, 
344        `sortOrder`, `sortProperty`, `pageNumber`, and `pageSize`
345    
346    :return: List of data for all tags
347
348    :raises KepHTTPError: If urllib provides an HTTPError
349    :raises KepURLError: If urllib provides an URLError
350    '''
351
352    path_obj = path_split(full_tag_path)
353    try:
354        url = f"{server.url}{channel._create_url(path_obj['channel'])}{device._create_url(path_obj['device'])}"
355        if 'tag_path' in path_obj:
356            for tg in path_obj['tag_path']:
357                url += _create_tag_groups_url(tag_group=tg)
358        url += _create_tags_url()
359    except KeyError as err:
360        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
361        raise KepError(err_msg)
362    except Exception as e:
363        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
364        raise KepError(err_msg)
365    r = server._config_get(url, params= options)
366    return r.payload
367
368def get_tag_group(server: server, tag_group_path: str) -> dict:
369    '''Returns the properties of the "tag group" object at a specific 
370    path in Kepware. Returned object is JSON.
371
372    :param server: instance of the `server` class
373    :param tag_group_path: path identifying location and tag group to retrieve properties. Standard Kepware address decimal 
374    notation string that tag groups exists such as "channel1.device1.tag_group1"
375
376    :return: Dict of data for the tag group requested
377
378    :raises KepHTTPError: If urllib provides an HTTPError
379    :raises KepURLError: If urllib provides an URLError
380    '''
381    path_obj = path_split(tag_group_path)
382    try:
383        url = server.url+channel._create_url(path_obj['channel'])+device._create_url(path_obj['device'])
384        for tg in path_obj['tag_path']:
385            url += _create_tag_groups_url(tag_group=tg)
386    except KeyError as err:
387        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
388        raise KepError(err_msg)
389    except Exception as e:
390        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
391        raise KepError(err_msg)
392    r = server._config_get(url)
393    return r.payload
394
395def get_all_tag_groups(server:server, tag_group_path: str, *, options: dict = None) -> list:
396    '''Returns the properties of all `"tag group"` objects at a specific 
397    path in Kepware.
398
399    :param server: instance of the `server` class
400    :param tag_group_path: path identifying location to retrieve tag group list and properties. Standard Kepware address decimal 
401    notation string that tag groups exists such as "channel1.device1.tag_group1"
402    :param options: *(optional)* Dict of parameters to filter, sort or pagenate the list of devices. Options are `filter`, 
403        `sortOrder`, `sortProperty`, `pageNumber`, and `pageSize`
404
405    :return: List of data for all tag groups
406
407    :raises KepHTTPError: If urllib provides an HTTPError
408    :raises KepURLError: If urllib provides an URLError
409    '''
410    path_obj = path_split(tag_group_path)
411    try:
412        url = f"{server.url}{channel._create_url(path_obj['channel'])}{device._create_url(path_obj['device'])}"
413        if 'tag_path' in path_obj:
414            for tg in path_obj['tag_path']:
415                url += _create_tag_groups_url(tag_group=tg)
416        url += _create_tag_groups_url()
417    except KeyError as err:
418        err_msg = 'Error: No key {} identified | Function: {}'.format(err, inspect.currentframe().f_code.co_name)
419        raise KepError(err_msg)
420    except Exception as e:
421        err_msg = 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
422        raise KepError(err_msg)
423    r = server._config_get(url, params= options)
424    return r.payload
425
426def get_full_tag_structure(server: server, path: str, *, recursive: bool = False, options: dict = None) -> dict:
427    '''Returns the properties of all `"tag"` and `"tag group"` objects at a specific 
428    path in Kepware. Returned object is a dict of tag list and tag group list.
429
430    Example:
431
432        {
433            'tags': [tag1_dict, tag2_dict,...],
434            'tag_groups':[tag_group1_dict, tag_group2_dict,...]
435        } 
436
437    If `recursive` is TRUE, then the call will iterate through all tag groups and get the tags and 
438    tag groups of all tag group children.This would be the equivilant of asking for all tags and tag groups
439    that exist below the `"path"` location. The returned object would look like below, nested based on how many 
440    levels the tag_group namespace has tags or tag_groups:
441
442    Example with Recursive True:
443
444        {
445            'tags': [tag1_dict, tag2_dict,...],
446            'tag_groups':[
447                {
448                    tag_group1_properties,
449                    'tags': [tag1_dict, tag2_dict,...]
450                    'tag_groups':[sub_group1, subgroup2,...]
451                }, 
452                {
453                    tag_group2_properties,
454                    'tags': [tag1_dict, tag2_dict,...]
455                    'tag_groups':[sub_group1, subgroup2,...]
456                },...]
457        } 
458
459    :param server: instance of the `server` class
460    :param path: path identifying location to retreive the tag structure. Standard Kepware address decimal 
461    notation string such as "channel1.device1.tag_group1" and must container at least the channel and device.
462    :param recursive: *(optional)* If True, returns structures within the tag groups found and all of their 
463    children. (default= False)
464    :param options: *(optional)* Dict of parameters to filter, sort or pagenate the list of tags and tag groups. 
465    Options are 'filter', 'sortOrder', and 'sortProperty' only.
466
467    :return: Dict of data for the tag structure requested at "path" location
468
469    :raises KepHTTPError: If urllib provides an HTTPError
470    :raises KepURLError: If urllib provides an URLError
471    '''
472    r = {}
473        
474    # Remove pagination options, if present. Not useful with this method since it is designed to get the full tree of items.
475    if options is not None:
476        remove_list = ['pageNumber','pageSize']
477        [options.pop(x) for x in remove_list]
478    
479    r['tags'] = get_all_tags(server, path, options= options)
480    r['tag_groups'] = get_all_tag_groups(server, path, options= options)
481    if recursive:
482        for group in r['tag_groups']:
483            res = get_full_tag_structure(server, path + '.' + group['common.ALLTYPES_NAME'], recursive= recursive, options= options)
484            group.update(res)
485    return r