diff --git a/README.md b/README.md index e0a9abe..0a58763 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A Portainer / docker compose stack that: -1. **mqtt-listener** — subscribes to MQTT topics on the form `//`, injects `imei` and a `device` name into the JSON payload, and stores it in PostgreSQL. +1. **mqtt-listener** — subscribes to MQTT topics on the form `//`, injects `imei` and a `device` name into each JSON object of the payload (a single object or an array of objects), and stores every object as its own row in PostgreSQL. 2. **api** — a FastAPI read API over the database, protected by static app tokens. 3. **db** — PostgreSQL 16 storing the messages (JSONB payloads). @@ -84,7 +84,8 @@ PASS: count endpoint returned {'count': N} - It looks the IMEI up in `configs/listener/imei_map.json`; on a match the device name is injected, otherwise the IMEI itself is used as the device name. - The injected keys are `imei` and `device` (configurable via `INJECT_IMEI_KEY` / `INJECT_DEVICE_KEY`). Any existing values in the payload are **overwritten** — the topic is authoritative. - A **device timestamp** is extracted at ingest time from the first parseable payload key in `PAYLOAD_TIME_KEYS` (default `timestamp,time,ts`; ISO 8601 strings, epoch seconds and epoch milliseconds all accepted) and stored in the `payload_time` column. A missing or unparseable timestamp does **not** skip the message — `payload_time` is simply NULL and `received_at` (server receipt time) remains authoritative. -- Messages whose payload is not a JSON object, or whose last topic segment is not a 15-digit IMEI, are skipped (with a warning log). +- The payload can be a **single JSON object or a JSON array of objects**. Each object in an array is injected and stored as its own database row (each with its own `payload_time`); array elements that are not objects are skipped. +- Messages whose payload contains no JSON object at all (invalid JSON, a scalar, an array without objects), or whose last topic segment is not a 15-digit IMEI, are skipped entirely (with a warning log). ## API reference diff --git a/listener/main.py b/listener/main.py index 24b49d7..27a9938 100644 --- a/listener/main.py +++ b/listener/main.py @@ -109,7 +109,7 @@ def main() -> None: topic = message.topic with imei_lock: current_map = imei_map - result = process_message( + results = process_message( topic, bytes(message.payload), current_map, @@ -117,15 +117,18 @@ def main() -> None: device_key=device_key, time_keys=time_keys, ) - if result is None: + if not results: logger.warning( - "skipping message on %r: payload is not a JSON object or last topic " + "skipping message on %r: no JSON object(s) in payload or last topic " "segment is not a 15-digit IMEI", topic, ) return - inserter.insert(topic, result.imei, result.device, result.payload, result.payload_time) - logger.info("stored %s/%s -> %s", result.imei, result.device, topic) + for result in results: + inserter.insert( + topic, result.imei, result.device, result.payload, result.payload_time + ) + logger.info("stored %s/%s -> %s", result.imei, result.device, topic) except Exception: logger.exception("message handling failed for topic %s", getattr(message, "topic", "?")) diff --git a/listener/process.py b/listener/process.py index 2c042c5..2d9c6a3 100644 --- a/listener/process.py +++ b/listener/process.py @@ -97,16 +97,20 @@ def process_message( imei_key: str = "imei", device_key: str = "device", time_keys: tuple[str, ...] = DEFAULT_TIME_KEYS, -) -> InjectionResult | None: +) -> list[InjectionResult]: """Parse a payload and inject IMEI + device into it, extracting the device timestamp. - Returns None when the payload is not valid JSON or not a JSON object — - those messages are skipped (logged by the caller). The original payload's - own imei/device fields are overwritten by the values derived from the - topic, so the topic is always authoritative. + The payload may be a single JSON object or a JSON array of objects: + every object yields one result (the caller stores one database row per + result). Returns an empty list when the payload is not valid JSON, + contains no JSON object at all (a scalar, or an array without objects) + or the last topic segment is not a 15-digit IMEI — those messages are + skipped (logged by the caller). The payload's own imei/device fields + are overwritten by the values derived from the topic, so the topic is + always authoritative. - A missing or unparseable timestamp does NOT skip the message: the + A missing or unparseable timestamp does NOT skip the object: its payload_time is simply None and only received_at (server receipt time) exists for the row. """ @@ -116,29 +120,42 @@ def process_message( normalized = normalize_imei(raw_imei) if normalized is None: - return None + return [] try: - payload = json.loads(payload_bytes.decode("utf-8")) + parsed = json.loads(payload_bytes.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): - return None - if not isinstance(payload, dict): - return None + return [] - payload = dict(payload) # do not mutate shared objects - payload[imei_key] = normalized - payload[device_key] = imei_map.get(normalized, normalized) + if isinstance(parsed, dict): + items: list[dict] = [parsed] + elif isinstance(parsed, list): + # Arrays may mix objects with non-objects; only the objects are + # messages. Nested arrays are NOT flattened. + items = [item for item in parsed if isinstance(item, dict)] + else: + return [] - payload_time = None - for key in time_keys: - if key in payload: - payload_time = parse_payload_time(payload[key]) - if payload_time is not None: - break + device = imei_map.get(normalized, normalized) + results: list[InjectionResult] = [] + for item in items: + payload = dict(item) # do not mutate shared objects + payload[imei_key] = normalized + payload[device_key] = device - return InjectionResult( - imei=normalized, - device=payload[device_key], - payload=payload, - payload_time=payload_time, - ) \ No newline at end of file + payload_time = None + for key in time_keys: + if key in payload: + payload_time = parse_payload_time(payload[key]) + if payload_time is not None: + break + + results.append( + InjectionResult( + imei=normalized, + device=payload[device_key], + payload=payload, + payload_time=payload_time, + ) + ) + return results \ No newline at end of file diff --git a/tools/test_listener.py b/tools/test_listener.py index 42d3e1e..76463f9 100644 --- a/tools/test_listener.py +++ b/tools/test_listener.py @@ -34,8 +34,8 @@ class TestNormalizeImei: ) def test_valid(self, value: str, expected: str) -> None: result = process_message(f"/up/{value}", b"{}", IMEI_MAP) - assert result is not None - assert result.imei == expected + assert len(result) == 1 + assert result[0].imei == expected @pytest.mark.parametrize( "value", @@ -50,53 +50,138 @@ class TestNormalizeImei: ], ) def test_invalid(self, value: str) -> None: - assert process_message(f"/up/{value}", b"{}", IMEI_MAP) is None + assert process_message(f"/up/{value}", b"{}", IMEI_MAP) == [] class TestProcessMessage: def test_injects_imei_and_mapped_device(self) -> None: payload = json.dumps({"temperature": 21.5}).encode() result = process_message("/vivab/350123456789012", payload, IMEI_MAP) - assert result == InjectionResult( - imei="350123456789012", - device="Greenhouse sensor A", - payload={"temperature": 21.5, "imei": "350123456789012", "device": "Greenhouse sensor A"}, - ) + assert result == [ + InjectionResult( + imei="350123456789012", + device="Greenhouse sensor A", + payload={"temperature": 21.5, "imei": "350123456789012", "device": "Greenhouse sensor A"}, + ) + ] def test_unmapped_imei_falls_back_to_imei(self) -> None: payload = json.dumps({"temperature": 21.5}).encode() result = process_message("/vivab/356938035643809", payload, IMEI_MAP) - assert result is not None - assert result.device == "356938035643809" - assert result.payload["device"] == "356938035643809" + assert len(result) == 1 + assert result[0].device == "356938035643809" + assert result[0].payload["device"] == "356938035643809" def test_overwrites_existing_keys(self) -> None: payload = json.dumps({"imei": "fake", "device": "fake", "seq": 1}).encode() result = process_message("/up/490154203237518", payload, IMEI_MAP) - assert result is not None - assert result.payload["imei"] == "490154203237518" - assert result.payload["device"] == "Water tank sensor B" + assert len(result) == 1 + assert result[0].payload["imei"] == "490154203237518" + assert result[0].payload["device"] == "Water tank sensor B" def test_does_not_mutate_caller_payload(self) -> None: original = {"seq": 1} result = process_message("/up/350123456789012", json.dumps(original).encode(), IMEI_MAP) - assert result is not None + assert len(result) == 1 assert original == {"seq": 1} - def test_invalid_json_returns_none(self) -> None: - assert process_message("/up/350123456789012", b"not json", IMEI_MAP) is None + def test_invalid_json_returns_empty(self) -> None: + assert process_message("/up/350123456789012", b"not json", IMEI_MAP) == [] - def test_non_object_json_returns_none(self) -> None: - assert process_message("/up/350123456789012", b"[1,2,3]", IMEI_MAP) is None + def test_scalar_json_returns_empty(self) -> None: + assert process_message("/up/350123456789012", b"\"just a string\"", IMEI_MAP) == [] + + def test_array_without_objects_returns_empty(self) -> None: + assert process_message("/up/350123456789012", b"[1,2,3]", IMEI_MAP) == [] + + def test_empty_array_returns_empty(self) -> None: + assert process_message("/up/350123456789012", b"[]", IMEI_MAP) == [] def test_custom_keys(self) -> None: payload = json.dumps({"seq": 1}).encode() result = process_message( "/up/350123456789012", payload, IMEI_MAP, imei_key="id", device_key="name" ) - assert result is not None - assert result.payload["id"] == "350123456789012" - assert result.payload["name"] == "Greenhouse sensor A" + assert len(result) == 1 + assert result[0].payload["id"] == "350123456789012" + assert result[0].payload["name"] == "Greenhouse sensor A" + + +class TestProcessMessageArray: + """JSON arrays of objects: one InjectionResult (and one DB row) per object.""" + + def test_single_object_array(self) -> None: + payload = json.dumps([{"temperature": 21.5}]).encode() + result = process_message("/up/350123456789012", payload, IMEI_MAP) + assert len(result) == 1 + assert result[0].imei == "350123456789012" + assert result[0].payload == { + "temperature": 21.5, + "imei": "350123456789012", + "device": "Greenhouse sensor A", + } + + def test_array_of_objects(self) -> None: + payload = json.dumps( + [ + {"seq": 1, "temperature": 20.0}, + {"seq": 2, "temperature": 21.0}, + {"seq": 3, "temperature": 22.0}, + ] + ).encode() + results = process_message("/up/350123456789012", payload, IMEI_MAP) + assert len(results) == 3 + assert [r.payload["seq"] for r in results] == [1, 2, 3] + # every array element got its own imei/device injection + assert all(r.imei == "350123456789012" for r in results) + assert all(r.device == "Greenhouse sensor A" for r in results) + assert all(r.payload["device"] == "Greenhouse sensor A" for r in results) + + def test_per_element_payload_time(self) -> None: + payload = json.dumps( + [ + {"timestamp": "2026-08-31T09:59:58Z"}, + {"temperature": 21.5}, # no timestamp -> None + {"timestamp": 1756646096}, + ] + ).encode() + results = process_message("/up/350123456789012", payload, IMEI_MAP) + assert len(results) == 3 + assert results[0].payload_time == datetime(2026, 8, 31, 9, 59, 58, tzinfo=timezone.utc) + assert results[1].payload_time is None # still stored, just without payload_time + assert results[2].payload_time == datetime.fromtimestamp(1756646096, tz=timezone.utc) + + def test_mixed_array_skips_non_objects(self) -> None: + payload = json.dumps([{"seq": 1}, "noise", 42, None, {"seq": 2}]).encode() + results = process_message("/up/350123456789012", payload, IMEI_MAP) + assert [r.payload["seq"] for r in results] == [1, 2] + + def test_nested_arrays_not_flattened(self) -> None: + payload = json.dumps([{"seq": 1}, [{"seq": 2}]]).encode() + results = process_message("/up/350123456789012", payload, IMEI_MAP) + # only top-level objects are messages; the nested array is skipped + assert [r.payload["seq"] for r in results] == [1] + + def test_array_overwrites_existing_keys(self) -> None: + payload = json.dumps([{"imei": "fake", "device": "fake", "seq": 1}]).encode() + results = process_message("/up/490154203237518", payload, IMEI_MAP) + assert len(results) == 1 + assert results[0].payload["imei"] == "490154203237518" + assert results[0].payload["device"] == "Water tank sensor B" + + def test_array_with_unmapped_imei_falls_back_to_imei(self) -> None: + payload = json.dumps([{"seq": 1}, {"seq": 2}]).encode() + results = process_message("/up/356938035643809", payload, IMEI_MAP) + assert len(results) == 2 + assert all(r.device == "356938035643809" for r in results) + + def test_array_does_not_mutate_caller_objects(self) -> None: + original = [{"seq": 1}] + results = process_message( + "/up/350123456789012", json.dumps(original).encode(), IMEI_MAP + ) + assert len(results) == 1 + assert original == [{"seq": 1}] class TestParsePayloadTime: @@ -139,36 +224,36 @@ class TestPayloadTimeExtraction: def test_timestamp_extracted_from_payload(self) -> None: payload = json.dumps({"temperature": 21.5, "timestamp": "2026-08-31T09:59:58Z"}).encode() result = process_message("/up/350123456789012", payload, IMEI_MAP) - assert result is not None - assert result.payload_time == datetime(2026, 8, 31, 9, 59, 58, tzinfo=timezone.utc) + assert len(result) == 1 + assert result[0].payload_time == datetime(2026, 8, 31, 9, 59, 58, tzinfo=timezone.utc) # the original payload value stays untouched - assert result.payload["timestamp"] == "2026-08-31T09:59:58Z" + assert result[0].payload["timestamp"] == "2026-08-31T09:59:58Z" def test_first_parseable_key_wins(self) -> None: payload = json.dumps({"timestamp": "garbage", "time": 1756646096}).encode() result = process_message("/up/350123456789012", payload, IMEI_MAP) - assert result is not None - assert result.payload_time == datetime.fromtimestamp(1756646096, tz=timezone.utc) + assert len(result) == 1 + assert result[0].payload_time == datetime.fromtimestamp(1756646096, tz=timezone.utc) def test_custom_time_keys(self) -> None: payload = json.dumps({"measured_at": "2026-08-31T09:59:58Z"}).encode() result = process_message( "/up/350123456789012", payload, IMEI_MAP, time_keys=("measured_at",) ) - assert result is not None - assert result.payload_time == datetime(2026, 8, 31, 9, 59, 58, tzinfo=timezone.utc) + assert len(result) == 1 + assert result[0].payload_time == datetime(2026, 8, 31, 9, 59, 58, tzinfo=timezone.utc) def test_missing_timestamp_yields_none_not_skip(self) -> None: payload = json.dumps({"temperature": 21.5}).encode() result = process_message("/up/350123456789012", payload, IMEI_MAP) - assert result is not None # message still stored - assert result.payload_time is None + assert len(result) == 1 # message still stored + assert result[0].payload_time is None def test_unparseable_timestamp_yields_none_not_skip(self) -> None: payload = json.dumps({"timestamp": "yesterday-ish"}).encode() result = process_message("/up/350123456789012", payload, IMEI_MAP) - assert result is not None - assert result.payload_time is None + assert len(result) == 1 + assert result[0].payload_time is None class TestLoadImeiMap: