hat.drivers.mms

 73async def connect(addr: tcp.Address,
 74                  local_detail_calling: int | None = None,
 75                  request_cb: RequestCb | None = None,
 76                  unconfirmed_cb: UnconfirmedCb | None = None,
 77                  **kwargs
 78                  ) -> 'Connection':
 79    """Connect to MMS server
 80
 81    Additional arguments are passed directly to `hat.drivers.acse.connect`
 82    (`syntax_name_list`, `app_context_name` and `user_data` are set by
 83    this coroutine).
 84
 85    """
 86    initiate_req = 'initiate-RequestPDU', {
 87        'proposedMaxServOutstandingCalling': 5,
 88        'proposedMaxServOutstandingCalled': 5,
 89        'initRequestDetail': {
 90            'proposedVersionNumber': 1,
 91            'proposedParameterCBB': _parameter_cbb,
 92            'servicesSupportedCalling': _service_support}}
 93
 94    if local_detail_calling is not None:
 95        initiate_req[1]['localDetailCalling'] = local_detail_calling
 96
 97    req_user_data = _encode(initiate_req)
 98    conn = await acse.connect(addr=addr,
 99                              syntax_name_list=[_mms_syntax_name],
100                              app_context_name=_mms_app_context_name,
101                              user_data=(_mms_syntax_name, req_user_data),
102                              **kwargs)
103
104    try:
105        res_syntax_name, res_user_data = conn.conn_res_user_data
106        if res_syntax_name != _mms_syntax_name:
107            raise Exception("invalid syntax name")
108
109        initiate_res = _decode(res_user_data)
110        if initiate_res[0] != 'initiate-ResponsePDU':
111            raise Exception("invalid initiate response")
112
113        return Connection(conn=conn,
114                          request_cb=request_cb,
115                          unconfirmed_cb=unconfirmed_cb)
116
117    except Exception:
118        await aio.uncancellable(conn.async_close())
119        raise

122async def listen(connection_cb: ConnectionCb,
123                 addr: tcp.Address = tcp.Address('0.0.0.0', 102),
124                 request_cb: RequestCb | None = None,
125                 unconfirmed_cb: UnconfirmedCb | None = None,
126                 *,
127                 bind_connections: bool = False,
128                 **kwargs
129                 ) -> 'Server':
130    """Create MMS listening server
131
132    Additional arguments are passed directly to `hat.drivers.acse.listen`.
133
134    Args:
135        connection_cb: new connection callback
136        request_cb: received request callback
137        addr: local listening address
138
139    """
140    server = Server()
141    server._connection_cb = connection_cb
142    server._request_cb = request_cb
143    server._unconfirmed_cb = unconfirmed_cb
144    server._bind_connections = bind_connections
145
146    server._log = logger.create_server_logger(mlog, kwargs.get('name'), None)
147
148    server._srv = await acse.listen(validate_cb=server._on_validate,
149                                    connection_cb=server._on_connection,
150                                    addr=addr,
151                                    bind_connections=False,
152                                    **kwargs)
153
154    server._log = logger.create_server_logger(mlog, server._srv.info.name,
155                                              server._srv.info)
156
157    return server

