# Consolidated Bus & Data Collection Specification

Data: 2026-03-20

Acest document consolidează specificațiile existente într-o referință unică pentru busuri, colectare de date și conformitate cu cerințele historian-ului. Nu înlocuiește documentele sursă, ci le leagă și stabilește deciziile rămase deschise.

Documentele sursă rămân canonice. Acest document este o hartă de navigare și un registru de decizii.

---

## 1. Arhitectura Pipeline-ului

```

Device / External System

    ↓

Protocol Adapter (Node-RED)

    ↓

Canonical MQTT Bus (home | energy)

    ↓

Historian Worker

    ↓

telemetry.ingest_measurement(...)   ← measurement path (implementat)

telemetry.ingest_counter(...)       ← counter path (contract stabilizat, neimplementat)

    ↓

PostgreSQL Historian

```

Documente relevante:

- `README.md` — arhitectura generală

- `addapters.md` — responsabilitățile adapterelor

- `historian_worker.md` — responsabilitățile worker-ului

---

## 2. Contractul MQTT Partajat

Definit în `mqtt_contract.md`. Toate busurile moștenesc aceste reguli.

### 2.1 Namespace

- Busuri semantice: `<site>/<bus>/...`

- Namespace operațional: `<site>/sys/...`

- `<site>` = site-id stabil, kebab-case (ex: `vad`)

- Busuri v1: `home`, `energy`

### 2.2 Stream-uri Canonice

| Stream | Scop | Retain | QoS |

|---|---|---|---|

| `value` | sample live hot-path | false | 1 |

| `last` | ultimul sample cu timestamp, pentru cold-start | true | 1 |

| `set` | comandă/request | false | 1 |

| `meta` | metadata statică sau slow-changing | true | 1 |

| `availability` | online/offline | true (+ LWT) | 1 |

Reguli:

- adaptoarele emit date live doar pe `value`

- adaptoarele deduplică `value` când valoarea semantică nu s-a schimbat

- adaptoarele actualizează `last` (retained) la fiecare schimbare de timestamp

- `state` și `event` sunt legacy, nu se introduc de adaptoare noi

- `set` nu se reține niciodată

- historian-ul ingestează doar `value`, ignoră `last`, `set`, `meta`, `availability`

### 2.3 Payload Profiles

**Profile A — Scalar (default pentru hot paths):**

```

23.6

41

true

on

```

- metadata pe `meta` separat (retained)

- `observed_at` = ingestion time (acceptabil pentru Phase 1)

**Profile B — Envelope JSON (opțional):**

```json

{

  "value": 23.6,

  "unit": "C",

  "observed_at": "2026-03-08T10:15:12Z",

  "quality": "good"

}

```

- câmpuri opționale: `published_at`, `source_seq`, `annotations`

- utilizat când fidelitatea timestamp-ului contează sau când quality trebuie propagat per-sample

### 2.4 Meta Contract

Fiecare topic retained `meta` descrie familia sibling de `value`/`last`.

Formă minimă recomandată:

```json

{

  "schema_ref": "mqbus.home.v1",

  "payload_profile": "scalar",

  "data_type": "number",

  "unit": "C",

  "adapter_id": "z2m-main",

  "source": "zigbee2mqtt",

  "source_ref": "0x00158d0008aa1111",

  "source_topic": "zigbee2mqtt/SENSOR/vad/bedroom/bedroom-sensor",

  "precision": 0.1,

  "historian": {

    "enabled": true,

    "mode": "sample"

  }

}

```

Câmpuri historian:

- `historian.enabled`: boolean

- `historian.mode`: `sample` | `state` | `event` | `ignore`

- `historian.retention_class`: opțional (`short`, `default`, `long`)

- `historian.sample_period_hint_s`: opțional

### 2.5 Time Semantics

Trei timestamp-uri distincte:

| Timestamp | Semnificație |

|---|---|

| `observed_at` | când sursa a observat/măsurat valoarea |

| `published_at` | când adapterul a publicat mesajul |

| `ingested_at` | când worker-ul a procesat mesajul |

Reguli:

- dacă sursa are timestamp, adapterul îl păstrează ca `observed_at`

- dacă nu, adapterul omite `observed_at`; worker-ul folosește `ingested_at`

- adaptoarele nu fabrică timestamp-uri

- dacă adapterul estimează timpul, folosește Profile B cu `quality=estimated`

### 2.6 Quality Model

