Skip to content

Low-level Client

Direct access to communication objects without schema knowledge. Use Controller for normal usage; drop to this level when you need raw C.O. reads/writes or want to bypass the configuration cache.

pycomap.protocol.ComApClient

ComApClient(transport: TransportT)

Speaks the ECDH/AES-encrypted EthernetMessage protocol over any Transport.

Generic over the concrete Transport type, so transport gives back the type you passed in (e.g. ComApClient[EthernetTransport] exposes .transport.host) instead of the narrower structural Transport protocol.

Pass any Transport implementation — typically EthernetTransport::

from ipaddress import IPv4Address

from pycomap.protocol.transport import EthernetTransport

async with ComApClient(EthernetTransport(IPv4Address("192.168.1.9"))) as client:
    await client.authenticate("0")

Parameters:

Name Type Description Default
transport TransportT

Byte-stream transport to use (typically EthernetTransport).

required

transport property

transport: TransportT

The underlying byte-stream transport.

connect async

connect() -> None

Open the transport and consume the controller's unsolicited VersionIB.

authenticate async

authenticate(access_code: str) -> None

Perform the ECDH handshake and verify access_code with the controller.

The AES session key is derived once from access_code and cannot be changed without reconnecting. Pass the base/anonymous code (often "0"), not a privileged write code. To unlock write-protected setpoints afterward, call elevate_access — re-calling this method with a different code would corrupt the cipher state.

Parameters:

Name Type Description Default
access_code str

The controller's base AccessCode (drives ECDH/AES key derivation).

required

Raises:

Type Description
ComApInvalidAccessCodeError

If the controller rejects the access code.

ComApProtocolError

If the ECDH exchange produces an unexpected response.

elevate_access async

elevate_access(password: int) -> None

Submit the controller's write-protection password to unlock password-protected setpoints for this session.

This is a completely separate credential from the access_code passed to authenticate — the AccessCode gates the TCP connection itself, while the Password (0-9999) gates individual setpoint writes based on each setpoint's configured accessLevel. Verified live against a real controller: write CommunicationObject.PASSWORD_FOR_WRITE (24524) with the 2-byte little-endian password value, then setpoint writes that would otherwise return ControllerError.INVALID_PASSWORD succeed.

The controller enforces brute-force protection (5 wrong attempts → 1 min block, doubling each time, 100 wrong → permanent lockout). Do not call this in a retry loop. See docs/protocol.md section 2.4.1 for the full picture.

read_object async

read_object(comm_obj: int, addr: int = 1) -> bytes

Read a communication object, handling SendToBlock continuation transparently.

Parameters:

Name Type Description Default
comm_obj int

Communication object number (C.O.).

required
addr int

Controller unit address; 1 for the primary unit.

1

Returns:

Type Description
bytes

Raw payload bytes.

Raises:

Type Description
ComApControllerError

If the controller responds with an error code.

ComApProtocolError

If an unexpected message operation is received.

write_object async

write_object(
    comm_obj: int, data: bytes, addr: int = 1
) -> bytes

Write a communication object.

Parameters:

Name Type Description Default
comm_obj int

Communication object number (C.O.).

required
data bytes

Raw payload bytes to write.

required
addr int

Controller unit address; 1 for the primary unit.

1

Returns:

Type Description
bytes

Any data carried back on the NEXT acknowledgment (usually empty).

Raises:

Type Description
ComApControllerError

If the controller responds with an error code.

execute_command async

execute_command(command: ControllerCommand) -> int

Execute a controller command and return the uint32 result code.

Writes the argument to COMMAND_ARGUMENT (24550), triggers via COMMAND (24551), then reads COMMAND_ARGUMENT back for the return value. Compare the result against command.expected_return (or use command.succeeded(result)) to determine success. A result of ControllerCommand.RESULT_REFUSED (0x02) means the controller state doesn't allow the action right now (e.g. engine start attempted outside MAN mode); RESULT_INVALID_ARGUMENT (0x01) means the command/argument combination is unrecognised.

Note: some commands require the session to be password-elevated first via elevate_access — the controller will raise ControllerError.INVALID_PASSWORD on the write if so.

read_datetime async

read_datetime() -> datetime.datetime | None

Read the controller's current date and time as a naive datetime.

Reads DATE (C.O. 24553) and TIME (C.O. 24554) in two round-trips. Returns None if the controller reports an invalid/unset clock (e.g. after a factory reset before time sync).

The value is the controller's local wall-clock time (no timezone info attached). DST is governed by the Summer Time Mode setpoint (8727) and the UTC offset by the Time Zone setpoint (24366).

write_datetime async

write_datetime(dt: datetime) -> None

Set the controller's clock to the wall-clock components of dt.

Writes DATE (C.O. 24553) then TIME (C.O. 24554). Seconds are included; sub-second precision is dropped. Year must be ≥ 2000.

Both setpoints have access_level=1 — call elevate_access first. Pass the correct local time directly.


pycomap.protocol.EthernetTransport

EthernetTransport(
    host: IPv4Address, port: int = DEFAULT_PORT
)

Bases: Transport

TCP transport for the ComAp native protocol (port 23).

Connects to host:port via plain TCP; the ComApClient layer adds the ECDH/AES framing on top.

Parameters:

Name Type Description Default
host IPv4Address

Controller IP address. Exposed back via host for e.g. probing reachability via discover_host.

required
port int

TCP port; defaults to 23 (ComAp native protocol port).

DEFAULT_PORT

host property

host: IPv4Address

The controller's configured IP address.

port property

port: int

The controller's configured TCP port.


pycomap.protocol.transport.Transport

Bases: ABC

Byte-stream transport used by ComApClient.

connect abstractmethod async

connect() -> None

Open the underlying connection.

close abstractmethod async

close() -> None

Close the underlying connection, suppressing already-closed errors.

read_exactly abstractmethod async

read_exactly(n: int) -> bytes

Read exactly n bytes, raising ComApConnectionError if the stream ends.

write abstractmethod async

write(data: bytes) -> None

Write data and flush.