MMS Connection

icspacket.proto.mms.connection.UnconfirmedServiceCallback

Type alias for callbacks that process unconfirmed MMS service elements.

Parameters:

  • conn (MMS_Connection) – Active MMS connection instance.

  • service (UnconfirmedService) – The unconfirmed MMS service payload.

Added in version 0.2.4.

alias of Callable[[MMS_Connection, UnconfirmedService], None]

icspacket.proto.mms.connection.UnconfirmedPDUCallback

Type alias for callbacks that process full unconfirmed MMS PDUs.

Parameters:

  • conn (MMS_Connection) – Active MMS connection instance.

  • pdu (Unconfirmed_PDU) – The unconfirmed MMS PDU as received from the remote peer.

Added in version 0.2.4.

alias of Callable[[MMS_Connection, Unconfirmed_PDU], None]

class icspacket.proto.mms.connection.UnconfirmedServiceHandler(service: PRESENT, func: Callable[[MMS_Connection, UnconfirmedService], None] | None = None)[source]

Utility class for dispatching unconfirmed MMS service elements.

Instances of this handler can be registered to react to specific unconfirmed MMS services. It acts as a callable object that can be directly invoked with an Unconfirmed_PDU.

Example
def on_status(conn, service):
    print("Received status report:", service)

handler = UnconfirmedServiceHandler(
    UnconfirmedService.PRESENT.PR_XXX,
    func=on_status
)

# later inside MMS_Connection
handler(conn, unconfirmed_pdu)
Parameters:

  • service (UnconfirmedService.PRESENT) – The UnconfirmedService.PRESENT discriminator that this handler should filter on.

  • func (UnconfirmedServiceCallback | None) – Optional callback invoked when a matching unconfirmed service is received.

Added in version 0.2.4.

on_pdu(conn: MMS_Connection, service: UnconfirmedService) → None[source]

Dispatch a matching unconfirmed service to the configured callback.

Parameters:

  • conn (MMS_Connection) – Active MMS connection instance.

  • service (UnconfirmedService) – The unconfirmed MMS service element matching the target type.

Added in version 0.2.4.

class icspacket.proto.mms.connection.MMS_Connection(cotp_conn: COTP_Connection | None = None, session_config: ISO_SessionSettings | None = None, presentation_config: ISO_PresentationSettings | None = None, auth: Authenticator | None = None, unconfirmed_cb: Callable[[MMS_Connection, Unconfirmed_PDU], None] | Iterable[Callable[[MMS_Connection, Unconfirmed_PDU], None]] | None = None)[source]

Implementation of the MMS (Manufacturing Message Specification) connection handling (ISO 9506-1,2).

This class provides a high-level MMS service endpoint on top of the ISO OSI stack. It integrates the following layers into a single connection object:

+-----------------------------+
| MMS (ISO 9506)              |
+-----------------------------+
| ACSE Association            |
+-----------------------------+
| ISO Presentation (ISO 8823) |
+-----------------------------+
| ISO Session (ISO 8327)      |
+-----------------------------+
| Transport (COTP, RFC 905)   |
+-----------------------------+
| TPKT / TCP                  |
+-----------------------------+

The class encapsulates these layers so that MMS service primitives such as associate(), release(), abort(), and close() can be directly invoked by the user without manually handling lower-layer PDUs.

Establishing an MMS association to a remote server:

from mms.connection import MMS_Connection

# Create MMS connection
mms_conn = MMS_Connection()

# Connect and associate with a peer MMS user
mms_conn.associate(("192.168.1.100", 102))
assert mms_conn.is_valid():

# Release the association in an orderly manner
mms_conn.release()
Parameters:

  • cotp_conn (COTP_Connection | None) – Optional transport layer (COTP) connection to use. If not provided, a new COTP_Connection is created using the default TPKT socket.

  • session_config (ISO_SessionSettings | None) – Optional ISO Session settings, passed down to the session layer.

  • presentation_config (ISO_PresentationSettings | None) – Optional ISO Presentation settings, passed to the presentation layer.

  • auth (Authenticator | None) – Optional ACSE Authenticator instance for authentication handling.

  • unconfirmed_cb – Callback or iterable of callbacks that will be invoked whenever an unconfirmed PDU is received from the peer. Each callback must conform to UnconfirmedPDUCallback.