Valori: `good`, `estimated`, `degraded`, `stale`, `invalid`

- `invalid` nu se emite pe bus semantic; merge pe `sys/.../error`

- `quality` se omite doar dacă `good` este implicit

---

## 3. Home Bus

Definit în `home_bus.md`. Bus room-centric pentru senzori, control, automatizare.

### 3.1 Topic Grammar

```

<site>/home/<location>/<capability>/<device_id>/<stream>

```

Exemple:

- `vad/home/bedroom/temperature/bedroom-sensor/value`

- `vad/home/living-room/motion/radar-south/value`

- `vad/home/kitchen/light/ceiling-switch/set`

### 3.2 Capability Catalog

**Environmental:** `temperature`, `humidity`, `pressure`, `illuminance`, `co2`, `voc`, `pm25`, `pm10`

**Presence/Safety:** `motion`, `presence`, `contact`, `water_leak`, `smoke`, `gas`, `tamper`

**Control:** `light`, `power`, `lock`, `cover_position`, `thermostat_mode`, `target_temperature`, `fan_mode`, `button`

**Device health:** `battery`, `battery_low`

Reguli:

- `power` pe `home` doar pentru control semantics (on/off)

- metrici electrice (`active_power`, `energy_total`) pe `energy`

- `linkquality` nu pe `home` (merge pe `sys` sau viitor `network`)

- `presence` = stare derivată de nivel mai înalt; detecția brută mmWave cu `fading_time=0` se publică ca `motion`

### 3.3 Historian Mapping

```

metric_name = <capability>

device_id   = <location>.<device_id>

```

Exemplu:

- Topic: `vad/home/bedroom/temperature/bedroom-sensor/value`

- `metric_name = temperature`

- `device_id = bedroom.bedroom-sensor`

- `value = 23.4` (scalar)

- `observed_at = ingested_at` (Profile A fallback)

---

## 4. Energy Bus

Definit în `energy_bus.md`. Bus pentru topologia electrică: producție, stocare, grid, load.

### 4.1 Topic Grammar

```

<site>/energy/<entity_type>/<entity_id>/<metric>/<stream>

```

Entity types: `source`, `storage`, `grid`, `load`, `transfer`

Exemple:

- `vad/energy/source/pv-roof-1/active_power/value`

- `vad/energy/storage/battery-main/soc/value`

- `vad/energy/grid/main-meter/import_power/value`

- `vad/energy/load/living-room-tv/active_power/value`

### 4.2 Metric Classes

**Measurement-style** (compatibile cu `ingest_measurement()` acum):

`active_power`, `voltage`, `current`, `frequency`, `soc`, `charge_power`, `discharge_power`

**Counter-style** (pe bus, dar NU prin measurement API — urmează `ingest_counter()`):

`energy_total`, `import_energy_total`, `export_energy_total`

### 4.3 Units Canonice

| Metric | Unit |

|---|---|

| power | `W` |

| energy | `Wh` sau `kWh` |

| voltage | `V` |

| current | `A` |

| frequency | `Hz` |

| state of charge | `%` |

### 4.4 Historian Mapping

```

metric_name = <metric>

device_id   = <entity_type>.<entity_id>

```

Exemplu:

- Topic: `vad/energy/storage/battery-main/soc/value`

- `metric_name = soc`

- `device_id = storage.battery-main`

### 4.5 Ownership Rule

- Dacă valoarea e pentru control utilizator într-o cameră → `home`

- Dacă valoarea e pentru contabilitate electrică → `energy`

- Un device fizic poate proiecta pe ambele busuri (smart plug: `power` pe `home`, `active_power` pe `energy`)

---

## 5. Namespace Operațional (sys)

Definit în `sys_bus.md`. Nu este un bus semantic.

### 5.1 Topic Grammar

```

<site>/sys/<producer_kind>/<instance_id>/<stream>

```

Producer kinds v1: `adapter`, `historian`

### 5.2 Stream-uri Operaționale

| Stream | Scop | Retain |

|---|---|---|

| `availability` | liveness-ul componentei | true + LWT |

| `stats` | contori operaționali, snapshot periodic | true |

| `error` | erori structurate pentru operator | false |

| `dlq` | mesaje dead-letter cu context | false |

Exemple:

- `vad/sys/adapter/z2m-main/availability` → `online`

- `vad/sys/historian/main/error` → `{"code":"unknown_metric",...}`

---

## 6. Adaptoare

Definit în `addapters.md` și `adapter_implementation_examples.md`.

