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