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