### 6.1 Responsabilități

1. Traducere topic-uri din format vendor în bus canonic

2. Normalizare payload (JSON → scalar, conversie unități)

3. Păstrare timestamp sursă (dacă există)

4. Fan-out: 1 mesaj inbound → N publicațiuni canonice

5. Proiecție multi-bus (smart plug → `home` + `energy`)

6. Publicare `meta` retained înaintea traficului live

7. Publicare `availability` retained + LWT

8. Routing erori pe `sys/.../error` și `sys/.../dlq`

### 6.2 NON-Responsabilități

- Nu implementează logica HomeKit

- Nu implementează reguli de automatizare

- Nu agregă senzori

- Nu stochează date istorice

- Nu repară mesaje malformed (le expune pe `sys`)

### 6.3 Publish Boundary

La publicare, mesajul conține DOAR:

```javascript

{ topic: normalizedTopic, payload: normalizedValue }

```

Structurile interne de normalizare se șterg înainte de publish.

### 6.4 Z2M Topic Convention

Friendly name: `<device_type>/<site>/<location>/<device_id>`

Exemplu:

- Z2M: `zigbee2mqtt/ZG-204ZV/vad/balcon/south`

- Canonical: `vad/home/balcon/illuminance/south/value`

Permite traducere deterministă fără lookup tables.

### 6.5 Z2M Field Mapping

| Z2M field | Bus | Canonical capability/metric |

|---|---|---|

| `temperature` | home | `temperature` |

| `humidity` | home | `humidity` |

| `pressure` | home | `pressure` |

| `illuminance` | home | `illuminance` |

| `contact` | home | `contact` |

| `occupancy` (PIR) | home | `motion` |

| `presence` (mmWave, `fading_time=0`) | home | `motion` |

| `battery` | home | `battery` |

| `state` (plug/switch) | home | `power` |

| `action` (remote) | home | `button` |

| `power` (smart plug) | energy | `active_power` |

| `voltage` (smart plug) | energy | `voltage` |

| `current` (smart plug) | energy | `current` |

| `energy` (smart plug) | energy | `energy_total` |

### 6.6 Mapping Registry

Adapoarele folosesc configurație declarativă. Câmpuri recomandate:

`source_system`, `source_topic_match`, `source_field`, `target_bus`, `target_location` / `target_entity_id`, `target_capability` / `target_metric`, `target_device_id`, `stream`, `payload_profile`, `unit`, `historian_enabled`, `historian_mode`

### 6.7 Auto-Provisioning

Adapoarele derivă dimensiunile semantice din structura topic-ului Z2M. Override-uri se aplică doar pentru excepții. Default-uri: `bus=home`, `location=unknown`, `stream=value`.

---

## 7. Historian Worker

Definit în `historian_worker.md`.

### 7.1 Subscription Model

```

+/home/+/+/+/value       ← ingestie primară

+/energy/+/+/+/value      ← ingestie primară

+/home/+/+/+/meta         ← cache local

+/energy/+/+/+/meta       ← cache local

+/home/+/+/+/last         ← ignorat pentru ingestie

+/energy/+/+/+/last       ← ignorat pentru ingestie

```

### 7.2 Topic → Historian Mapping (sumar)

| Bus | `metric_name` | `device_id` |

|---|---|---|

| `home` | `<capability>` | `<location>.<device_id>` |

| `energy` | `<metric>` | `<entity_type>.<entity_id>` |

### 7.3 Payload Handling

**Scalar (Profile A):**

- `value` = payload parsat ca number/boolean

- `observed_at` = ingestion time

- `unit` din meta cache

**Envelope (Profile B):**

- `value` = `payload.value`

- `observed_at` = `payload.observed_at` sau ingestion time

- `unit` = `payload.unit` sau meta cache

### 7.4 Type Routing

| Tip payload | Overload DB |

|---|---|

| numeric | `ingest_measurement(..., value::double precision, ...)` |

| boolean | `ingest_measurement(..., value::boolean, ...)` |

| string enum | SKIP (nu se ingestează fără encoding policy explicit) |

| counter cumulative | SKIP measurement path; rutare pe counter pipeline |

### 7.5 Ordering

- Serializare scrieri per `(metric_name, device_id)` — obligatoriu

- Paralelism acceptabil între device-uri/metrici diferite

- Retry-urile păstrează ordinea pe stream

### 7.6 Error Handling

