Metadata-Version: 2.4
Name: aacommpy
Version: 2.0.0
Summary: Runtime-free Python client for Agito AAComm motion controllers (native ctypes, no .NET)
Author-email: Agito / Akribis <daniel.brousser@akribis-sys.com>
License: MIT
Project-URL: Homepage, https://pypi.org/project/aacommpy/
Keywords: agito,aacomm,motion-control,ctypes,native
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: Microsoft :: Windows
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# aacommpy

A small, Pythonic, **runtime-free** client for Agito **AAComm** motion controllers.

`aacommpy` drives a controller through the native `aacomm.*` C library and a bundled native
`AACommServerNative` daemon — both NativeAOT, **zero .NET runtime, no Python.NET, no NuGet**. `pip
install` and connect. Samples use `172.1.1.101`, the controller's factory-default address —
substitute yours.

```python
import aacommpy
from aacommpy import Ethernet

with aacommpy.connect(Ethernet("172.1.1.101")) as c:
    print(c.query("AIdentity[1]").text)   # controller type, e.g. "10"
    print(c.get("APos"))                  # parameter value as a number
    c.set("AVel[1]", 1000)                # assign a parameter
    fut = c.send("AReset")                # async -> concurrent.futures.Future
    fut.result(timeout=15)
    c.on_push(lambda msg: print("push:", msg))
```

## Install

```bash
pip install aacommpy
```

Platform wheels bundle the native library (`aacomm.dll` / `libaacomm.so`) **and** the AACS daemon
(`AACommServerNative`), so there is nothing else to install — no .NET, no server setup. (win-x64
first; other platforms as their wheels land.)

## API

| Call | Meaning |
|---|---|
| `connect(conn, *, server=None, default_timeout_ms=0) -> Client` | Start the bundled daemon, connect, return a context-managed `Client`. Raises on failure. |
| `Ethernet(ip, port=0)` · `Serial(port, baud=0)` · `Can(node, channel=0, baud=0)` | Connection configs. |
| `Client.query(message, timeout_ms=None) -> Reply` | Synchronous send/receive. |
| `Client.send(message, timeout_ms=None) -> Future[Reply]` | Asynchronous send; the `Future` resolves with the reply. |
| `Client.get(param) -> int\|float` · `Client.set(param, value) -> Reply` | Convenience get/set over `query`. |
| `Client.identity(index=1) -> Reply` | `AIdentity[index]` (default `[1]` = controller type). |
| `Client.record(params, duration_s, gap_ms, triggers, *, force=False, trigger_mode=2, trigger_position=0, logic01=0, logic12=0, timeout_ms=10000) -> Recording` | Capture a controller data recording. Needs ≥1 `Trigger`; `force=True` completes immediately; `timeout_ms` bounds the wait-for-completion poll (default 10 s). |
| `Client.download_firmware(conn, fw_path, *, ver_path=None, password="160412", verify_currents=False, target=0, timeout_ms=180000, on_progress=None)` | Flash firmware; blocks until done. Ethernet (AGM800) / RS-232 only — never CAN. |
| `Client.download_fpga(conn, fpga_path, ver_path, *, password="160412", timeout_ms=180000, on_progress=None)` | Flash FPGA (`ver_path` required); blocks until done. RS-232 / Ethernet, non-AGM800, normal mode. |
| `Client.download_user_program(path, *, timeout_ms=180000, on_progress=None)` | Download a compiled user program to the **already-connected** controller; blocks until done. |
| `Client.get_user_units_enabled(axis) -> bool` · `set_user_units_enabled(axis, enabled)` | Query / request per-axis user-units enable (e.g. `axis="A"`). |
| `Client.on_push / on_error / on_longop_started / on_longop_ended(cb)` | Register session-event callbacks. |
| `Client.test_connection(conn) -> ConnectResult` · `connect` · `disconnect` · `is_connected` · `close_server` · `close` | Explicit lifecycle. |

### Recording

```python
import aacommpy
from aacommpy import Ethernet, Trigger

with aacommpy.connect(Ethernet("172.1.1.101")) as c:
    rec = c.record(
        ["APos", "AVel[1]"], duration_s=0.05, gap_ms=1.0,
        triggers=[Trigger(source="APos", type=5, value=0.0, mask=-1)],
        force=True,                    # complete immediately; the condition is irrelevant
    )
    print(rec.rec_length, rec.sample_rate_hz)
    print(rec["APos"].values)          # samples for one vector
    print(rec.time_s)                  # shared time axis, seconds
```

A recording always needs at least one `Trigger` — `type`/`mask` are raw controller values
(`RecTrigTyp`/`RecTrigMask`), `value`/`max_value` are in user units. `force=True` drives completion
via a forced trigger regardless of the condition; omit it to wait for the condition to fire (up to
`timeout_ms`). Multi-trigger logic (serial/logical, `trigger_mode`/`logic01`/`logic12`) mirrors the
controller's RecData trigger model. The parse, float resolution, and user-unit scaling are done in the
managed engine — the result `Recording` carries per-vector `Vector`s already in user units.

### Firmware download

