hat.gateway.devices.iec61850.client

   1"""IEC 61850 client device"""
   2
   3from collections.abc import Collection
   4import asyncio
   5import collections
   6import datetime
   7import logging
   8import math
   9
  10from hat import aio
  11from hat import json
  12from hat import util
  13from hat.drivers import iec61850
  14from hat.drivers import tcp
  15import hat.event.common
  16
  17from hat.gateway import common
  18
  19
  20mlog: logging.Logger = logging.getLogger(__name__)
  21
  22
  23termination_timeout: int = 100
  24
  25report_segments_timeout: int = 100
  26
  27
  28async def create(conf: common.DeviceConf,
  29                 eventer_client: hat.event.eventer.Client,
  30                 event_type_prefix: common.EventTypePrefix
  31                 ) -> 'Iec61850ClientDevice':
  32
  33    value_types = {}
  34    value_types_61850 = {}
  35    for vt in conf['value_types']:
  36        ref = (vt['logical_device'],
  37               vt['logical_node'],
  38               vt['fc'],
  39               vt['name'])
  40        value_types_61850[ref] = _vtype_61850_from_vtype_conf(vt['type'])
  41        value_types[ref] = _vtype_from_vtype_conf(vt['type'])
  42
  43    rcb_confs = {i['report_id']: i for i in conf['rcbs']}
  44    dataset_confs = {_dataset_ref_from_conf(ds_conf['ref']): ds_conf
  45                     for ds_conf in conf['datasets']}
  46    value_ref_data_names = collections.defaultdict(collections.deque)
  47    data_value_types = {}
  48    for data_conf in conf['data']:
  49        data_v_ref = _value_ref_from_conf(data_conf['value'])
  50        data_value_types[data_v_ref] = _value_type_from_ref(
  51            value_types, data_v_ref)
  52
  53        q_ref = (_value_ref_from_conf(data_conf['quality'])
  54                 if data_conf.get('quality') else None)
  55        if data_conf.get('quality'):
  56            q_type = _value_type_from_ref(value_types, q_ref)
  57            if q_type != iec61850.AcsiValueType.QUALITY:
  58                raise Exception(f"invalid quality type {q_ref}")
  59
  60        t_ref = (_value_ref_from_conf(data_conf['timestamp'])
  61                 if data_conf.get('timestamp') else None)
  62        if t_ref:
  63            t_type = _value_type_from_ref(value_types, t_ref)
  64            if t_type != iec61850.AcsiValueType.TIMESTAMP:
  65                raise Exception(f"invalid timestamp type {t_ref}")
  66
  67        seld_ref = (_value_ref_from_conf(data_conf['selected'])
  68                    if data_conf.get('selected') else None)
  69        if seld_ref:
  70            seld_type = _value_type_from_ref(value_types, seld_ref)
  71            if seld_type != iec61850.BasicValueType.BOOLEAN:
  72                raise Exception(f"invalid selected type {seld_ref}")
  73
  74        rcb_conf = rcb_confs[data_conf['report_id']]
  75        ds_ref = _dataset_ref_from_conf(rcb_conf['dataset'])
  76        ds_conf = dataset_confs[ds_ref]
  77        for value_conf in ds_conf['values']:
  78            value_ref = _value_ref_from_conf(value_conf)
  79            if (_refs_match(data_v_ref, value_ref) or
  80                    (q_ref and _refs_match(q_ref, value_ref)) or
  81                    (t_ref and _refs_match(t_ref, value_ref)) or
  82                    (seld_ref and _refs_match(seld_ref, value_ref))):
  83                value_ref_data_names[value_ref].append(data_conf['name'])
  84
  85    dataset_values_ref_type = {}
  86    for ds_conf in conf['datasets']:
  87        for val_ref_conf in ds_conf['values']:
  88            value_ref = _value_ref_from_conf(val_ref_conf)
  89            value_type = _value_type_from_ref(value_types, value_ref)
  90            dataset_values_ref_type[value_ref] = value_type
  91
  92    command_ref_value_type = {}
  93    for cmd_conf in conf['commands']:
  94        cmd_ref = iec61850.CommandRef(**cmd_conf['ref'])
  95        value_type = _value_type_from_ref(value_types, cmd_ref)
  96        command_ref_value_type[cmd_ref] = value_type
  97
  98    change_ref_value_type = {}
  99    for change_conf in conf['changes']:
 100        value_ref = _value_ref_from_conf(change_conf['ref'])
 101        value_type = _value_type_from_ref(value_types, value_ref)
 102        change_ref_value_type[value_ref] = value_type
 103
 104    entry_id_event_types = [
 105        (*event_type_prefix, 'gateway', 'entry_id', i['report_id'])
 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_conf(i['ref']) for i in conf['changes']}
 138    device._rcb_type = {rcb_conf['report_id']: rcb_conf['ref']['type']
 139                        for rcb_conf in conf['rcbs']}
 140    device._persist_dyn_datasets = set(_get_persist_dyn_datasets(conf))
 141    device._dyn_datasets_values = dict(_get_dyn_datasets_values(conf))
 142    device._report_data_refs = collections.defaultdict(collections.deque)
 143    for rcb_conf in device._conf['rcbs']:
 144        report_id = rcb_conf['report_id']
 145        ds_ref = _dataset_ref_from_conf(rcb_conf['dataset'])
 146        for value_conf in dataset_confs[ds_ref]['values']:
 147            value_ref = _value_ref_from_conf(value_conf)
 148            device._report_data_refs[report_id].append(value_ref)
 149
 150    device._dataset_change_value_types = dict(
 151        _get_dataset_change_value_types(conf, value_types_61850))
 152    device._cmd_value_types = dict(
 153        _get_cmd_value_types(conf, value_types_61850))
 154
 155    device._log = _create_logger_adapter(conf['name'])
 156
 157    device._async_group = aio.Group()
 158    device._async_group.spawn(device._connection_loop)
 159    device._loop = asyncio.get_running_loop()
 160
 161    return device
 162
 163
 164info: common.DeviceInfo = common.DeviceInfo(
 165    type="iec61850_client",
 166    create=create,
 167    json_schema_id="hat-gateway://iec61850.yaml#/$defs/client",
 168    json_schema_repo=common.json_schema_repo)
 169
 170
 171class Iec61850ClientDevice(common.Device):
 172
 173    @property
 174    def async_group(self) -> aio.Group:
 175        return self._async_group
 176
 177    async def process_events(self, events: Collection[hat.event.common.Event]):
 178        try:
 179            for event in events:
 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_conf(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_conf(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(rcb_conf['report_id'])
 547            if rcb_conf.get('purge_buffer') or entry_id is None:
 548                await self._set_rcb(
 549                    ref, [(iec61850.RcbAttrType.PURGE_BUFFER, True)])
 550
 551            else:
 552                try:
 553                    await self._set_rcb(
 554                        ref, [(iec61850.RcbAttrType.ENTRY_ID, entry_id)],
 555                        critical=True)
 556
 557                except Exception as e:
 558                    self._log.warning('%s', e, exc_info=e)
 559                    # try setting entry id to 0 in order to resynchronize
 560                    await self._set_rcb(
 561                        ref, [(iec61850.RcbAttrType.ENTRY_ID, b'\x00')])
 562
 563        attrs = collections.deque()
 564        if 'trigger_options' in rcb_conf:
 565            attrs.append((iec61850.RcbAttrType.TRIGGER_OPTIONS,
 566                          set(iec61850.TriggerCondition[i]
 567                              for i in rcb_conf['trigger_options'])))
 568        if 'optional_fields' in rcb_conf:
 569            attrs.append((iec61850.RcbAttrType.OPTIONAL_FIELDS,
 570                          set(iec61850.OptionalField[i]
 571                              for i in rcb_conf['optional_fields'])))
 572        if 'buffer_time' in rcb_conf:
 573            attrs.append((iec61850.RcbAttrType.BUFFER_TIME,
 574                          rcb_conf['buffer_time']))
 575        if 'integrity_period' in rcb_conf:
 576            attrs.append((iec61850.RcbAttrType.INTEGRITY_PERIOD,
 577                          rcb_conf['integrity_period']))
 578        if attrs:
 579            await self._set_rcb(ref, attrs)
 580
 581        await self._set_rcb(
 582            ref, [(iec61850.RcbAttrType.REPORT_ENABLE, True)], critical=True)
 583        await self._set_rcb(
 584            ref, [(iec61850.RcbAttrType.GI, True)], critical=True)
 585        self._log.debug('rcb %s initiated', ref)
 586
 587    async def _set_rcb(self, ref, attrs, critical=False):
 588        try:
 589            resp = await self._conn.set_rcb_attrs(ref, attrs)
 590            attrs_failed = set((attr, attr_res)
 591                               for attr, attr_res in resp.items()
 592                               if isinstance(attr_res, iec61850.ServiceError))
 593            if attrs_failed:
 594                raise Exception(f"set attribute errors: {attrs_failed}")
 595
 596        except Exception as e:
 597            if critical:
 598                raise Exception(f'set rcb {ref} failed') from e
 599
 600            else:
 601                self._log.warning('set rcb %s failed: %s', ref, e, exc_info=e)
 602
 603    async def _create_dynamic_datasets(self):
 604        existing_ds_refs = set()
 605        for ds_ref in self._persist_dyn_datasets:
 606            ld = ds_ref.logical_device
 607            res = await self._conn.get_persisted_dataset_refs(ld)
 608            if isinstance(res, iec61850.ServiceError):
 609                raise Exception(f'get datasets for ld {ld} failed: {res}')
 610
 611            existing_ds_refs.update(res)
 612
 613        existing_persisted_ds_refs = existing_ds_refs.intersection(
 614            self._persist_dyn_datasets)
 615        for ds_ref, ds_value_refs in self._dyn_datasets_values.items():
 616            if ds_ref in existing_persisted_ds_refs:
 617                res = await self._conn.get_dataset_data_refs(ds_ref)
 618                if isinstance(res, iec61850.ServiceError):
 619                    raise Exception(f'get ds {ds_ref} data refs failed: {res}')
 620                else:
 621                    exist_ds_value_refs = res
 622
 623                if ds_value_refs == list(exist_ds_value_refs):
 624                    self._log.debug('dataset %s already exists', ds_ref)
 625                    continue
 626
 627                raise Exception('persisted dataset changed')
 628
 629            res = await self._conn.create_dataset(ds_ref, ds_value_refs)
 630            if res is not None:
 631                raise Exception(f'create dataset {ds_ref} failed: {res}')
 632
 633            self._log.debug("dataset %s crated", ds_ref)
 634
 635    async def _register_events(self, events):
 636        try:
 637            await self._eventer_client.register(events)
 638
 639        except ConnectionError:
 640            self.close()
 641            raise
 642
 643    async def _register_status(self, status):
 644        if status == self._conn_status:
 645            return
 646
 647        event = hat.event.common.RegisterEvent(
 648            type=(*self._event_type_prefix, 'gateway', 'status'),
 649            source_timestamp=None,
 650            payload=hat.event.common.EventPayloadJson(status))
 651        await self._register_events([event])
 652        self._conn_status = status
 653        self._log.debug('registered status %s', status)
 654
 655
 656def _value_ref_from_conf(value_ref_conf):
 657    return iec61850.DataRef(
 658        logical_device=value_ref_conf['logical_device'],
 659        logical_node=value_ref_conf['logical_node'],
 660        fc=value_ref_conf['fc'],
 661        names=tuple(value_ref_conf['names']))
 662
 663
 664def _dataset_ref_from_conf(ds_conf):
 665    if isinstance(ds_conf, str):
 666        return iec61850.NonPersistedDatasetRef(ds_conf)
 667
 668    elif isinstance(ds_conf, dict):
 669        return iec61850.PersistedDatasetRef(**ds_conf)
 670
 671
 672def _refs_match(ref1, ref2):
 673    if ref1.logical_device != ref2.logical_device:
 674        return False
 675
 676    if ref1.logical_node != ref2.logical_node:
 677        return False
 678
 679    if ref1.fc != ref2.fc:
 680        return False
 681
 682    if len(ref1.names) == len(ref2.names):
 683        return ref1.names == ref2.names
 684
 685    if len(ref1.names) < len(ref2.names):
 686        names1, names2 = ref1.names, ref2.names
 687    else:
 688        names1, names2 = ref2.names, ref1.names
 689
 690    return names2[:len(names1)] == names1
 691
 692
 693def _get_persist_dyn_datasets(conf):
 694    for ds_conf in conf['datasets']:
 695        if not ds_conf['dynamic']:
 696            continue
 697
 698        ds_ref = _dataset_ref_from_conf(ds_conf['ref'])
 699        if isinstance(ds_ref, iec61850.PersistedDatasetRef):
 700            yield ds_ref
 701
 702
 703def _get_dyn_datasets_values(conf):
 704    for ds_conf in conf['datasets']:
 705        if not ds_conf['dynamic']:
 706            continue
 707
 708        ds_ref = _dataset_ref_from_conf(ds_conf['ref'])
 709        yield ds_ref, [_value_ref_from_conf(val_conf)
 710                       for val_conf in ds_conf['values']]
 711
 712
 713def _vtype_61850_from_vtype_conf(vt_conf):
 714    if isinstance(vt_conf, str):
 715        return _value_type_from_str(vt_conf)
 716
 717    if vt_conf['type'] == 'ARRAY':
 718        return iec61850.ArrayValueType(
 719            type=_vtype_61850_from_vtype_conf(vt_conf['element_type']),
 720            length=vt_conf['length'])
 721
 722    if vt_conf['type'] == 'STRUCT':
 723        return iec61850.StructValueType(
 724            [(el_conf['name'], _vtype_61850_from_vtype_conf(el_conf['type']))
 725             for el_conf in vt_conf['elements']])
 726
 727    raise Exception('unsupported value type')
 728
 729
 730def _vtype_from_vtype_conf(vt_conf):
 731    if isinstance(vt_conf, str):
 732        return _value_type_from_str(vt_conf)
 733
 734    if vt_conf['type'] == 'ARRAY':
 735        return iec61850.ArrayValueType(
 736            type=_vtype_from_vtype_conf(vt_conf['element_type']),
 737            length=vt_conf['length'])
 738
 739    if vt_conf['type'] == 'STRUCT':
 740        return {el_conf['name']: _vtype_from_vtype_conf(el_conf['type'])
 741                for el_conf in vt_conf['elements']}
 742
 743    raise Exception('unsupported value type')
 744
 745
 746def _value_type_from_str(vt_conf):
 747    if vt_conf in ['BOOLEAN',
 748                   'INTEGER',
 749                   'UNSIGNED',
 750                   'FLOAT',
 751                   'BIT_STRING',
 752                   'OCTET_STRING',
 753                   'VISIBLE_STRING',
 754                   'MMS_STRING']:
 755        return iec61850.BasicValueType(vt_conf)
 756
 757    if vt_conf in ['QUALITY',
 758                   'TIMESTAMP',
 759                   'DOUBLE_POINT',
 760                   'DIRECTION',
 761                   'SEVERITY',
 762                   'ANALOGUE',
 763                   'VECTOR',
 764                   'STEP_POSITION',
 765                   'BINARY_CONTROL']:
 766        return iec61850.AcsiValueType(vt_conf)
 767
 768    raise Exception('unsupported value type')
 769
 770
 771def _value_type_from_ref(value_types, ref):
 772    left_names = []
 773    if isinstance(ref, iec61850.CommandRef):
 774        left_names = ['Oper', 'ctlVal']
 775        key = (ref.logical_device, ref.logical_node, 'CO', ref.name)
 776
 777    elif isinstance(ref, iec61850.DataRef):
 778        left_names = ref.names[1:]
 779        key = (ref.logical_device, ref.logical_node, ref.fc, ref.names[0])
 780
 781    else:
 782        raise Exception('unexpected reference')
 783
 784    if key not in value_types:
 785        return
 786
 787    value_type = value_types[key]
 788    while left_names:
 789        name = left_names[0]
 790        left_names = left_names[1:]
 791        if isinstance(name, str):
 792            if isinstance(value_type, iec61850.StructValueType):
 793                _, value_type = util.first(
 794                    value_type.elements, lambda i: i[0] == name)
 795
 796            elif value_type == iec61850.AcsiValueType.ANALOGUE:
 797                if name == 'i':
 798                    value_type = iec61850.BasicValueType.INTEGER
 799                elif name == 'f':
 800                    value_type = iec61850.BasicValueType.FLOAT
 801                else:
 802                    raise Exception("analogue attribute should be 'i' or 'f'")
 803
 804            elif value_type == iec61850.AcsiValueType.VECTOR:
 805                if name in ('mag', 'ang'):
 806                    value_type = iec61850.AcsiValueType.ANALOGUE
 807                else:
 808                    raise Exception(
 809                        "vector attribute should be 'mag' or 'ang'")
 810
 811            elif value_type == iec61850.AcsiValueType.STEP_POSITION:
 812                if name == 'posVal':
 813                    value_type = iec61850.BasicValueType.INTEGER
 814                elif name == 'transInd':
 815                    value_type = iec61850.BasicValueType.BOOLEAN
 816                else:
 817                    raise Exception("step position attribute should be "
 818                                    "'posVal' or 'transInd'")
 819
 820            else:
 821                value_type = value_type[name]
 822
 823        if isinstance(name, int):
 824            value_type = value_type.type
 825
 826    return value_type
 827
 828
 829def _get_dataset_change_value_types(conf, value_types):
 830    for ds_conf in conf['datasets']:
 831        for val_ref_conf in ds_conf['values']:
 832            value_ref = _value_ref_from_conf(val_ref_conf)
 833            value_type = _value_type_from_ref(value_types, value_ref)
 834            yield value_ref, value_type
 835
 836    for change_conf in conf['changes']:
 837        value_ref = _value_ref_from_conf(change_conf['ref'])
 838        value_type = _value_type_from_ref(value_types, value_ref)
 839        yield value_ref, value_type
 840
 841
 842def _get_cmd_value_types(conf, value_types):
 843    for cmd_conf in conf['commands']:
 844        cmd_ref = iec61850.CommandRef(**cmd_conf['ref'])
 845        value_type = _value_type_from_ref(value_types, cmd_ref)
 846        yield cmd_ref, value_type
 847
 848
 849_epoch_start = datetime.datetime.fromtimestamp(0, datetime.timezone.utc)
 850
 851
 852def _command_from_event(event, cmd_conf, control_number, value_type):
 853    if cmd_conf['with_operate_time']:
 854        operate_time = iec61850.Timestamp(
 855            value=_epoch_start,
 856            leap_second=False,
 857            clock_failure=False,
 858            not_synchronized=False,
 859            accuracy=0)
 860    else:
 861        operate_time = None
 862    return iec61850.Command(
 863        value=_value_from_json(event.payload.data['value'], value_type),
 864        operate_time=operate_time,
 865        origin=iec61850.Origin(
 866            category=iec61850.OriginCategory[
 867                event.payload.data['origin']['category']],
 868            identification=event.payload.data[
 869                'origin']['identification'].encode('utf-8')),
 870        control_number=control_number,
 871        t=_timestamp_from_event_timestamp(event.timestamp),
 872        test=event.payload.data['test'],
 873        checks=set(iec61850.Check[i] for i in event.payload.data['checks']))
 874
 875
 876def _timestamp_from_event_timestamp(timestamp):
 877    return iec61850.Timestamp(
 878        value=hat.event.common.timestamp_to_datetime(timestamp),
 879        leap_second=False,
 880        clock_failure=False,
 881        not_synchronized=False,
 882        accuracy=None)
 883
 884
 885def _cmd_resp_to_event(event_type_prefix, cmd_name, event_session_id, action,
 886                       resp):
 887    success = resp is None
 888    payload = {'session_id': event_session_id,
 889               'action': action,
 890               'success': success}
 891    if not success:
 892        if resp.service_error is not None:
 893            payload['service_error'] = resp.service_error.name
 894        if resp.additional_cause is not None:
 895            payload['additional_cause'] = resp.additional_cause.name
 896        if resp.test_error is not None:
 897            payload['test_error'] = resp.test_error.name
 898
 899    return hat.event.common.RegisterEvent(
 900        type=(*event_type_prefix, 'gateway', 'command', cmd_name),
 901        source_timestamp=None,
 902        payload=hat.event.common.EventPayloadJson(payload))
 903
 904
 905def _write_data_resp_to_event(event_type_prefix, value_name, session_id, resp):
 906    success = resp is None
 907    payload = {'session_id': session_id,
 908               'success': success}
 909    if not success:
 910        payload['error'] = resp.name
 911    return hat.event.common.RegisterEvent(
 912        type=(*event_type_prefix, 'gateway', 'change', value_name),
 913        source_timestamp=None,
 914        payload=hat.event.common.EventPayloadJson(payload))
 915
 916
 917def _get_command_session_id(cmd_ref, cmd):
 918    return (cmd_ref, cmd.control_number)
 919
 920
 921def _value_from_json(event_value, value_type):
 922    if isinstance(value_type, iec61850.BasicValueType):
 923        if value_type == iec61850.BasicValueType.OCTET_STRING:
 924            return bytes.fromhex(event_value)
 925
 926        elif value_type == iec61850.BasicValueType.FLOAT:
 927            return float(event_value)
 928
 929        else:
 930            return event_value
 931
 932    if value_type == iec61850.AcsiValueType.QUALITY:
 933        return iec61850.Quality(
 934            validity=iec61850.QualityValidity[event_value['validity']],
 935            details={iec61850.QualityDetail[i]
 936                     for i in event_value['details']},
 937            source=iec61850.QualitySource[event_value['source']],
 938            test=event_value['test'],
 939            operator_blocked=event_value['operator_blocked'])
 940
 941    if value_type == iec61850.AcsiValueType.TIMESTAMP:
 942        return iec61850.Timestamp(
 943            value=datetime.datetime.fromtimestamp(event_value['value'],
 944                                                  datetime.timezone.utc),
 945            leap_second=event_value['leap_second'],
 946            clock_failure=event_value['clock_failure'],
 947            not_synchronized=event_value['not_synchronized'],
 948            accuracy=event_value.get('accuracy'))
 949
 950    if value_type == iec61850.AcsiValueType.DOUBLE_POINT:
 951        return iec61850.DoublePoint[event_value]
 952
 953    if value_type == iec61850.AcsiValueType.DIRECTION:
 954        return iec61850.Direction[event_value]
 955
 956    if value_type == iec61850.AcsiValueType.SEVERITY:
 957        return iec61850.Severity[event_value]
 958
 959    if value_type == iec61850.AcsiValueType.ANALOGUE:
 960        return iec61850.Analogue(
 961            i=event_value.get('i'),
 962            f=(float(event_value['f']) if 'f' in event_value else None))
 963
 964    if value_type == iec61850.AcsiValueType.VECTOR:
 965        return iec61850.Vector(
 966            magnitude=_value_from_json(event_value['magnitude'],
 967                                       iec61850.AcsiValueType.ANALOGUE),
 968            angle=(_value_from_json(event_value['angle'],
 969                                    iec61850.AcsiValueType.ANALOGUE)
 970                   if 'angle' in event_value else None))
 971
 972    if value_type == iec61850.AcsiValueType.STEP_POSITION:
 973        return iec61850.StepPosition(value=event_value['value'],
 974                                     transient=event_value.get('transient'))
 975
 976    if value_type == iec61850.AcsiValueType.BINARY_CONTROL:
 977        return iec61850.BinaryControl[event_value]
 978
 979    if isinstance(value_type, iec61850.ArrayValueType):
 980        return [_value_from_json(val, value_type.type)
 981                for val in event_value]
 982
 983    if isinstance(value_type, dict):
 984        return {k: _value_from_json(v, value_type[k])
 985                for k, v in event_value.items()}
 986
 987    raise Exception('unsupported value type')
 988
 989
 990def _value_to_json(data_value, value_type):
 991    if isinstance(value_type, iec61850.BasicValueType):
 992        if value_type == iec61850.BasicValueType.OCTET_STRING:
 993            return data_value.hex()
 994
 995        elif value_type == iec61850.BasicValueType.BIT_STRING:
 996            return list(data_value)
 997
 998        elif value_type == iec61850.BasicValueType.FLOAT:
 999            return data_value if math.isfinite(data_value) else str(data_value)
1000
1001        else:
1002            return data_value
1003
1004    if isinstance(value_type, iec61850.AcsiValueType):
1005        if value_type == iec61850.AcsiValueType.QUALITY:
1006            return {'validity': data_value.validity.name,
1007                    'details': [i.name for i in data_value.details],
1008                    'source': data_value.source.name,
1009                    'test': data_value.test,
1010                    'operator_blocked': data_value.operator_blocked}
1011
1012        if value_type == iec61850.AcsiValueType.TIMESTAMP:
1013            val = {'value': data_value.value.timestamp(),
1014                   'leap_second': data_value.leap_second,
1015                   'clock_failure': data_value.clock_failure,
1016                   'not_synchronized': data_value.not_synchronized}
1017            if data_value.accuracy is not None:
1018                val['accuracy'] = data_value.accuracy
1019            return val
1020
1021        if value_type in [iec61850.AcsiValueType.DOUBLE_POINT,
1022                          iec61850.AcsiValueType.DIRECTION,
1023                          iec61850.AcsiValueType.SEVERITY,
1024                          iec61850.AcsiValueType.BINARY_CONTROL]:
1025            return data_value.name
1026
1027        if value_type == iec61850.AcsiValueType.ANALOGUE:
1028            val = {}
1029            if data_value.i is not None:
1030                val['i'] = data_value.i
1031            if data_value.f is not None:
1032                val['f'] = (data_value.f if math.isfinite(data_value.f)
1033                            else str(data_value.f))
1034            return val
1035
1036        if value_type == iec61850.AcsiValueType.VECTOR:
1037            val = {'mag': _value_to_json(
1038                        data_value.magnitude, iec61850.AcsiValueType.ANALOGUE)}
1039            if data_value.angle is not None:
1040                val['ang'] = _value_to_json(
1041                    data_value.angle, iec61850.AcsiValueType.ANALOGUE)
1042            return val
1043
1044        if value_type == iec61850.AcsiValueType.STEP_POSITION:
1045            val = {'posVal': data_value.value}
1046            if data_value.transient is not None:
1047                val['transInd'] = data_value.transient
1048            return val
1049
1050    if isinstance(value_type, iec61850.ArrayValueType):
1051        return [_value_to_json(i, value_type.type) for i in data_value]
1052
1053    if isinstance(value_type, dict):
1054        return {
1055            child_name: _value_to_json(child_value, value_type[child_name])
1056            for child_name, child_value in data_value.items()}
1057
1058    raise Exception('unsupported value type')
1059
1060
1061def _value_json_to_event_json(data_value, value_type):
1062    if value_type == iec61850.AcsiValueType.VECTOR:
1063        val = {'magnitude': data_value['mag']}
1064        if 'ang' in data_value:
1065            val['angle'] = data_value['ang']
1066        return val
1067
1068    if value_type == iec61850.AcsiValueType.STEP_POSITION:
1069        val = {'value': data_value['posVal']}
1070        if 'transInd' in data_value:
1071            val['transient'] = data_value['transInd']
1072        return val
1073
1074    if isinstance(value_type, iec61850.ArrayValueType):
1075        return [_value_json_to_event_json(i, value_type.type)
1076                for i in data_value]
1077
1078    if isinstance(value_type, dict):
1079        return {
1080            child_name: _value_json_to_event_json(
1081                child_value, value_type[child_name])
1082            for child_name, child_value in data_value.items()}
1083
1084    return data_value
1085
1086
1087def _update_ctl_num(ctl_num, action, cmd_model):
1088    if action == 'SELECT' or (
1089        action == 'OPERATE' and
1090        cmd_model in ['DIRECT_WITH_NORMAL_SECURITY',
1091                      'DIRECT_WITH_ENHANCED_SECURITY']):
1092        return (ctl_num + 1) % 256
1093
1094    return ctl_num
1095
1096
1097def _validate_get_rcb_response(get_rcb_resp, rcb_conf):
1098    for k, v in get_rcb_resp.items():
1099        if isinstance(v, iec61850.ServiceError):
1100            raise Exception(f"get {k.name} failed: {v}")
1101
1102        if (k == iec61850.RcbAttrType.REPORT_ID and
1103                v != rcb_conf['report_id']):
1104            raise Exception(f"rcb report id {v} different from "
1105                            f"configured {rcb_conf['report_id']}")
1106
1107        if (k == iec61850.RcbAttrType.DATASET and
1108                v != _dataset_ref_from_conf(rcb_conf['dataset'])):
1109            raise Exception(f"rcb dataset {v} different from "
1110                            f"configured {rcb_conf['report_id']}")
1111
1112        if (k == iec61850.RcbAttrType.CONF_REVISION and
1113                v != rcb_conf['conf_revision']):
1114            raise Exception(
1115                f"Conf revision {v} different from "
1116                f"the configuration defined {rcb_conf['conf_revision']}")
1117
1118
1119def _conf_ref_to_path(conf_ref):
1120    return [conf_ref['logical_device'],
1121            conf_ref['logical_node'],
1122            conf_ref['fc'],
1123            *conf_ref['names']]
1124
1125
1126def _create_logger_adapter(name):
1127    extra = {'meta': {'type': 'Iec61850ClientDevice',
1128                      'name': name}}
1129
1130    return logging.LoggerAdapter(mlog, extra)