**Erori permanente** (log + DLQ, nu retry):

- topic incompatibil cu grammar-ul busului

- tip payload incompatibil

- metric necunoscut

- out-of-order measurement

- valoare respinsă de policy

**Erori tranziente** (retry cu ordine):

- PostgreSQL indisponibil

- deconectare rețea

- deconectare broker temporară

### 7.7 Meta Cache

- Cheie: stem topic (fără ultimul segment stream)

- Câmpuri cache: `payload_profile`, `data_type`, `unit`, `historian.enabled`, `historian.mode`, `schema_ref`, `adapter_id`, `source`, `source_ref`

- Missing meta = mod degradat, nu stop

### 7.8 Deployment

Model recomandat inițial: un singur worker per site consumând `home` + `energy`.

---

## 8. Measurement Ingestion API

Definit în `tdb_ingestion/mqtt_ingestion_api.md`. Implementat și funcțional.

### 8.1 Semnătură

```sql

-- Numeric

SELECT * FROM telemetry.ingest_measurement(

    p_metric_name => $1,

    p_device_id   => $2,

    p_value       => $3::double precision,

    p_observed_at => $4::timestamptz

);

-- Boolean

SELECT * FROM telemetry.ingest_measurement(

    p_metric_name => $1,

    p_device_id   => $2,

    p_value       => $3::boolean,

    p_observed_at => $4::timestamptz

);

```

### 8.2 Comportament Automat

Gestionat de PostgreSQL, worker-ul nu reimplementează:

- Auto-provisioning device (`ensure_device()`)

- Append-only enforcement (watermark per `metric_name, device_id`)

- Policy lookup din `telemetry.metric_policies`

- Gap detection la sosirea următorului measurement

- Segment splitting (value change, NULL, policy change, gap)

- Tail NULL simulation la query time

### 8.3 Response Actions

`opened`, `opened_null`, `extended`, `extended_null`, `split`, `null_to_value`, `value_to_null`, `gap_split`, `gap_to_null`

Worker-ul le loghează, nu branșează logica pe ele.

### 8.4 Idempotency

Limitare curentă: `ingest_measurement()` nu acceptă `idempotency_key`. Replay după network failure ambiguă poate produce eroare out-of-order. Worker-ul tratează aceasta ca duplicate operațional.

### 8.5 Advisory Locking

- Un advisory lock per `(metric_name, device_id)` pe tranzacție

- Implementat ca wait (nu NOWAIT)

- Recomandare: writer affinity per `(metric_name, device_id)` via routing determinist

---

## 9. Counter Ingestion API

Definit în `tdb_ingestion/counter_ingestion_api.md`. Contractul este acum materializat în repo și reprezintă ținta de implementare pentru worker și backend.

### 9.1 Semnătură

```sql

SELECT * FROM telemetry.ingest_counter(

    p_metric_name     => $1,

    p_device_id       => $2,

    p_counter_value   => $3::numeric,

    p_observed_at     => $4::timestamptz,

    p_source_sequence => $5,  -- nullable

    p_idempotency_key => $6,  -- nullable

    p_snapshot_id     => $7   -- nullable

);

```

### 9.2 Reguli Core

- Append-only, ordonat per `(metric_name, device_id)`

- `counter_value` nu acceptă NULL; dacă sursa nu are valoare, observația se omite

- Gap-uri derivate la query time din freshness, nu scrieri sintetice NULL

- Scăderea valorii = boundary semantic (`reset_boundary`, `rollover_boundary`, `invalid_drop`)

- Delta și rate nu traversează boundary-uri de reset/rollover

- Worker-ul rutează aceste metrici pe `ingest_counter()`, nu pe `ingest_measurement()`

### 9.3 Stream Identity

```

stream = (metric_name, device_id)

```

Nu există `stream_id` suplimentar. Ordering, freshness, replay metadata — toate per `(metric_name, device_id)`.

### 9.4 Reliability Levels

| Nivel | Cerințe | Garanții |

|---|---|---|

| `degraded / at-most-once` | fără `source_sequence`, fără `idempotency_key` | retry nesigur |

| `recommended / at-least-once` | `source_sequence` sau `idempotency_key` | replay detectabil |

| `stronger replay safety` | ambele | deduplicare explicită |

### 9.5 Reporting Modes

| Mode | Semantică |

|---|---|

| `periodic` | cadență așteptată; lipsa update-ului = gap |

| `on_change` | publicare la schimbare; tăcerea nu = delta 0 |

