hat.drivers.mms

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

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

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

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

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

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

36class AaSpecificObjectName(typing.NamedTuple):
37    identifier: str

40class DomainSpecificObjectName(typing.NamedTuple):
41    domain_id: str
42    item_id: str

45class VmdSpecificObjectName(typing.NamedTuple):
46    identifier: str

56class AaSpecificObjectScope(typing.NamedTuple):
57    pass

60class DomainSpecificObjectScope(typing.NamedTuple):
61    identifier: str

64class VmdSpecificObjectScope(typing.NamedTuple):
65    pass

75class ArrayData(typing.NamedTuple):
76    elements: Collection['Data']

79class BcdData(typing.NamedTuple):
80    value: int

83class BinaryTimeData(typing.NamedTuple):
84    value: datetime.datetime

87class BitStringData(typing.NamedTuple):
88    value: Collection[bool]

91class BooleanData(typing.NamedTuple):
92    value: bool

95class BooleanArrayData(typing.NamedTuple):
96    value: Collection[bool]

 99class FloatingPointData(typing.NamedTuple):
100    value: float

103class GeneralizedTimeData(typing.NamedTuple):
104    value: str

107class IntegerData(typing.NamedTuple):
108    value: int

111class MmsStringData(typing.NamedTuple):
112    value: str

115class ObjIdData(typing.NamedTuple):
116    value: asn1.ObjectIdentifier

119class OctetStringData(typing.NamedTuple):
120    value: util.Bytes

123class StructureData(typing.NamedTuple):
124    elements: Collection['Data']

127class UnsignedData(typing.NamedTuple):
128    value: int

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

140class VisibleStringData(typing.NamedTuple):
141    value: str

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

169class BcdTypeDescription(typing.NamedTuple):
170    xyz: int

173class BinaryTimeTypeDescription(typing.NamedTuple):
174    xyz: bool

177class BitStringTypeDescription(typing.NamedTuple):
178    xyz: int

181class BooleanTypeDescription(typing.NamedTuple):
182    pass

185class FloatingPointTypeDescription(typing.NamedTuple):
186    format_width: int
187    exponent_width: int

190class GeneralizedTimeTypeDescription(typing.NamedTuple):
191    pass

194class IntegerTypeDescription(typing.NamedTuple):
195    xyz: int

198class MmsStringTypeDescription(typing.NamedTuple):
199    xyz: int

202class ObjIdTypeDescription(typing.NamedTuple):
203    pass

206class OctetStringTypeDescription(typing.NamedTuple):
207    xyz: int

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

215class UnsignedTypeDescription(typing.NamedTuple):
216    xyz: int

219class UtcTimeTypeDescription(typing.NamedTuple):
220    pass

223class VisibleStringTypeDescription(typing.NamedTuple):
224    xyz: int

246class AddressVariableSpecification(typing.NamedTuple):
247    address: int | str | util.Bytes

250class InvalidatedVariableSpecification(typing.NamedTuple):
251    pass

254class NameVariableSpecification(typing.NamedTuple):
255    name: ObjectName

258class ScatteredAccessDescriptionVariableSpecification(typing.NamedTuple):
259    specifications: Collection['VariableSpecification']

262class VariableDescriptionVariableSpecification(typing.NamedTuple):
263    address: int | str | util.Bytes
264    type_specification: TypeDescription | ObjectName

277class StatusRequest(typing.NamedTuple):
278    pass

281class GetNameListRequest(typing.NamedTuple):
282    object_class: ObjectClass
283    object_scope: ObjectScope
284    continue_after: str | None

287class IdentifyRequest(typing.NamedTuple):
288    pass

291class GetVariableAccessAttributesRequest(typing.NamedTuple):
292    value: ObjectName | int | str | util.Bytes