160class Server(aio.Resource):
161    """MMS listening server
162
163    For creating new server see `listen`.
164
165    """
166
167    @property
168    def async_group(self) -> aio.Group:
169        """Async group"""
170        return self._srv.async_group
171
172    @property
173    def info(self) -> tcp.ServerInfo:
174        """Server info"""
175        return self._srv.info
176
177    async def _on_validate(self, syntax_names, user_data):
178        syntax_name, req_user_data = user_data
179        if syntax_name != _mms_syntax_name:
180            raise Exception('invalid mms syntax name')
181
182        initiate_req = _decode(req_user_data)
183        if initiate_req[0] != 'initiate-RequestPDU':
184            raise Exception('invalid initiate request')
185
186        initiate_res = 'initiate-ResponsePDU', {
187            'negotiatedMaxServOutstandingCalling': 5,
188            'negotiatedMaxServOutstandingCalled': 5,
189            'negotiatedDataStructureNestingLevel': 4,  # TODO compatibility
190            'initResponseDetail': {
191                'negotiatedVersionNumber': 1,
192                'negotiatedParameterCBB': _parameter_cbb,
193                'servicesSupportedCalled': _service_support}}
194        if 'localDetailCalling' in initiate_req[1]:
195            initiate_res[1]['localDetailCalled'] = \
196                initiate_req[1]['localDetailCalling']
197
198        res_user_data = _encode(initiate_res)
199        return _mms_syntax_name, res_user_data
200
201    async def _on_connection(self, acse_conn):
202        try:
203            try:
204                conn = Connection(conn=acse_conn,
205                                  request_cb=self._request_cb,
206                                  unconfirmed_cb=self._unconfirmed_cb)
207
208            except Exception:
209                await aio.uncancellable(acse_conn.async_close())
210                raise
211
212            try:
213                await aio.call(self._connection_cb, conn)
214
215            except BaseException:
216                await aio.uncancellable(conn.async_close())
217                raise
218
219        except Exception as e:
220            self._log.error("error creating new incomming connection: %s",
221                            e, exc_info=e)
222            return
223
224        if not self._bind_connections:
225            return
226
227        try:
228            await conn.wait_closed()
229
230        except BaseException:
231            await aio.uncancellable(conn.async_close())
232            raise

