MMS

Python implementation

class hat.drivers.mms.Address(host, port)

Bases: tuple

Create new instance of Address(host, port)

host

host name

Type

str

port

TCP port

Type

int

class hat.drivers.mms.ConnectionInfo(local_addr, local_tsel, local_ssel, local_psel, local_ap_title, local_ae_qualifier, remote_addr, remote_tsel, remote_ssel, remote_psel, remote_ap_title, remote_ae_qualifier)

Bases: tuple

Create new instance of ConnectionInfo(local_addr, local_tsel, local_ssel, local_psel, local_ap_title, local_ae_qualifier, remote_addr, remote_tsel, remote_ssel, remote_psel, remote_ap_title, remote_ae_qualifier)

local_addr

local address

Type

Address

local_ae_qualifier

local AE qualifier

Type

Optional[int]

local_ap_title

local AP title

Type

Optional[asn1.ObjectIdentifier]

local_psel

local COPP selector

Type

Optional[int]

local_ssel

local COSP selector

Type

Optional[int]

local_tsel

local COTP selector

Type

Optional[int]

remote_addr

remote address

Type

Address

remote_ae_qualifier

remote AE qualifier

Type

Optional[int]

remote_ap_title

remote AP title

Type

Optional[asn1.ObjectIdentifier]

remote_psel

remote COPP selector

Type

Optional[int]

remote_ssel

remote COSP selector

Type

Optional[int]

remote_tsel

remote COTP selector

Type

Optional[int]

hat.drivers.mms.RequestCb

The central part of internal API.

This represents a generic version of type ‘origin’ with type arguments ‘params’. There are two kind of these aliases: user defined and special. The special ones are wrappers around builtin collections and ABCs in collections.abc. These must have ‘name’ always set. If ‘inst’ is False, then the alias can’t be instantiated, this is used by e.g. typing.List and typing.Dict.

alias of Callable[[Connection, abc.Request], Union[abc.Response, Awaitable[abc.Response]]]

hat.drivers.mms.ConnectionCb

The central part of internal API.

This represents a generic version of type ‘origin’ with type arguments ‘params’. There are two kind of these aliases: user defined and special. The special ones are wrappers around builtin collections and ABCs in collections.abc. These must have ‘name’ always set. If ‘inst’ is False, then the alias can’t be instantiated, this is used by e.g. typing.List and typing.Dict.

alias of Callable[[Connection], Union[None, Awaitable[None]]]

async hat.drivers.mms.connect(request_cb, addr, local_tsel=None, remote_tsel=None, local_ssel=None, remote_ssel=None, local_psel=None, remote_psel=None, local_ap_title=None, remote_ap_title=None, local_ae_qualifier=None, remote_ae_qualifier=None, user_data=None)

Connect to ACSE server

Parameters

  • request_cb (RequestCb) – received request callback

  • addr (Address) – remote address

  • local_tsel (Optional[int]) – local COTP selector

  • remote_tsel (Optional[int]) – remote COTP selector

  • local_ssel (Optional[int]) – local COSP selector

  • remote_ssel (Optional[int]) – remote COSP selector

  • local_psel (Optional[int]) – local COPP selector

  • remote_psel (Optional[int]) – remote COPP selector

  • local_ap_title (Optional[asn1.ObjectIdentifier]) – local AP title

  • remote_ap_title (Optional[asn1.ObjectIdentifier]) – remote AP title

  • local_ae_qualifier (Optional[int]) – local AE qualifier

  • remote_ae_qualifier (Optional[int]) – remote AE qualifier

  • user_data (Optional[IdentifiedEntity]]) – connect request user data

Returns

Connection

async hat.drivers.mms.listen(connection_cb, request_cb, addr=Address(host='0.0.0.0', port=102))

Create MMS listening server

Parameters

  • connection_cb (ConnectionCb) – new connection callback

  • request_cb (RequestCb) – received request callback

  • addr (Address) – local listening address

Returns

Server

class hat.drivers.mms.Server

Bases: object

MMS listening server

For creating new server see listen()

property addresses

listening addresses

Type

List[Address]

property closed

closed future

Type

asyncio.Future

async async_close()

Close listening socket

Calling this method doesn’t close active incomming connections

class hat.drivers.mms.Connection

Bases: object

MMS connection

For creating new connection see connect()

property info

connection info

Type

ConnectionInfo

property closed

closed future

Type

asyncio.Future

async async_close()

Async close

async receive_unconfirmed()

Receive unconfirmed message

Returns

common.Unconfirmed

Raises

ConnectionError – in case connection is in closing or closed state

send_unconfirmed(unconfirmed)

Send unconfirmed message

Parameters

unconfirmed (common.Unconfirmed) – unconfirmed message

async send_confirmed(req)

Send confirmed request and wait for response

