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
 394        except Exception as e:
 395            self._log.warning('report %s ignored: %s',
 396                              report.report_id, e, exc_info=e)
 397            return
 398
 399        if not events:
 400            return
 401
 402        await self._register_events(events)
 403
 404        if self._rcb_type[report.report_id] == 'BUFFERED':
 405            self._rcbs_entry_ids[report.report_id] = report.entry_id
 406
 407    def _events_from_report(self, report):
 408        report_id = report.report_id
 409        if report_id not in self._report_data_refs:
 410            raise Exception(f'unexpected report {report_id}')
 411
 412        segm_id = (report_id, report.sequence_number)
 413        if report.more_segments_follow:
 414            if segm_id in self._reports_segments:
 415                segment_data, timeout_timer = self._reports_segments[segm_id]
 416                timeout_timer.cancel()
 417            else:
 418                segment_data = collections.deque()
 419
 420            segment_data.extend(report.data)
 421            timeout_timer = self._loop.call_later(
 422                report_segments_timeout, self._reports_segments.pop, segm_id)
 423            self._reports_segments[segm_id] = (segment_data, timeout_timer)
 424            return
 425
 426        if segm_id in self._reports_segments:
 427            report_data, timeout_timer = self._reports_segments.pop(segm_id)
 428            timeout_timer.cancel()
 429            report_data.extend(report.data)
 430
 431        else:
 432            report_data = report.data
 433
 434        yield from self._events_from_report_data(report_data, report_id)
 435
 436        if self._rcb_type[report_id] == 'BUFFERED':
 437            yield hat.event.common.RegisterEvent(
 438                type=(*self._event_type_prefix, 'gateway',
 439                      'entry_id', report_id),
 440                source_timestamp=None,
 441                payload=hat.event.common.EventPayloadJson(
 442                    report.entry_id.hex()
 443                    if report.entry_id is not None else None))
 444
 445    def _events_from_report_data(self, report_data, report_id):
 446        data_values_json = collections.defaultdict(dict)
 447        data_reasons = collections.defaultdict(set)
 448        for rv in report_data:
 449            if rv.ref not in self._value_ref_data_names:
 450                continue
 451
 452            value_type = self._dataset_values_ref_type[rv.ref]
 453            if value_type is None:
 454                self._log.warning('report data ignored: unknown value type')
 455                continue
 456
 457            value_json = _value_to_json(rv.value, value_type)
 458            for data_name in self._value_ref_data_names[rv.ref]:
 459                value_path = [rv.ref.logical_device, rv.ref.logical_node,
 460                              rv.ref.fc, *rv.ref.names]
 461                data_values_json[data_name] = json.set_(
 462                    data_values_json[data_name], value_path, value_json)
 463                if rv.reasons:
 464                    data_reasons[data_name].update(
 465                        reason.name for reason in rv.reasons)
 466
 467        for data_name, values_json in data_values_json.items():
 468            payload = {'reasons': list(data_reasons[data_name])}
 469            data_conf = self._data_name_confs[data_name]
 470            value_path = _conf_ref_to_path(data_conf['value'])
 471            value_json = json.get(values_json, value_path)
 472            if value_json is not None:
 473                value_ref = _value_ref_from_json(data_conf['value'])
 474                value_type = self._data_value_types[value_ref]
 475                payload['value'] = _value_json_to_event_json(
 476                    value_json, value_type)
 477
 478            if 'quality' in data_conf:
 479                quality_path = _conf_ref_to_path(data_conf['quality'])
 480                quality_json = json.get(values_json, quality_path)
 481                if quality_json is not None:
 482                    payload['quality'] = quality_json
 483
 484            if 'timestamp' in data_conf:
 485                timestamp_path = _conf_ref_to_path(data_conf['timestamp'])
 486                timestamp_json = json.get(values_json, timestamp_path)
 487                if timestamp_json is not None:
 488                    payload['timestamp'] = timestamp_json
 489
 490            if 'selected' in data_conf:
 491                selected_path = _conf_ref_to_path(data_conf['selected'])
 492                selected_json = json.get(values_json, selected_path)
 493                if selected_json is not None:
 494                    payload['selected'] = selected_json
 495
 496            yield hat.event.common.RegisterEvent(
 497                type=(*self._event_type_prefix, 'gateway',
 498                      'data', data_name),
 499                source_timestamp=None,
 500                payload=hat.event.common.EventPayloadJson(payload))
 501
 502    def _on_termination(self, termination):
 503        cmd_session_id = _get_command_session_id(
 504            termination.ref, termination.cmd)
 505        if cmd_session_id not in self._terminations:
 506            self._log.warning('unexpected termination dropped')
 507            return
 508
 509        term_future = self._terminations[cmd_session_id]
 510        if not term_future.done():
 511            self._terminations[cmd_session_id].set_result(termination)
 512
 513    async def _init_rcb(self, rcb_conf):
 514        ref = iec61850.RcbRef(
 515            logical_device=rcb_conf['ref']['logical_device'],
 516            logical_node=rcb_conf['ref']['logical_node'],
 517            type=iec61850.RcbType[rcb_conf['ref']['type']],
 518            name=rcb_conf['ref']['name'])
 519        self._log.debug('initiating rcb %s', ref)
 520
 521        get_attrs = collections.deque([iec61850.RcbAttrType.REPORT_ID])
 522        dataset_ref = _dataset_ref_from_json(rcb_conf['dataset'])
 523        if dataset_ref not in self._dyn_datasets_values:
 524            get_attrs.append(iec61850.RcbAttrType.DATASET)
 525        if 'conf_revision' in rcb_conf:
 526            get_attrs.append(iec61850.RcbAttrType.CONF_REVISION)
 527
 528        get_rcb_resp = await self._conn.get_rcb_attrs(ref, get_attrs)
 529        _validate_get_rcb_response(get_rcb_resp, rcb_conf)
 530
 531        if ref.type == iec61850.RcbType.BUFFERED:
 532            if 'reservation_time' in rcb_conf:
 533                await self._set_rcb(
 534                    ref, [(iec61850.RcbAttrType.RESERVATION_TIME,
 535                           rcb_conf['reservation_time'])])
 536        elif ref.type == iec61850.RcbType.UNBUFFERED:
 537            await self._set_rcb(ref, [(iec61850.RcbAttrType.RESERVE, True)])
 538        else:
 539            raise Exception('unexpected rcb type')
 540
 541        await self._set_rcb(ref, [(iec61850.RcbAttrType.REPORT_ENABLE, False)])
 542
 543        if dataset_ref in self._dyn_datasets_values:
 544            await self._set_rcb(
 545                ref, [(iec61850.RcbAttrType.DATASET, dataset_ref)],
 546                critical=True)
 547
 548        if ref.type == iec61850.RcbType.BUFFERED:
 549            entry_id = self._rcbs_entry_ids.get(
 550                _report_id_from_rcb_conf(rcb_conf))
 551            if rcb_conf.get('purge_buffer') or entry_id is None:
 552                await self._set_rcb(
 553                    ref, [(iec61850.RcbAttrType.PURGE_BUFFER, True)])
 554
 555            else:
 556                try:
 557                    await self._set_rcb(
 558                        ref, [(iec61850.RcbAttrType.ENTRY_ID, entry_id)],
 559                        critical=True)
 560
 561                except Exception as e:
 562                    self._log.warning('%s', e, exc_info=e)
 563                    # try setting entry id to 0 in order to resynchronize
 564                    await self._set_rcb(
 565                        ref, [(iec61850.RcbAttrType.ENTRY_ID, b'\x00')])
 566
 567        attrs = collections.deque()
 568        if 'trigger_options' in rcb_conf:
 569            attrs.append((iec61850.RcbAttrType.TRIGGER_OPTIONS,
 570                          set(iec61850.TriggerCondition[i]
 571                              for i in rcb_conf['trigger_options'])))
 572        if 'optional_fields' in rcb_conf:
 573            attrs.append((iec61850.RcbAttrType.OPTIONAL_FIELDS,
 574                          set(iec61850.OptionalField[i]
 575                              for i in rcb_conf['optional_fields'])))
 576        if 'buffer_time' in rcb_conf:
 577            attrs.append((iec61850.RcbAttrType.BUFFER_TIME,
 578                          rcb_conf['buffer_time']))
 579        if 'integrity_period' in rcb_conf:
 580            attrs.append((iec61850.RcbAttrType.INTEGRITY_PERIOD,
 581                          rcb_conf['integrity_period']))
 582        if attrs:
 583            await self._set_rcb(ref, attrs)
 584
 585        await self._set_rcb(
 586            ref, [(iec61850.RcbAttrType.REPORT_ENABLE, True)], critical=True)
 587        await self._set_rcb(
 588            ref, [(iec61850.RcbAttrType.GI, True)], critical=True)
 589        self._log.debug('rcb %s initiated', ref)
 590
 591    async def _set_rcb(self, ref, attrs, critical=False):
 592        try:
 593            resp = await self._conn.set_rcb_attrs(ref, attrs)
 594            attrs_failed = set((attr, attr_res)
 595                               for attr, attr_res in resp.items()
 596                               if isinstance(attr_res, iec61850.ServiceError))
 597            if attrs_failed:
 598                raise Exception(f"set attribute errors: {attrs_failed}")
 599
 600        except Exception as e:
 601            if critical:
 602                raise Exception(f'set rcb {ref} failed') from e
 603
 604            else:
 605                self._log.warning('set rcb %s failed: %s', ref, e, exc_info=e)
 606
 607    async def _create_dynamic_datasets(self):
 608        existing_ds_refs = set()
 609        for ds_ref in self._persist_dyn_datasets:
 610            ld = ds_ref.logical_device
 611            res = await self._conn.get_persisted_dataset_refs(ld)
 612            if isinstance(res, iec61850.ServiceError):
 613                raise Exception(f'get datasets for ld {ld} failed: {res}')
 614
 615            existing_ds_refs.update(res)
 616
 617        existing_persisted_ds_refs = existing_ds_refs.intersection(
 618            self._persist_dyn_datasets)
 619        for ds_ref, ds_value_refs in self._dyn_datasets_values.items():
 620            if ds_ref in existing_persisted_ds_refs:
 621                res = await self._conn.get_dataset_data_refs(ds_ref)
 622                if isinstance(res, iec61850.ServiceError):
 623                    raise Exception(f'get ds {ds_ref} data refs failed: {res}')
 624                else:
 625                    exist_ds_value_refs = res
 626
 627                if ds_value_refs == list(exist_ds_value_refs):
 628                    self._log.debug('dataset %s already exists', ds_ref)
 629                    continue
 630
 631                raise Exception('persisted dataset changed')
 632
 633            res = await self._conn.create_dataset(ds_ref, ds_value_refs)
 634            if res is not None:
 635                raise Exception(f'create dataset {ds_ref} failed: {res}')
 636
 637            self._log.debug("dataset %s crated", ds_ref)
 638
 639    async def _register_events(self, events):
 640        try:
 641            await self._eventer_client.register(events)
 642
 643        except ConnectionError:
 644            self._log.debug('eventer connection closed')
 645            self.close()
 646
 647    async def _register_status(self, status):
 648        if status == self._conn_status:
 649            return
 650
 651        event = hat.event.common.RegisterEvent(
 652            type=(*self._event_type_prefix, 'gateway', 'status'),
 653            source_timestamp=None,
 654            payload=hat.event.common.EventPayloadJson(status))
 655        await self._register_events([event])
 656        self._conn_status = status
 657        self._log.debug('registered status %s', status)
 658
 659
 660def _value_ref_from_json(value_ref_conf):
 661    return iec61850.DataRef(
 662        logical_device=value_ref_conf['logical_device'],
 663        logical_node=value_ref_conf['logical_node'],
 664        fc=value_ref_conf['fc'],
 665        names=tuple(value_ref_conf['names']))
 666
 667
 668def _dataset_ref_from_json(ds_conf):
 669    if isinstance(ds_conf, str):
 670        return iec61850.NonPersistedDatasetRef(ds_conf)
 671
 672    elif isinstance(ds_conf, dict):
 673        return iec61850.PersistedDatasetRef(**ds_conf)
 674
 675
 676def _rcb_ref_from_json(rcb_ref_conf):
 677    return iec61850.RcbRef(
 678        logical_device=rcb_ref_conf['logical_device'],
 679        logical_node=rcb_ref_conf['logical_node'],
 680        type=iec61850.RcbType[rcb_ref_conf['type']],
 681        name=rcb_ref_conf['name'])
 682
 683
 684def _refs_match(ref1, ref2):
 685    if ref1.logical_device != ref2.logical_device:
 686        return False
 687
 688    if ref1.logical_node != ref2.logical_node:
 689        return False
 690
 691    if ref1.fc != ref2.fc:
 692        return False
 693
 694    if len(ref1.names) == len(ref2.names):
 695        return ref1.names == ref2.names
 696
 697    if len(ref1.names) < len(ref2.names):
 698        names1, names2 = ref1.names, ref2.names
 699    else:
 700        names1, names2 = ref2.names, ref1.names
 701
 702    return names2[:len(names1)] == names1
 703
 704
 705def _get_persist_dyn_datasets(conf):
 706    for ds_conf in conf['datasets']:
 707        if not ds_conf['dynamic']:
 708            continue
 709
 710        ds_ref = _dataset_ref_from_json(ds_conf['ref'])
 711        if isinstance(ds_ref, iec61850.PersistedDatasetRef):
 712            yield ds_ref
 713
 714
 715def _get_dyn_datasets_values(conf):
 716    for ds_conf in conf['datasets']:
 717        if not ds_conf['dynamic']:
 718            continue
 719
 720        ds_ref = _dataset_ref_from_json(ds_conf['ref'])
 721        yield ds_ref, [_value_ref_from_json(val_conf)
 722                       for val_conf in ds_conf['values']]
 723
 724
 725def _vtype_61850_from_vtype_conf(vt_conf):
 726    if isinstance(vt_conf, str):
 727        return _value_type_from_str(vt_conf)
 728
 729    if vt_conf['type'] == 'ARRAY':
 730        return iec61850.ArrayValueType(
 731            type=_vtype_61850_from_vtype_conf(vt_conf['element_type']),
 732            length=vt_conf['length'])
 733
 734    if vt_conf['type'] == 'STRUCT':
 735        return iec61850.StructValueType(
 736            [(el_conf['name'], _vtype_61850_from_vtype_conf(el_conf['type']))
 737             for el_conf in vt_conf['elements']])
 738
 739    raise Exception('unsupported value type')
 740
 741
 742def _vtype_from_vtype_conf(vt_conf):
 743    if isinstance(vt_conf, str):
 744        return _value_type_from_str(vt_conf)
 745
 746    if vt_conf['type'] == 'ARRAY':
 747        return iec61850.ArrayValueType(
 748            type=_vtype_from_vtype_conf(vt_conf['element_type']),
 749            length=vt_conf['length'])
 750
 751    if vt_conf['type'] == 'STRUCT':
 752        return {el_conf['name']: _vtype_from_vtype_conf(el_conf['type'])
 753                for el_conf in vt_conf['elements']}
 754
 755    raise Exception('unsupported value type')
 756
 757
 758def _value_type_from_str(vt_conf):
 759    if vt_conf in ['BOOLEAN',
 760                   'INTEGER',
 761                   'UNSIGNED',
 762                   'FLOAT',
 763                   'BIT_STRING',
 764                   'OCTET_STRING',
 765                   'VISIBLE_STRING',
 766                   'MMS_STRING']:
 767        return iec61850.BasicValueType(vt_conf)
 768
 769    if vt_conf in ['QUALITY',
 770                   'TIMESTAMP',
 771                   'DOUBLE_POINT',
 772                   'DIRECTION',
 773                   'SEVERITY',
 774                   'ANALOGUE',
 775                   'VECTOR',
 776                   'STEP_POSITION',
 777                   'BINARY_CONTROL']:
 778        return iec61850.AcsiValueType(vt_conf)
 779
 780    raise Exception('unsupported value type')
 781
 782
 783def _value_type_from_ref(value_types, ref):
 784    left_names = []
 785    if isinstance(ref, iec61850.CommandRef):
 786        left_names = ['Oper', 'ctlVal']
 787        key = (ref.logical_device, ref.logical_node, 'CO', ref.name)
 788
 789    elif isinstance(ref, iec61850.DataRef):
 790        left_names = ref.names[1:]
 791        key = (ref.logical_device, ref.logical_node, ref.fc, ref.names[0])
 792
 793    else:
 794        raise Exception('unexpected reference')
 795
 796    if key not in value_types:
 797        return
 798
 799    value_type = value_types[key]
 800    while left_names:
 801        name = left_names[0]
 802        left_names = left_names[1:]
 803        if isinstance(name, str):
 804            if isinstance(value_type, iec61850.StructValueType):
 805                _, value_type = util.first(
 806                    value_type.elements, lambda i: i[0] == name)
 807
 808            elif value_type == iec61850.AcsiValueType.ANALOGUE:
 809                if name == 'i':
 810                    value_type = iec61850.BasicValueType.INTEGER
 811                elif name == 'f':
 812                    value_type = iec61850.BasicValueType.FLOAT
 813                else:
 814                    raise Exception("analogue attribute should be 'i' or 'f'")
 815
 816            elif value_type == iec61850.AcsiValueType.VECTOR:
 817                if name in ('mag', 'ang'):
 818                    value_type = iec61850.AcsiValueType.ANALOGUE
 819                else:
 820                    raise Exception(
 821                        "vector attribute should be 'mag' or 'ang'")
 822
 823            elif value_type == iec61850.AcsiValueType.STEP_POSITION:
 824                if name == 'posVal':
 825                    value_type = iec61850.BasicValueType.INTEGER
 826                elif name == 'transInd':
 827                    value_type = iec61850.BasicValueType.BOOLEAN
 828                else:
 829                    raise Exception("step position attribute should be "
 830                                    "'posVal' or 'transInd'")
 831
 832            else:
 833                value_type = value_type[name]
 834
 835        if isinstance(name, int):
 836            value_type = value_type.type
 837
 838    return value_type
 839
 840
 841def _get_dataset_change_value_types(conf, value_types):
 842    for ds_conf in conf['datasets']:
 843        for val_ref_conf in ds_conf['values']:
 844            value_ref = _value_ref_from_json(val_ref_conf)
 845            value_type = _value_type_from_ref(value_types, value_ref)
 846            yield value_ref, value_type
 847
 848    for change_conf in conf['changes']:
 849        value_ref = _value_ref_from_json(change_conf['ref'])
 850        value_type = _value_type_from_ref(value_types, value_ref)
 851        yield value_ref, value_type
 852
 853
 854def _get_cmd_value_types(conf, value_types):
 855    for cmd_conf in conf['commands']:
 856        cmd_ref = iec61850.CommandRef(**cmd_conf['ref'])
 857        value_type = _value_type_from_ref(value_types, cmd_ref)
 858        yield cmd_ref, value_type
 859
 860
 861_epoch_start = datetime.datetime.fromtimestamp(0, datetime.timezone.utc)
 862
 863
 864def _command_from_event(event, cmd_conf, control_number, value_type):
 865    if cmd_conf['with_operate_time']:
 866        operate_time = iec61850.Timestamp(
 867            value=_epoch_start,
 868            leap_second=False,
 869            clock_failure=False,
 870            not_synchronized=False,
 871            accuracy=0)
 872    else:
 873        operate_time = None
 874    return iec61850.Command(
 875        value=_value_from_json(event.payload.data['value'], value_type),
 876        operate_time=operate_time,
 877        origin=iec61850.Origin(
 878            category=iec61850.OriginCategory[
 879                event.payload.data['origin']['category']],
 880            identification=event.payload.data[
 881                'origin']['identification'].encode('utf-8')),
 882        control_number=control_number,
 883        t=_timestamp_from_event_timestamp(event.timestamp),
 884        test=event.payload.data['test'],
 885        checks=set(iec61850.Check[i] for i in event.payload.data['checks']))
 886
 887
 888def _timestamp_from_event_timestamp(timestamp):
 889    return iec61850.Timestamp(
 890        value=hat.event.common.timestamp_to_datetime(timestamp),
 891        leap_second=False,
 892        clock_failure=False,
 893        not_synchronized=False,
 894        accuracy=None)
 895
 896
 897def _cmd_resp_to_event(event_type_prefix, cmd_name, event_session_id, action,
 898                       resp):
 899    success = resp is None
 900    payload = {'session_id': event_session_id,
 901               'action': action,
 902               'success': success}
 903    if not success:
 904        if resp.service_error is not None:
 905            payload['service_error'] = resp.service_error.name
 906        if resp.additional_cause is not None:
 907            payload['additional_cause'] = resp.additional_cause.name
 908        if resp.test_error is not None:
 909            payload['test_error'] = resp.test_error.name
 910
 911    return hat.event.common.RegisterEvent(
 912        type=(*event_type_prefix, 'gateway', 'command', cmd_name),
 913        source_timestamp=None,
 914        payload=hat.event.common.EventPayloadJson(payload))
 915
 916
 917def _write_data_resp_to_event(event_type_prefix, value_name, session_id, resp):
 918    success = resp is None
 919    payload = {'session_id': session_id,
 920               'success': success}
 921    if not success:
 922        payload['error'] = resp.name
 923    return hat.event.common.RegisterEvent(
 924        type=(*event_type_prefix, 'gateway', 'change', value_name),
 925        source_timestamp=None,
 926        payload=hat.event.common.EventPayloadJson(payload))
 927
 928
 929def _get_command_session_id(cmd_ref, cmd):
 930    return (cmd_ref, cmd.control_number)
 931
 932
 933def _value_from_json(event_value, value_type):
 934    if isinstance(value_type, iec61850.BasicValueType):
 935        if value_type == iec61850.BasicValueType.OCTET_STRING:
 936            return bytes.fromhex(event_value)
 937
 938        elif value_type == iec61850.BasicValueType.FLOAT:
 939            return float(event_value)
 940
 941        else:
 942            return event_value
 943
 944    if value_type == iec61850.AcsiValueType.QUALITY:
 945        return iec61850.Quality(
 946            validity=iec61850.QualityValidity[event_value['validity']],
 947            details={iec61850.QualityDetail[i]
 948                     for i in event_value['details']},
 949            source=iec61850.QualitySource[event_value['source']],
 950            test=event_value['test'],
 951            operator_blocked=event_value['operator_blocked'])
 952
 953    if value_type == iec61850.AcsiValueType.TIMESTAMP:
 954        return iec61850.Timestamp(
 955            value=datetime.datetime.fromtimestamp(event_value['value'],
 956                                                  datetime.timezone.utc),
 957            leap_second=event_value['leap_second'],
 958            clock_failure=event_value['clock_failure'],
 959            not_synchronized=event_value['not_synchronized'],
 960            accuracy=event_value.get('accuracy'))
 961
 962    if value_type == iec61850.AcsiValueType.DOUBLE_POINT:
 963        return iec61850.DoublePoint[event_value]
 964
 965    if value_type == iec61850.AcsiValueType.DIRECTION:
 966        return iec61850.Direction[event_value]
 967
 968    if value_type == iec61850.AcsiValueType.SEVERITY:
 969        return iec61850.Severity[event_value]
 970
 971    if value_type == iec61850.AcsiValueType.ANALOGUE:
 972        return iec61850.Analogue(
 973            i=event_value.get('i'),
 974            f=(float(event_value['f']) if 'f' in event_value else None))
 975
 976    if value_type == iec61850.AcsiValueType.VECTOR:
 977        return iec61850.Vector(
 978            magnitude=_value_from_json(event_value['magnitude'],
 979                                       iec61850.AcsiValueType.ANALOGUE),
 980            angle=(_value_from_json(event_value['angle'],
 981                                    iec61850.AcsiValueType.ANALOGUE)
 982                   if 'angle' in event_value else None))
 983
 984    if value_type == iec61850.AcsiValueType.STEP_POSITION:
 985        return iec61850.StepPosition(value=event_value['value'],
 986                                     transient=event_value.get('transient'))
 987
 988    if value_type == iec61850.AcsiValueType.BINARY_CONTROL:
 989        return iec61850.BinaryControl[event_value]
 990
 991    if isinstance(value_type, iec61850.ArrayValueType):
 992        return [_value_from_json(val, value_type.type)
 993                for val in event_value]
 994
 995    if isinstance(value_type, dict):
 996        return {k: _value_from_json(v, value_type[k])
 997                for k, v in event_value.items()}
 998
 999    raise Exception('unsupported value type')
