hat.gateway.devices.iec104.master

  1"""IEC 60870-5-104 master device"""
  2
  3from collections.abc import Collection
  4import asyncio
  5import collections
  6import contextlib
  7import datetime
  8import enum
  9import logging
 10
 11from hat import aio
 12from hat.drivers import iec104
 13from hat.drivers import tcp
 14import hat.event.common
 15import hat.event.eventer
 16
 17from hat.gateway.devices.iec104 import common
 18from hat.gateway.devices.iec104 import ssl
 19
 20
 21mlog: logging.Logger = logging.getLogger(__name__)
 22
 23
 24class Iec104MasterDevice(common.Device):
 25
 26    def __init__(self,
 27                 conf: common.DeviceConf,
 28                 eventer_client: hat.event.eventer.Client,
 29                 event_type_prefix: common.EventTypePrefix):
 30        self._eventer_client = eventer_client
 31        self._event_type_prefix = event_type_prefix
 32        self._conn = None
 33        self._async_group = aio.Group()
 34
 35        ssl_ctx = (
 36            ssl.create_ssl_ctx(conf['security'], ssl.SslProtocol.TLS_CLIENT)
 37            if conf['security'] else None)
 38
 39        self.async_group.spawn(self._connection_loop, conf, ssl_ctx)
 40
 41    @property
 42    def async_group(self) -> aio.Group:
 43        return self._async_group
 44
 45    async def process_events(self, events: Collection[hat.event.common.Event]):
 46        msgs = collections.deque()
 47        for event in events:
 48            try:
 49                mlog.debug('received event: %s', event)
 50                msg = _msg_from_event(self._event_type_prefix, event)
 51                msgs.append(msg)
 52
 53            except Exception as e:
 54                mlog.warning('error processing event: %s',
 55                             e, exc_info=e)
 56                continue
 57
 58        if not msgs:
 59            return
 60
 61        if not self._conn or not self._conn.is_open:
 62            mlog.warning('connection closed: %s events ignored',
 63                         len(msgs))
 64            return
 65
 66        try:
 67            await self._conn.send(msgs)
 68            mlog.debug('%s messages sent', len(msgs))
 69
 70        except ConnectionError as e:
 71            mlog.warning('error sending messages: %s', e, exc_info=e)
 72
 73    async def _connection_loop(self, conf, ssl_ctx):
 74
 75        async def cleanup():
 76            with contextlib.suppress(ConnectionError):
 77                await self._register_status('DISCONNECTED')
 78
 79            if self._conn:
 80                await self._conn.async_close()
 81
 82        try:
 83            while True:
 84                await self._register_status('CONNECTING')
 85                for address in conf['remote_addresses']:
 86                    try:
 87                        self._conn = await iec104.connect(
 88                            addr=tcp.Address(host=address['host'],
 89                                             port=address['port']),
 90                            response_timeout=conf['response_timeout'],
 91                            supervisory_timeout=conf['supervisory_timeout'],
 92                            test_timeout=conf['test_timeout'],
 93                            send_window_size=conf['send_window_size'],
 94                            receive_window_size=conf['receive_window_size'],
 95                            ssl=ssl_ctx)
 96
 97                        if conf['security']:
 98                            try:
 99                                ssl.init_security(conf['security'], self._conn)
