hat.drivers.mms
Manufacturing Message Specification
1"""Manufacturing Message Specification""" 2 3from hat.drivers.mms.connection import ( 4 ConnectionCb, 5 RequestCb, 6 UnconfirmedCb, 7 connect, 8 listen, 9 Server, 10 Connection) 11 12from hat.drivers.mms.common import ( 13 DataAccessError, 14 ObjectClass, 15 ErrorClass, 16 StatusRequest, 17 GetNameListRequest, 18 IdentifyRequest, 19 GetVariableAccessAttributesRequest, 20 GetNamedVariableListAttributesRequest, 21 ReadRequest, 22 WriteRequest, 23 DefineNamedVariableListRequest, 24 DeleteNamedVariableListRequest, 25 Request, 26 ErrorResponse, 27 StatusResponse, 28 GetNameListResponse, 29 IdentifyResponse, 30 GetVariableAccessAttributesResponse, 31 GetNamedVariableListAttributesResponse, 32 ReadResponse, 33 WriteResponse, 34 DefineNamedVariableListResponse, 35 DeleteNamedVariableListResponse, 36 Response, 37 EventNotificationUnconfirmed, 38 InformationReportUnconfirmed, 39 UnsolicitedStatusUnconfirmed, 40 Unconfirmed, 41 ArrayData, 42 BcdData, 43 BinaryTimeData, 44 BitStringData, 45 BooleanData, 46 BooleanArrayData, 47 FloatingPointData, 48 GeneralizedTimeData, 49 IntegerData, 50 MmsStringData, 51 ObjIdData, 52 OctetStringData, 53 StructureData, 54 UnsignedData, 55 UtcTimeData, 56 VisibleStringData, 57 Data, 58 ArrayTypeDescription, 59 BcdTypeDescription, 60 BinaryTimeTypeDescription, 61 BitStringTypeDescription, 62 BooleanTypeDescription, 63 FloatingPointTypeDescription, 64 GeneralizedTimeTypeDescription, 65 IntegerTypeDescription, 66 MmsStringTypeDescription, 67 ObjIdTypeDescription, 68 OctetStringTypeDescription, 69 StructureTypeDescription, 70 UnsignedTypeDescription, 71 UtcTimeTypeDescription, 72 VisibleStringTypeDescription, 73 TypeDescription, 74 AaSpecificObjectName, 75 DomainSpecificObjectName, 76 VmdSpecificObjectName, 77 ObjectName, 78 AaSpecificObjectScope, 79 DomainSpecificObjectScope, 80 VmdSpecificObjectScope, 81 ObjectScope, 82 AddressVariableSpecification, 83 InvalidatedVariableSpecification, 84 NameVariableSpecification, 85 ScatteredAccessDescriptionVariableSpecification, 86 VariableDescriptionVariableSpecification, 87 VariableSpecification) 88 89 90__all__ = [ 91 'ConnectionCb', 92 'RequestCb', 93 'UnconfirmedCb', 94 'connect', 95 'listen', 96 'Server', 97 'Connection', 98 'DataAccessError', 99 'ObjectClass', 100 'ErrorClass', 101 'StatusRequest', 102 'GetNameListRequest', 103 'IdentifyRequest', 104 'GetVariableAccessAttributesRequest', 105 'GetNamedVariableListAttributesRequest', 106 'ReadRequest', 107 'WriteRequest', 108 'DefineNamedVariableListRequest', 109 'DeleteNamedVariableListRequest', 110 'Request', 111 'ErrorResponse', 112 'StatusResponse', 113 'GetNameListResponse', 114 'IdentifyResponse', 115 'GetVariableAccessAttributesResponse', 116 'GetNamedVariableListAttributesResponse', 117 'ReadResponse', 118 'WriteResponse', 119 'DefineNamedVariableListResponse', 120 'DeleteNamedVariableListResponse', 121 'Response', 122 'EventNotificationUnconfirmed', 123 'InformationReportUnconfirmed', 124 'UnsolicitedStatusUnconfirmed', 125 'Unconfirmed', 126 'ArrayData', 127 'BcdData', 128 'BinaryTimeData', 129 'BitStringData', 130 'BooleanData', 131 'BooleanArrayData', 132 'FloatingPointData', 133 'GeneralizedTimeData', 134 'IntegerData', 135 'MmsStringData', 136 'ObjIdData', 137 'OctetStringData', 138 'StructureData', 139 'UnsignedData', 140 'UtcTimeData', 141 'VisibleStringData', 142 'Data', 143 'ArrayTypeDescription', 144 'BcdTypeDescription', 145 'BinaryTimeTypeDescription', 146 'BitStringTypeDescription', 147 'BooleanTypeDescription', 148 'FloatingPointTypeDescription', 149 'GeneralizedTimeTypeDescription', 150 'IntegerTypeDescription', 151 'MmsStringTypeDescription', 152 'ObjIdTypeDescription', 153 'OctetStringTypeDescription', 154 'StructureTypeDescription', 155 'UnsignedTypeDescription', 156 'UtcTimeTypeDescription', 157 'VisibleStringTypeDescription', 158 'TypeDescription', 159 'AaSpecificObjectName', 160 'DomainSpecificObjectName', 161 'VmdSpecificObjectName', 162 'ObjectName', 163 'AaSpecificObjectScope', 164 'DomainSpecificObjectScope', 165 'VmdSpecificObjectScope', 166 'ObjectScope', 167 'AddressVariableSpecification', 168 'InvalidatedVariableSpecification', 169 'NameVariableSpecification', 170 'ScatteredAccessDescriptionVariableSpecification', 171 'VariableDescriptionVariableSpecification', 172 'VariableSpecification']
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
Connect to MMS server
Additional arguments are passed directly to hat.drivers.acse.connect
(syntax_name_list
, app_context_name
and user_data
are set by
this coroutine).
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
Create MMS listening server
Additional arguments are passed directly to hat.drivers.acse.listen
.
Arguments:
- connection_cb: new connection callback
- request_cb: received request callback
- addr: local listening address
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
MMS listening server
For creating new server see listen
.
155 @property 156 def async_group(self) -> aio.Group: 157 """Async group""" 158 return self._srv.async_group
Async group
160 @property 161 def addresses(self) -> list[tcp.Address]: 162 """Listening addresses""" 163 return self._srv.addresses
Listening addresses
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)
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)
247 @property 248 def async_group(self) -> aio.Group: 249 """Async group""" 250 return self._async_group
Async group
252 @property 253 def info(self) -> acse.ConnectionInfo: 254 """Connection info""" 255 return self._conn.info
Connection info
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)
Send unconfirmed message
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)
Send confirmed request and wait for response
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
An enumeration.
76class ObjectClass(enum.Enum): 77 NAMED_VARIABLE = 0 78 NAMED_VARIABLE_LIST = 2 79 JOURNAL = 8 80 DOMAIN = 9 81 UNDEFINED = 0xFF
An enumeration.
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'
An enumeration.
StatusRequest()
105@request 106class GetNameListRequest(typing.NamedTuple): 107 object_class: ObjectClass 108 object_scope: ObjectScope 109 continue_after: str | None
GetNameListRequest(object_class, object_scope, continue_after)
Create new instance of GetNameListRequest(object_class, object_scope, continue_after)
IdentifyRequest()
117@request 118class GetVariableAccessAttributesRequest(typing.NamedTuple): 119 value: ObjectName | int | str | util.Bytes
GetVariableAccessAttributesRequest(value,)
122@request 123class GetNamedVariableListAttributesRequest(typing.NamedTuple): 124 value: ObjectName
GetNamedVariableListAttributesRequest(value,)
127@request 128class ReadRequest(typing.NamedTuple): 129 value: list[VariableSpecification] | ObjectName
ReadRequest(value,)
132@request 133class WriteRequest(typing.NamedTuple): 134 specification: list[VariableSpecification] | ObjectName 135 data: list[Data]
WriteRequest(specification, data)
138@request 139class DefineNamedVariableListRequest(typing.NamedTuple): 140 name: ObjectName 141 specification: list[VariableSpecification]
DefineNamedVariableListRequest(name, specification)
DeleteNamedVariableListRequest(names,)
Helper class that provides a standard way to create an ABC using inheritance.
ErrorResponse(error_class, value)
Create new instance of ErrorResponse(error_class, value)
StatusResponse(logical, physical)
161@response 162class GetNameListResponse(typing.NamedTuple): 163 identifiers: list[str] 164 more_follows: bool
GetNameListResponse(identifiers, more_follows)
167@response 168class IdentifyResponse(typing.NamedTuple): 169 vendor: str 170 model: str 171 revision: str 172 syntaxes: list[asn1.ObjectIdentifier] | None
IdentifyResponse(vendor, model, revision, syntaxes)
175@response 176class GetVariableAccessAttributesResponse(typing.NamedTuple): 177 mms_deletable: bool 178 type_description: TypeDescription
GetVariableAccessAttributesResponse(mms_deletable, type_description)
181@response 182class GetNamedVariableListAttributesResponse(typing.NamedTuple): 183 mms_deletable: bool 184 specification: list[VariableSpecification]
GetNamedVariableListAttributesResponse(mms_deletable, specification)
ReadResponse(results,)
Create new instance of ReadResponse(results,)
WriteResponse(results,)
DefineNamedVariableListResponse()
202@response 203class DeleteNamedVariableListResponse(typing.NamedTuple): 204 matched: int 205 deleted: int
DeleteNamedVariableListResponse(matched, deleted)
Helper class that provides a standard way to create an ABC using inheritance.
208@unconfirmed 209class EventNotificationUnconfirmed(typing.NamedTuple): 210 enrollment: ObjectName 211 condition: ObjectName 212 severity: int 213 time: Data | int | None
EventNotificationUnconfirmed(enrollment, condition, severity, time)
216@unconfirmed 217class InformationReportUnconfirmed(typing.NamedTuple): 218 specification: list[VariableSpecification] | ObjectName 219 data: list[DataAccessError | Data]
InformationReportUnconfirmed(specification, data)
Create new instance of InformationReportUnconfirmed(specification, data)
222@unconfirmed 223class UnsolicitedStatusUnconfirmed(typing.NamedTuple): 224 logical: int 225 physical: int
UnsolicitedStatusUnconfirmed(logical, physical)
Helper class that provides a standard way to create an ABC using inheritance.
ArrayData(elements,)
BcdData(value,)
BinaryTimeData(value,)
BitStringData(value,)
BooleanData(value,)
BooleanArrayData(value,)
FloatingPointData(value,)
GeneralizedTimeData(value,)
IntegerData(value,)
MmsStringData(value,)
ObjIdData(value,)
OctetStringData(value,)
StructureData(elements,)
UnsignedData(value,)
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]"""
UtcTimeData(value, leap_second, clock_failure, not_synchronized, accuracy)
VisibleStringData(value,)
Helper class that provides a standard way to create an ABC using inheritance.
313@type_description 314class ArrayTypeDescription(typing.NamedTuple): 315 number_of_elements: int 316 element_type: TypeDescription | ObjectName
ArrayTypeDescription(number_of_elements, element_type)
BcdTypeDescription(xyz,)
BinaryTimeTypeDescription(xyz,)
BitStringTypeDescription(xyz,)
BooleanTypeDescription()
339@type_description 340class FloatingPointTypeDescription(typing.NamedTuple): 341 format_width: int 342 exponent_width: int
FloatingPointTypeDescription(format_width, exponent_width)
GeneralizedTimeTypeDescription()
IntegerTypeDescription(xyz,)
MmsStringTypeDescription(xyz,)
ObjIdTypeDescription()
OctetStringTypeDescription(xyz,)
370@type_description 371class StructureTypeDescription(typing.NamedTuple): 372 components: list[tuple[str | None, TypeDescription | ObjectName]]
StructureTypeDescription(components,)
UnsignedTypeDescription(xyz,)
UtcTimeTypeDescription()
VisibleStringTypeDescription(xyz,)
Helper class that provides a standard way to create an ABC using inheritance.
AaSpecificObjectName(identifier,)
395@object_name 396class DomainSpecificObjectName(typing.NamedTuple): 397 domain_id: str 398 item_id: str
DomainSpecificObjectName(domain_id, item_id)
VmdSpecificObjectName(identifier,)
Helper class that provides a standard way to create an ABC using inheritance.
AaSpecificObjectScope()
DomainSpecificObjectScope(identifier,)
VmdSpecificObjectScope()
Helper class that provides a standard way to create an ABC using inheritance.
421@variable_specification 422class AddressVariableSpecification(typing.NamedTuple): 423 address: int | str | asn1.Bytes
AddressVariableSpecification(address,)
InvalidatedVariableSpecification()
431@variable_specification 432class NameVariableSpecification(typing.NamedTuple): 433 name: ObjectName
NameVariableSpecification(name,)
436@variable_specification 437class ScatteredAccessDescriptionVariableSpecification(typing.NamedTuple): 438 specifications: list[VariableSpecification]
ScatteredAccessDescriptionVariableSpecification(specifications,)
441@variable_specification 442class VariableDescriptionVariableSpecification(typing.NamedTuple): 443 address: int | str | asn1.Bytes 444 type_specification: TypeDescription | ObjectName
VariableDescriptionVariableSpecification(address, type_specification)
Helper class that provides a standard way to create an ABC using inheritance.