hat.gateway.devices.iec61850.client

   1"""IEC 61850 client device"""
   2
   3from collections.abc import Collection
   4import asyncio
   5import collections
   6import datetime
   7import logging
   8import math
   9
  10from hat import aio
  11from hat import json
  12from hat import util
  13from hat.drivers import iec61850
  14from hat.drivers import tcp
  15import hat.event.common
  16
  17from hat.gateway import common
  18
  19
  20mlog: logging.Logger = logging.getLogger(__name__)
  21
  22
  23termination_timeout: int = 100
  24
  25report_segments_timeout: int = 100
  26
  27
  28async def create(conf: common.DeviceConf,
  29                 eventer_client: hat.event.eventer.Client,
  30                 event_type_prefix: common.EventTypePrefix
  31                 ) -> 'Iec61850ClientDevice':
  32
  33    value_types = {}
  34    value_types_61850 = {}
  35    for vt in conf['value_types']:
  36        ref = (vt['logical_device'],
  37               vt['logical_node'],
  38               vt['fc'],
  39               vt['name'])
  40        value_types_61850[ref] = _vtype_61850_from_vtype_conf(vt['type'])
  41        value_types[ref] = _vtype_from_vtype_conf(vt['type'])
  42
  43    rcb_confs = {i['report_id']: i for i in conf['rcbs']}
  44    dataset_confs = {_dataset_ref_from_conf(ds_conf['ref']): ds_conf
  45                     for ds_conf in conf['datasets']}
  46    value_ref_data_name = {}
  47    for data_conf in conf['data']:
  48        data_v_ref = _value_ref_from_conf(data_conf['value'])
  49
  50        q_ref = (_value_ref_from_conf(data_conf['quality'])
  51                 if data_conf.get('quality') else None)
  52        if data_conf.get('quality'):
  53            q_type = _value_type_from_ref(value_types, q_ref)
  54            if q_type != iec61850.AcsiValueType.QUALITY:
  55                raise Exception(f"invalid quality type {q_ref}")
  56
  57        t_ref = (_value_ref_from_conf(data_conf['timestamp'])
  58                 if data_conf.get('timestamp') else None)
  59        if t_ref:
  60            t_type = _value_type_from_ref(value_types, t_ref)
  61            if t_type != iec61850.AcsiValueType.TIMESTAMP:
  62                raise Exception(f"invalid timestamp type {t_ref}")
  63
  64        seld_ref = (_value_ref_from_conf(data_conf['selected'])
  65                    if data_conf.get('selected') else None)
  66        if seld_ref:
  67            seld_type = _value_type_from_ref(value_types, seld_ref)
  68            if seld_type != iec61850.BasicValueType.BOOLEAN:
  69                raise Exception(f"invalid selected type {seld_ref}")
  70
  71        rcb_conf = rcb_confs[data_conf['report_id']]
  72        ds_ref = _dataset_ref_from_conf(rcb_conf['dataset'])
  73        ds_conf = dataset_confs[ds_ref]
  74        for value_conf in ds_conf['values']:
  75            value_ref = _value_ref_from_conf(value_conf)
  76            if (_refs_match(data_v_ref, value_ref) or
  77                    (q_ref and _refs_match(q_ref, value_ref)) or
  78                    (t_ref and _refs_match(t_ref, value_ref)) or
  79                    (seld_ref and _refs_match(seld_ref, value_ref))):
  80                value_ref_data_name[value_ref] = data_conf['name']
  81
  82    dataset_values_ref_type = {}
  83    for ds_conf in conf['datasets']:
  84        for val_ref_conf in ds_conf['values']:
  85            value_ref = _value_ref_from_conf(val_ref_conf)
  86            value_type = _value_type_from_ref(value_types, value_ref)
  87            dataset_values_ref_type[value_ref] = value_type
  88
  89    command_ref_value_type = {}
  90    for cmd_conf in conf['commands']:
  91        cmd_ref = iec61850.CommandRef(**cmd_conf['ref'])
  92        value_type = _value_type_from_ref(value_types, cmd_ref)
  93        command_ref_value_type[cmd_ref] = value_type
  94
  95    change_ref_value_type = {}
  96    for change_conf in conf['changes']:
  97        value_ref = _value_ref_from_conf(change_conf['ref'])
  98        value_type = _value_type_from_ref(value_types, value_ref)
  99        change_ref_value_type[value_ref] = value_type
 100
 101    entry_id_event_types = [
 102        (*event_type_prefix, 'gateway', 'entry_id', i['report_id'])
 103        for i in conf['rcbs'] if i['ref']['type'] == 'BUFFERED']
 104    if entry_id_event_types:
 105        result = await eventer_client.query(
 106            hat.event.common.QueryLatestParams(entry_id_event_types))
 107        rcbs_entry_ids = {
 108            event.type[5]: (bytes.fromhex(event.payload.data)
 109                            if event.payload.data is not None else None)
 110            for event in result.events}
 111    else:
 112        rcbs_entry_ids = {}
 113
 114    device = Iec61850ClientDevice()
 115
 116    device._rcbs_entry_ids = rcbs_entry_ids
 117    device._conf = conf
 118    device._eventer_client = eventer_client
 119    device._event_type_prefix = event_type_prefix
 120    device._conn = None
 121    device._conn_status = None
 122    device._terminations = {}
 123    device._reports_segments = {}
 124
 125    device._value_ref_data_name = value_ref_data_name
 126    device._dataset_values_ref_type = dataset_values_ref_type
 127    device._command_ref_value_type = command_ref_value_type
 128    device._change_ref_value_type = change_ref_value_type
 129    device._data_name_confs = {i['name']: i for i in conf['data']}
 130    device._command_name_confs = {i['name']: i for i in conf['commands']}
 131    device._command_name_ctl_nums = {i['name']: 0 for i in conf['commands']}
 132    device._change_name_value_refs = {
 133        i['name']: _value_ref_from_conf(i['ref']) for i in conf['changes']}
 134    device._rcb_type = {rcb_conf['report_id']: rcb_conf['ref']['type']
 135                        for rcb_conf in conf['rcbs']}
 136    device._persist_dyn_datasets = set(_get_persist_dyn_datasets(conf))
 137    device._dyn_datasets_values = dict(_get_dyn_datasets_values(conf))
 138    device._report_data_refs = collections.defaultdict(collections.deque)
 139    for rcb_conf in device._conf['rcbs']:
 140        report_id = rcb_conf['report_id']
 141        ds_ref = _dataset_ref_from_conf(rcb_conf['dataset'])
 142        for value_conf in dataset_confs[ds_ref]['values']:
 143            value_ref = _value_ref_from_conf(value_conf)
 144            device._report_data_refs[report_id].append(value_ref)
 145
 146    device._data_value_types = dict(
 147        _get_data_value_types(conf, value_types_61850))
 148    device._cmd_value_types = dict(
 149        _get_cmd_value_types(conf, value_types_61850))
 150
 151    device._async_group = aio.Group()
 152    device._async_group.spawn(device._connection_loop)
 153    device._loop = asyncio.get_running_loop()
 154
 155    return device
 156
 157
 158info: common.DeviceInfo = common.DeviceInfo(
 159    type="iec61850_client",
 160    create=create,
 161    json_schema_id="hat-gateway://iec61850.yaml#/$defs/client",
 162    json_schema_repo=common.json_schema_repo)
 163
 164
 165class Iec61850ClientDevice(common.Device):
 166
 167    @property
 168    def async_group(self) -> aio.Group:
 169        return self._async_group
 170
 171    async def process_events(self, events: Collection[hat.event.common.Event]):
 172        try:
 173            for event in events:
 174                suffix = event.type[len(self._event_type_prefix):]
 175
 176                if suffix[:2] == ('system', 'command'):
 177                    cmd_name, = suffix[2:]
 178                    await self._process_cmd_req(event, cmd_name)
 179
 180                elif suffix[:2] == ('system', 'change'):
 181                    val_name, = suffix[2:]
 182                    await self._process_change_req(event, val_name)
 183
 184                else:
 185                    raise Exception('unsupported event type')
 186
 187        except Exception as e:
 188            mlog.warning('error processing event %s: %s',
 189                         event.type, e, exc_info=e)
 190
 191    async def _connection_loop(self):
 192
 193        async def cleanup():
 194            await self._register_status('DISCONNECTED')
 195            if self._conn:
 196                await self._conn.async_close()
 197
 198        conn_conf = self._conf['connection']
 199        try:
 200            while True:
 201                await self._register_status('CONNECTING')
 202                try:
 203                    mlog.debug('connecting to %s:%s',
 204                               conn_conf['host'], conn_conf['port'])
 205                    self._conn = await aio.wait_for(
 206                        iec61850.connect(
 207                            addr=tcp.Address(conn_conf['host'],
 208                                             conn_conf['port']),
 209                            data_value_types=self._data_value_types,
 210                            cmd_value_types=self._cmd_value_types,
 211                            report_data_refs=self._report_data_refs,
 212                            report_cb=self._on_report,
 213                            termination_cb=self._on_termination,
 214                            status_delay=conn_conf['status_delay'],
 215                            status_timeout=conn_conf['status_timeout'],
 216                            local_tsel=conn_conf.get('local_tsel'),
 217                            remote_tsel=conn_conf.get('remote_tsel'),
 218                            local_ssel=conn_conf.get('local_ssel'),
 219                            remote_ssel=conn_conf.get('remote_ssel'),
 220                            local_psel=conn_conf.get('local_psel'),
 221                            remote_psel=conn_conf.get('remote_psel'),
 222                            local_ap_title=conn_conf.get('local_ap_title'),
 223                            remote_ap_title=conn_conf.get('remote_ap_title'),
 224                            local_ae_qualifier=conn_conf.get(
 225                                'local_ae_qualifier'),
 226                            remote_ae_qualifier=conn_conf.get(
 227                                'remote_ae_qualifier'),
 228                            local_detail_calling=conn_conf.get(
 229                                'local_detail_calling')),
 230                        conn_conf['connect_timeout'])
 231
 232                except Exception as e:
 233                    mlog.warning('connnection failed: %s', e, exc_info=e)
 234                    await self._register_status('DISCONNECTED')
 235                    await asyncio.sleep(conn_conf['reconnect_delay'])
 236                    continue
 237
 238                mlog.debug('connected')
 239                await self._register_status('CONNECTED')
 240
 241                initialized = False
 242                try:
 243                    await self._create_dynamic_datasets()
 244                    for rcb_conf in self._conf['rcbs']:
 245                        await self._init_rcb(rcb_conf)
 246                    initialized = True
 247
 248                except Exception as e:
 249                    mlog.warning(
 250                        'initialization failed: %s, closing connection',
 251                        e, exc_info=e)
 252                    self._conn.close()
 253
 254                await self._conn.wait_closing()
 255                await self._register_status('DISCONNECTED')
 256                await self._conn.wait_closed()
 257                self._conn = None
 258                self._terminations = {}
 259                self._reports_segments = {}
 260                if not initialized:
 261                    await asyncio.sleep(conn_conf['reconnect_delay'])
 262
 263        except Exception as e:
 264            mlog.error('connection loop error: %s', e, exc_info=e)
 265
 266        finally:
 267            mlog.debug('closing connection loop')
 268            self.close()
 269            await aio.uncancellable(cleanup())
 270
 271    async def _process_cmd_req(self, event, cmd_name):
 272        if not self._conn or not self._conn.is_open:
 273            raise Exception('no connection')
 274
 275        if cmd_name not in self._command_name_confs:
 276            raise Exception('unexpected command name')
 277
 278        cmd_conf = self._command_name_confs[cmd_name]
 279        cmd_ref = iec61850.CommandRef(**cmd_conf['ref'])
 280        action = event.payload.data['action']
 281        evt_session_id = event.payload.data['session_id']
 282        if (action == 'SELECT' and
 283                cmd_conf['model'] == 'SBO_WITH_NORMAL_SECURITY'):
 284            cmd = None
 285        else:
 286            ctl_num = self._command_name_ctl_nums[cmd_name]
 287            ctl_num = _update_ctl_num(ctl_num, action, cmd_conf['model'])
 288            self._command_name_ctl_nums[cmd_name] = ctl_num
 289            value_type = self._command_ref_value_type[cmd_ref]
 290            if value_type is None:
 291                raise Exception('value type undefined')
 292
 293            cmd = _command_from_event(event, cmd_conf, ctl_num, value_type)
 294
 295        term_future = None
 296        if (action == 'OPERATE' and
 297                cmd_conf['model'] in ['DIRECT_WITH_ENHANCED_SECURITY',
 298                                      'SBO_WITH_ENHANCED_SECURITY']):
 299            term_future = self._loop.create_future()
 300            self._conn.async_group.spawn(
 301                self._wait_cmd_term, cmd_name, cmd_ref, cmd, evt_session_id,
 302                term_future)
 303
 304        try:
 305            resp = await aio.wait_for(
 306                self._send_command(action, cmd_ref, cmd),
 307                self._conf['connection']['response_timeout'])
 308
 309        except (asyncio.TimeoutError, ConnectionError) as e:
 310            mlog.warning('send command failed: %s', e, exc_info=e)
 311            if term_future and not term_future.done():
 312                term_future.cancel()
 313            return
 314
 315        if resp is not None:
 316            if term_future and not term_future.done():
 317                term_future.cancel()
 318
 319        event = _cmd_resp_to_event(
 320            self._event_type_prefix, cmd_name, evt_session_id, action, resp)
 321        await self._register_events([event])
 322
 323    async def _send_command(self, action, cmd_ref, cmd):
 324        if action == 'SELECT':
 325            return await self._conn.select(cmd_ref, cmd)
 326
 327        if action == 'CANCEL':
 328            return await self._conn.cancel(cmd_ref, cmd)
 329
 330        if action == 'OPERATE':
 331            return await self._conn.operate(cmd_ref, cmd)
 332
 333        raise Exception('unsupported action')
 334
 335    async def _wait_cmd_term(self, cmd_name, cmd_ref, cmd, session_id, future):
 336        cmd_session_id = _get_command_session_id(cmd_ref, cmd)
 337        self._terminations[cmd_session_id] = future
 338        try:
 339            term = await aio.wait_for(future, termination_timeout)
 340            event = _cmd_resp_to_event(
 341                self._event_type_prefix, cmd_name, session_id, 'TERMINATION',
 342                term.error)
 343            await self._register_events([event])
 344
 345        except asyncio.TimeoutError:
 346            mlog.warning('command termination timeout')
 347
 348        finally:
 349            del self._terminations[cmd_session_id]
 350
 351    async def _process_change_req(self, event, value_name):
 352        if not self._conn or not self._conn.is_open:
 353            raise Exception('no connection')
 354
 355        if value_name not in self._change_name_value_refs:
 356            raise Exception('unexpected change name')
 357
 358        ref = self._change_name_value_refs[value_name]
 359        value_type = self._change_ref_value_type[ref]
 360        if value_type is None:
 361            raise Exception('value type undefined')
 362
 363        value = _value_from_json(event.payload.data['value'], value_type)
 364        try:
 365            resp = await aio.wait_for(
 366                self._conn.write_data(ref, value),
 367                self._conf['connection']['response_timeout'])
 368
 369        except asyncio.TimeoutError:
 370            mlog.warning('write data response timeout')
 371            return
 372
 373        except ConnectionError as e:
 374            mlog.warning('connection error on write data: %s', e, exc_info=e)
 375            return
 376
 377        session_id = event.payload.data['session_id']
 378        event = _write_data_resp_to_event(
 379            self._event_type_prefix, value_name, session_id, resp)
 380        await self._register_events([event])
 381
 382    async def _on_report(self, report):
 383        try:
 384            events = list(self._events_from_report(report))
 385            if not events:
 386                return
 387
 388            await self._register_events(events)
 389            if self._rcb_type[report.report_id] == 'BUFFERED':
 390                self._rcbs_entry_ids[report.report_id] = report.entry_id
 391
 392        except Exception as e:
 393            mlog.warning('report %s ignored: %s',
 394                         report.report_id, e, exc_info=e)
 395
 396    def _events_from_report(self, report):
 397        report_id = report.report_id
 398        if report_id not in self._report_data_refs:
 399            raise Exception(f'unexpected report {report_id}')
 400
 401        segm_id = (report_id, report.sequence_number)
 402        if report.more_segments_follow:
 403            if segm_id in self._reports_segments:
 404                segment_data, timeout_timer = self._reports_segments[segm_id]
 405                timeout_timer.cancel()
 406            else:
 407                segment_data = collections.deque()
 408
 409            segment_data.extend(report.data)
 410            timeout_timer = self._loop.call_later(
 411                report_segments_timeout, self._reports_segments.pop, segm_id)
 412            self._reports_segments[segm_id] = (segment_data, timeout_timer)
 413            return
 414
 415        if segm_id in self._reports_segments:
 416            report_data, timeout_timer = self._reports_segments.pop(segm_id)
 417            timeout_timer.cancel()
 418            report_data.extend(report.data)
 419
 420        else:
 421            report_data = report.data
 422
 423        yield from self._events_from_report_data(report_data, report_id)
 424
 425        if self._rcb_type[report_id] == 'BUFFERED':
 426            yield hat.event.common.RegisterEvent(
 427                type=(*self._event_type_prefix, 'gateway',
 428                      'entry_id', report_id),
 429                source_timestamp=None,
 430                payload=hat.event.common.EventPayloadJson(
 431                    report.entry_id.hex()
 432                    if report.entry_id is not None else None))
 433
 434    def _events_from_report_data(self, report_data, report_id):
 435        data_values_json = collections.defaultdict(dict)
 436        data_reasons = collections.defaultdict(set)
 437        for rv in report_data:
 438            if rv.ref not in self._value_ref_data_name:
 439                continue
 440
 441            data_name = self._value_ref_data_name[rv.ref]
 442            value_type = self._dataset_values_ref_type[rv.ref]
 443            if value_type is None:
 444                mlog.warning('report data ignored: unknown value type')
 445                continue
 446
 447            value_json = _value_to_json(rv.value, value_type)
 448            value_path = [rv.ref.logical_device, rv.ref.logical_node,
 449                          rv.ref.fc, *rv.ref.names]
 450            data_values_json[data_name] = json.set_(
 451                data_values_json[data_name], value_path, value_json)
 452            if rv.reasons:
 453                data_reasons[data_name].update(
 454                    reason.name for reason in rv.reasons)
 455
 456        for data_name, values_json in data_values_json.items():
 457            payload = {'reasons': list(data_reasons[data_name])}
 458            data_conf = self._data_name_confs[data_name]
 459            value_path = _conf_ref_to_path(data_conf['value'])
 460            value_json = json.get(values_json, value_path)
 461            if value_json is not None:
 462                payload['value'] = value_json
 463
 464            if 'quality' in data_conf:
 465                quality_path = _conf_ref_to_path(data_conf['quality'])
 466                quality_json = json.get(values_json, quality_path)
 467                if quality_json is not None:
 468                    payload['quality'] = quality_json
 469
 470            if 'timestamp' in data_conf:
 471                timestamp_path = _conf_ref_to_path(data_conf['timestamp'])
 472                timestamp_json = json.get(values_json, timestamp_path)
 473                if timestamp_json is not None:
 474                    payload['timestamp'] = timestamp_json
 475
 476            if 'selected' in data_conf:
 477                selected_path = _conf_ref_to_path(data_conf['selected'])
 478                selected_json = json.get(values_json, selected_path)
 479                if selected_json is not None:
 480                    payload['selected'] = selected_json
 481
 482            yield hat.event.common.RegisterEvent(
 483                type=(*self._event_type_prefix, 'gateway',
 484                      'data', data_name),
 485                source_timestamp=None,
 486                payload=hat.event.common.EventPayloadJson(payload))
 487
 488    def _on_termination(self, termination):
 489        cmd_session_id = _get_command_session_id(
 490            termination.ref, termination.cmd)
 491        if cmd_session_id not in self._terminations:
 492            mlog.warning('unexpected termination dropped')
 493            return
 494
 495        term_future = self._terminations[cmd_session_id]
 496        if not term_future.done():
 497            self._terminations[cmd_session_id].set_result(termination)
 498
 499    async def _init_rcb(self, rcb_conf):
 500        ref = iec61850.RcbRef(
 501            logical_device=rcb_conf['ref']['logical_device'],
 502            logical_node=rcb_conf['ref']['logical_node'],
 503            type=iec61850.RcbType[rcb_conf['ref']['type']],
 504            name=rcb_conf['ref']['name'])
 505        mlog.debug('initiating rcb %s', ref)
 506
 507        get_attrs = collections.deque([iec61850.RcbAttrType.REPORT_ID])
 508        dataset_ref = _dataset_ref_from_conf(rcb_conf['dataset'])
 509        if dataset_ref not in self._dyn_datasets_values:
 510            get_attrs.append(iec61850.RcbAttrType.DATASET)
 511        if 'conf_revision' in rcb_conf:
 512            get_attrs.append(iec61850.RcbAttrType.CONF_REVISION)
 513
 514        get_rcb_resp = await self._conn.get_rcb_attrs(ref, get_attrs)
 515        _validate_get_rcb_response(get_rcb_resp, rcb_conf)
 516
 517        await self._set_rcb(ref, [(iec61850.RcbAttrType.REPORT_ENABLE, False)])
 518
 519        if dataset_ref in self._dyn_datasets_values:
 520            await self._set_rcb(
 521                ref, [(iec61850.RcbAttrType.DATASET, dataset_ref)],
 522                critical=True)
 523
 524        if ref.type == iec61850.RcbType.BUFFERED:
 525            if 'reservation_time' in rcb_conf:
 526                await self._set_rcb(
 527                    ref, [(iec61850.RcbAttrType.RESERVATION_TIME,
 528                           rcb_conf['reservation_time'])])
 529
 530            entry_id = self._rcbs_entry_ids.get(rcb_conf['report_id'])
 531            if rcb_conf.get('purge_buffer') or entry_id is None:
 532                await self._set_rcb(
 533                    ref, [(iec61850.RcbAttrType.PURGE_BUFFER, True)])
 534
 535            else:
 536                try:
 537                    await self._set_rcb(
 538                        ref, [(iec61850.RcbAttrType.ENTRY_ID, entry_id)],
 539                        critical=True)
 540
 541                except Exception as e:
 542                    mlog.warning('%s', e, exc_info=e)
 543                    # try setting entry id to 0 in order to resynchronize
 544                    await self._set_rcb(
 545                        ref, [(iec61850.RcbAttrType.ENTRY_ID, b'\x00')])
 546
 547        elif ref.type == iec61850.RcbType.UNBUFFERED:
 548            await self._set_rcb(ref, [(iec61850.RcbAttrType.RESERVE, True)])
 549
 550        attrs = collections.deque()
 551        if 'trigger_options' in rcb_conf:
 552            attrs.append((iec61850.RcbAttrType.TRIGGER_OPTIONS,
 553                          set(iec61850.TriggerCondition[i]
 554                              for i in rcb_conf['trigger_options'])))
 555        if 'optional_fields' in rcb_conf:
 556            attrs.append((iec61850.RcbAttrType.OPTIONAL_FIELDS,
 557                          set(iec61850.OptionalField[i]
 558                              for i in rcb_conf['optional_fields'])))
 559        if 'buffer_time' in rcb_conf:
 560            attrs.append((iec61850.RcbAttrType.BUFFER_TIME,
 561                          rcb_conf['buffer_time']))
 562        if 'integrity_period' in rcb_conf:
 563            attrs.append((iec61850.RcbAttrType.INTEGRITY_PERIOD,
 564                          rcb_conf['integrity_period']))
 565        if attrs:
 566            await self._set_rcb(ref, attrs)
 567
 568        await self._set_rcb(
 569            ref, [(iec61850.RcbAttrType.REPORT_ENABLE, True)], critical=True)
 570        await self._set_rcb(
 571            ref, [(iec61850.RcbAttrType.GI, True)], critical=True)
 572        mlog.debug('rcb %s initiated', ref)
 573
 574    async def _set_rcb(self, ref, attrs, critical=False):
 575        try:
 576            resp = await self._conn.set_rcb_attrs(ref, attrs)
 577            attrs_failed = set((attr, attr_res)
 578                               for attr, attr_res in resp.items()
 579                               if isinstance(attr_res, iec61850.ServiceError))
 580            if attrs_failed:
 581                raise Exception(f"set attribute errors: {attrs_failed}")
 582
 583        except Exception as e:
 584            if critical:
 585                raise Exception(f'set rcb {ref} failed') from e
 586
 587            else:
 588                mlog.warning('set rcb %s failed: %s', ref, e, exc_info=e)
 589
 590    async def _create_dynamic_datasets(self):
 591        existing_ds_refs = set()
 592        for ds_ref in self._persist_dyn_datasets:
 593            ld = ds_ref.logical_device
 594            res = await self._conn.get_persisted_dataset_refs(ld)
 595            if isinstance(res, iec61850.ServiceError):
 596                raise Exception(f'get datasets for ld {ld} failed: {res}')
 597
 598            existing_ds_refs.update(res)
 599
 600        existing_persisted_ds_refs = existing_ds_refs.intersection(
 601            self._persist_dyn_datasets)
 602        for ds_ref, ds_value_refs in self._dyn_datasets_values.items():
 603            if ds_ref in existing_persisted_ds_refs:
 604                res = await self._conn.get_dataset_data_refs(ds_ref)
 605                if isinstance(res, iec61850.ServiceError):
 606                    raise Exception(f'get ds {ds_ref} data refs failed: {res}')
 607                else:
 608                    exist_ds_value_refs = res
 609
 610                if ds_value_refs == list(exist_ds_value_refs):
 611                    mlog.debug('dataset %s already exists', ds_ref)
 612                    continue
 613
 614                mlog.debug("dataset %s exists, but different", ds_ref)
 615                res = await self._conn.delete_dataset(ds_ref)
 616                if res is not None:
 617                    raise Exception(f'delete dataset {ds_ref} failed: {res}')
 618                mlog.debug("dataset %s deleted", ds_ref)
 619
 620            res = await self._conn.create_dataset(ds_ref, ds_value_refs)
 621            if res is not None:
 622                raise Exception(f'create dataset {ds_ref} failed: {res}')
 623
 624            mlog.debug("dataset %s crated", ds_ref)
 625
 626    async def _register_events(self, events):
 627        try:
 628            await self._eventer_client.register(events)
 629
 630        except ConnectionError:
 631            self.close()
 632            raise
 633
 634    async def _register_status(self, status):
 635        if status == self._conn_status:
 636            return
 637
 638        event = hat.event.common.RegisterEvent(
 639            type=(*self._event_type_prefix, 'gateway', 'status'),
 640            source_timestamp=None,
 641            payload=hat.event.common.EventPayloadJson(status))
 642        await self._register_events([event])
 643        self._conn_status = status
 644        mlog.debug('registered status %s', status)
 645
 646
 647def _value_ref_from_conf(value_ref_conf):
 648    return iec61850.DataRef(
 649        logical_device=value_ref_conf['logical_device'],
 650        logical_node=value_ref_conf['logical_node'],
 651        fc=value_ref_conf['fc'],
 652        names=tuple(value_ref_conf['names']))
 653
 654
 655def _dataset_ref_from_conf(ds_conf):
 656    if isinstance(ds_conf, str):
 657        return iec61850.NonPersistedDatasetRef(ds_conf)
 658
 659    elif isinstance(ds_conf, dict):
 660        return iec61850.PersistedDatasetRef(**ds_conf)
 661
 662
 663def _refs_match(ref1, ref2):
 664    if ref1.logical_device != ref2.logical_device:
 665        return False
 666
 667    if ref1.logical_node != ref2.logical_node:
 668        return False
 669
 670    if ref1.fc != ref2.fc:
 671        return False
 672
 673    if len(ref1.names) == len(ref2.names):
 674        return ref1.names == ref2.names
 675
 676    if len(ref1.names) < len(ref2.names):
 677        names1, names2 = ref1.names, ref2.names
 678    else:
 679        names1, names2 = ref2.names, ref1.names
 680
 681    return names2[:len(names1)] == names1
 682
 683
 684def _get_persist_dyn_datasets(conf):
 685    for ds_conf in conf['datasets']:
 686        if not ds_conf['dynamic']:
 687            continue
 688
 689        ds_ref = _dataset_ref_from_conf(ds_conf['ref'])
 690        if isinstance(ds_ref, iec61850.PersistedDatasetRef):
 691            yield ds_ref
 692
 693
 694def _get_dyn_datasets_values(conf):
 695    for ds_conf in conf['datasets']:
 696        if not ds_conf['dynamic']:
 697            continue
 698
 699        ds_ref = _dataset_ref_from_conf(ds_conf['ref'])
 700        yield ds_ref, [_value_ref_from_conf(val_conf)
 701                       for val_conf in ds_conf['values']]
 702
 703
 704def _vtype_61850_from_vtype_conf(vt_conf):
 705    if isinstance(vt_conf, str):
 706        return _value_type_from_str(vt_conf)
 707
 708    if vt_conf['type'] == 'ARRAY':
 709        return iec61850.ArrayValueType(
 710            type=_vtype_61850_from_vtype_conf(vt_conf['element_type']),
 711            length=vt_conf['length'])
 712
 713    if vt_conf['type'] == 'STRUCT':
 714        return iec61850.StructValueType(
 715            [(el_conf['name'], _vtype_61850_from_vtype_conf(el_conf['type']))
 716             for el_conf in vt_conf['elements']])
 717
 718    raise Exception('unsupported value type')
 719
 720
 721def _vtype_from_vtype_conf(vt_conf):
 722    if isinstance(vt_conf, str):
 723        return _value_type_from_str(vt_conf)
 724
 725    if vt_conf['type'] == 'ARRAY':
 726        return iec61850.ArrayValueType(
 727            type=_vtype_from_vtype_conf(vt_conf['element_type']),
 728            length=vt_conf['length'])
 729
 730    if vt_conf['type'] == 'STRUCT':
 731        return {el_conf['name']: _vtype_from_vtype_conf(el_conf['type'])
 732                for el_conf in vt_conf['elements']}
 733
 734    raise Exception('unsupported value type')
 735
 736
 737def _value_type_from_str(vt_conf):
 738    if vt_conf in ['BOOLEAN',
 739                   'INTEGER',
 740                   'UNSIGNED',
 741                   'FLOAT',
 742                   'BIT_STRING',
 743                   'OCTET_STRING',
 744                   'VISIBLE_STRING',
 745                   'MMS_STRING']:
 746        return iec61850.BasicValueType(vt_conf)
 747
 748    if vt_conf in ['QUALITY',
 749                   'TIMESTAMP',
 750                   'DOUBLE_POINT',
 751                   'DIRECTION',
 752                   'SEVERITY',
 753                   'ANALOGUE',
 754                   'VECTOR',
 755                   'STEP_POSITION',
 756                   'BINARY_CONTROL']:
 757        return iec61850.AcsiValueType(vt_conf)
 758
 759    raise Exception('unsupported value type')
 760
 761
 762def _value_type_from_ref(value_types, ref):
 763    left_names = []
 764    if isinstance(ref, iec61850.CommandRef):
 765        left_names = []
 766        key = (ref.logical_device, ref.logical_node, 'CO', ref.name)
 767
 768    elif isinstance(ref, iec61850.DataRef):
 769        left_names = ref.names[1:]
 770        key = (ref.logical_device, ref.logical_node, ref.fc, ref.names[0])
 771
 772    else:
 773        raise Exception('unexpected reference')
 774
 775    if key not in value_types:
 776        return
 777
 778    value_type = value_types[key]
 779    while left_names:
 780        name = left_names[0]
 781        left_names = left_names[1:]
 782        if isinstance(name, str):
 783            if isinstance(value_type, iec61850.StructValueType):
 784                _, value_type = util.first(
 785                    value_type.elements, lambda i: i[0] == name)
 786            else:
 787                value_type = value_type[name]
 788
 789        if isinstance(name, int):
 790            value_type = value_type.type
 791
 792    return value_type
 793
 794
 795def _get_data_value_types(conf, value_types):
 796    for ds_conf in conf['datasets']:
 797        for val_ref_conf in ds_conf['values']:
 798            value_ref = _value_ref_from_conf(val_ref_conf)
 799            value_type = _value_type_from_ref(value_types, value_ref)
 800            yield value_ref, value_type
 801
 802    for change_conf in conf['changes']:
 803        value_ref = _value_ref_from_conf(change_conf['ref'])
 804        value_type = _value_type_from_ref(value_types, value_ref)
 805        yield value_ref, value_type
 806
 807
 808def _get_cmd_value_types(conf, value_types):
 809    for cmd_conf in conf['commands']:
 810        cmd_ref = iec61850.CommandRef(**cmd_conf['ref'])
 811        value_type = _value_type_from_ref(value_types, cmd_ref)
 812        yield cmd_ref, value_type
 813
 814
 815_epoch_start = datetime.datetime.fromtimestamp(0, datetime.timezone.utc)
 816
 817
 818def _command_from_event(event, cmd_conf, control_number, value_type):
 819    if cmd_conf['with_operate_time']:
 820        operate_time = iec61850.Timestamp(
 821            value=_epoch_start,
 822            leap_second=False,
 823            clock_failure=False,
 824            not_synchronized=False,
 825            accuracy=0)
 826    else:
 827        operate_time = None
 828    return iec61850.Command(
 829        value=_value_from_json(event.payload.data['value'], value_type),
 830        operate_time=operate_time,
 831        origin=iec61850.Origin(
 832            category=iec61850.OriginCategory[
 833                event.payload.data['origin']['category']],
 834            identification=event.payload.data[
 835                'origin']['identification'].encode('utf-8')),
 836        control_number=control_number,
 837        t=_timestamp_from_event_timestamp(event.timestamp),
 838        test=event.payload.data['test'],
 839        checks=set(iec61850.Check[i] for i in event.payload.data['checks']))
 840
 841
 842def _timestamp_from_event_timestamp(timestamp):
 843    return iec61850.Timestamp(
 844        value=hat.event.common.timestamp_to_datetime(timestamp),
 845        leap_second=False,
 846        clock_failure=False,
 847        not_synchronized=False,
 848        accuracy=None)
 849
 850
 851def _cmd_resp_to_event(event_type_prefix, cmd_name, event_session_id, action,
 852                       resp):
 853    success = resp is None
 854    payload = {'session_id': event_session_id,
 855               'action': action,
 856               'success': success}
 857    if not success:
 858        if resp.service_error is not None:
 859            payload['service_error'] = resp.service_error.name
 860        if resp.additional_cause is not None:
 861            payload['additional_cause'] = resp.additional_cause.name
 862        if resp.test_error is not None:
 863            payload['test_error'] = resp.test_error.name
 864
 865    return hat.event.common.RegisterEvent(
 866        type=(*event_type_prefix, 'gateway', 'command', cmd_name),
 867        source_timestamp=None,
 868        payload=hat.event.common.EventPayloadJson(payload))
 869
 870
 871def _write_data_resp_to_event(event_type_prefix, value_name, session_id, resp):
 872    success = resp is None
 873    payload = {'session_id': session_id,
 874               'success': success}
 875    if not success:
 876        payload['error'] = resp.name
 877    return hat.event.common.RegisterEvent(
 878        type=(*event_type_prefix, 'gateway', 'change', value_name),
 879        source_timestamp=None,
 880        payload=hat.event.common.EventPayloadJson(payload))
 881
 882
 883def _get_command_session_id(cmd_ref, cmd):
 884    return (cmd_ref, cmd.control_number)
 885
 886
 887def _value_from_json(event_value, value_type):
 888    if isinstance(value_type, iec61850.BasicValueType):
 889        if value_type == iec61850.BasicValueType.OCTET_STRING:
 890            return bytes.fromhex(event_value)
 891
 892        elif value_type == iec61850.BasicValueType.FLOAT:
 893            return float(event_value)
 894
 895        else:
 896            return event_value
 897
 898    if value_type == iec61850.AcsiValueType.QUALITY:
 899        return iec61850.Quality(
 900            validity=iec61850.QualityValidity[event_value['validity']],
 901            details={iec61850.QualityDetail[i]
 902                     for i in event_value['details']},
 903            source=iec61850.QualitySource[event_value['source']],
 904            test=event_value['test'],
 905            operator_blocked=event_value['operator_blocked'])
 906
 907    if value_type == iec61850.AcsiValueType.TIMESTAMP:
 908        return iec61850.Timestamp(
 909            value=datetime.datetime.fromtimestamp(event_value['value'],
 910                                                  datetime.timezone.utc),
 911            leap_second=event_value['leap_second'],
 912            clock_failure=event_value['clock_failure'],
 913            not_synchronized=event_value['not_synchronized'],
 914            accuracy=event_value.get('accuracy'))
 915
 916    if value_type == iec61850.AcsiValueType.DOUBLE_POINT:
 917        return iec61850.DoublePoint[event_value]
 918
 919    if value_type == iec61850.AcsiValueType.DIRECTION:
 920        return iec61850.Direction[event_value]
 921
 922    if value_type == iec61850.AcsiValueType.SEVERITY:
 923        return iec61850.Severity[event_value]
 924
 925    if value_type == iec61850.AcsiValueType.ANALOGUE:
 926        return iec61850.Analogue(
 927            i=event_value.get('i'),
 928            f=(float(event_value['f']) if 'f' in event_value else None))
 929
 930    if value_type == iec61850.AcsiValueType.VECTOR:
 931        return iec61850.Vector(
 932            magnitude=_value_from_json(event_value['magnitude'],
 933                                       iec61850.AcsiValueType.ANALOGUE),
 934            angle=(_value_from_json(event_value['angle'],
 935                                    iec61850.AcsiValueType.ANALOGUE)
 936                   if 'angle' in event_value else None))
 937
 938    if value_type == iec61850.AcsiValueType.STEP_POSITION:
 939        return iec61850.StepPosition(value=event_value['value'],
 940                                     transient=event_value.get('transient'))
 941
 942    if value_type == iec61850.AcsiValueType.BINARY_CONTROL:
 943        return iec61850.BinaryControl[event_value]
 944
 945    if isinstance(value_type, iec61850.ArrayValueType):
 946        return [_value_from_json(val, value_type.type)
 947                for val in event_value]
 948
 949    if isinstance(value_type, dict):
 950        return {k: _value_from_json(v, value_type[k])
 951                for k, v in event_value.items()}
 952
 953    raise Exception('unsupported value type')
 954
 955
 956def _value_to_json(data_value, value_type):
 957    if isinstance(value_type, iec61850.BasicValueType):
 958        if value_type == iec61850.BasicValueType.OCTET_STRING:
 959            return data_value.hex()
 960
 961        elif value_type == iec61850.BasicValueType.BIT_STRING:
 962            return list(data_value)
 963
 964        elif value_type == iec61850.BasicValueType.FLOAT:
 965            return data_value if math.isfinite(data_value) else str(data_value)
 966
 967        else:
 968            return data_value
 969
 970    if isinstance(value_type, iec61850.AcsiValueType):
 971        if value_type == iec61850.AcsiValueType.QUALITY:
 972            return {'validity': data_value.validity.name,
 973                    'details': [i.name for i in data_value.details],
 974                    'source': data_value.source.name,
 975                    'test': data_value.test,
 976                    'operator_blocked': data_value.operator_blocked}
 977
 978        if value_type == iec61850.AcsiValueType.TIMESTAMP:
 979            val = {'value': data_value.value.timestamp(),
 980                   'leap_second': data_value.leap_second,
 981                   'clock_failure': data_value.clock_failure,
 982                   'not_synchronized': data_value.not_synchronized}
 983            if data_value.accuracy is not None:
 984                val['accuracy'] = data_value.accuracy
 985            return val
 986
 987        if value_type in [iec61850.AcsiValueType.DOUBLE_POINT,
 988                          iec61850.AcsiValueType.DIRECTION,
 989                          iec61850.AcsiValueType.SEVERITY,
 990                          iec61850.AcsiValueType.BINARY_CONTROL]:
 991            return data_value.name
 992
 993        if value_type == iec61850.AcsiValueType.ANALOGUE:
 994            val = {}
 995            if data_value.i is not None:
 996                val['i'] = data_value.i
 997            if data_value.f is not None:
 998                val['f'] = (data_value.f if math.isfinite(data_value.f)
 999                            else str(data_value.f))