100
101                            except Exception:
102                                await aio.uncancellable(
103                                    self._conn.async_close())
104                                raise
105
106                        break
107
108                    except Exception as e:
109                        mlog.warning('connection failed: %s', e, exc_info=e)
110
111                else:
112                    await self._register_status('DISCONNECTED')
113                    await asyncio.sleep(conf['reconnect_delay'])
114                    continue
115
116                await self._register_status('CONNECTED')
117                self.async_group.spawn(self._receive_loop, self._conn)
118                if conf['time_sync_delay'] is not None:
119                    self.async_group.spawn(self._time_sync_loop, self._conn,
120                                           conf['time_sync_delay'])
121
122                await self._conn.wait_closed()
123                await self._register_status('DISCONNECTED')
124                self._conn = None
125
126        except ConnectionError:
127            pass
128
129        except Exception as e:
130            mlog.error('connection loop error: %s', e, exc_info=e)
131
132        finally:
133            mlog.debug('closing connection loop')
134            self.close()
135            await aio.uncancellable(cleanup())
136
137    async def _receive_loop(self, conn):
138        try:
139            while True:
140                try:
141                    msgs = await conn.receive()
142
143                except iec104.AsduTypeError as e:
144                    mlog.warning("asdu type error: %s", e)
145                    continue
146
147                events = collections.deque()
148                for msg in msgs:
149                    try:
150                        mlog.debug('received message: %s', msg)
151                        if isinstance(msg, iec104.ClockSyncMsg):
152                            continue
153
154                        event = _msg_to_event(self._event_type_prefix, msg)
155                        events.append(event)
156
157                    except Exception as e:
158                        mlog.warning('error processing message: %s',
159                                     e, exc_info=e)
160                        continue
161
162                if not events:
163                    continue
164
165                await self._eventer_client.register(events)
166                mlog.debug('%s events registered', len(events))
167
168        except ConnectionError:
169            mlog.debug('connection closed')
170
171        except Exception as e:
172            mlog.error('receive loop error: %s', e, exc_info=e)
173
174        finally:
175            mlog.debug('closing receive loop')
176            conn.close()
177
178    async def _time_sync_loop(self, conn, time_sync_delay):
179        try:
180            while True:
181                time_now = datetime.datetime.now(datetime.timezone.utc)
182                time_iec104_now = iec104.time_from_datetime(time_now)
183
184                msg = iec104.ClockSyncMsg(
185                    is_test=False,
186                    originator_address=0,
187                    asdu_address=0xFFFF,
188                    time=time_iec104_now,
189                    is_negative_confirm=False,
190                    cause=iec104.ClockSyncReqCause.ACTIVATION)
191                await conn.send([msg])
192
193                await conn.drain()
194                mlog.debug('time sync sent %s', time_iec104_now)
195
196                await asyncio.sleep(time_sync_delay)
197
198        except ConnectionError:
199            mlog.debug('connection closed')
200
201        except Exception as e:
202            mlog.error('time sync loop error: %s', e, exc_info=e)
203
204        finally:
205            mlog.debug('closing time sync loop')
206            conn.close()
207
208    async def _register_status(self, status):
209        event = hat.event.common.RegisterEvent(
210            type=(*self._event_type_prefix, 'gateway', 'status'),
211            source_timestamp=None,
212            payload=hat.event.common.EventPayloadJson(status))
213
214        await self._eventer_client.register([event])
215        mlog.debug('registered status %s', status)
216
217
218info: common.DeviceInfo = common.DeviceInfo(
219    type="iec104_master",
220    create=Iec104MasterDevice,
221    json_schema_id="hat-gateway://iec104.yaml#/$defs/master",
222    json_schema_repo=common.json_schema_repo)
223
224
225def _msg_to_event(event_type_prefix, msg):
226    if isinstance(msg, iec104.DataMsg):
227        return _data_to_event(event_type_prefix, msg)
228
229    if isinstance(msg, iec104.CommandMsg):
230        return _command_to_event(event_type_prefix, msg)
231
232    if isinstance(msg, iec104.InterrogationMsg):
233        return _interrogation_to_event(event_type_prefix, msg)
234
235    if isinstance(msg, iec104.CounterInterrogationMsg):
236        return _counter_interrogation_to_event(event_type_prefix, msg)
237
238    raise Exception('unsupported message type')
239
240
241def _data_to_event(event_type_prefix, msg):
242    data_type = common.get_data_type(msg.data)
243    cause = _cause_to_json(iec104.DataResCause, msg.cause)
244    data = common.data_to_json(msg.data)
245    event_type = (*event_type_prefix, 'gateway', 'data', data_type.value,
246                  str(msg.asdu_address), str(msg.io_address))
247    source_timestamp = common.time_to_source_timestamp(msg.time)
248
249    return hat.event.common.RegisterEvent(
250        type=event_type,
251        source_timestamp=source_timestamp,
252        payload=hat.event.common.EventPayloadJson({
253            'is_test': msg.is_test,
254            'cause': cause,
255            'data': data}))
256
257
258def _command_to_event(event_type_prefix, msg):
259    command_type = common.get_command_type(msg.command)
260    cause = _cause_to_json(iec104.CommandResCause, msg.cause)
261    command = common.command_to_json(msg.command)
262    event_type = (*event_type_prefix, 'gateway', 'command', command_type.value,
263                  str(msg.asdu_address), str(msg.io_address))
264    source_timestamp = common.time_to_source_timestamp(msg.time)
265
266    return hat.event.common.RegisterEvent(
267        type=event_type,
268        source_timestamp=source_timestamp,
269        payload=hat.event.common.EventPayloadJson({
270            'is_test': msg.is_test,
271            'is_negative_confirm': msg.is_negative_confirm,
272            'cause': cause,
273            'command': command}))
274
275
276def _interrogation_to_event(event_type_prefix, msg):
277    cause = _cause_to_json(iec104.CommandResCause, msg.cause)
278    event_type = (*event_type_prefix, 'gateway', 'interrogation',
279                  str(msg.asdu_address))
280
281    return hat.event.common.RegisterEvent(
282        type=event_type,
283        source_timestamp=None,
284        payload=hat.event.common.EventPayloadJson({
285            'is_test': msg.is_test,
286            'is_negative_confirm': msg.is_negative_confirm,
287            'request': msg.request,
288            'cause': cause}))
289
290
291def _counter_interrogation_to_event(event_type_prefix, msg):
292    cause = _cause_to_json(iec104.CommandResCause, msg.cause)
293    event_type = (*event_type_prefix, 'gateway', 'counter_interrogation',
294                  str(msg.asdu_address))
295
296    return hat.event.common.RegisterEvent(
297        type=event_type,
298        source_timestamp=None,
299        payload=hat.event.common.EventPayloadJson({
300            'is_test': msg.is_test,
301            'is_negative_confirm': msg.is_negative_confirm,
302            'request': msg.request,
303            'freeze': msg.freeze.name,
304            'cause': cause}))
305
306
307def _msg_from_event(event_type_prefix, event):
308    suffix = event.type[len(event_type_prefix):]
309
310    if suffix[:2] == ('system', 'command'):
311        cmd_type_str, asdu_address_str, io_address_str = suffix[2:]
312        cmd_key = common.CommandKey(cmd_type=common.CommandType(cmd_type_str),
313                                    asdu_address=int(asdu_address_str),
314                                    io_address=int(io_address_str))
315        return _command_from_event(cmd_key, event)
316
317    if suffix[:2] == ('system', 'interrogation'):
318        asdu_address = int(suffix[2])
319        return _interrogation_from_event(asdu_address, event)
320
321    if suffix[:2] == ('system', 'counter_interrogation'):
322        asdu_address = int(suffix[2])
323        return _counter_interrogation_from_event(asdu_address, event)
324
325    raise Exception('unsupported event type')
326
327
328def _command_from_event(cmd_key, event):
329    time = common.time_from_source_timestamp(event.source_timestamp)
330    cause = _cause_from_json(iec104.CommandReqCause,
331                             event.payload.data['cause'])
332    command = common.command_from_json(cmd_key.cmd_type,
333                                       event.payload.data['command'])
334
335    return iec104.CommandMsg(is_test=event.payload.data['is_test'],
336                             originator_address=0,
337                             asdu_address=cmd_key.asdu_address,
338                             io_address=cmd_key.io_address,
339                             command=command,
340                             is_negative_confirm=False,
341                             time=time,
342                             cause=cause)
343
344
345def _interrogation_from_event(asdu_address, event):
346    cause = _cause_from_json(iec104.CommandReqCause,
347                             event.payload.data['cause'])
348
349    return iec104.InterrogationMsg(is_test=event.payload.data['is_test'],
350                                   originator_address=0,
351                                   asdu_address=asdu_address,
352                                   request=event.payload.data['request'],
353                                   is_negative_confirm=False,
354                                   cause=cause)
355
356
357def _counter_interrogation_from_event(asdu_address, event):
358    freeze = iec104.FreezeCode[event.payload.data['freeze']]
359    cause = _cause_from_json(iec104.CommandReqCause,
360                             event.payload.data['cause'])
361
362    return iec104.CounterInterrogationMsg(
363        is_test=event.payload.data['is_test'],
364        originator_address=0,
365        asdu_address=asdu_address,
366        request=event.payload.data['request'],
367        freeze=freeze,
368        is_negative_confirm=False,
369        cause=cause)
370
371
372def _cause_to_json(cls, cause):
373    return (cause.name if isinstance(cause, cls) else
374            cause.value if isinstance(cause, enum.Enum) else
375            cause)
376
377
378def _cause_from_json(cls, cause):
379    return cls[cause] if isinstance(cause, str) else cause