hat.drivers.mms

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

115async def listen(connection_cb: ConnectionCb,
116                 addr: tcp.Address = tcp.Address('0.0.0.0', 102),
117                 request_cb: RequestCb | None = None,
118                 unconfirmed_cb: UnconfirmedCb | None = None,
119                 *,
120                 bind_connections: bool = False,
121                 **kwargs
122                 ) -> 'Server':
123    """Create MMS listening server
124
125    Additional arguments are passed directly to `hat.drivers.acse.listen`.
126
127    Args:
128        connection_cb: new connection callback
129        request_cb: received request callback
130        addr: local listening address
131
132    """
133    server = Server()
134    server._connection_cb = connection_cb
135    server._request_cb = request_cb
136    server._unconfirmed_cb = unconfirmed_cb
137    server._bind_connections = bind_connections
138
139    server._srv = await acse.listen(server._on_validate,
140                                    server._on_connection,
141                                    addr,
142                                    bind_connections=False,
143                                    **kwargs)
144
145    return server

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

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

61class DataAccessError(enum.Enum):
62    OBJECT_INVALIDATED = 0
63    HARDWARE_FAULT = 1
64    TEMPORARILY_UNAVAILABLE = 2
65    OBJECT_ACCESS_DENIED = 3
66    OBJECT_UNDEFINED = 4
67    INVALID_ADDRESS = 5
68    TYPE_UNSUPPORTED = 6
69    TYPE_INCONSISTENT = 7
70    OBJECT_ATTRIBUTE_INCONSISTENT = 8
71    OBJECT_ACCESS_UNSUPPORTED = 9
72    OBJECT_NON_EXISTENT = 10
73    OBJECT_VALUE_INVALID = 11

76class ObjectClass(enum.Enum):
77    NAMED_VARIABLE = 0
78    NAMED_VARIABLE_LIST = 2
79    JOURNAL = 8
80    DOMAIN = 9
81    UNDEFINED = 0xFF

84class ErrorClass(enum.Enum):
85    ACCESS = 'access'
86    APPLICATION_REFERENCE = 'application-reference'
87    CANCEL = 'cancel'
88    CONCLUDE = 'conclude'
89    DEFINITION = 'definition'
90    FILE = 'file'
91    INITIATE = 'initiate'
92    OTHERS = 'others'
93    RESOURCE = 'resource'
94    SERVICE = 'service'
95    SERVICE_PREEMPT = 'service-preempt'
96    TIME_RESOLUTION = 'time-resolution'
97    VMD_STATE = 'vmd-state'

100@request
101class StatusRequest(typing.NamedTuple):
102    pass

105@request
106class GetNameListRequest(typing.NamedTuple):
107    object_class: ObjectClass
108    object_scope: ObjectScope
109    continue_after: str | None

112@request
113class IdentifyRequest(typing.NamedTuple):
114    pass

117@request
118class GetVariableAccessAttributesRequest(typing.NamedTuple):
119    value: ObjectName | int | str | util.Bytes

122@request
123class GetNamedVariableListAttributesRequest(typing.NamedTuple):
124    value: ObjectName

127@request
128class ReadRequest(typing.NamedTuple):
129    value: list[VariableSpecification] | ObjectName

132@request
133class WriteRequest(typing.NamedTuple):
134    specification: list[VariableSpecification] | ObjectName
135    data: list[Data]

138@request
139class DefineNamedVariableListRequest(typing.NamedTuple):
140    name: ObjectName
141    specification: list[VariableSpecification]

144@request
145class DeleteNamedVariableListRequest(typing.NamedTuple):
146    names: list[ObjectName]

149@response
150class ErrorResponse(typing.NamedTuple):
151    error_class: ErrorClass
152    value: int

155@response
156class StatusResponse(typing.NamedTuple):
157    logical: int
158    physical: int

161@response
162class GetNameListResponse(typing.NamedTuple):
163    identifiers: list[str]
164    more_follows: bool

167@response
168class IdentifyResponse(typing.NamedTuple):
169    vendor: str
170    model: str
171    revision: str
172    syntaxes: list[asn1.ObjectIdentifier] | None

175@response
176class GetVariableAccessAttributesResponse(typing.NamedTuple):
177    mms_deletable: bool
178    type_description: TypeDescription

181@response
182class GetNamedVariableListAttributesResponse(typing.NamedTuple):
183    mms_deletable: bool
184    specification: list[VariableSpecification]

187@response
188class ReadResponse(typing.NamedTuple):
189    results: list[DataAccessError | Data]

192@response
193class WriteResponse(typing.NamedTuple):
194    results: list[DataAccessError | None]

197@response
198class DefineNamedVariableListResponse(typing.NamedTuple):
199    pass

202@response
203class DeleteNamedVariableListResponse(typing.NamedTuple):
204    matched: int
205    deleted: int

208@unconfirmed
209class EventNotificationUnconfirmed(typing.NamedTuple):
210    enrollment: ObjectName
211    condition: ObjectName
212    severity: int
213    time: Data | int | None

216@unconfirmed
217class InformationReportUnconfirmed(typing.NamedTuple):
218    specification: list[VariableSpecification] | ObjectName
219    data: list[DataAccessError | Data]

222@unconfirmed
223class UnsolicitedStatusUnconfirmed(typing.NamedTuple):
224    logical: int
225    physical: int

228@data
229class ArrayData(typing.NamedTuple):
230    elements: list[Data]

233@data
234class BcdData(typing.NamedTuple):
235    value: int

238@data
239class BinaryTimeData(typing.NamedTuple):
240    value: datetime.datetime

243@data
244class BitStringData(typing.NamedTuple):
245    value: list[bool]

248@data
249class BooleanData(typing.NamedTuple):
250    value: bool

253@data
254class BooleanArrayData(typing.NamedTuple):
255    value: list[bool]

258@data
259class FloatingPointData(typing.NamedTuple):
260    value: float

263@data
264class GeneralizedTimeData(typing.NamedTuple):
265    value: str

268@data
269class IntegerData(typing.NamedTuple):
270    value: int

273@data
274class MmsStringData(typing.NamedTuple):
275    value: str

278@data
279class ObjIdData(typing.NamedTuple):
280    value: asn1.ObjectIdentifier

283@data
284class OctetStringData(typing.NamedTuple):
285    value: asn1.Bytes

288@data
289class StructureData(typing.NamedTuple):
290    elements: list[Data]

293@data
294class UnsignedData(typing.NamedTuple):
295    value: int

298@data
299class UtcTimeData(typing.NamedTuple):
300    value: datetime.datetime
301    leap_second: bool
302    clock_failure: bool
303    not_synchronized: bool
304    accuracy: int | None
305    """accurate fraction bits [0,24]"""

308@data
309class VisibleStringData(typing.NamedTuple):
310    value: str

313@type_description
314class ArrayTypeDescription(typing.NamedTuple):
315    number_of_elements: int
316    element_type: TypeDescription | ObjectName

319@type_description
320class BcdTypeDescription(typing.NamedTuple):
321    xyz: int

324@type_description
325class BinaryTimeTypeDescription(typing.NamedTuple):
326    xyz: bool

329@type_description
330class BitStringTypeDescription(typing.NamedTuple):
331    xyz: int

334@type_description
335class BooleanTypeDescription(typing.NamedTuple):
336    pass

339@type_description
340class FloatingPointTypeDescription(typing.NamedTuple):
341    format_width: int
342    exponent_width: int