hat.gateway.devices.iec101.slave

  1"""IEC 60870-5-104 slave device"""
  2
  3from collections.abc import Collection, Iterable
  4import collections
  5import contextlib
  6import functools
  7import itertools
  8import logging
  9
 10from hat import aio
 11from hat import json
 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                 ) -> 'Iec101SlaveDevice':
 28    device = Iec101SlaveDevice()
 29    device._conf = conf
 30    device._eventer_client = eventer_client
 31    device._event_type_prefix = event_type_prefix
 32    device._next_conn_ids = itertools.count(1)
 33    device._conns = {}
 34    device._buffers = {}
 35    device._data_msgs = {}
 36    device._data_buffers = {}
 37
 38    init_buffers(buffers_conf=conf['buffers'],
 39                 buffers=device._buffers)
 40
 41    await init_data(data_conf=conf['data'],
 42                    data_msgs=device._data_msgs,
 43                    data_buffers=device._data_buffers,
 44                    buffers=device._buffers,
 45                    eventer_client=eventer_client,
 46                    event_type_prefix=event_type_prefix)
 47
 48    device._slave = await link.unbalanced.create_slave(
 49        port=conf['port'],
 50        addrs=conf['addresses'],
 51        connection_cb=device._on_connection,
 52        baudrate=conf['baudrate'],
 53        bytesize=serial.ByteSize[conf['bytesize']],
 54        parity=serial.Parity[conf['parity']],
 55        stopbits=serial.StopBits[conf['stopbits']],
 56        xonxoff=conf['flow_control']['xonxoff'],
 57        rtscts=conf['flow_control']['rtscts'],
 58        dsrdtr=conf['flow_control']['dsrdtr'],
 59        silent_interval=conf['silent_interval'],
 60        address_size=link.AddressSize[conf['device_address_size']],
 61        keep_alive_timeout=conf['keep_alive_timeout'])
 62
 63    try:
 64        await device._register_connections()
 65
 66    except BaseException:
 67        await aio.uncancellable(device.async_close())
 68        raise
 69
 70    return device
 71
 72
 73info: common.DeviceInfo = common.DeviceInfo(
 74    type="iec101_slave",
 75    create=create,
 76    json_schema_id="hat-gateway://iec101.yaml#/$defs/slave",
 77    json_schema_repo=common.json_schema_repo)
 78
 79
 80class Buffer:
 81
 82    def __init__(self, size: int):
 83        self._size = size
 84        self._data = collections.OrderedDict()
 85
 86    def add(self,
 87            event_id: hat.event.common.EventId,
 88            data_msg: iec101.DataMsg):
 89        self._data[event_id] = data_msg
 90        while len(self._data) > self._size:
 91            self._data.popitem(last=False)
 92
 93    def remove(self, event_id: hat.event.common.EventId):
 94        self._data.pop(event_id, None)
 95
 96    def get_event_id_data_msgs(self) -> Iterable[tuple[hat.event.common.EventId,  # NOQA
 97                                                       iec101.DataMsg]]:
 98        return self._data.items()
 99
