Metadata-Version: 2.4
Name: agtag-gateway
Version: 0.1.0
Summary: AgTag Gateway ingestion, BLE processing, and AGSEG binary data tools
Author: AgriGatesLLC
License: MIT
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: bleak>=0.21
Requires-Dist: psycopg2-binary>=2.9
Requires-Dist: zstandard>=0.22

# AgTag gateway — ingestion & PostgreSQL source

## Loading a bundle into PostgreSQL for AgNotate — start here

You do not need to parse segments yourself. `tools/agseg_load.py` takes an export bundle
and a database and produces what the gateway used to hand over as a full `pg_dump`: the
six metadata tables, `sensor_data` in exactly today's 19-column shape, and the per-animal
views AgNotate reads. Point AgNotate at the result. It does not need to know the gateway
stopped writing rows.

```bash
pip install zstandard psycopg2-binary           # plus pg_restore on the PATH
python3 tools/agseg_load.py <bundle_dir>     --dsn "host=127.0.0.1 dbname=agtag user=agtag password=agtag"
```

An **empty** database is a valid target — the loader creates `sensor_data` itself. Running
the same bundle twice is harmless (inserts are `ON CONFLICT DO NOTHING` on the primary key;
metadata is replaced, since every bundle carries a full snapshot). Bundles can be loaded in
any order.

What it does, in order: verifies every file against `manifest.json` and refuses a bundle
that does not check out; creates `sensor_data` if absent; `pg_restore`s `meta.dump`;
decodes every segment, **re-verifying the tag's own CRC on every record**, and inserts
the rows scaled to the same physical units the gateway wrote; rebuilds the per-animal
views from `agnotate_link` exactly as the gateway UI creates them.

`the_date` / `the_time` are the legacy local-wall-clock columns and part of the primary
key. They derive from each record's UTC instant via `--tz` (default `UTC`); pass the
gateway's zone, e.g. `--tz America/New_York`, to reproduce what that gateway would have
written. `epoch_ms` is the canonical instant either way. `--edge-id` stamps
`sensor_data.edge_id` (default 1).

**Validated against a live gateway**, not fixtures: 60,000 real rows replayed through
ingest, exported as a real bundle, loaded into an empty database and diffed against the
originals — all 19 channels and `epoch_ms` identical on every row, metadata at the live
counts, per-animal views partitioning a tag's rows exactly. One thing a loaded table
cannot carry: the legacy `the_time` column on a live gateway holds microseconds of
*ingest* jitter (`datetime.now()` at receive), while the tag clock and the format are
millisecond resolution. Those digits are not the animal's time and are not recoverable.
Do not load bundles into a database that also received the same samples through the
old row-store path — the primary key includes `the_time`, so the jitter would let one
sample appear twice. A gateway is on one path or the other.

## Opening a `.agseg` file by hand

`.agseg` is a binary container we defined, not text: a 64-byte header, then
zstd-compressed frames of 72-byte records, then an index trailer. It will not open in a
text editor and there is nothing about it on the internet — the whole format is
`docs/agtag_binary_segment_format_v1.md` in this repo, and the reader is next to it.

```bash
pip install zstandard
python3 tools/agseg_dump.py info   tests/vectors/three_records.agseg
python3 tools/agseg_dump.py dump   tests/vectors/three_records.agseg --scaled
python3 tools/agseg_dump.py verify tests/vectors/multiblock.agseg
```

`info` prints the header, `dump` decodes records to CSV (raw ints, or `--scaled` to
physical units), `verify` re-checks every record's wire CRC. To look at the raw bytes,
`xxd tests/vectors/three_records.agseg | head` — the first four are `AGSG`, the magic.

Shared with AgriGates by ARG, 2 Sep 2026, for the gateway data-storage discussion.
This is the complete path from a packed BLE record to a PostgreSQL row and back
out again, plus the binary segment format that replaces the row-per-sample store.

> **Updated 4 Sep 2026.** The wire contract here was originally a stale snapshot: it
> described `agtag_sample_t` as 46 bytes with a uint16 `seq`. The firmware actually
> sends **52 bytes** — `seq` is uint32, and the record carries `battery_mv`,
> `cal_status` and `flags` after the channels. `shared/agtag_packet.py` is now correct
> and is authoritative. Anything built against the 46-byte shape computes the CRC over
> the wrong bytes and rejects every genuine sample.

## Where to start, depending on what you are doing

**Writing the segment converter (the current work).** Start at
`docs/agtag_binary_segment_format_v1.md`, then `tools/agseg_dump.py` as a worked reader
and `tools/agseg_to_csv.py` as a worked converter, and check whatever you write against
`tests/vectors/`. See "The binary segment format (v1)" below.

**Understanding how the gateway ingests today.** The table below walks the current
row-per-sample path, which is what ships on the OnLogic units now. That path is not going
away while those units are in the field, and `sensor_data` looks the same to AgNotate
either way.