1000            return val
1001
1002        if value_type == iec61850.AcsiValueType.VECTOR:
1003            val = {'magnitude': _value_to_json(
1004                        data_value.magnitude, iec61850.AcsiValueType.ANALOGUE)}
1005            if data_value.angle is not None:
1006                val['angle'] = _value_to_json(
1007                    data_value.angle, iec61850.AcsiValueType.ANALOGUE)
1008            return val
1009
1010        if value_type == iec61850.AcsiValueType.STEP_POSITION:
1011            val = {'value': data_value.value}
1012            if data_value.transient is not None:
1013                val['transient'] = data_value.transient
1014            return val
1015
1016    if isinstance(value_type, iec61850.ArrayValueType):
1017        return [_value_to_json(i, value_type.type) for i in data_value]
1018
1019    if isinstance(value_type, dict):
1020        return {
1021            child_name: _value_to_json(child_value, value_type[child_name])
1022            for child_name, child_value in data_value.items()}
1023
1024    raise Exception('unsupported value type')
1025
1026
1027def _update_ctl_num(ctl_num, action, cmd_model):
1028    if action == 'SELECT' or (
1029        action == 'OPERATE' and
1030        cmd_model in ['DIRECT_WITH_NORMAL_SECURITY',
1031                      'DIRECT_WITH_ENHANCED_SECURITY']):
1032        return (ctl_num + 1) % 256
1033
1034    return ctl_num
1035
1036
1037def _validate_get_rcb_response(get_rcb_resp, rcb_conf):
1038    for k, v in get_rcb_resp.items():
1039        if isinstance(v, iec61850.ServiceError):
1040            raise Exception(f"get {k.name} failed: {v}")
1041
1042        if (k == iec61850.RcbAttrType.REPORT_ID and
1043                v != rcb_conf['report_id']):
1044            raise Exception(f"rcb report id {v} different from "
1045                            f"configured {rcb_conf['report_id']}")
1046
1047        if (k == iec61850.RcbAttrType.DATASET and
1048                v != _dataset_ref_from_conf(rcb_conf['dataset'])):
1049            raise Exception(f"rcb dataset {v} different from "
1050                            f"configured {rcb_conf['report_id']}")
1051
1052        if (k == iec61850.RcbAttrType.CONF_REVISION and
1053                v != rcb_conf['conf_revision']):
1054            raise Exception(
1055                f"Conf revision {v} different from "
1056                f"the configuration defined {rcb_conf['conf_revision']}")
1057
1058
1059def _conf_ref_to_path(conf_ref):
1060    return [conf_ref['logical_device'],
1061            conf_ref['logical_node'],
1062            conf_ref['fc'],
1063            *conf_ref['names']]