| `hybrid` | on_change + heartbeat periodic |

### 9.6 Freshness Defaults

| Reporting mode | Input | Default `stale_after_s` |

|---|---|---|

| `periodic` | `expected_interval_s` | `expected_interval_s * 2` |

| `on_change` | `heartbeat_interval_s` | `heartbeat_interval_s * 2` |

| `hybrid` | `heartbeat_interval_s` | `heartbeat_interval_s * 2` |

`stale_after_s` explicit în policy overridează default-ul.

---

## 10. Decizii Luate

### 10.1 Closed for Phase 1

Aceste decizii sunt stabile și se poate implementa pe baza lor.

| # | Decizie | Status |

|---|---|---|

| 1 | Measurement path: numeric + boolean prin `ingest_measurement()` | Acceptat, implementat |

| 2 | Counter path: separat de measurement, prin `ingest_counter()` | Contract stabilizat, neimplementat |

| 3 | `last` nu se ingestează ca time-series | Închis |

| 4 | Ordering per `(metric_name, device_id)` obligatoriu | Închis |

| 5 | Missing source timestamp → fallback la `ingested_at` | Închis |

| 6 | Missing `meta` = mod degradat, nu hard stop | Închis |

| 7 | String enum states: nu se ingestează fără encoding policy | Închis |

| 8 | Counter `stream_id = (metric_name, device_id)` — fără ID suplimentar | Închis |

| 9 | Counter: NULL never, gaps derivate la query time | Închis |

| 10 | Counter: un model comun cu profile de domeniu, nu separare energy/traffic | Închis |

| 11 | Permanent errors → DLQ + log, nu retry | Închis |

| 12 | Transient errors → retry cu ordine păstrată | Închis |

### 10.2 Decizii Luate Acum (în acest document)

| # | Decizie | Alegere | Rațiune |

|---|---|---|---|

| D1 | Snapshot pe bus vs poller→worker | **Snapshot rămâne pe boundary-ul poller→worker.** Bus-ul semantic păstrează publicații per-stream. | Busul semantic e optimizat pentru publicații lightweight per-topic. Snapshot-ul e un artefact de colectare, nu o semantică de bus. Dacă mai târziu se dovedește necesar, se poate evolua contractul explicit. |

| D2 | Freshness defaults | **`stale_after_s = expected_interval * 2`** (conform cu draft-ul existent) | Multiplicatorul 2x din draft e suficient de conservator. Override explicit per policy rămâne disponibil. |

| D3 | Profile energy/traffic counter | **Amânat până la date reale.** | Nu avem network bus; profilele de traffic nu sunt relevante acum. Energy counter profile se definesc când implementăm `ingest_counter()`. |

---

## 11. Puncte Deschise (non-blocante)

Aceste puncte nu blochează Phase 1 dar vor necesita atenție:

| # | Punct | Când devine relevant |

|---|---|---|

| O1 | `idempotency_key` pe measurement API | Când avem nevoie de replay sigur pe measurement path |

| O2 | Encoding policy pentru string enum states | Când vrem să persistăm HVAC modes sau alte enums |

| O3 | Forma fizică a reset/rollover boundary în counter storage | La implementarea `ingest_counter()` |

| O4 | Batch SQL pentru counter ingestion | Când volumul de counters justifică optimizarea |

| O5 | Completitudine snapshot (cum se declară explicit) | Când se adaugă surse bulk polling |

---

## 12. Testing & Validation

### 12.1 Prima Matrice de Test

| # | Device | Capabilities | Bus | Historian |

|---|---|---|---|---|

| 1 | Room sensor | `temperature`, `humidity`, `battery` | home | Da (numeric) |

| 2 | Door sensor | `contact`, `battery` | home | Da (boolean + numeric) |

| 3 | Smart plug | `power` (control) | home | Nu (string enum) |

| 3 | Smart plug | `active_power`, `voltage`, `current` | energy | Da (numeric) |

| 3 | Smart plug | `energy_total` | energy | Nu (counter path) |

| 4 | Remote | `button` / `action` | home | Nu (event, skip default) |

### 12.2 Acceptance Criteria Phase 1

1. Historian stochează samples `home` + `energy` (measurement) fără parsing vendor topics

2. `value` topics ingestabile doar cu topic parsing + opțional `meta` cache

3. Smart plug proiectat pe ambele busuri fără ambiguitate

4. `meta.historian.mode` respectat (sample/state/event/ignore)