## Three things people hit on a live gateway

**`sensor_data` is in the `agtag` database, not `agrigates`.** The tables are created at
install time by `deploy/install.sh` — it creates the database, then runs `migrate.py up`,
and `0001_baseline.sql` creates `sensor_data` and the rest in schema `public`. Nothing is
created on demand. A converted gateway can still have the legacy AgriGates `agrigates`
database sitting on the same server and port, which is an easy place to look and find
nothing:

```bash
psql "host=127.0.0.1 dbname=agtag user=agtag password=agtag" -c '\dt'
```

**Ingest is a systemd service, not a cron job.** `agtag-ingest.service` runs
`ble_ingest.py` continuously as a BLE central — it stays connected and drains tags when
they notify, rather than being woken on a schedule. It starts at boot
(`WantedBy=multi-user.target`, after `postgresql` and `bluetooth`) and restarts on exit
(`Restart=always`, `RestartSec=3`):

```bash
systemctl status agtag-ingest
journalctl -u agtag-ingest -f
```

The units that *are* periodic are `agtag-btreset.timer` and `agtag-blewatch.timer` —
Bluetooth housekeeping, not ingestion.

**Parsing is immediate; the data usually is not.** There is no schedule and no polling.
BlueZ delivers a GATT notification, which fires `on_notify` synchronously: one notification
carries N back-to-back 52-byte records, `_parse_notification()` slices it in `SAMPLE_SIZE`
steps, `parse_sample()` CRC-checks each one, and `store_samples()` does a single batched
INSERT for that notification. Records failing CRC, or a trailing fragment too short to be a
whole record, are dropped there and never reach PostgreSQL.

But the samples themselves are usually not recent. A tag records to its own flash ring
whether or not the gateway is connected; on connect we send `CMD_START` and the tag
**replays that buffer**. A tag out of range all afternoon hands over the whole afternoon in
one session. It is a drain, not a live feed.

That is why stored timestamps are not arrival time. Each session sets the tag clock, reads
its time base back (`epoch_now_ms`, `t_ms_now`, `seq_at_boot`), and back-dates every sample
to its true capture instant:

```
absolute = epoch_now_ms - (t_ms_now - sample.t_ms)
```

Without it, an hour of buffered data would land on one arrival second and offline
video/audio alignment in AgNotate would be impossible. Records buffered across a tag reboot
— where the uptime counter has since reset, so the delta is meaningless — are anchored at
receipt instead. **Anything consuming this data should treat capture order and arrival order
as unrelated.**

Finally, silence is normal: motion-gated tags sleep when the animal is still and send
nothing for long stretches. No notifications does not mean ingest is stuck.

## Read in this order

| # | File | What it is |
|---|------|-----------|
| 1 | `shared/agtag_packet.py` | **Start here.** The whole wire contract — frame format, the 52-byte `agtag_sample_t` record, CRC16-CCITT, the BNO08x Q-point scaling, and the incremental de-framer. ~160 lines. |
| 2 | `shared/agtag_packet.h` | The C mirror of the same contract, kept byte-identical across the tag, dongle and gateway repos. Authoritative for struct layout. |
| 3 | `shared/schema.sql` | The designed schema with the rationale comments — why `seq` is in the primary key, why identity is derived, what changed from the inherited schema. |
| 4 | `gateway/ingestor.py` | **The expansion point.** `store_samples()` is where a parsed record becomes a `sensor_data` row: identity derivation, timestamp back-dating, and the upsert. Shared by both ingest paths. |
| 5 | `gateway/ble_ingest.py` | The live path — a `bleak` central that drains tags over the gateway's own Bluetooth (no dongle). `_parse_notification()` slices notifications into 52-byte records; `on_notify()` hands them to `store_samples()`. |
| 6 | `gateway/export_job.py` | The outbound side — per-device gzipped CSV in the AgriGates data standard, plus the full `pg_dump -Fc` used for the AgNotate handoff. |
| 7 | `gateway/deploy/schema.sql` | The **actual deployed** schema, as a `pg_dump` from a live gateway (PostgreSQL 16.14 / Ubuntu 24.04). Use this over #3 if the two disagree. |
| 8 | `gateway/test_pipeline.py`, `gateway/test_export.py` | Round-trip tests — the shortest executable description of the expected behaviour. |

`gateway/line_ingest.py` is included for completeness but is **not** the production
path: it parses the bring-up firmware's human-readable text output. The binary path
is `ingestor.py`. Both derive the same `sn`, so they write the same device rows.

## The flow in one line

```
tag flash ring (52 B/sample packed)
  -> BLE notify, N records back-to-back
  -> _parse_notification()      ble_ingest.py
  -> parse_sample()             agtag_packet.py   CRC16 check, int16 Q-point -> float
  -> store_samples()            ingestor.py       timestamps + INSERT
  -> sensor_data                PostgreSQL        19 double-precision columns
  -> _write_device_csv()        export_job.py     gzipped CSV, AgriGates standard
```

