kepconfig.connectivity.egd.range

  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"""`ranges` exposes an API to allow modifications (add, delete, modify) to 
  9range objects in exchanges for EGD devices within the Kepware Configuration API
 10"""
 11
 12from typing import Union
 13from .. import egd as EGD
 14from ...connection import server
 15from ...error import KepError, KepHTTPError
 16from ...utils import _url_parse_object
 17
 18RANGES_ROOT = '/ranges'
 19
 20def _create_url(device_path, ex_type, exchange_name, range = None):
 21    '''Creates url object for the "range" branch of Kepware's project tree. Used 
 22    to build a part of Kepware Configuration API URL structure
 23
 24    Returns the range specific url when a value is passed as the range name.
 25    '''
 26    exchange_root = EGD.exchange._create_url(device_path, ex_type, exchange_name)
 27
 28    if range == None:
 29        return '{}{}'.format(exchange_root, RANGES_ROOT)
 30    else:
 31        return '{}{}/{}'.format(exchange_root, RANGES_ROOT, _url_parse_object(range))
 32
 33def add_range(server: server, device_path: str, ex_type: str, exchange_name: str, DATA: Union[dict, list]) -> Union[bool, list]:
 34    '''Add a `"range"` or multiple `"range"` objects to Kepware. This allows you to 
 35    create a range or multiple ranges all in one function, if desired.
 36
 37    When passing multiple ranges, they will be populated in the same order
 38    in the list sent. Ensure you provide the list in the order desired.
 39
 40    :param server: instance of the `server` class
 41    :param device_path: path to EGD device. Standard Kepware address decimal 
 42    notation string such as `"channel1.device1"`
 43    :param ex_type: type of exchange, either `CONSUMER` or `PRODUCER`
 44    :param exchange_name: name of exchange that range is located
 45    :param DATA: Dict or List of Dicts of the range(s) to add
 46
 47    :return: True - If a "HTTP 201 - Created" is received from Kepware server
 48    :return: If a "HTTP 207 - Multi-Status" is received from Kepware with a list of dict error responses for all 
 49    ranges added that failed.
 50        
 51    :raises KepHTTPError: If urllib provides an HTTPError
 52    :raises KepURLError: If urllib provides an URLError
 53    '''
 54
 55    r = server._config_add(server.url + _create_url(device_path, ex_type, exchange_name), DATA)
 56    if r.code == 201: return True
 57    elif r.code == 207:
 58        errors = [] 
 59        for item in r.payload:
 60            if item['code'] != 201:
 61                errors.append(item)
 62        return errors
 63    else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
 64
 65def del_range(server: server, device_path: str, ex_type: str, exchange_name: str, range_name: str) -> bool:
 66    '''Delete a `"range"` object in Kepware.
 67    
 68    :param server: instance of the `server` class
 69    :param device_path: path to EGD device. Standard Kepware address decimal 
 70    notation string such as `"channel1.device1"`
 71    :param ex_type: type of exchange, either `CONSUMER` or `PRODUCER`
 72    :param exchange_name: name of exchange that range is located
 73    :param range_name: name of range to delete
 74    
 75    :return: True - If a "HTTP 200 - OK" is received from Kepware server
 76
 77    :raises KepHTTPError: If urllib provides an HTTPError
 78    :raises KepURLError: If urllib provides an URLError
 79    '''
 80
 81    r = server._config_del(server.url + _create_url(device_path, ex_type, exchange_name, range_name))
 82    if r.code == 200: return True 
 83    else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
 84
 85def modify_range(server: server, device_path: str, ex_type: str, exchange_name: str, DATA: dict, *, range_name: str = None, force: bool = False) -> bool:
 86    '''Modify a `"range"` object and it's properties in Kepware. If a `"range_name"` is not provided as an input,
 87    you need to identify the range in the *'common.ALLTYPES_NAME'* property field in the `"DATA"`. It will 
 88    assume that is the range that is to be modified.
 89
 90    :param server: instance of the `server` class
 91    :param device_path: path to EGD device. Standard Kepware address decimal 
 92    notation string such as `"channel1.device1"`
 93    :param ex_type: type of exchange, either `CONSUMER` or `PRODUCER`
 94    :param exchange_name: name of exchange that range is located
 95    :param DATA: Dict of the range properties to be modified.
 96    :param range_name: *(optional)* name of range to to modify. Only needed if not existing in `"DATA"`
 97    :param force: *(optional)* if True, will force the configuration update to the Kepware server
 98    
 99    :return: True - If a "HTTP 200 - OK" is received from Kepware server
100
101    :raises KepHTTPError: If urllib provides an HTTPError
102    :raises KepURLError: If urllib provides an URLError
103    '''
104    
105    range_data = server._force_update_check(force, DATA)
106    if range_name == None:
107        try:
108            r = server._config_update(server.url + _create_url(device_path, ex_type, exchange_name, range_data['common.ALLTYPES_NAME']), range_data)
109            if r.code == 200: return True 
110            else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
111        except KeyError as err:
112            err_msg = 'Error: No range identified in DATA | Key Error: {}'.format(err)
113            raise KepError(err_msg)
114    else:
115        r = server._config_update(server.url + _create_url(device_path, ex_type, exchange_name, range_name), range_data)
116        if r.code == 200: return True 
117        else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
118
119def get_range(server: server, device_path: str, ex_type: str, exchange_name: str, range_name: str = None, *, options: dict = None) -> Union[dict, list]:
120    '''Returns the properties of the `"range"` object or a list of all ranges.
121    
122    :param server: instance of the `server` class
123    :param device_path: path to EGD device. Standard Kepware address decimal 
124    notation string such as `"channel1.device1"`
125    :param ex_type: type of exchange, either `CONSUMER` or `PRODUCER`
126    :param exchange_name: name of exchange that range is located
127    :param DATA: Dict of the range properties to be modified.
128    :param range_name: *(optional)* name of range to retrieve. If not defined, get all ranges
129    :param options: *(optional)* Dict of parameters to filter, sort or pagenate the list of exchanges. Options are 'filter', 
130    'sortOrder', 'sortProperty', 'pageNumber', and 'pageSize'. Only used when range_name is not defined.
131    
132    :return: Dict of properties for the range requested or a List of ranges and their properties
133
134    :raises KepHTTPError: If urllib provides an HTTPError
135    :raises KepURLError: If urllib provides an URLError
136    '''
137    if range_name == None:
138        r = server._config_get(f'{server.url}{_create_url(device_path, ex_type, exchange_name)}', params= options)
139    else:
140        r = server._config_get(f'{server.url}{_create_url(device_path, ex_type, exchange_name, range_name)}')
141    return r.payload