1000
1001
1002def _value_to_json(data_value, value_type):
1003    if isinstance(value_type, iec61850.BasicValueType):
1004        if value_type == iec61850.BasicValueType.OCTET_STRING:
1005            return data_value.hex()
1006
1007        elif value_type == iec61850.BasicValueType.BIT_STRING:
1008            return list(data_value)
1009
1010        elif value_type == iec61850.BasicValueType.FLOAT:
1011            return data_value if math.isfinite(data_value) else str(data_value)
1012
1013        else:
1014            return data_value
1015
1016    if isinstance(value_type, iec61850.AcsiValueType):
1017        if value_type == iec61850.AcsiValueType.QUALITY:
1018            return {'validity': data_value.validity.name,
1019                    'details': [i.name for i in data_value.details],
1020                    'source': data_value.source.name,
1021                    'test': data_value.test,
1022                    'operator_blocked': data_value.operator_blocked}
1023
1024        if value_type == iec61850.AcsiValueType.TIMESTAMP:
1025            val = {'value': data_value.value.timestamp(),
1026                   'leap_second': data_value.leap_second,
1027                   'clock_failure': data_value.clock_failure,
1028                   'not_synchronized': data_value.not_synchronized}
1029            if data_value.accuracy is not None:
1030                val['accuracy'] = data_value.accuracy
1031            return val
1032
1033        if value_type in [iec61850.AcsiValueType.DOUBLE_POINT,
1034                          iec61850.AcsiValueType.DIRECTION,
1035                          iec61850.AcsiValueType.SEVERITY,
1036                          iec61850.AcsiValueType.BINARY_CONTROL]:
1037            return data_value.name
1038
1039        if value_type == iec61850.AcsiValueType.ANALOGUE:
1040            val = {}
1041            if data_value.i is not None:
1042                val['i'] = data_value.i
1043            if data_value.f is not None:
1044                val['f'] = (data_value.f if math.isfinite(data_value.f)
1045                            else str(data_value.f))
1046            return val
1047
1048        if value_type == iec61850.AcsiValueType.VECTOR:
1049            val = {'mag': _value_to_json(
1050                        data_value.magnitude, iec61850.AcsiValueType.ANALOGUE)}
1051            if data_value.angle is not None:
1052                val['ang'] = _value_to_json(
1053                    data_value.angle, iec61850.AcsiValueType.ANALOGUE)
1054            return val
1055
1056        if value_type == iec61850.AcsiValueType.STEP_POSITION:
1057            val = {'posVal': data_value.value}
1058            if data_value.transient is not None:
1059                val['transInd'] = data_value.transient
1060            return val
1061
1062    if isinstance(value_type, iec61850.ArrayValueType):
1063        return [_value_to_json(i, value_type.type) for i in data_value]
1064
1065    if isinstance(value_type, dict):
1066        return {
1067            child_name: _value_to_json(child_value, value_type[child_name])
1068            for child_name, child_value in data_value.items()}
1069
1070    raise Exception('unsupported value type')
1071
1072
1073def _value_json_to_event_json(data_value, value_type):
1074    if value_type == iec61850.AcsiValueType.VECTOR:
1075        val = {'magnitude': data_value['mag']}
1076        if 'ang' in data_value:
1077            val['angle'] = data_value['ang']
1078        return val
1079
1080    if value_type == iec61850.AcsiValueType.STEP_POSITION:
1081        val = {'value': data_value['posVal']}
1082        if 'transInd' in data_value:
1083            val['transient'] = data_value['transInd']
1084        return val
1085
1086    if isinstance(value_type, iec61850.ArrayValueType):
1087        return [_value_json_to_event_json(i, value_type.type)
1088                for i in data_value]
1089
1090    if isinstance(value_type, dict):
1091        return {
1092            child_name: _value_json_to_event_json(
1093                child_value, value_type[child_name])
1094            for child_name, child_value in data_value.items()}
1095
1096    return data_value
1097
1098
1099def _update_ctl_num(ctl_num, action, cmd_model):
1100    if action == 'SELECT' or (
1101        action == 'OPERATE' and
1102        cmd_model in ['DIRECT_WITH_NORMAL_SECURITY',
1103                      'DIRECT_WITH_ENHANCED_SECURITY']):
1104        return (ctl_num + 1) % 256
1105
1106    return ctl_num
1107
1108
1109def _validate_get_rcb_response(get_rcb_resp, rcb_conf):
1110    for k, v in get_rcb_resp.items():
1111        if isinstance(v, iec61850.ServiceError):
1112            raise Exception(f"get {k.name} failed: {v}")
1113
1114        if (k == iec61850.RcbAttrType.REPORT_ID and
1115                v != rcb_conf['report_id']):
1116            raise Exception(f"rcb report id {v} different from "
1117                            f"configured {rcb_conf['report_id']}")
1118
1119        if (k == iec61850.RcbAttrType.DATASET and
1120                v != _dataset_ref_from_json(rcb_conf['dataset'])):
1121            raise Exception(f"rcb dataset {v} different from "
1122                            f"configured {rcb_conf['dataset']}")
1123
1124        if (k == iec61850.RcbAttrType.CONF_REVISION and
1125                v != rcb_conf['conf_revision']):
1126            raise Exception(
1127                f"Conf revision {v} different from "
1128                f"the configuration defined {rcb_conf['conf_revision']}")
1129
1130
1131def _conf_ref_to_path(conf_ref):
1132    return [conf_ref['logical_device'],
1133            conf_ref['logical_node'],
1134            conf_ref['fc'],
1135            *conf_ref['names']]
1136
1137
1138def _report_id_from_rcb_conf(rcb_conf):
1139    return (rcb_conf['report_id'] if rcb_conf['report_id'] else
1140            _report_id_from_rcb_ref(rcb_conf['ref']))
1141
1142
1143def _report_id_from_rcb_ref(rcb_ref):
1144    return (f"{rcb_ref['logical_device']}/"
1145            f"{rcb_ref['logical_node']}$"
1146            f"{iec61850.RcbType[rcb_ref['type']].value}$"
1147            f"{rcb_ref['name']}")
1148
1149
1150def _create_logger_adapter(name):
1151    extra = {'meta': {'type': 'Iec61850ClientDevice',
1152                      'name': name}}
1153
1154    return logging.LoggerAdapter(mlog, extra)