The binary exists only in flight. Nothing is currently stored as `bytea` — every
record is expanded at ingest, which is the thing under discussion.

## Three details worth knowing before you read the code

**Identity is derived, not assigned.** `sn = crc32(ble_mac) & 0x7FFFFFFF`
(`agtag_packet.derive_sn`). There is no hand-managed serial list. Note that
`ble_mac` is stored in LE byte order (`ble_ingest._le_mac`) so both ingest paths
agree.

**`seq` is part of the primary key.** `(sensor_id, the_date, the_time, seq)`. The
inherited schema keyed on the first three and silently dropped any two samples
landing in the same second. `seq` also makes replay idempotent via
`ON CONFLICT ... DO NOTHING`.

**Timestamps are back-dated, not receipt-stamped.** Samples come off a ring buffer,
so receipt time is not capture time. After `SETTIME` the gateway reads the tag's
time-base characteristic for `(epoch_now_ms, t_ms_now, seq_at_boot)` and
reconstructs each sample's true wall-clock, with guards for u32 uptime wrap and for
records buffered across a tag reboot. See the comments in `store_samples()`. This
is what makes offline video/audio alignment possible, and it is the part most
easily got wrong by a reimplementation.

## The binary segment format (v1)

`docs/agtag_binary_segment_format_v1.md` is the normative spec. Motion samples stop
becoming one PostgreSQL row each and are written as packed 72-byte records inside zstd
`.agseg` segments; tag registry, health, RSSI history, configuration and animal pairing
stay relational and travel in the bundle as a `pg_dump -Fc`.

| File | What it is |
|---|---|
| `docs/agtag_binary_segment_format_v1.md` | The spec. Where it and any code here disagree, the spec wins. |
| `tools/agseg_load.py` | **The converter.** Bundle → AgNotate-ready PostgreSQL. Start here. |
| `tools/agseg_dump.py` | Reference **reader**: `info` / `verify` / `dump` (CSV or JSONL, raw or scaled) / `bundle`. Self-contained — stdlib plus a zstd binding. |
| `tools/agseg_to_csv.py` | Reference **converter**: segments → the AgriGates data-standard CSV, one gzipped file per device. |
| `tests/vectors/` | Committed golden vectors, including a torn `.part`. Anything you write should agree with these. |
| `tests/test_agseg_format.py` | Conformance tests over those vectors. |

### On `tools/agseg_to_csv.py`

**This is a reference design, not the converter.** The production converter into AgNotate
is AgriGates' — we do not know your ingestion contract (file layout, naming, whether you
load to a database), and that is deliberately left with you. This exists so both sides
have something concrete to agree on, and so ARG could validate the binary path end to end
without blocking on it.

Please write yours against `docs/agtag_binary_segment_format_v1.md` and check it against
`tests/vectors/`, rather than against this file. A converter built against our code rather
than the spec is precisely how the 46-byte discrepancy above went unnoticed for as long as
it did — the reader, the generator and the vectors all agreed with each other and
disagreed with the firmware.

### It produces exactly what the gateway produces today

`agseg_to_csv.py` reproduces, from segments alone, the same gzipped per-device CSV
`gateway/export_job.py` writes from the `sensor_data` table — same header, same ISO 8601
UTC `datetime` with milliseconds, same 14 standard columns, same values. ARG ran both
paths over the same live rows on a gateway (device `0x4A47B0EC`, 13,673 rows): the header
matched and the two exports were **line-for-line identical** over the window.

Separately, replaying 60,000 real rows through the ingest path and reading the segments
back reproduced every channel value exactly, with all 60,000 records re-verifying against
the tag's own CRC.

### Verify a record without trusting the gateway

Every record stores the tag's original CRC16 **and** every byte that CRC covers. Re-pack
`<II3h3h3h4h3h3hHBB` from `t_ms`, `seq`, the 19 channels, `battery_mv`, `cal_status` and
`tag_flags`, take CRC16-CCITT over those 50 bytes, and it must equal the stored `crc16`.
`record_crc_ok()` in `agseg_dump.py` is the whole recipe. Assert it on every record.

## Notes

- Units already match the published AgriGates data standard: accel m/s², gyro rad/s,
  mag µT, quaternion `[w, x, y, z]`.
- Format strings are explicitly little-endian (`<`), so records are portable across
  architectures.
- `DSN` defaults to `host=127.0.0.1 dbname=agtag user=agtag password=agtag` — a
  loopback-only default for the gateway's local database, overridable via the
  `AGTAG_DSN` environment variable.
- A corrupt record fails its CRC in `parse_sample()` and is dropped at ingest, so it
  never reaches storage.
- `AGTAG_PROTO_VERSION` is currently a compile-time constant only — it is **not**
  written into the records themselves. That matters if stored binary is ever going
  to outlive the code that wrote it.
