1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
"""IOmeter reading."""
import json
from dataclasses import dataclass
from datetime import datetime
from typing import List
@dataclass
class Register:
"""Represents a meter register reading."""
obis: str
value: float
unit: str
@dataclass
class MeterReading:
"""Represents a point-in-time reading."""
time: datetime
registers: List[Register]
def get_register_by_obis(self, obis: str) -> Register | None:
"""Get register by OBIS code."""
return next((reg for reg in self.registers if reg.obis == obis), None)
@dataclass
class Meter:
"""Represents the meter device."""
number: str
reading: MeterReading
@dataclass
class Reading:
"""Top level class representing a complete meter reading."""
meter: Meter
typename: str = "iometer.reading.v1"
# OBIS code constants
TOTAL_CONSUMPTION_OBIS = "01-00:01.08.00*ff"
TOTAL_PRODUCTION_OBIS = "01-00:02.08.00*ff"
CURRENT_POWER_OBIS = "01-00:10.07.00*ff"
CURRENT_POWER_OBIS_ALT = "01-00:24.07.00*ff"
@classmethod
def from_json(cls, json_str: str) -> "Reading":
"""Create Reading instance from JSON string."""
data = json.loads(json_str)
# Create registers
registers = [
Register(obis=reg["obis"], value=reg["value"], unit=reg["unit"])
for reg in data["meter"]["reading"]["registers"]
]
# Create meter reading
meter_reading = MeterReading(
time=datetime.fromisoformat(
data["meter"]["reading"]["time"].replace("Z", "+00:00")
),
registers=registers,
)
# Create meter
meter = Meter(number=data["meter"]["number"], reading=meter_reading)
# Create reading
return cls(meter=meter)
def to_json(self) -> str:
"""Convert the status to JSON string"""
return json.dumps(
{
"__typename": self.typename,
"meter": {
"number": self.meter.number,
"reading": {
"time": self.meter.reading.time.strftime("%Y-%m-%dT%H:%M:%SZ"),
"registers": [
{
"obis": register.obis,
"value": register.value,
"unit": register.unit,
}
for register in self.meter.reading.registers
],
},
},
}
)
def get_total_consumption(self) -> float | None:
"""Get total consumption in Wh."""
register = self.meter.reading.get_register_by_obis(self.TOTAL_CONSUMPTION_OBIS)
return register.value if register else None
def get_total_production(self) -> float | None:
"""Get total production in Wh.
Returns None if OBIS is not found, otherwise the value in Wh as float.
"""
register = self.meter.reading.get_register_by_obis(self.TOTAL_PRODUCTION_OBIS)
return register.value if register else None
def get_current_power(self) -> float | None:
"""Get current power consumption in W.
Prefer 10.07.00 OBIS, it is missing in some meters.
These meters falsely report 24.07.00 OBIS as current power.
Returns None if neither OBIS is found, otherwise the value in W as float.
"""
register = self.meter.reading.get_register_by_obis(self.CURRENT_POWER_OBIS)
if register:
return register.value
register_alt = self.meter.reading.get_register_by_obis(
self.CURRENT_POWER_OBIS_ALT
)
return register_alt.value if register_alt else None
def __str__(self) -> str:
return self.to_json()
|