Parameters

req (common.Request) – request

Returns

common.Response

Raises

ConnectionError – in case connection is in closing or closed state

class hat.drivers.mms.DataAccessError(value)

Bases: enum.Enum

An enumeration.

  • OBJECT_INVALIDATED

  • HARDWARE_FAULT

  • TEMPORARILY_UNAVAILABLE

  • OBJECT_ACCESS_DENIED

  • OBJECT_UNDEFINED

  • INVALID_ADDRESS

  • TYPE_UNSUPPORTED

  • TYPE_INCONSISTENT

  • OBJECT_ATTRIBUTE_INCONSISTENT

  • OBJECT_ACCESS_UNSUPPORTED

  • OBJECT_NON_EXISTENT

  • OBJECT_VALUE_INVALID

OBJECT_INVALIDATED
HARDWARE_FAULT
TEMPORARILY_UNAVAILABLE
OBJECT_ACCESS_DENIED
OBJECT_UNDEFINED
INVALID_ADDRESS
TYPE_UNSUPPORTED
TYPE_INCONSISTENT
OBJECT_ATTRIBUTE_INCONSISTENT
OBJECT_ACCESS_UNSUPPORTED
OBJECT_NON_EXISTENT
OBJECT_VALUE_INVALID
class hat.drivers.mms.ObjectClass(value)

Bases: enum.Enum

An enumeration.

  • NAMED_VARIABLE

  • NAMED_VARIABLE_LIST

  • JOURNAL

  • DOMAIN

  • UNDEFINED

NAMED_VARIABLE
NAMED_VARIABLE_LIST
JOURNAL
DOMAIN
UNDEFINED
class hat.drivers.mms.ErrorClass(value)

Bases: enum.Enum

An enumeration.

  • ACCESS

  • APPLICATION_REFERENCE

  • CANCEL

  • CONCLUDE

  • DEFINITION

  • FILE

  • INITIATE

  • OTHERS

  • RESOURCE

  • SERVICE

  • SERVICE_PREEMPT

  • TIME_RESOLUTION

  • VMD_STATE

ACCESS
APPLICATION_REFERENCE
CANCEL
CONCLUDE
DEFINITION
FILE
INITIATE
OTHERS
RESOURCE
SERVICE
SERVICE_PREEMPT
TIME_RESOLUTION
VMD_STATE
class hat.drivers.mms.StatusRequest

Bases: tuple

Create new instance of StatusRequest()

class hat.drivers.mms.GetNameListRequest(object_class, object_scope, continue_after)

Bases: tuple

Create new instance of GetNameListRequest(object_class, object_scope, continue_after)

continue_after

Optional[str]

object_class

ObjectClass

object_scope

ObjectScope

class hat.drivers.mms.IdentifyRequest

Bases: tuple

Create new instance of IdentifyRequest()

class hat.drivers.mms.GetVariableAccessAttributesRequest(value)

Bases: tuple

Create new instance of GetVariableAccessAttributesRequest(value,)

value

Union[ObjectName,int,str,hat.asn1.Data]

class hat.drivers.mms.GetNamedVariableListAttributesRequest(value)

Bases: tuple

Create new instance of GetNamedVariableListAttributesRequest(value,)

value

ObjectName

class hat.drivers.mms.ReadRequest(value)

Bases: tuple

Create new instance of ReadRequest(value,)

value

Union[List[VariableSpecification],ObjectName]

class hat.drivers.mms.WriteRequest(specification, data)

Bases: tuple

Create new instance of WriteRequest(specification, data)

data

List[Data]

specification

Union[List[VariableSpecification],ObjectName]

class hat.drivers.mms.DefineNamedVariableListRequest(name, specification)

Bases: tuple

Create new instance of DefineNamedVariableListRequest(name, specification)

name

ObjectName

specification

List[VariableSpecification]

class hat.drivers.mms.DeleteNamedVariableListRequest(names)

Bases: tuple

Create new instance of DeleteNamedVariableListRequest(names,)

names

List[ObjectName]

class hat.drivers.mms.Request

Bases: abc.ABC

class hat.drivers.mms.ErrorResponse(error_class, value)

Bases: tuple

Create new instance of ErrorResponse(error_class, value)

error_class

ErrorClass

value

int

class hat.drivers.mms.StatusResponse(logical, physical)

Bases: tuple

Create new instance of StatusResponse(logical, physical)

logical

int

physical

int

class hat.drivers.mms.GetNameListResponse(identifiers, more_follows)

Bases: tuple

Create new instance of GetNameListResponse(identifiers, more_follows)

identifiers

List[str]

more_follows

bool

class hat.drivers.mms.IdentifyResponse(vendor, model, revision, syntaxes)

Bases: tuple

