hat.gateway.devices.iec101.master

  1"""IEC 60870-5-101 master device"""
  2
  3from collections.abc import Collection
  4import asyncio
  5import collections
  6import contextlib
  7import datetime
  8import functools
  9import logging
 10
 11from hat import aio
 12from hat.drivers import iec101
 13from hat.drivers import serial
 14from hat.drivers.iec60870 import link
 15import hat.event.common
 16import hat.event.eventer
 17
 18from hat.gateway.devices.iec101 import common
 19
 20
 21mlog: logging.Logger = logging.getLogger(__name__)
 22
 23
 24async def create(conf: common.DeviceConf,
 25                 eventer_client: hat.event.eventer.Client,
 26                 event_type_prefix: common.EventTypePrefix
 27                 ) -> 'Iec101MasterDevice':
 28    event_types = [(*event_type_prefix, 'system', 'remote_device',
 29                    str(i['address']), 'enable')
 30                   for i in conf['remote_devices']]
 31    params = hat.event.common.QueryLatestParams(event_types)
 32    result = await eventer_client.query(params)
 33
 34    device = Iec101MasterDevice(conf=conf,
 35                                eventer_client=eventer_client,
 36                                event_type_prefix=event_type_prefix)
 37    try:
 38        await device.process_events(result.events)
 39
 40    except BaseException:
 41        await aio.uncancellable(device.async_close())
 42        raise
 43
 44    return device
 45
 46
 47info: common.DeviceInfo = common.DeviceInfo(
 48    type="iec101_master",
 49    create=create,
 50    json_schema_id="hat-gateway://iec101.yaml#/$defs/master",
 51    json_schema_repo=common.json_schema_repo)
 52
 53
 54class Iec101MasterDevice(common.Device):
 55
 56    def __init__(self,
 57                 conf: common.DeviceConf,
 58                 eventer_client: hat.event.eventer.Client,
 59                 event_type_prefix: common.EventTypePrefix,
 60                 send_queue_size: int = 1024):
 61        self._conf = conf
 62        self._event_type_prefix = event_type_prefix
 63        self._eventer_client = eventer_client
 64        self._master = None
 65        self._conns = {}
 66        self._send_queue = aio.Queue(send_queue_size)
 67        self._async_group = aio.Group()
 68        self._remote_enabled = {i['address']: False
 69                                for i in conf['remote_devices']}
 70        self._remote_confs = {i['address']: i
 71                              for i in conf['remote_devices']}
 72        self._remote_groups = {}
 73
 74        self.async_group.spawn(self._create_link_master_loop)
 75        self.async_group.spawn(self._send_loop)
 76
 77    @property
 78    def async_group(self) -> aio.Group:
 79        return self._async_group
 80
 81    async def process_events(self, events: Collection[hat.event.common.Event]):
 82        for event in events:
 83            try:
 84                await self._process_event(event)
 85
 86            except Exception as e:
 87                mlog.warning('error processing event: %s', e, exc_info=e)
 88
 89    async def _create_link_master_loop(self):
 90
 91        async def cleanup():
 92            with contextlib.suppress(ConnectionError):
 93                await self._register_status('DISCONNECTED')
 94
 95            if self._master:
 96                await self._master.async_close()
 97
 98        try:
 99            while True:
