126 lines
4.8 KiB
Python
126 lines
4.8 KiB
Python
"""Tests for hameter.meter."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from hameter.config import MeterConfig
|
|
from hameter.meter import parse_rtlamr_line
|
|
|
|
FIXTURES_DIR = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_lines() -> list[str]:
|
|
"""Load sample rtlamr JSON lines from the fixture file."""
|
|
path = FIXTURES_DIR / "sample_rtlamr_output.jsonl"
|
|
return [line.strip() for line in path.read_text().splitlines() if line.strip()]
|
|
|
|
|
|
@pytest.fixture
|
|
def electric_meter() -> MeterConfig:
|
|
return MeterConfig(
|
|
id=23040293,
|
|
protocol="scm",
|
|
name="Electric Meter",
|
|
unit_of_measurement="kWh",
|
|
device_class="energy",
|
|
multiplier=0.1156,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def meters_by_id(electric_meter) -> dict[int, MeterConfig]:
|
|
return {electric_meter.id: electric_meter}
|
|
|
|
|
|
class TestParseRtlamrLine:
|
|
def test_scm_matching_meter(self, sample_lines, meters_by_id):
|
|
"""SCM line for configured meter should parse and apply multiplier."""
|
|
# First line is SCM for meter 23040293 with Consumption=516012
|
|
reading = parse_rtlamr_line(sample_lines[0], meters_by_id)
|
|
|
|
assert reading is not None
|
|
assert reading.meter_id == 23040293
|
|
assert reading.protocol == "SCM"
|
|
assert reading.raw_consumption == 516012
|
|
assert reading.calibrated_consumption == round(516012 * 0.1156, 4)
|
|
|
|
def test_scm_second_reading(self, sample_lines, meters_by_id):
|
|
"""Second SCM line should show updated consumption."""
|
|
reading = parse_rtlamr_line(sample_lines[1], meters_by_id)
|
|
assert reading is not None
|
|
assert reading.raw_consumption == 516030
|
|
assert reading.calibrated_consumption == round(516030 * 0.1156, 4)
|
|
|
|
def test_idm_matching_meter(self, sample_lines, meters_by_id):
|
|
"""IDM line for same meter should parse using ERTSerialNumber."""
|
|
reading = parse_rtlamr_line(sample_lines[2], meters_by_id)
|
|
assert reading is not None
|
|
assert reading.meter_id == 23040293
|
|
assert reading.protocol == "IDM"
|
|
assert reading.raw_consumption == 2124513
|
|
|
|
def test_scm_plus_unconfigured_meter(self, sample_lines, meters_by_id):
|
|
"""SCM+ line for a meter NOT in our config should return None."""
|
|
reading = parse_rtlamr_line(sample_lines[3], meters_by_id)
|
|
assert reading is None
|
|
|
|
def test_r900_unconfigured_meter(self, sample_lines, meters_by_id):
|
|
"""R900 line for a meter NOT in our config should return None."""
|
|
reading = parse_rtlamr_line(sample_lines[4], meters_by_id)
|
|
assert reading is None
|
|
|
|
def test_scm_unknown_meter_filtered(self, sample_lines, meters_by_id):
|
|
"""SCM line for meter 99999999 should be filtered out."""
|
|
reading = parse_rtlamr_line(sample_lines[5], meters_by_id)
|
|
assert reading is None
|
|
|
|
def test_discovery_mode_accepts_all(self, sample_lines):
|
|
"""With empty meters dict (discovery mode), all lines should parse."""
|
|
results = []
|
|
for line in sample_lines:
|
|
r = parse_rtlamr_line(line, meters={})
|
|
if r:
|
|
results.append(r)
|
|
|
|
# All 6 lines should produce readings in discovery mode.
|
|
assert len(results) == 6
|
|
|
|
# Check different protocols parsed correctly.
|
|
protocols = {r.protocol for r in results}
|
|
assert protocols == {"SCM", "IDM", "SCM+", "R900"}
|
|
|
|
def test_multiplier_default_in_discovery(self, sample_lines):
|
|
"""In discovery mode, multiplier should default to 1.0."""
|
|
reading = parse_rtlamr_line(sample_lines[0], meters={})
|
|
assert reading is not None
|
|
# raw == calibrated when multiplier is 1.0
|
|
assert reading.calibrated_consumption == float(reading.raw_consumption)
|
|
|
|
def test_invalid_json_raises(self):
|
|
"""Non-JSON input should raise JSONDecodeError."""
|
|
with pytest.raises(json.JSONDecodeError):
|
|
parse_rtlamr_line("not json", meters={})
|
|
|
|
def test_unknown_type_returns_none(self):
|
|
"""Unknown message Type should return None."""
|
|
line = json.dumps({"Type": "UNKNOWN", "Message": {"ID": 1, "Consumption": 100}})
|
|
assert parse_rtlamr_line(line, meters={}) is None
|
|
|
|
def test_timestamp_from_rtlamr(self, sample_lines, meters_by_id):
|
|
"""Timestamp should come from the rtlamr Time field."""
|
|
reading = parse_rtlamr_line(sample_lines[0], meters_by_id)
|
|
assert reading is not None
|
|
assert reading.timestamp == "2026-03-01T15:18:58Z"
|
|
|
|
def test_r900_fields(self, sample_lines):
|
|
"""R900 reading should extract ID and Consumption correctly."""
|
|
# Line index 4 is R900
|
|
reading = parse_rtlamr_line(sample_lines[4], meters={})
|
|
assert reading is not None
|
|
assert reading.meter_id == 55512345
|
|
assert reading.raw_consumption == 12345678
|
|
assert reading.protocol == "R900"
|