cta2045 0.1.0

CTA-2045 (ANSI/CTA-2045-B) protocol library — encode/decode of the SGD↔UCM demand-response interface

python-cta2045

A CTA-2045 (ANSI/CTA-2045-B) protocol library in Python — encode/decode of the demand-response interface between a Smart Grid Device (SGD, e.g. a water heater) and a Universal Communications Module (UCM).

Status: pre-alpha. The core protocol — Basic DR and Intermediate DR encode/decode, plus an abstract UCM interface — is implemented and tested. The public API may still change before 1.0, and the package is not yet published to PyPI.

Not certified. This is an independent implementation of the published protocol. It has not been tested or certified under any conformance program (EcoPort or otherwise), and carries no warranty of interoperability with any certified device.

What is CTA-2045?

CTA-2045 (consumer-facing brand: EcoPort) standardizes a modular communications socket on an appliance (the SGD) into which a UCM plugs to provide grid demand-response. CTA-2045 specifies only the SGD↔UCM link; a UCM's upward (network) interface is vendor-specific. This library implements the CTA-2045 message layer itself, independent of any vendor or transport: bytes (or ASCII-hex) in, structured Python objects out, and back. It has no runtime dependencies and does no I/O.

See References for the standard and the EcoPort program.

Install

Not yet on PyPI. For now, install from source:

pip install -e ".[dev]"

Requires Python 3.10+.

Quick start

Encode a command to send to a device, and decode messages received from one:

from cta2045 import app
from cta2045.codec import bytes_to_hex

# Encode a 10-minute shed command (UCM -> SGD). Durations are in MINUTES.
msg = app.shed(10)
msg.to_bytes()                       # b'\x08\x01\x00\x02\x01\x11'
bytes_to_hex(msg.to_bytes())         # '080100020111'

# Decode messages received from a device (one or more concatenated frames):
for m in app.decode_hex('080100021302'):
    print(m.category, m.operational_state)
    # BasicDRCategory.State_Query_Response OperationalState.Running_Curtailed

Basic DR commands

app.shed(10)            # curtail load for 10 minutes
app.end()               # end the current shed event
app.load_up(30)         # store energy for 30 minutes
app.critical_peak(60)   # critical peak event
app.grid_emergency(15)  # grid emergency event
app.power_level(50)     # request 50% power

Each returns a BasicDR object; call .to_bytes() for the wire frame. Durations are minutes; pass an explicit cta2045.codec.Duration for the Unknown / Too Long sentinels or for second-level control. Note the wire format quantizes durations (it can only carry 2·n² seconds) — use Duration.nearest() to see the value that will actually be transmitted.

Advanced Load Up

from cta2045 import app
from cta2045.codec import bytes_to_hex
from cta2045.enums import AdvancedLoadUpUnits

# Store 500 Wh (5 × 100 Wh) of extra energy over 60 minutes (CTA-2045-B § 11.6)
cmd = app.advanced_load_up(60, 5, AdvancedLoadUpUnits.Wh_100)
bytes_to_hex(cmd.to_bytes())         # '080200070C00003C000502'

Decoding device replies

decode_all(bytes) / decode_hex(str) return a list of message objects — BasicDR, IntermediateDR (whose .body is a CommodityReadReply, GetInformationReply, AdvancedLoadUp, or ThermostatResponse), or UnknownMessage for message types this library doesn't yet decode. Decoding is lenient: an unrecognized opcode or enum value is preserved (as category=None with a raw opcode1, or as a raw int) rather than raising — structural errors (truncated frames) still raise cta2045.codec.CodecError.

from cta2045 import app

for m in app.decode_all(raw_bytes):
    if isinstance(m, app.IntermediateDR) and isinstance(m.body, app.CommodityReadReply):
        for r in m.body.reports:
            print(r.code, r.instantaneous, r.cumulative)

Implementing a UCM binding

cta2045.ucm.Ucm is an abstract interface that turns the codec into a UCM client. Subclass it, implement the single transport primitive transmit(), and you get the full DR command set plus inbound decoding for free:

from cta2045.ucm import Ucm
from cta2045.enums import AdvancedLoadUpUnits

class MyUcm(Ucm):
    def transmit(self, frame: bytes) -> None:
        ...  # send `frame` to the SGD over your serial link

    def on_message(self, message) -> None:
        ...  # handle one decoded inbound message

ucm = MyUcm()
ucm.shed(10)                                       # build + transmit a shed command
ucm.advanced_load_up(60, 5, AdvancedLoadUpUnits.Wh_100)
ucm.receive(inbound_bytes)                         # decode + dispatch to on_message()

Vendor-proprietary UCM bindings (which add a specific UCM's network API) live in separate, non-open packages.

Package layout

• cta2045.enums — on-the-wire enumerations (message/DR-command types, device types, operational states, commodity codes, capabilities, …).
• cta2045.app — application-layer messages: Basic DR, Intermediate DR (commodity/energy reads, GetInformation, Advanced Load Up), with encoders and decoders.
• cta2045.codec — ASCII-hex ↔ bytes, frame header parse/build, and field encodings (e.g. the event-duration byte).
• cta2045.link — reserved for a future RS-485 link-layer implementation (enabling a Pi/RS-485 "own-UCM").
• cta2045.ucm — abstract UCM interface; vendor-proprietary bindings live in separate packages.

Development

pip install -e ".[dev]"
pytest
ruff check . && ruff format --check .

These three checks are the quality gate; all must pass before a change is merged. See CONTRIBUTING.md for how to file issues, propose changes, and the project's spec-conformance posture.

Scope & boundaries

This is a pure codec for the CTA-2045 message/application layer — no transport, no I/O. The CTA-2045 link layer (RS-485 or SPI serial framing, ACK/NAK, CRC) is a separate concern; the cta2045.link namespace is reserved for it.

References

• ANSI/CTA-2045-B — Modular Communications Interface for Energy Management, the standard this library implements: overview · purchase / download. The Intermediate DR application messages come from the companion ANSI/CTA-2045.3.
• EcoPort — the consumer-facing brand for CTA-2045-certified products; certification is run by the OpenADR Alliance.
• Certification & testing — OpenADR and CTA-2045 / EcoPort testing (UL Solutions).

License

MIT — see LICENSE.