235class Connection(aio.Resource):
236    """MMS connection
237
238    For creating new connection see `connect` or `listen`.
239
240    """
241
242    def __init__(self,
243                 conn: acse.Connection,
244                 request_cb: RequestCb,
245                 unconfirmed_cb: UnconfirmedCb):
246        self._conn = conn
247        self._request_cb = request_cb
248        self._unconfirmed_cb = unconfirmed_cb
249        self._loop = asyncio.get_running_loop()
250        self._next_invoke_ids = itertools.count(0)
251        self._response_futures = {}
252        self._close_pdu = 'conclude-RequestPDU', None
253        self._async_group = aio.Group()
254        self._log = logger.create_connection_logger(mlog, conn.info)
255        self._comm_log = logger.CommunicationLogger(mlog, conn.info)
256
257        self.async_group.spawn(aio.call_on_cancel, self._on_close)
258        self.async_group.spawn(self._receive_loop)
259        self.async_group.spawn(aio.call_on_done, conn.wait_closing(),
260                               self.close)
261
262        self.async_group.spawn(aio.call_on_cancel, self._comm_log.log,
263                               common.CommLogAction.CLOSE)
264        self._comm_log.log(common.CommLogAction.OPEN)
265
266    @property
267    def async_group(self) -> aio.Group:
268        """Async group"""
269        return self._async_group
270
271    @property
272    def info(self) -> acse.ConnectionInfo:
273        """Connection info"""
274        return self._conn.info
275
276    async def send_unconfirmed(self, unconfirmed: common.Unconfirmed):
277        """Send unconfirmed message"""
278        if not self.is_open:
279            raise ConnectionError()
280
281        pdu = 'unconfirmed-PDU', {
282            'service': encoder.encode_unconfirmed(unconfirmed)}
283        data = _mms_syntax_name, _encode(pdu)
284
285        self._comm_log.log(common.CommLogAction.SEND, unconfirmed)
286
287        await self._conn.send(data)
288
289    async def send_confirmed(self,
290                             req: common.Request
291                             ) -> common.Response | common.Error:
292        """Send confirmed request and wait for response"""
293        if not self.is_open:
294            raise ConnectionError()
295
296        invoke_id = next(self._next_invoke_ids)
297        pdu = 'confirmed-RequestPDU', {
298            'invokeID': invoke_id,
299            'service': encoder.encode_request(req)}
300        data = _mms_syntax_name, _encode(pdu)
301
302        self._comm_log.log(common.CommLogAction.SEND, req)
303
304        await self._conn.send(data)
305
306        if not self.is_open:
307            raise ConnectionError()
308
309        try:
310            future = self._loop.create_future()
311            self._response_futures[invoke_id] = future
312            return await future
313
314        finally:
315            self._response_futures.pop(invoke_id, None)
316
317    async def _on_close(self):
318        if self._conn.is_open:
319            try:
320                data = _mms_syntax_name, _encode(self._close_pdu)
321
322                await self._conn.send(data)
323                await self._conn.drain()
324
325                # TODO: wait for response in case of conclude-RequestPDU
326
327            except Exception as e:
328                self._log.error("on close error: %s", e, exc_info=e)
329
330        await self._conn.async_close()
331
332    async def _receive_loop(self):
333        try:
334            while True:
335                syntax_name, entity = await self._conn.receive()
336                if syntax_name != _mms_syntax_name:
337                    continue
338
339                name, data = _decode(entity)
340
341                if name == 'unconfirmed-PDU':
342                    await self._process_unconfirmed(data)
343
344                elif name == 'confirmed-RequestPDU':
345                    await self._process_request(data)
346
347                elif name == 'confirmed-ResponsePDU':
348                    await self._process_response(data)
349
350                elif name == 'confirmed-ErrorPDU':
351                    await self._process_error(data)
352
353                elif name == 'conclude-RequestPDU':
354                    self._close_pdu = 'conclude-ResponsePDU', None
355                    break
356
357                else:
358                    raise Exception('unsupported pdu')
359
360        except ConnectionError:
361            pass
362
363        except Exception as e:
364            self._log.error("receive loop error: %s", e, exc_info=e)
365
366        finally:
367            self.close()
368
369            for response_future in self._response_futures.values():
370                if not response_future.done():
371                    response_future.set_exception(ConnectionError())
372
373    async def _process_unconfirmed(self, data):
374        unconfirmed = encoder.decode_unconfirmed(data['service'])
375
376        self._comm_log.log(common.CommLogAction.RECEIVE, unconfirmed)
377
378        if self._unconfirmed_cb is None:
379            raise Exception('unconfirmed_cb not defined')
380
381        await aio.call(self._unconfirmed_cb, self, unconfirmed)
382
383    async def _process_request(self, data):
384        invoke_id = data['invokeID']
385        req = encoder.decode_request(data['service'])
386
387        self._comm_log.log(common.CommLogAction.RECEIVE, req)
388
389        if self._request_cb is None:
390            raise Exception('request_cb not defined')
391
392        res = await aio.call(self._request_cb, self, req)
393
394        if isinstance(res, common.Response):
395            res_pdu = 'confirmed-ResponsePDU', {
396                'invokeID': invoke_id,
397                'service': encoder.encode_response(res)}
398
399        elif isinstance(res, common.Error):
400            res_pdu = 'confirmed-ErrorPDU', {
401                'invokeID': invoke_id,
402                'serviceError': encoder.encode_error(res)}
403
404        else:
405            TypeError('unsupported response/error type')
406
407        res_data = _mms_syntax_name, _encode(res_pdu)
408
409        self._comm_log.log(common.CommLogAction.SEND, res)
410
411        await self._conn.send(res_data)
412
413    async def _process_response(self, data):
414        invoke_id = data['invokeID']
415        res = encoder.decode_response(data['service'])
416
417        self._comm_log.log(common.CommLogAction.RECEIVE, res)
418
419        future = self._response_futures.get(invoke_id)
420        if not future or future.done():
421            self._log.warning("dropping confirmed response (invoke_id: %s)",
422                              invoke_id)
423            return
424
425        future.set_result(res)
426
427    async def _process_error(self, data):
428        invoke_id = data['invokeID']
429        error = encoder.decode_error(data['serviceError'])
430
431        self._comm_log.log(common.CommLogAction.RECEIVE, error)
432
433        future = self._response_futures.get(invoke_id)
434        if not future or future.done():
435            self._log.warning("dropping confirmed error (invoke_id: %s)",
436                              invoke_id)
437            return
438
439        future.set_result(error)

13class DataAccessError(enum.Enum):
14    OBJECT_INVALIDATED = 0
15    HARDWARE_FAULT = 1
16    TEMPORARILY_UNAVAILABLE = 2
17    OBJECT_ACCESS_DENIED = 3
18    OBJECT_UNDEFINED = 4
19    INVALID_ADDRESS = 5
20    TYPE_UNSUPPORTED = 6
21    TYPE_INCONSISTENT = 7
22    OBJECT_ATTRIBUTE_INCONSISTENT = 8
23    OBJECT_ACCESS_UNSUPPORTED = 9
24    OBJECT_NON_EXISTENT = 10
25    OBJECT_VALUE_INVALID = 11