Create new instance of IdentifyResponse(vendor, model, revision, syntaxes)

model

str

revision

str

syntaxes

Optional[List[asn1.ObjectIdentifier]]

vendor

str

class hat.drivers.mms.GetVariableAccessAttributesResponse(mms_deletable, type_description)

Bases: tuple

Create new instance of GetVariableAccessAttributesResponse(mms_deletable, type_description)

mms_deletable

bool

type_description

TypeDescription

class hat.drivers.mms.GetNamedVariableListAttributesResponse(mms_deletable, specification)

Bases: tuple

Create new instance of GetNamedVariableListAttributesResponse(mms_deletable, specification)

mms_deletable

bool

specification

List[VariableSpecification

class hat.drivers.mms.ReadResponse(results)

Bases: tuple

Create new instance of ReadResponse(results,)

results

List[Union[DataAccessError,Data]]

class hat.drivers.mms.WriteResponse(results)

Bases: tuple

Create new instance of WriteResponse(results,)

results

List[Optional[DataAccessError]]

class hat.drivers.mms.DefineNamedVariableListResponse

Bases: tuple

Create new instance of DefineNamedVariableListResponse()

class hat.drivers.mms.DeleteNamedVariableListResponse(matched, deleted)

Bases: tuple

Create new instance of DeleteNamedVariableListResponse(matched, deleted)

deleted

int

matched

int

class hat.drivers.mms.Response

Bases: abc.ABC

class hat.drivers.mms.EventNotificationUnconfirmed(enrollment, condition, severity, time)

Bases: tuple

Create new instance of EventNotificationUnconfirmed(enrollment, condition, severity, time)

condition

ObjectName

enrollment

ObjectName

severity

int

time

Optional[Union[Data,int]]

class hat.drivers.mms.InformationReportUnconfirmed(specification, data)

Bases: tuple

Create new instance of InformationReportUnconfirmed(specification, data)

data

List[Union[DataAccessError,Data]]

specification

Union[List[VariableSpecification],ObjectName]

class hat.drivers.mms.UnsolicitedStatusUnconfirmed(logical, physical)

Bases: tuple

Create new instance of UnsolicitedStatusUnconfirmed(logical, physical)

logical

int

physical

int

class hat.drivers.mms.Unconfirmed

Bases: abc.ABC

class hat.drivers.mms.ArrayData(elements)

Bases: tuple

Create new instance of ArrayData(elements,)

elements

List[Data]

class hat.drivers.mms.BcdData(value)

Bases: tuple

Create new instance of BcdData(value,)

value

int

class hat.drivers.mms.BinaryTimeData(value)

Bases: tuple

Create new instance of BinaryTimeData(value,)

value

datetime.datetime

class hat.drivers.mms.BitStringData(value)

Bases: tuple

Create new instance of BitStringData(value,)

value

List[bool]

class hat.drivers.mms.BooleanData(value)

Bases: tuple

Create new instance of BooleanData(value,)

value

bool

class hat.drivers.mms.BooleanArrayData(value)

Bases: tuple

Create new instance of BooleanArrayData(value,)

value

List[bool]

class hat.drivers.mms.FloatingPointData(value)

Bases: tuple

Create new instance of FloatingPointData(value,)

value

float

class hat.drivers.mms.GeneralizedTimeData(value)

Bases: tuple

Create new instance of GeneralizedTimeData(value,)

value

str

class hat.drivers.mms.IntegerData(value)

Bases: tuple

Create new instance of IntegerData(value,)

value

int

class hat.drivers.mms.MmsStringData(value)

Bases: tuple

Create new instance of MmsStringData(value,)

value

str

class hat.drivers.mms.ObjIdData(value)

Bases: tuple

Create new instance of ObjIdData(value,)

value

asn1.ObjectIdentifier

class hat.drivers.mms.OctetStringData(value)

Bases: tuple

Create new instance of OctetStringData(value,)

value

hat.asn1.Data

class hat.drivers.mms.StructureData(elements)

Bases: tuple

Create new instance of StructureData(elements,)

elements

List[Data]

class hat.drivers.mms.UnsignedData(value)

Bases: tuple

Create new instance of UnsignedData(value,)

value

int

class hat.drivers.mms.UtcTimeData(value, leap_second, clock_failure, not_synchronized, accuracy)

Bases: tuple

Create new instance of UtcTimeData(value, leap_second, clock_failure, not_synchronized, accuracy)

accuracy

accurate fraction bits [0,24]

Type

Optional[int]

clock_failure

bool

leap_second

bool

not_synchronized

bool

value

datetime.datetime

class hat.drivers.mms.VisibleStringData(value)

Bases: tuple

Create new instance of VisibleStringData(value,)

value

str

class hat.drivers.mms.Data

Bases: abc.ABC

class hat.drivers.mms.ArrayTypeDescription(number_of_elements, element_type)

Bases: tuple

Create new instance of ArrayTypeDescription(number_of_elements, element_type)

element_type

Union[TypeDescription,ObjectName]

number_of_elements

int

class hat.drivers.mms.BcdTypeDescription(xyz)

Bases: tuple

Create new instance of BcdTypeDescription(xyz,)

xyz

int

class hat.drivers.mms.BinaryTimeTypeDescription(xyz)

Bases: tuple

Create new instance of BinaryTimeTypeDescription(xyz,)

xyz

bool

class hat.drivers.mms.BitStringTypeDescription(xyz)

Bases: tuple

Create new instance of BitStringTypeDescription(xyz,)

xyz

int

class hat.drivers.mms.BooleanTypeDescription

Bases: tuple

Create new instance of BooleanTypeDescription()

class hat.drivers.mms.FloatingPointTypeDescription(format_width, exponent_width)

Bases: tuple

Create new instance of FloatingPointTypeDescription(format_width, exponent_width)

exponent_width

int

format_width

int

class hat.drivers.mms.GeneralizedTimeTypeDescription

Bases: tuple

Create new instance of GeneralizedTimeTypeDescription()

class hat.drivers.mms.IntegerTypeDescription(xyz)

Bases: tuple

Create new instance of IntegerTypeDescription(xyz,)

xyz

int

class hat.drivers.mms.MmsStringTypeDescription(xyz)

Bases: tuple

Create new instance of MmsStringTypeDescription(xyz,)

xyz

int

class hat.drivers.mms.ObjIdTypeDescription

Bases: tuple

Create new instance of ObjIdTypeDescription()

class hat.drivers.mms.OctetStringTypeDescription(xyz)

Bases: tuple

Create new instance of OctetStringTypeDescription(xyz,)

xyz

int

class hat.drivers.mms.StructureTypeDescription(components)

Bases: tuple

Create new instance of StructureTypeDescription(components,)

components

List[Tuple[Optional[str],Union[TypeDescription,ObjectName]]]

class hat.drivers.mms.UnsignedTypeDescription(xyz)

Bases: tuple

Create new instance of UnsignedTypeDescription(xyz,)

xyz

int

class hat.drivers.mms.UtcTimeTypeDescription

Bases: tuple

Create new instance of UtcTimeTypeDescription()

class hat.drivers.mms.VisibleStringTypeDescription(xyz)

Bases: tuple

Create new instance of VisibleStringTypeDescription(xyz,)

xyz

int

class hat.drivers.mms.TypeDescription

Bases: abc.ABC

class hat.drivers.mms.AaSpecificObjectName(identifier)

Bases: tuple

Create new instance of AaSpecificObjectName(identifier,)

identifier

str

class hat.drivers.mms.DomainSpecificObjectName(domain_id, item_id)

Bases: tuple

Create new instance of DomainSpecificObjectName(domain_id, item_id)

domain_id

str

item_id

str

class hat.drivers.mms.VmdSpecificObjectName(identifier)

Bases: tuple

Create new instance of VmdSpecificObjectName(identifier,)

identifier

str

class hat.drivers.mms.ObjectName

Bases: abc.ABC

class hat.drivers.mms.AaSpecificObjectScope

Bases: tuple

Create new instance of AaSpecificObjectScope()

class hat.drivers.mms.DomainSpecificObjectScope(identifier)

Bases: tuple

Create new instance of DomainSpecificObjectScope(identifier,)

identifier

str

class hat.drivers.mms.VmdSpecificObjectScope

Bases: tuple

Create new instance of VmdSpecificObjectScope()

class hat.drivers.mms.ObjectScope

Bases: abc.ABC

class hat.drivers.mms.AddressVariableSpecification(address)

Bases: tuple

Create new instance of AddressVariableSpecification(address,)

address

Union[int,str,hat.asn1.Data]

class hat.drivers.mms.InvalidatedVariableSpecification

Bases: tuple

Create new instance of InvalidatedVariableSpecification()

class hat.drivers.mms.NameVariableSpecification(name)

Bases: tuple

Create new instance of NameVariableSpecification(name,)

name

ObjectName

class hat.drivers.mms.ScatteredAccessDescriptionVariableSpecification(specifications)

Bases: tuple

Create new instance of ScatteredAccessDescriptionVariableSpecification(specifications,)

specifications

List[VariableSpecification]

class hat.drivers.mms.VariableDescriptionVariableSpecification(address, type_specification)

Bases: tuple

Create new instance of VariableDescriptionVariableSpecification(address, type_specification)

address

Union[int,str,hat.asn1.Data]

type_specification

Union[TypeDescription,ObjectName]

class hat.drivers.mms.VariableSpecification

Bases: abc.ABC