100
101def init_buffers(buffers_conf: json.Data,
102                 buffers: dict[str, Buffer]):
103    for buffer_conf in buffers_conf:
104        buffers[buffer_conf['name']] = Buffer(buffer_conf['size'])
105
106
107async def init_data(data_conf: json.Data,
108                    data_msgs: dict[common.DataKey, iec101.DataMsg],
109                    data_buffers: dict[common.DataKey, Buffer],
110                    buffers: dict[str, Buffer],
111                    eventer_client: hat.event.eventer.Client,
112                    event_type_prefix: common.EventTypePrefix):
113    for data in data_conf:
114        data_key = common.DataKey(data_type=common.DataType[data['data_type']],
115                                  asdu_address=data['asdu_address'],
116                                  io_address=data['io_address'])
117        data_msgs[data_key] = None
118        if data['buffer']:
119            data_buffers[data_key] = buffers[data['buffer']]
120
121    event_types = [(*event_type_prefix, 'system', 'data', '*')]
122    params = hat.event.common.QueryLatestParams(event_types)
123    result = await eventer_client.query(params)
124
125    for event in result.events:
126        try:
127            data_type_str, asdu_address_str, io_address_str = \
128                event.type[len(event_type_prefix)+2:]
129            data_key = common.DataKey(data_type=common.DataType(data_type_str),
130                                      asdu_address=int(asdu_address_str),
131                                      io_address=int(io_address_str))
132            if data_key not in data_msgs:
133                raise Exception(f'data {data_key} not configured')
134
135            data_msgs[data_key] = data_msg_from_event(data_key, event)
136
137        except Exception as e:
138            mlog.debug('skipping initial data: %s', e, exc_info=e)
139
140
141class Iec101SlaveDevice(common.Device):
142
143    @property
144    def async_group(self) -> aio.Group:
145        return self._slave.async_group
146
147    async def process_events(self, events: Collection[hat.event.common.Event]):
148        for event in events:
149            try:
150                await self._process_event(event)
151
152            except Exception as e:
153                mlog.warning('error processing event: %s', e, exc_info=e)
154
155    def _on_connection(self, conn):
156        self.async_group.spawn(self._connection_loop, conn)
157
158    async def _connection_loop(self, conn):
159        conn_id = next(self._next_conn_ids)
160
161        try:
162            conn = iec101.SlaveConnection(
163                conn=conn,
164                cause_size=iec101.CauseSize[self._conf['cause_size']],
165                asdu_address_size=iec101.AsduAddressSize[self._conf['asdu_address_size']],  # NOQA
166                io_address_size=iec101.IoAddressSize[self._conf['io_address_size']])  # NOQA
167
168            self._conns[conn_id] = conn
169            await self._register_connections()
170
171            with contextlib.suppress(Exception):
172                for buffer in self._buffers.values():
173                    for event_id, data_msg in buffer.get_event_id_data_msgs():
174                        self._send_data_msg(conn, buffer, event_id, data_msg)
175
176            while True:
177                msgs = await conn.receive()
178
179                for msg in msgs:
180                    try:
181                        mlog.debug('received message: %s', msg)
182                        await self._process_msg(conn_id, conn, msg)
183
184                    except Exception as e:
185                        mlog.warning('error processing message: %s',
186                                     e, exc_info=e)
187
188        except ConnectionError:
189            mlog.debug('connection close')
190
191        except Exception as e:
192            mlog.warning('connection error: %s', e, exc_info=e)
193
194        finally:
195            mlog.debug('closing connection')
196            conn.close()
197
198            with contextlib.suppress(Exception):
199                self._conns.pop(conn_id)
200                await aio.uncancellable(self._register_connections())
201
202    async def _register_connections(self):
203        payload = [{'connection_id': conn_id,
204                    'address': conn.address}
205                   for conn_id, conn in self._conns.items()]
206
207        event = hat.event.common.RegisterEvent(
208            type=(*self._event_type_prefix, 'gateway', 'connections'),
209            source_timestamp=None,
210            payload=hat.event.common.EventPayloadJson(payload))
211
212        await self._eventer_client.register([event])
213
214    async def _process_event(self, event):
215        suffix = event.type[len(self._event_type_prefix):]
216
217        if suffix[:2] == ('system', 'data'):
218            data_type_str, asdu_address_str, io_address_str = suffix[2:]
219            data_key = common.DataKey(data_type=common.DataType(data_type_str),
220                                      asdu_address=int(asdu_address_str),
221                                      io_address=int(io_address_str))
222
223            await self._process_data_event(data_key, event)
224
225        elif suffix[:2] == ('system', 'command'):
226            cmd_type_str, asdu_address_str, io_address_str = suffix[2:]
227            cmd_key = common.CommandKey(
228                cmd_type=common.CommandType(cmd_type_str),
229                asdu_address=int(asdu_address_str),
230                io_address=int(io_address_str))
231
232            await self._process_command_event(cmd_key, event)
233
234        else:
235            raise Exception('unsupported event type')
236
237    async def _process_data_event(self, data_key, event):
238        if data_key not in self._data_msgs:
239            raise Exception('data not configured')
240
241        data_msg = data_msg_from_event(data_key, event)
242        self._data_msgs[data_key] = data_msg
243
244        buffer = self._data_buffers.get(data_key)
245        if buffer:
246            buffer.add(event.id, data_msg)
247
248        for conn in self._conns.values():
249            self._send_data_msg(conn, buffer, event.id, data_msg)
250
251    async def _process_command_event(self, cmd_key, event):
252        cmd_msg = cmd_msg_from_event(cmd_key, event)
253        conn_id = event.payload.data['connection_id']
254        conn = self._conns[conn_id]
255        conn.send([cmd_msg])
256
257    async def _process_msg(self, conn_id, conn, msg):
258        if isinstance(msg, iec101.CommandMsg):
259            await self._process_command_msg(conn_id, conn, msg)
260
261        elif isinstance(msg, iec101.InterrogationMsg):
262            self._process_interrogation_msg(conn_id, conn, msg)
263
264        elif isinstance(msg, iec101.CounterInterrogationMsg):
265            self._process_counter_interrogation_msg(conn_id, conn, msg)
266
267        elif isinstance(msg, iec101.ReadMsg):
268            self._process_read_msg(conn_id, conn, msg)
269
270        elif isinstance(msg, iec101.ClockSyncMsg):
271            self._process_clock_sync_msg(conn_id, conn, msg)
272
273        elif isinstance(msg, iec101.TestMsg):
274            self._process_test_msg(conn_id, conn, msg)
275
276        elif isinstance(msg, iec101.ResetMsg):
277            self._process_reset_msg(conn_id, conn, msg)
278
279        elif isinstance(msg, iec101.ParameterMsg):
280            self._process_parameter_msg(conn_id, conn, msg)
281
282        elif isinstance(msg, iec101.ParameterActivationMsg):
283            self._process_parameter_activation_msg(conn_id, conn, msg)
284
285        else:
286            raise Exception('unsupported message')
287
288    async def _process_command_msg(self, conn_id, conn, msg):
289        if isinstance(msg.cause, iec101.CommandReqCause):
290            event = cmd_msg_to_event(self._event_type_prefix, conn_id, msg)
291            await self._eventer_client.register([event])
292
293        else:
294            res = msg._replace(cause=iec101.CommandResCause.UNKNOWN_CAUSE,
295                               is_negative_confirm=True)
296            conn.send([res])
297
298    def _process_interrogation_msg(self, conn_id, conn, msg):
299        if msg.cause == iec101.CommandReqCause.ACTIVATION:
300            res = msg._replace(
301                cause=iec101.CommandResCause.ACTIVATION_CONFIRMATION,
302                is_negative_confirm=False)
303            conn.send([res])
304
305            data_msgs = [
306                data_msg._replace(
307                    is_test=msg.is_test,
308                    cause=iec101.DataResCause.INTERROGATED_STATION)
309                for data_msg in self._data_msgs.values()
310                if (data_msg and
311                    (msg.asdu_address == 0xFFFF or
312                     msg.asdu_address == data_msg.asdu_address) and
313                    not isinstance(data_msg.data, iec101.BinaryCounterData))]
314            conn.send(data_msgs)
315
316            res = msg._replace(
317                cause=iec101.CommandResCause.ACTIVATION_TERMINATION,
318                is_negative_confirm=False)
319            conn.send([res])
320
321        elif msg.cause == iec101.CommandReqCause.DEACTIVATION:
322            res = msg._replace(
323                cause=iec101.CommandResCause.DEACTIVATION_CONFIRMATION,
324                is_negative_confirm=True)
325            conn.send([res])
326
327        else:
328            res = msg._replace(cause=iec101.CommandResCause.UNKNOWN_CAUSE,
329                               is_negative_confirm=True)
330            conn.send([res])
331
332    def _process_counter_interrogation_msg(self, conn_id, conn, msg):
333        if msg.cause == iec101.CommandReqCause.ACTIVATION:
334            res = msg._replace(
335                cause=iec101.CommandResCause.ACTIVATION_CONFIRMATION,
336                is_negative_confirm=False)
337            conn.send([res])
338
339            data_msgs = [
340                data_msg._replace(
341                    is_test=msg.is_test,
342                    cause=iec101.DataResCause.INTERROGATED_COUNTER)
343                for data_msg in self._data_msgs.values()
344                if (data_msg and
345                    (msg.asdu_address == 0xFFFF or
346                     msg.asdu_address == data_msg.asdu_address) and
347                    isinstance(data_msg.data, iec101.BinaryCounterData))]
348            conn.send(data_msgs)
349
350            res = msg._replace(
351                cause=iec101.CommandResCause.ACTIVATION_TERMINATION,
352                is_negative_confirm=False)
353            conn.send([res])
354
355        elif msg.cause == iec101.CommandReqCause.DEACTIVATION:
356            res = msg._replace(
357                cause=iec101.CommandResCause.DEACTIVATION_CONFIRMATION,
358                is_negative_confirm=True)
359            conn.send([res])
360
361        else:
362            res = msg._replace(cause=iec101.CommandResCause.UNKNOWN_CAUSE,
363                               is_negative_confirm=True)
364            conn.send([res])
365
366    def _process_read_msg(self, conn_id, conn, msg):
367        res = msg._replace(cause=iec101.ReadResCause.UNKNOWN_TYPE)
368        conn.send([res])
369
370    def _process_clock_sync_msg(self, conn_id, conn, msg):
371        if isinstance(msg.cause, iec101.ClockSyncReqCause):
372            res = msg._replace(
373                cause=iec101.ClockSyncResCause.ACTIVATION_CONFIRMATION,
374                is_negative_confirm=True)
375            conn.send([res])
376
377        else:
378            res = msg._replace(cause=iec101.ClockSyncResCause.UNKNOWN_CAUSE,
379                               is_negative_confirm=True)
380            conn.send([res])
381
382    def _process_test_msg(self, conn_id, conn, msg):
383        res = msg._replace(cause=iec101.ActivationResCause.UNKNOWN_TYPE)
384        conn.send([res])
385
386    def _process_reset_msg(self, conn_id, conn, msg):
387        res = msg._replace(cause=iec101.ActivationResCause.UNKNOWN_TYPE)
388        conn.send([res])
389
390    def _process_parameter_msg(self, conn_id, conn, msg):
391        res = msg._replace(cause=iec101.ParameterResCause.UNKNOWN_TYPE)
392        conn.send([res])
393
394    def _process_parameter_activation_msg(self, conn_id, conn, msg):
395        res = msg._replace(
396            cause=iec101.ParameterActivationResCause.UNKNOWN_TYPE)
397        conn.send([res])
398
399    def _send_data_msg(self, conn, buffer, event_id, data_msg):
400        sent_cb = (functools.partial(buffer.remove, event_id)
401                   if buffer else None)
402        conn.send([data_msg], sent_cb=sent_cb)
403
404
405def cmd_msg_to_event(event_type_prefix: hat.event.common.EventType,
406                     conn_id: int,
407                     msg: iec101.CommandMsg
408                     ) -> hat.event.common.RegisterEvent:
409    command_type = common.get_command_type(msg.command)
410    cause = common.cause_to_json(iec101.CommandReqCause, msg.cause)
411    command = common.command_to_json(msg.command)
412    event_type = (*event_type_prefix, 'gateway', 'command', command_type.value,
413                  str(msg.asdu_address), str(msg.io_address))
414
415    return hat.event.common.RegisterEvent(
416        type=event_type,
417        source_timestamp=None,
418        payload=hat.event.common.EventPayloadJson({
419            'connection_id': conn_id,
420            'is_test': msg.is_test,
421            'cause': cause,
422            'command': command}))
423
424
425def data_msg_from_event(data_key: common.DataKey,
426                        event: hat.event.common.Event
427                        ) -> iec101.DataMsg:
428    time = common.time_from_source_timestamp(event.source_timestamp)
429    cause = common.cause_from_json(iec101.DataResCause,
430                                   event.payload.data['cause'])
431    data = common.data_from_json(data_key.data_type,
432                                 event.payload.data['data'])
433
434    return iec101.DataMsg(is_test=event.payload.data['is_test'],
435                          originator_address=0,
436                          asdu_address=data_key.asdu_address,
437                          io_address=data_key.io_address,
438                          data=data,
439                          time=time,
440                          cause=cause)
441
442
443def cmd_msg_from_event(cmd_key: common.CommandKey,
444                       event: hat.event.common.Event
445                       ) -> iec101.CommandMsg:
446    cause = common.cause_from_json(iec101.CommandResCause,
447                                   event.payload.data['cause'])
448    command = common.command_from_json(cmd_key.cmd_type,
449                                       event.payload.data['command'])
450    is_negative_confirm = event.payload.data['is_negative_confirm']
451
452    return iec101.CommandMsg(is_test=event.payload.data['is_test'],
453                             originator_address=0,
454                             asdu_address=cmd_key.asdu_address,
455                             io_address=cmd_key.io_address,
456                             command=command,
457                             is_negative_confirm=is_negative_confirm,
458                             cause=cause)