property session: ISO_Session

The ISO Session layer object.

Returns:

Underlying ISO Session instance.

Return type:

ISO_Session

property transport: COTP_Connection

The transport layer (COTP) connection.

Returns:

Underlying COTP transport connection.

Return type:

COTP_Connection

property presentation: ISO_Presentation

The ISO Presentation layer object.

Returns:

The presentation layer instance for this MMS connection.

Return type:

ISO_Presentation

property association: Association

The ACSE Association handler for this MMS connection.

Returns:

Association object responsible for ACSE/MMS binding.

Return type:

Association

property next_invoke_id: int

Increment and return the next available invoke-id.

This value is used to uniquely identify outstanding MMS service requests.

Returns:

Next invoke identifier.

Return type:

int

property invoke_id: int

The current invoke-id counter.

Returns:

Current invoke identifier.

Return type:

int

property unconfirmed_cb: list[Callable[[MMS_Connection, Unconfirmed_PDU], None]]

List of registered unconfirmed PDU callbacks.

Each callback is invoked in registration order whenever an Unconfirmed_PDU is received.

Returns:

List of registered callback callables.

Return type:

list[UnconfirmedPDUCallback]

Added in version 0.2.4.

connect(address: tuple[str, int]) → None[source]

Establish a transport/session/presentation connection.

If the connection is already established, this call is ignored.

Parameters:

address (tuple[str, int]) – Remote address (host, port) to connect to.

associate(address: tuple[str, int] | None = None, request: Initiate_RequestPDU | None = None) → None[source]

6.9.1 M-ASSOCIATE Service (ISO 9506-1).

Establishes an MMS association with a peer MMS-user.

This method performs the ACSE and MMS-level handshake by sending an Initiate Request PDU and validating the Initiate Response PDU.

Parameters:

  • address (tuple[str, int] | None) – Optional peer address (host, port). If provided, a transport connection is established first.

  • request (Initiate_RequestPDU | None) – Optional Initiate Request PDU. If not provided, a default one is created using new_initiate_request().

Raises:

  • MMSConnectionError – If the peer responds with an MMS error PDU.

  • TypeError – If the peer responds with an unexpected PDU type.

release(reason: VALUES | None = None, graceful: bool = False) → None[source]

6.9.2 M-RELEASE Service (ISO 9506-1).

Terminate the association in an orderly manner.

Parameters:

  • reason (Release_request_reason.VALUES | None) – Optional release reason.

  • graceful (bool) – If True, attempt a graceful release.

abort(source: VALUES | None = None) → None[source]

6.9.4 M-U-ABORT Service (ISO 9506-1).

Abruptly terminate the association without completing an orderly release handshake.

Parameters:

source (ABRT_source.VALUES | None) – Optional abort source identifier.

close() → None[source]

Close the MMS association and underlying presentation connection.

This performs a Conclude handshake (Conclude-Request / Conclude-Response). If the peer responds with an error, MMSConnectionError is raised.

Raises:

  • MMSConnectionError – If the peer responds with a conclude error.

  • TypeError – If the peer responds with an unexpected PDU.

send_data(octets: bytes, /) → None[source]

Send raw BER-encoded MMS data.

Parameters:

octets (bytes) – BER-encoded MMS PDU.

Raises:

AssertionError – If the connection is not established.

send_mms_data(pdu: MMSpdu, /) → None[source]

Send an MMS PDU after BER encoding.

Parameters:

pdu (MMSpdu) – MMS PDU to send.

recv_mms_data() → MMSpdu[source]

Receive and decode an MMS PDU.

Returns:

Decoded MMS PDU.

Return type:

MMSpdu

Raises:

TypeError – If the received data is not an MMS PDU.

get_status(ex_derivation: bool = False) → tuple[vmdLogicalStatus_VALUES, vmdPhysicalStatus_VALUES][source]

10.3 Status Service

The Status service is used by an MMS client to determine the general condition or health of a VMD (Virtual Manufacturing Device).

