hat.drivers.mms

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

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

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

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