hat.drivers.mms

 73async def connect(addr: net.StreamAddress,
 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: net.StreamAddress = net.TcpAddress('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    server._log = mlog
146
147    server._srv = await acse.listen(validate_cb=server._on_validate,
148                                    connection_cb=server._on_connection,
149                                    addr=addr,
150                                    bind_connections=False,
151                                    **kwargs)
152
153    server._log = logger.create_server_logger(mlog, server._srv.info)
154
155    return server

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

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

260class ScatteredAccessDescriptionVariableSpecification(typing.NamedTuple):
261    specifications: Collection['VariableSpecification']