Parameters:

ex_derivation (bool) – If True, requests an extended derivation of the status response. This influences how the server derives the logical and physical VMD status (see ISO 9506-1, 10.3.1.1.1).

Returns:

A tuple (logical_status, physical_status), where:

  • logical_status is of type StatusResponse.vmdLogicalStatus_VALUES.

  • physical_status is of type StatusResponse.vmdPhysicalStatus_VALUES.

Return type:

tuple

Example:

>>> logical, physical = mms_conn.get_status()
>>> print(f"Logical VMD status: {logical!r}, Physical VMD status: {physical!r}")
get_name_list(object_class: ObjectClass, scope: objectScope_TYPE | None = None) → list[str][source]

10.5 GetNameList Service

The GetNameList service requests the list (or a subset) of object names defined at the VMD. The server may return the list in segments if it is too long to fit in a single response.

Parameters:

  • object_class (ObjectClass) – The class of objects for which names are requested. For example: ObjectClass.namelist.domain.

  • scope (GetNameList_Request.objectScope_TYPE or None) – Optional scope of the name list request. If not provided, defaults to VMD-specific scope.

Returns:

A list of object identifiers.

Return type:

list[str]

Raises:

ValueError – If the MMS server returns an invalid response.

Example:

>>> domains = mms_conn.get_name_list(ObjectClass.domain)
>>> print("Available domains:", domains)
identify() → tuple[str, str, str][source]

10.6 Identify Service

The Identify service retrieves the vendor, model, and revision information of the remote MMS VMD.

Returns:

A tuple (vendor_name, model_name, revision) with all fields mandatory as per ISO 9506-2.

Return type:

tuple[str, str, str]

>>> vendor, model, revision = mms_conn.identify()
>>> print(f"Device: {vendor} {model} (rev {revision})")
get_capabilities() → list[str][source]

10.8 GetCapabilityList Service

The GetCapabilityList service requests the list of services or features supported by the remote MMS server. Like GetNameList, the response may be segmented and require multiple follow-up requests.

Returns:

A list of capability strings supported by the server.

Return type:

list[str]

Example:

>>> caps = mms_conn.get_capabilities()
>>> print("Server capabilities:", caps)
read_variables(*pos_variables: Member_TYPE, variables: list[Member_TYPE] | None = None, spec_in_result: bool = False, list_name: ObjectName | None = None) → list[AccessResult][source]

14.6 Variable Read Service

Reads one or more variables or an entire variable list from the remote MMS server.

Parameters:

  • pos_variables (VariableAccessItem) – Positional list of variables to read.

  • variables (list[VariableAccessItem] or None) – Alternative way of specifying variables via keyword.

  • spec_in_result (bool) – If True, requests that the server return the VariableAccessSpecification in the response (14.6.1.1.1). Usually False for efficiency.

  • list_name (ObjectName or None) – If set, references a named variable list to read instead of individual variables.

Returns:

A list of AccessResult objects, one per variable read.

Return type:

list[AccessResult]

Raises:

ValueError – If no variables or list_name is provided, or if both are specified at once.

Example (read two variables):

>>> var1 = VariableAccessItem(name=ObjectName(vmd_specific="TempSensor1"))
>>> var2 = VariableAccessItem(name=ObjectName(vmd_specific="TempSensor2"))
>>> results = mms_conn.read_variables(var1, var2)
>>> for res in results:
...     print(res.success)  # AccessResult can be success or failure
write_variable(value: Data | list[Data], /, variable: Member_TYPE | None = None, list_name: ObjectName | None = None) → DataAccessError | None[source]

14.7 Variable Write Service

Writes the value of a single variable at the MMS server.

Parameters:

  • variable (VariableAccessSpecification.listOfVariable_TYPE.Member_TYPE) – The target variable to be written.

  • value (Data | list[Data]) – The value to write, encoded as MMS Data.

Returns:

None on success, or a DataAccessError if the write failed for this variable.

Return type:

DataAccessError or None

Example (write a new integer value):