100                await self._register_status('CONNECTING')
101
102                try:
103                    self._master = await link.unbalanced.create_master(
104                        port=self._conf['port'],
105                        baudrate=self._conf['baudrate'],
106                        bytesize=serial.ByteSize[self._conf['bytesize']],
107                        parity=serial.Parity[self._conf['parity']],
108                        stopbits=serial.StopBits[self._conf['stopbits']],
109                        xonxoff=self._conf['flow_control']['xonxoff'],
110                        rtscts=self._conf['flow_control']['rtscts'],
111                        dsrdtr=self._conf['flow_control']['dsrdtr'],
112                        silent_interval=self._conf['silent_interval'],
113                        address_size=link.AddressSize[
114                            self._conf['device_address_size']])
115
116                except Exception as e:
117                    mlog.warning('link master (endpoint) failed to create: %s',
118                                 e, exc_info=e)
119                    await self._register_status('DISCONNECTED')
120                    await asyncio.sleep(self._conf['reconnect_delay'])
121                    continue
122
123                await self._register_status('CONNECTED')
124
125                for address, enabled in self._remote_enabled.items():
126                    if enabled:
127                        self._enable_remote(address)
128
129                await self._master.wait_closed()
130                await self._register_status('DISCONNECTED')
131                self._master = None
132
133        except Exception as e:
134            mlog.error('create link master error: %s', e, exc_info=e)
135
136        finally:
137            mlog.debug('closing link master loop')
138            self.close()
139            self._conns = {}
140            await aio.uncancellable(cleanup())
141
142    async def _send_loop(self):
143        while True:
144            msg, address = await self._send_queue.get()
145
146            conn = self._conns.get(address)
147            if not conn or not conn.is_open:
148                mlog.warning('msg %s not sent, connection to %s closed',
149                             msg, address)
150                continue
151
152            try:
153                await conn.send([msg])
154                mlog.debug('msg sent asdu=%s', msg.asdu_address)
155
156            except ConnectionError:
157                mlog.warning('msg %s not sent, connection to %s closed',
158                             msg, address)
159
160    async def _connection_loop(self, group, address):
161
162        async def cleanup():
163            with contextlib.suppress(ConnectionError):
164                await self._register_rmt_status(address, 'DISCONNECTED')
165
166            conn = self._conns.pop(address, None)
167            if conn:
168                await conn.async_close()
169
170        remote_conf = self._remote_confs[address]
171        try:
172            while True:
173                await self._register_rmt_status(address, 'CONNECTING')
174
175                try:
176                    master_conn = await self._master.connect(
177                        addr=address,
178                        response_timeout=remote_conf['response_timeout'],
179                        send_retry_count=remote_conf['send_retry_count'],
180                        poll_class1_delay=remote_conf['poll_class1_delay'],
181                        poll_class2_delay=remote_conf['poll_class2_delay'])
182
183                except Exception as e:
184                    mlog.error('connection error to address %s: %s',
185                               address, e, exc_info=e)
186                    await self._register_rmt_status(address, 'DISCONNECTED')
187                    await asyncio.sleep(remote_conf['reconnect_delay'])
188                    continue
189
190                await self._register_rmt_status(address, 'CONNECTED')
191
192                conn = iec101.MasterConnection(
193                    conn=master_conn,
194                    cause_size=iec101.CauseSize[self._conf['cause_size']],
195                    asdu_address_size=iec101.AsduAddressSize[
196                        self._conf['asdu_address_size']],
197                    io_address_size=iec101.IoAddressSize[
198                        self._conf['io_address_size']])
199                self._conns[address] = conn
200                group.spawn(self._receive_loop, conn, address)
201
202                if remote_conf['time_sync_delay'] is not None:
203                    group.spawn(self._time_sync_loop, conn,
204                                remote_conf['time_sync_delay'])
205
206                await conn.wait_closed()
207                await self._register_rmt_status(address, 'DISCONNECTED')
208                self._conns.pop(address)
209
210        except Exception as e:
211            mlog.error('connection loop error: %s', e, exc_info=e)
212
213        finally:
214            mlog.debug('closing remote device %s', address)
215            group.close()
216            await aio.uncancellable(cleanup())
217
218    async def _receive_loop(self, conn, address):
219        try:
220            while True:
221                try:
222                    msgs = await conn.receive()
223
224                except iec101.AsduTypeError as e:
225                    mlog.warning("asdu type error: %s", e)
226                    continue
227
228                events = collections.deque()
229                for msg in msgs:
230                    if isinstance(msg, iec101.ClockSyncMsg):
231                        continue
232
233                    try:
234                        event = _msg_to_event(self._event_type_prefix, address,
235                                              msg)
236                        events.append(event)
237
238                    except Exception as e:
239                        mlog.warning('message %s ignored due to: %s',
240                                     msg, e, exc_info=e)
241
242                if not events:
243                    continue
244
245                await self._eventer_client.register(events)
246                for e in events:
247                    mlog.debug('registered event %s', e)
248
249        except ConnectionError:
250            mlog.debug('connection closed')
251
252        except Exception as e:
253            mlog.error('receive loop error: %s', e, exc_info=e)
254
255        finally:
256            conn.close()
257
258    async def _time_sync_loop(self, conn, delay):
259        try:
260            while True:
261                time_now = datetime.datetime.now(datetime.timezone.utc)
262                time_iec101 = iec101.time_from_datetime(time_now)
263                msg = iec101.ClockSyncMsg(
264                    is_test=False,
265                    originator_address=0,
266                    asdu_address={
267                        'ONE': 0xFF,
268                        'TWO': 0xFFFF}[self._conf['asdu_address_size']],
269                    time=time_iec101,
270                    is_negative_confirm=False,
271                    cause=iec101.ClockSyncReqCause.ACTIVATION)
272                await conn.send([msg])
273                mlog.debug('time sync sent %s', time_iec101)
274
275                await asyncio.sleep(delay)
276
277        except ConnectionError:
278            mlog.debug('connection closed')
279
280        except Exception as e:
281            mlog.error('time sync loop error: %s', e, exc_info=e)
282
283        finally:
284            conn.close()
285
286    async def _process_event(self, event):
287        match_type = functools.partial(hat.event.common.matches_query_type,
288                                       event.type)
289
290        prefix = (*self._event_type_prefix, 'system', 'remote_device', '?')
291        if not match_type((*prefix, '*')):
292            raise Exception('unexpected event type')
293
294        address = int(event.type[len(prefix) - 1])
295        suffix = event.type[len(prefix):]
296
297        if match_type((*prefix, 'enable')):
298            self._process_event_enable(address, event)
299
300        elif match_type((*prefix, 'command', '?', '?', '?')):
301            cmd_key = common.CommandKey(
302                cmd_type=common.CommandType(suffix[1]),
303                asdu_address=int(suffix[2]),
304                io_address=int(suffix[3]))
305            msg = _command_from_event(cmd_key, event)
306
307            await self._send_queue.put((msg, address))
308            mlog.debug('command asdu=%s io=%s prepared for sending',
309                       cmd_key.asdu_address, cmd_key.io_address)
310
311        elif match_type((*prefix, 'interrogation', '?')):
312            asdu_address = int(suffix[1])
313            msg = _interrogation_from_event(asdu_address, event)
314
315            await self._send_queue.put((msg, address))
316            mlog.debug("interrogation request asdu=%s prepared for sending",
317                       asdu_address)
318
319        elif match_type((*prefix, 'counter_interrogation', '?')):
320            asdu_address = int(suffix[1])
321            msg = _counter_interrogation_from_event(asdu_address, event)
322
323            await self._send_queue.put((msg, address))
324            mlog.debug("counter interrogation request asdu=%s prepared for "
325                       "sending", asdu_address)
326
327        else:
328            raise Exception('unexpected event type')
329
330    def _process_event_enable(self, address, event):
331        if address not in self._remote_enabled:
332            raise Exception('invalid remote device address')
333
334        enable = event.payload.data
335        if not isinstance(enable, bool):
336            raise Exception('invalid enable event payload')
337
338        if address not in self._remote_enabled:
339            mlog.warning('received enable for unexpected remote device')
340            return
341
342        self._remote_enabled[address] = enable
343
344        if not enable:
345            self._disable_remote(address)
346
347        elif not self._master:
348            return
349
350        else:
351            self._enable_remote(address)
352
353    def _enable_remote(self, address):
354        mlog.debug('enabling device %s', address)
355        remote_group = self._remote_groups.get(address)
356        if remote_group and remote_group.is_open:
357            mlog.debug('device %s is already running', address)
358            return
359
360        remote_group = self._async_group.create_subgroup()
361        self._remote_groups[address] = remote_group
362        remote_group.spawn(self._connection_loop, remote_group, address)
363
364    def _disable_remote(self, address):
365        mlog.debug('disabling device %s', address)
366        if address in self._remote_groups:
367            remote_group = self._remote_groups.pop(address)
368            remote_group.close()
369
370    async def _register_status(self, status):
371        event = hat.event.common.RegisterEvent(
372            type=(*self._event_type_prefix, 'gateway', 'status'),
373            source_timestamp=None,
374            payload=hat.event.common.EventPayloadJson(status))
375        await self._eventer_client.register([event])
376        mlog.debug('device status -> %s', status)
377
378    async def _register_rmt_status(self, address, status):
379        event = hat.event.common.RegisterEvent(
380            type=(*self._event_type_prefix, 'gateway', 'remote_device',
381                  str(address), 'status'),
382            source_timestamp=None,
383            payload=hat.event.common.EventPayloadJson(status))
384        await self._eventer_client.register([event])
385        mlog.debug('remote device %s status -> %s', address, status)
386
387
388def _msg_to_event(event_type_prefix, address, msg):
389    if isinstance(msg, iec101.DataMsg):
390        return _data_to_event(event_type_prefix, address, msg)
391
392    if isinstance(msg, iec101.CommandMsg):
393        return _command_to_event(event_type_prefix, address, msg)
394
395    if isinstance(msg, iec101.InterrogationMsg):
396        return _interrogation_to_event(event_type_prefix, address, msg)
397
398    if isinstance(msg, iec101.CounterInterrogationMsg):
399        return _counter_interrogation_to_event(event_type_prefix, address, msg)
400
401    raise Exception('unsupported message type')
402
403
404def _data_to_event(event_type_prefix, address, msg):
405    data_type = common.get_data_type(msg.data)
406    cause = common.cause_to_json(iec101.DataResCause, msg.cause)
407    data = common.data_to_json(msg.data)
408    event_type = (*event_type_prefix, 'gateway', 'remote_device', str(address),
409                  'data', data_type.value, str(msg.asdu_address),
410                  str(msg.io_address))
411    source_timestamp = common.time_to_source_timestamp(msg.time)
412
413    return hat.event.common.RegisterEvent(
414        type=event_type,
415        source_timestamp=source_timestamp,
416        payload=hat.event.common.EventPayloadJson({'is_test': msg.is_test,
417                                                   'cause': cause,
418                                                   'data': data}))
419
420
421def _command_to_event(event_type_prefix, address, msg):
422    command_type = common.get_command_type(msg.command)
423    cause = common.cause_to_json(iec101.CommandResCause, msg.cause)
424    command = common.command_to_json(msg.command)
425    event_type = (*event_type_prefix, 'gateway', 'remote_device', str(address),
426                  'command', command_type.value, str(msg.asdu_address),
427                  str(msg.io_address))
428
429    return hat.event.common.RegisterEvent(
430        type=event_type,
431        source_timestamp=None,
432        payload=hat.event.common.EventPayloadJson({
433            'is_test': msg.is_test,
434            'is_negative_confirm': msg.is_negative_confirm,
435            'cause': cause,
436            'command': command}))
437
438
439def _interrogation_to_event(event_type_prefix, address, msg):
440    cause = common.cause_to_json(iec101.CommandResCause, msg.cause)
441    event_type = (*event_type_prefix, 'gateway', 'remote_device', str(address),
442                  'interrogation', str(msg.asdu_address))
443
444    return hat.event.common.RegisterEvent(
445        type=event_type,
446        source_timestamp=None,
447        payload=hat.event.common.EventPayloadJson({
448            'is_test': msg.is_test,
449            'is_negative_confirm': msg.is_negative_confirm,
450            'request': msg.request,
451            'cause': cause}))
452
453
454def _counter_interrogation_to_event(event_type_prefix, address, msg):
455    cause = common.cause_to_json(iec101.CommandResCause, msg.cause)
456    event_type = (*event_type_prefix, 'gateway', 'remote_device', str(address),
457                  'counter_interrogation', str(msg.asdu_address))
458
459    return hat.event.common.RegisterEvent(
460        type=event_type,
461        source_timestamp=None,
462        payload=hat.event.common.EventPayloadJson({
463            'is_test': msg.is_test,
464            'is_negative_confirm': msg.is_negative_confirm,
465            'request': msg.request,
466            'freeze': msg.freeze.name,
467            'cause': cause}))
468
469
470def _command_from_event(cmd_key, event):
471    cause = common.cause_from_json(iec101.CommandReqCause,
472                                   event.payload.data['cause'])
473    command = common.command_from_json(cmd_key.cmd_type,
474                                       event.payload.data['command'])
475
476    return iec101.CommandMsg(is_test=event.payload.data['is_test'],
477                             originator_address=0,
478                             asdu_address=cmd_key.asdu_address,
479                             io_address=cmd_key.io_address,
480                             command=command,
481                             is_negative_confirm=False,
482                             cause=cause)
483
484
485def _interrogation_from_event(asdu_address, event):
486    cause = common.cause_from_json(iec101.CommandReqCause,
487                                   event.payload.data['cause'])
488
489    return iec101.InterrogationMsg(is_test=event.payload.data['is_test'],
490                                   originator_address=0,
491                                   asdu_address=asdu_address,
492                                   request=event.payload.data['request'],
493                                   is_negative_confirm=False,
494                                   cause=cause)
495
496
497def _counter_interrogation_from_event(asdu_address, event):
498    freeze = iec101.FreezeCode[event.payload.data['freeze']]
499    cause = common.cause_from_json(iec101.CommandReqCause,
500                                   event.payload.data['cause'])
501
502    return iec101.CounterInterrogationMsg(
503        is_test=event.payload.data['is_test'],
504        originator_address=0,
505        asdu_address=asdu_address,
506        request=event.payload.data['request'],
507        freeze=freeze,
508        is_negative_confirm=False,
509        cause=cause)