hat.gateway.devices.iec61850.client

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