```python
import aacommpy
from aacommpy import Ethernet

c = aacommpy.Client()
try:
    c.start_server()
    c.download_firmware(
        Ethernet("172.1.1.101"), r"C:\fw\AGM800-3.8.0.0.bin",
        on_progress=lambda permille, status: print(status or f"{permille/10:.0f}%"),
    )
finally:
    c.close()
```

The session must **not** be connected — the engine opens its own connection and leaves it
disconnected. The controller type is auto-discovered so the AGM800 `.bin` path is chosen (pass
`controller_type=` on the connection if the controller may be in BOOT mode). `ver_path` is the
optional `.ver` validation file (AGM800 flashes without it). Runs over **Ethernet (AGM800)** or
**RS-232** — CAN is rejected. `download_firmware` blocks until the flash completes (default 3-minute
bound) and raises `AACommError` on any failure; `on_progress` reports `permille` 0–1000 (status
lines arrive with `permille == -1`). **Destructive** — it reboots the controller.

### FPGA download

`download_fpga(conn, fpga_path, ver_path, ...)` is the FPGA counterpart with the same disconnected-
session contract, progress callback, and blocking semantics. Differences: the `.ver` companion
(`ver_path`) is **required** (it carries the algo/data sizes, CRC and product type), there is no
`verify_currents`/`target`, and FPGA download runs over **RS-232** or **Ethernet** (default IP only) —
never CAN, never AGM800 — and requires **normal (non-BOOT)** mode. **Destructive** — it resets the
controller.

### User-program download

```python
import aacommpy
from aacommpy import Serial

with aacommpy.connect(Serial("COM4")) as c:   # must already be connected — unlike FW/FPGA
    c.download_user_program(r"C:\up\program.cupb2")
```

`download_user_program(path, ...)` downloads a compiled user program (`.cupb2`). **Unlike** firmware/FPGA
it takes **no `Connection`** and the client must **already be connected** — the engine uses the live
session and rejects a disconnected one. Binary auto-detect plus file-version and target-firmware match
are validated in the engine: the program must be compiled for the connected controller's firmware
version. Same progress callback and blocking semantics (default 3-minute bound); raises `AACommError` on
any failure. **Non-destructive** — a user program is reversible.

### User units

`get_user_units_enabled(axis)` / `set_user_units_enabled(axis, enabled)` query and toggle the controller's
per-axis user-units scaling (`axis` is a letter, e.g. `"A"`). The connect handshake reads the user-units
data, so the getter is a local read of that cache — `False` when the feature is unsupported, the session is
disconnected, or user units are being ignored. The setter is confirmed asynchronously by the controller, so
re-query the getter for the applied state.

### Errors and replies (the `requests` model)

- **Transport / lifecycle failures raise** — `connect`/`start_server` raise `ConnectionError` /
  `ServerError` (subclasses of `AACommError`).
- **A controller `ERR NN` reply is data, not an exception.** `query`/`send` return a `Reply` you
  inspect (`reply.is_error`, `reply.text`); call `reply.raise_for_error()` to opt into
  a `ControllerError`. `get`/`set` call it for you (they must return a value). A `Reply` is falsy when
  it is an error, so `if c.query(m):` works.

## Development (running from source, no wheel)

The native binaries aren't checked in. Point the package at your local NativeAOT publishes:

```powershell
$env:PYTHONPATH    = "<repo>/AACommComponents/AACommPy"
$env:AACOMM_LIB    = "<repo>/AACommComponents/AACommNative/bin/Release/net8.0/win-x64/publish/AACommNative.dll"
$env:AACOMM_SERVER = "<repo>/AACommComponents/AACommServerNative/bin/Release/net10.0/win-x64/publish/AACommServerNative.exe"
python -c "import aacommpy; from aacommpy import Ethernet; print(aacommpy.connect(Ethernet('172.1.1.101')).query('AIdentity[1]').text)"
```

`AACOMM_LIB` / `AACOMM_SERVER` override the bundled binaries; `AACOMM_NATIVE_DIR` overrides the whole
bundle directory. Build the two native artifacts per `../AACommNative/packaging/README.md` and
`../AACommServerNative/README.md`.

### Building a wheel

`packaging/build-wheel.ps1` publishes both native artifacts, drops them in `aacommpy/_bundle/`, and
builds a platform wheel.

## Tests

```bash
pip install pytest
pytest aacommpy/tests -m "not hardware"        # pure-Python unit tests, no hardware
# hardware smoke (needs a reachable controller + native binaries via the env vars above):
$env:AACOMM_RIG_AGM800="172.1.1.101"; pytest aacommpy/tests -m hardware
```

## Notes

- Firmware, FPGA, and user-program download are supported: `Client.download_firmware(conn, fw_path)`,
  `Client.download_fpga(conn, fpga_path, ver_path)` (Ethernet / RS-232; never CAN), and
  `Client.download_user_program(path)` (on an already-connected client).
- Wheels are built per host RID — NativeAOT cannot cross-compile, so a release publishes the
  platforms whose build hosts exist. win-x64 today.

## Release notes

`CHANGELOG.md` — the wheel is versioned independently of the `Agito.AAComm` NuGet.