>>> var = VariableAccessItem(name=ObjectName(vmd_specific="SetPoint"))
>>> data = Data()
>>> data.integer = 42
>>> err = mms_conn.write_variable(data, variable)
>>> print("Write successful" if err is None else f"Failed: {err}")

Changed in version 0.2.2: Multiple Data objects allowed within the value parameter. Ordering of parameters has changed.

variable_attributes(*, name: ObjectName | None = None, address: Address | None = None) → GetVariableAccessAttributes_Response[source]

14.9 GetVariableAccessAttributes Service

Retrieves metadata describing a variable, such as its type, address, and whether it is deletable or modifiable.

Parameters:

  • name (ObjectName or None) – The object name of the variable to query.

  • address (Address or None) – The variable’s address (alternative to name).

Returns:

The variable access attributes response from the MMS server.

Return type:

GetVariableAccessAttributes_Response

Raises:

ValueError – If neither name nor address is specified, or if both are given at the same time.

Example (query attributes by name):

>>> obj = ObjectName(vmd_specific="TempSensor1")
>>> attrs = mms_conn.variable_attributes(name=obj)
>>> print("Type description:", attrs.typeDescription)
variable_list_attributes(name: ObjectName, /) → GetNamedVariableListAttributes_Response[source]

Added in version 0.2.2.

obtain_file(source: str | Path, destination: str | Path, remote: ApplicationReference | None = None) → None[source]

Perform the MMS ObtainFile service to transfer a file between the client and a remote MMS server.

The ObtainFile service may be used by an MMS client to instruct an MMS server to obtain a specified file from a file server. Depending on the usage, this can either be a request to pull a file from a remote source or to serve a local file to the requesting MMS peer.

The method implements the complete reques-response sequence defined in IEC 61850 Annex C, including FileOpen, FileRead, FileClose, and the final ObtainFile confirmation.

Parameters:

  • source (str | Path) – Path to the source file. If remote is provided, this denotes the file path to be retrieved from the remote server. Otherwise, this must point to a local file which will be transferred.

  • destination (str | Path) – Path where the file should be stored on the remote server. For local serving, this is the name advertised to the requesting peer.

  • remote (ApplicationReference | None) – Optional reference to an external application server from which the file should be obtained. If None, this MMS connection instance will act as the file source.

Raises:

  • MMSServiceError – If the MMS server responds with an unexpected PDU type or an error occurs during the transfer sequence.

  • OSError – If the local file system access fails (e.g., file not found or permission denied).

Added in version 0.2.4.

file_transfer(source: str | Path, destination: str | Path, remote: ApplicationReference | None = None) → None

Added in version 0.2.4.

file_open(name: FileName, offset: int = 0) → FileHandle[source]

D.3 FileOpen Service

Opens a file in the MMS server’s virtual filestore and returns a handle used for subsequent file operations (read, close, etc.).

Parameters:

  • name (FileName) – The file name to open.

  • offset (int, optional) – Initial byte offset from which to start reading. Defaults to 0 (beginning of the file).

Returns:

A file handle including the FRSM ID and file attributes.

Return type:

FileHandle

Example (complete): >>> # Open a file for reading >>> handle = mms_conn.file_open(FileName([“/logs/system.log”])) >>> # Read complete file >>> data = mms_conn.file_read(handle) >>> print(data.decode(“utf-8”)[:200]) # show first 200 characters >>> # Close the file handle >>> mms_conn.file_close(handle)

file_read_chunk(handle: FileHandle | int) → tuple[bytes, bool][source]

D.4 FileRead (chunked)

Reads a chunk of data from an open file handle.

Parameters:

handle (FileHandle or int) – A FileHandle or raw FRSM ID.

Returns:

A tuple (data, more_follows) where: - data is a bytes object containing the chunk read. - more_follows is a boolean indicating if further reads are required.

Return type:

tuple[bytes, bool]

>>> chunk, more = mms_conn.file_read_chunk(handle)
>>> print(len(chunk), "bytes read, more follows?", more)
file_read(handle: FileHandle | int) → bytes[source]

D.4 FileRead (complete)

Reads the entire content of a file by repeatedly fetching chunks until moreFollows is false.

Parameters:

handle (FileHandle or int) – A FileHandle or raw FRSM ID.

