Metadata-Version: 2.4
Name: cta2045
Version: 0.1.0
Summary: CTA-2045 (ANSI/CTA-2045-B) protocol library — encode/decode of the SGD↔UCM demand-response interface
Author: Clark Communications Corporation
License-Expression: MIT
Project-URL: Homepage, https://ebus.energy
Project-URL: Repository, https://github.com/electrification-bus/python-cta2045
Project-URL: Issues, https://github.com/electrification-bus/python-cta2045/issues
Keywords: cta-2045,cta2045,ecoport,demand-response,smart-grid,water-heater,ucm,sgd
Classifier: Development Status :: 2 - Pre-Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Home Automation
Classifier: Topic :: System :: Hardware
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Dynamic: license-file

# python-cta2045

[![PyPI version](https://img.shields.io/pypi/v/cta2045.svg)](https://pypi.org/project/cta2045/)
[![Python versions](https://img.shields.io/pypi/pyversions/cta2045.svg)](https://pypi.org/project/cta2045/)
[![CI](https://github.com/electrification-bus/python-cta2045/actions/workflows/ci.yml/badge.svg)](https://github.com/electrification-bus/python-cta2045/actions/workflows/ci.yml)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

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](#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:

```python
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

```python
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

```python
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`.

```python
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:

```python
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](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](https://www.cta.tech/standards/ansicta-2045-b/) · [purchase / download](https://shop.cta.tech/products/modular-communications-interface-for-energy-management). 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](https://www.openadr.org/ecoport).
- **Certification & testing** — [OpenADR and CTA-2045 / EcoPort testing (UL Solutions)](https://www.ul.com/resources/openadr-and-cta-2045ecoport-testing).

## License

MIT — see [LICENSE](LICENSE).
