kepconfig.connectivity.egd.exchange

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