Returns:

Complete file contents as bytes.

Return type:

bytes

file_close(handle: FileHandle | int) → None[source]

D.5 FileClose Service

Closes an open file handle. This releases the FRSM ID allocated by the server.

Parameters:

handle (FileHandle or int) – A FileHandle or raw FRSM ID.

file_rename(old_name: FileName, new_name: FileName) → None[source]

D.6 FileRename Service

Renames a file in the server’s virtual filestore.

Parameters:

  • old_name (FileName) – The current file name.

  • new_name (FileName) – The new file name to assign.

Example:

>>> mms_conn.file_rename(FileName(["/logs/old.log"]),
...                      FileName(["/logs/new.log"]))
file_delete(name: FileName) → None[source]

D.7 FileDelete Service

Deletes a file from the MMS server’s virtual filestore.

Parameters:

name (FileName) – The file name to delete.

Example:

>>> mms_conn.file_delete(FileName(["/logs/obsolete.log"]))
list_directory(name: FileName | None = None) → list[DirectoryEntry][source]

D.8 FileDirectory Service

Retrieves directory listings or file attributes from the server’s virtual filestore.

Parameters:

name (FileName or None) – Optional file or directory name pattern. If omitted, the server returns an implementation-defined default set.

Returns:

List of directory entries, possibly spanning multiple responses.

Return type:

list[DirectoryEntry]

Example:

>>> entries = mms_conn.list_directory()
>>> for e in entries:
...     print(e.fileName[0], e.fileAttributes.fileSize)
additional_service_request(request: AdditionalService_Request) → AdditionalService_Response[source]

Issues an MMS Additional Service request (ISO 9506-1:1990, Clause 19).

This is a specialized helper that wraps an AdditionalService_Request in a ConfirmedServiceRequest and dispatches it via service_request().

It is rarely used directly, but provides a template for invoking “non-standardized” or vendor-specific services.

Parameters:

request (AdditionalService_Request) – The additional service request to send.

Returns:

The decoded additional service response from the server.

Return type:

AdditionalService_Response

Example:

>>> req = AdditionalService_Request()
>>> # populate request with service-specific fields
>>> resp = mms_conn.additional_service_request(req)
>>> print(resp)
service_request(request: ConfirmedServiceRequest, need_response: bool = True) → ConfirmedServiceResponse[source]

Issues a generic MMS Confirmed Service Request and returns the corresponding Confirmed Service Response.

This is the central method through which all MMS services are ultimately invoked. Higher-level convenience wrappers (e.g., read_variables(), file_open(), etc.) construct the appropriate ConfirmedServiceRequest and delegate to this method.

Workflow:

  1. A new Confirmed_RequestPDU is allocated and

    tagged with the next available invokeID.

  2. The caller-supplied request is embedded into the PDU.

  3. The request is wrapped into an MMSpdu.

  4. Transmission is performed using

    send_mms_data().

  5. A response is awaited via recv_mms_data().

  6. Any detected service error is normalized using

    _error_from_service_response().

  7. The decoded ConfirmedServiceResponse

    is returned to the caller.

Parameters:

  • request (ConfirmedServiceRequest) – The service request PDU, e.g. Read_Request, FileOpen_Request, etc., wrapped in a ConfirmedServiceRequest.

  • need_response (bool, optional) – Whether a response is required. If False, the method will still wait for the server response (ISO 9506-2 requires this) but certain services may ignore the payload.

Raises:

  • MMSConnectionError – If the response indicates a service error.

  • MMSServiceError – For generic or protocol-level failures.

  • MMSUnknownServiceError – If the service response cannot be mapped to a known type.

Returns:

The service component of the decoded ConfirmedServiceResponse, containing the service-specific result (e.g. a Read_Response).

Return type:

ConfirmedServiceResponse

Example:

>>> # Manually constructing a Read request
>>> read_req = Read_Request()
>>> read_req.variableAccessSpecification.listOfVariable = [var]
>>> csr = ConfirmedServiceRequest(read=read_req)
>>> resp = mms_conn.service_request(csr)
>>> print(resp.read.listOfAccessResult)