5. Missing timestamps nu blochează ingestia

6. Missing `meta` degradează, nu blochează

7. Erori adapter pe `sys`, nu pe bus semantic

8. Counter totals coexistă pe bus fără a fi confundate cu measurement path

---

## 13. Hartă de Navigare a Documentelor

| Document | Ce conține | Când îl citești |

|---|---|---|

| `mqtt_contract.md` | Contract partajat: payload, meta, time, quality, delivery | Când construiești orice componentă care publică sau consumă de pe bus |

| `home_bus.md` | Topic grammar, capabilities, historian mapping home | Când lucrezi pe adaptoare home sau pe historian worker |

| `energy_bus.md` | Topic grammar, entity types, metrics, historian mapping energy | Când lucrezi pe surse de energie |

| `sys_bus.md` | Namespace operațional, stream-uri sys | Când adaugi observabilitate la adaptoare sau worker |

| `addapters.md` | Responsabilități adapter, fan-out, multi-bus, Z2M convention | Când construiești un adapter |

| `adapter_implementation_examples.md` | Patterns Node-RED, cold-start, failure modes | Când implementezi concret în Node-RED |

| `historian_worker.md` | Subscription, mapping, payload handling, ordering, errors | Când construiești historian worker-ul |

| `tdb_ingestion/mqtt_ingestion_api.md` | API measurement implementat, error handling, ordering | Când scrii cod de ingestie measurement |

| `tdb_ingestion/counter_ingestion_api.md` | Contract counter stabilizat (neimplementat) | Când implementezi counter ingestion |

---

## 14. End-to-End Trace: De la Device la PostgreSQL

### Exemplu 1: Temperatură cameră (measurement, Profile A)

```

[Z2M Device] zigbee2mqtt/SENSOR/vad/bedroom/bedroom-sensor

  payload: {"temperature": 23.4, "humidity": 41, "battery": 87}

[Adapter] fan-out → 3 publicațiuni:

  vad/home/bedroom/temperature/bedroom-sensor/value  → 23.4

  vad/home/bedroom/humidity/bedroom-sensor/value     → 41

  vad/home/bedroom/battery/bedroom-sensor/value      → 87

[Worker] topic parse:

  metric_name = "temperature"

  device_id   = "bedroom.bedroom-sensor"

  value       = 23.4

  observed_at = ingested_at (Profile A, no source timestamp)

[PostgreSQL]

  SELECT * FROM telemetry.ingest_measurement(

    'temperature', 'bedroom.bedroom-sensor',

    23.4::double precision, now()

  );

  → action: extended

```

### Exemplu 2: Smart plug dual projection (measurement + counter)

```

[Z2M Device] zigbee2mqtt/PLUG/vad/living-room/tv-plug

  payload: {"state":"on", "power":126.8, "energy":4.72, "voltage":229.4, "current":0.58}

[Adapter] fan-out → 5 publicațiuni pe 2 busuri:

  vad/home/living-room/power/tv-plug/value                → "on"          (home, string enum, skip historian)

  vad/energy/load/living-room-tv/active_power/value       → 126.8         (energy, measurement)

  vad/energy/load/living-room-tv/voltage/value            → 229.4         (energy, measurement)

  vad/energy/load/living-room-tv/current/value            → 0.58          (energy, measurement)

  vad/energy/load/living-room-tv/energy_total/value       → 4.72          (energy, counter — skip measurement path)

[Worker]

  active_power → ingest_measurement('active_power', 'load.living-room-tv', 126.8, now())  ✓

  voltage      → ingest_measurement('voltage', 'load.living-room-tv', 229.4, now())       ✓

  current      → ingest_measurement('current', 'load.living-room-tv', 0.58, now())        ✓

  energy_total → skip measurement path, route to counter pipeline                         ⏳

  "on"         → skip (string enum, no encoding policy)                                   ⊘

```

### Exemplu 3: Contact sensor (boolean measurement)

```

[Z2M Device] zigbee2mqtt/SNZB-04P/vad/entrance/front-door

  payload: {"contact": false, "battery": 94}

[Adapter] fan-out → 2 publicațiuni:

  vad/home/entrance/contact/front-door/value  → false

  vad/home/entrance/battery/front-door/value  → 94

[Worker]

  contact → ingest_measurement('contact', 'entrance.front-door', false, now())  ✓

  battery → ingest_measurement('battery', 'entrance.front-door', 94, now())     ✓

```