28class ObjectClass(enum.Enum):
29    NAMED_VARIABLE = 0
30    NAMED_VARIABLE_LIST = 2
31    JOURNAL = 8
32    DOMAIN = 9
33    UNDEFINED = 0xFF

38class AaSpecificObjectName(typing.NamedTuple):
39    identifier: str

42class DomainSpecificObjectName(typing.NamedTuple):
43    domain_id: str
44    item_id: str

47class VmdSpecificObjectName(typing.NamedTuple):
48    identifier: str

58class AaSpecificObjectScope(typing.NamedTuple):
59    pass

62class DomainSpecificObjectScope(typing.NamedTuple):
63    identifier: str

66class VmdSpecificObjectScope(typing.NamedTuple):
67    pass

77class ArrayData(typing.NamedTuple):
78    elements: Collection['Data']

81class BcdData(typing.NamedTuple):
82    value: int

85class BinaryTimeData(typing.NamedTuple):
86    value: datetime.datetime

89class BitStringData(typing.NamedTuple):
90    value: Collection[bool]

93class BooleanData(typing.NamedTuple):
94    value: bool

97class BooleanArrayData(typing.NamedTuple):
98    value: Collection[bool]

101class FloatingPointData(typing.NamedTuple):
102    value: float

105class GeneralizedTimeData(typing.NamedTuple):
106    value: str

109class IntegerData(typing.NamedTuple):
110    value: int

113class MmsStringData(typing.NamedTuple):
114    value: str

117class ObjIdData(typing.NamedTuple):
118    value: asn1.ObjectIdentifier

121class OctetStringData(typing.NamedTuple):
122    value: util.Bytes

125class StructureData(typing.NamedTuple):
126    elements: Collection['Data']

129class UnsignedData(typing.NamedTuple):
130    value: int

133class UtcTimeData(typing.NamedTuple):
134    value: datetime.datetime
135    leap_second: bool
136    clock_failure: bool
137    not_synchronized: bool
138    accuracy: int | None
139    """accurate fraction bits [0,24]"""

142class VisibleStringData(typing.NamedTuple):
143    value: str

166class ArrayTypeDescription(typing.NamedTuple):
167    number_of_elements: int
168    element_type: typing.Union['TypeDescription', ObjectName]

171class BcdTypeDescription(typing.NamedTuple):
172    xyz: int

175class BinaryTimeTypeDescription(typing.NamedTuple):
176    xyz: bool

179class BitStringTypeDescription(typing.NamedTuple):
180    xyz: int

183class BooleanTypeDescription(typing.NamedTuple):
184    pass

187class FloatingPointTypeDescription(typing.NamedTuple):
188    format_width: int
189    exponent_width: int

192class GeneralizedTimeTypeDescription(typing.NamedTuple):
193    pass

196class IntegerTypeDescription(typing.NamedTuple):
197    xyz: int

200class MmsStringTypeDescription(typing.NamedTuple):
201    xyz: int

204class ObjIdTypeDescription(typing.NamedTuple):
205    pass

208class OctetStringTypeDescription(typing.NamedTuple):
209    xyz: int

212class StructureTypeDescription(typing.NamedTuple):
213    components: Collection[tuple[str | None,
214                                 typing.Union['TypeDescription', ObjectName]]]

217class UnsignedTypeDescription(typing.NamedTuple):
218    xyz: int

221class UtcTimeTypeDescription(typing.NamedTuple):
222    pass

225class VisibleStringTypeDescription(typing.NamedTuple):
226    xyz: int

248class AddressVariableSpecification(typing.NamedTuple):
249    address: int | str | util.Bytes

252class InvalidatedVariableSpecification(typing.NamedTuple):
253    pass

256class NameVariableSpecification(typing.NamedTuple):
257    name: ObjectName