feat(measurement): integrate typed energy quality and capacity APIs

Port the locally backed-up measurement extensions to main async storage and PR #1256 device maps. Preserve runtime capacity estimates across #1305 bulk updates. Confirm JSON singleton restore defect on unchanged main and add regression. No production configuration or measurements included.

Co-authored-by: Andreas <drbacke@gmx.de>
This commit is contained in:
Andreas
2026-09-16 12:24:48 +02:00
parent d9c434f5d4
commit 7338baff56
13 changed files with 1789 additions and 10 deletions
@@ -11,6 +11,7 @@ from numpydantic import NDArray, Shape
from pydantic import Field, computed_field, field_validator, model_validator
from akkudoktoreos.devices.settings.devicebasesettings import DevicesBaseSettings
from akkudoktoreos.measurement.batterycapacity import BatteryCapacityEstimate, BatteryCapacityEstimationSettings
if TYPE_CHECKING:
from akkudoktoreos.devices.genetic0.genetic0battery import (
@@ -38,6 +39,12 @@ class BatteriesCommonSettings(DevicesBaseSettings):
Note: Used for the GENETIC and GENETIC0 algorithm.
"""
capacity_estimation: Optional[BatteryCapacityEstimationSettings] = None
capacity_estimate: Optional[BatteryCapacityEstimate] = Field(
default=None,
description="Last explicitly stored capacity estimate; does not override capacity_wh.",
)
capacity_wh: int = Field(
default=8000,
gt=0,
@@ -0,0 +1,207 @@
"""Estimate effective model capacity from independent SoC anchors and DC power."""
from datetime import datetime
from math import isfinite
from typing import TYPE_CHECKING, Literal
from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, model_validator
from akkudoktoreos.config.configabc import SettingsBaseModel
from akkudoktoreos.measurement.quality import SampleQuality
if TYPE_CHECKING:
from akkudoktoreos.measurement.measurement import MeasurementChannelSettings
class BatteryCapacityEstimationSettings(SettingsBaseModel):
"""Only a battery-terminal DC power channel is supported, not inverter AC."""
power_key: str = Field(min_length=1)
positive_power: Literal["charging", "discharging"]
measurement_boundary: Literal["battery_dc"] = "battery_dc"
min_soc_change_percentage: float = Field(default=20, gt=0, le=100)
max_duration_hours: float = Field(default=168, gt=0, le=744)
class CapacityModel(BaseModel):
model_config = ConfigDict(extra="forbid", allow_inf_nan=False)
class BatteryCapacityRequest(CapacityModel):
"""Caller attests both anchors are independent of the capacity being fitted.
End is the first confirmed full point when using the default end SoC.
Voltage/current anchors must be independently established for the chemistry.
"""
start: AwareDatetime
end: AwareDatetime
start_soc_percentage: float = Field(ge=0, le=100)
end_soc_percentage: float = Field(default=100, ge=0, le=100)
soc_reference: Literal["bms", "external_calibration", "voltage_current_anchor"]
store_estimate: bool = False
@model_validator(mode="after")
def ordered(self):
if self.end.timestamp() <= self.start.timestamp():
raise ValueError("end must be after start.")
return self
class BatteryCapacityEstimate(CapacityModel):
battery_id: str
start: AwareDatetime
end: AwareDatetime
soc_reference: str
start_soc_percentage: float
end_soc_percentage: float
estimated_capacity_wh: float
configured_capacity_wh: float
capacity_change_percentage: float
model_end_soc_percentage_unclipped: float
model_soc_error_percentage_points: float
charge_energy_wh: float
discharge_energy_wh: float
stored_energy_change_wh: float
charging_efficiency: float
discharging_efficiency: float
power_key: str
positive_power: str
integration_method: str
coverage_seconds: float
samples_used: int
warnings: list[str]
def estimate_capacity(
request: BatteryCapacityRequest,
settings: BatteryCapacityEstimationSettings,
channel: "MeasurementChannelSettings",
samples: list[tuple[datetime, float | None, SampleQuality]],
*,
battery_id: str,
capacity_wh: float,
charging_efficiency: float,
discharging_efficiency: float,
) -> BatteryCapacityEstimate:
"""Fit C in C * delta_soc = eta_c * E_charge - E_discharge / eta_d.
No clipping at 0/100%, extrapolation, gap filling or efficiency fitting.
Linear segments are split at zero before applying directional efficiencies.
"""
left, right = request.start.timestamp(), request.end.timestamp()
if right - left > settings.max_duration_hours * 3600:
raise ValueError("Requested period exceeds capacity_estimation.max_duration_hours.")
delta_soc = (request.end_soc_percentage - request.start_soc_percentage) / 100
if abs(delta_soc) * 100 + 1e-9 < settings.min_soc_change_percentage:
raise ValueError(
"Independent SoC change is too small; full-to-full cannot identify capacity."
)
if channel.quantity != "power":
raise ValueError("Capacity estimation requires signed battery DC power in W or kW.")
if not isfinite(capacity_wh) or capacity_wh <= 0:
raise ValueError("Configured capacity must be positive and finite.")
if not all(isfinite(e) and 0 < e <= 1 for e in (charging_efficiency, discharging_efficiency)):
raise ValueError("Battery efficiencies must be finite and in (0, 1].")
points = []
for time, value, quality in samples:
if time.tzinfo is None or time.utcoffset() is None:
raise ValueError("Explicit sample timezone required.")
points.append((time.timestamp(), value, quality))
points.sort(key=lambda row: row[0])
if any(a[0] == b[0] for a, b in zip(points, points[1:])):
raise ValueError("Duplicate power sample timestamps.")
factor = (1000 if channel.unit == "kW" else 1) * (
1 if settings.positive_power == "charging" else -1
)
charge = discharge = coverage = 0.0
cumulative_net = 0.0
minimum_net = maximum_net = 0.0
used = set()
cursor = left
for (a, va, qa), (b, vb, qb) in zip(points, points[1:]):
lo, hi = max(left, a), min(right, b)
if hi <= lo:
continue
if lo > cursor + 1e-6 or b - a > channel.max_gap_seconds:
raise ValueError("Incomplete power coverage or sample gap exceeds max_gap_seconds.")
if (
any(v is None or isinstance(v, bool) or not isfinite(v) for v in (va, vb))
or qa.status != "measured"
or qb.status != "measured"
):
raise ValueError(
"Capacity estimation requires finite measured power samples throughout."
)
if qb.reset or qa.generation != qb.generation:
raise ValueError("Power sensor reset or generation change within the period.")
p, q = va * factor, vb * factor
if channel.integration_method == "linear":
slope = (q - p) / (b - a)
p, q = p + slope * (lo - a), p + slope * (hi - a)
else:
q = p
duration = hi - lo
if p * q < 0:
first = duration * abs(p) / (abs(p) + abs(q))
parts = [(p * first / 7200), (q * (duration - first) / 7200)]
else:
parts = [(p + q) * duration / 7200]
charge += sum(max(0, energy) for energy in parts)
discharge += sum(max(0, -energy) for energy in parts)
for energy in parts:
cumulative_net += (
energy * charging_efficiency if energy >= 0 else energy / discharging_efficiency
)
minimum_net = min(minimum_net, cumulative_net)
maximum_net = max(maximum_net, cumulative_net)
coverage += duration
cursor = hi
used.update((a, b))
if cursor < right - 1e-6 or abs(coverage - (right - left)) > 1e-6:
raise ValueError("Incomplete power coverage; no extrapolation to the SoC anchors.")
net = charge * charging_efficiency - discharge / discharging_efficiency
capacity = net / delta_soc
if not isfinite(capacity) or capacity <= 0:
raise ValueError("Energy flow disagrees with SoC change; check polarity and anchors.")
if (
request.start_soc_percentage + minimum_net / capacity * 100 < -1e-6
or request.start_soc_percentage + maximum_net / capacity * 100 > 100 + 1e-6
):
raise ValueError(
"Fitted SoC leaves 0..100% within the period; check anchors and use the first full point."
)
model_end = request.start_soc_percentage + net / capacity_wh * 100
warnings = [
"Conditional estimate: SoC anchors, DC measurement boundary and configured efficiencies must be correct.",
"A single interval cannot independently identify both capacity and efficiencies.",
"The active capacity_wh is unchanged.",
]
if request.soc_reference == "voltage_current_anchor":
warnings.append(
"Voltage/current anchors depend on chemistry, temperature and operating conditions."
)
return BatteryCapacityEstimate(
battery_id=battery_id,
start=request.start,
end=request.end,
soc_reference=request.soc_reference,
start_soc_percentage=request.start_soc_percentage,
end_soc_percentage=request.end_soc_percentage,
estimated_capacity_wh=capacity,
configured_capacity_wh=capacity_wh,
capacity_change_percentage=(capacity / capacity_wh - 1) * 100,
model_end_soc_percentage_unclipped=model_end,
model_soc_error_percentage_points=model_end - request.end_soc_percentage,
charge_energy_wh=charge,
discharge_energy_wh=discharge,
stored_energy_change_wh=net,
charging_efficiency=charging_efficiency,
discharging_efficiency=discharging_efficiency,
power_key=settings.power_key,
positive_power=settings.positive_power,
integration_method=channel.integration_method,
coverage_seconds=coverage,
samples_used=len(used),
warnings=warnings,
)
+190
View File
@@ -0,0 +1,190 @@
"""Quality-aware energy conversion of raw measurement samples.
No extrapolation beyond the last sample. Intervals use elapsed UTC seconds.
"""
from dataclasses import dataclass
from datetime import datetime, timezone
from math import isfinite
from typing import TYPE_CHECKING, Iterable, Literal
from akkudoktoreos.measurement.quality import SampleQuality
if TYPE_CHECKING:
from akkudoktoreos.measurement.measurement import MeasurementChannelSettings
@dataclass(frozen=True)
class EnergyInterval:
"""Energy and actual temporal support; missing energy is never zero-filled."""
start: datetime
end: datetime
energy_wh: float | None
observed_energy_wh: float | None
coverage_seconds: float
coverage_status: Literal["complete", "partial", "missing", "invalid"]
methods: tuple[str, ...]
flags: tuple[str, ...]
# Retain the support for subsequent multi-channel balance intersection.
coverage_ranges: tuple[tuple[datetime, datetime], ...]
def _timestamp(value: datetime) -> float:
if value.tzinfo is None or value.utcoffset() is None:
raise ValueError("Explicit timezone required.")
return value.timestamp()
def _date(value: float) -> datetime:
return datetime.fromtimestamp(value, timezone.utc)
def energy_intervals(
samples: Iterable[tuple[datetime, float | None]],
channel: "MeasurementChannelSettings",
start: datetime,
end: datetime,
interval_seconds: int = 900,
quality: dict[float, SampleQuality] | None = None,
) -> list[EnergyInterval]:
"""Convert finite samples, preserving gaps, resets and partial coverage.
Hold applies only between consecutive observations within max_gap_seconds.
Null/invalid observations break continuity. Meter resets invalidate that segment.
Fixed interval energy is uniformly allocated when a target cuts its source interval.
"""
left, right = _timestamp(start), _timestamp(end)
if right <= left or type(interval_seconds) is not int or interval_seconds <= 0:
raise ValueError("Require end > start and positive integer interval_seconds.")
points = sorted(((_timestamp(t), v) for t, v in samples), key=lambda point: point[0])
if any(a[0] == b[0] for a, b in zip(points, points[1:])):
raise ValueError("Duplicate sample timestamps must be resolved before conversion.")
scale = 1000 if channel.unit in ("kW", "kWh") else 1
quality = quality or {}
def sample_quality(time: float) -> SampleQuality:
return quality.get(time, SampleQuality())
def number(value: float | None) -> float | None:
if isinstance(value, bool) or not isinstance(value, (int, float)) or not isfinite(value):
return None
return float(value) * scale
# Segment: start, end, start/end power in W, method, error flag.
segments = []
if channel.quantity == "interval_energy":
duration = channel.interval_seconds
if duration is None:
raise ValueError("Interval duration is required.")
previous_end = None
for time, raw in points:
a = time if channel.timestamp_reference == "start" else time - duration
b = a + duration
if previous_end is not None and a < previous_end:
raise ValueError("Overlapping source energy intervals.")
previous_end = b
value = number(raw)
if sample_quality(time).status in ("invalid", "unavailable"):
value = None
power = value * 3600 / duration if value is not None else None
segments.append(
(a, b, power, power, "interval_energy", "invalid_sample" if power is None else None)
)
else:
for (a, raw_a), (b, raw_b) in zip(points, points[1:]):
va, vb = number(raw_a), number(raw_b)
qa, qb = sample_quality(a), sample_quality(b)
if qa.status in ("invalid", "unavailable"):
va = None
if qb.status in ("invalid", "unavailable"):
vb = None
flag = None
if channel.max_gap_seconds is not None and b - a > channel.max_gap_seconds:
flag = "gap_too_large"
elif va is None or (
vb is None
and (
channel.quantity == "cumulative_energy"
or channel.integration_method == "linear"
)
):
flag = "invalid_sample"
if channel.quantity == "cumulative_energy":
method = "meter_difference"
if va is not None and vb is not None and vb < va:
flag = "meter_reset"
if qb.reset or qa.generation != qb.generation:
flag = "meter_reset"
pa = pb = (
(vb - va) * 3600 / (b - a)
if flag is None and va is not None and vb is not None
else None
)
else:
method = "integrated_power"
pa = va
pb = vb if channel.integration_method == "linear" else va
segments.append((a, b, pa, pb, method, flag))
result = []
slot = left
segment_index = 0
while slot < right:
stop = min(slot + interval_seconds, right)
total, coverage = 0.0, 0.0
methods: set[str] = set()
flags: set[str] = set()
ranges: list[tuple[datetime, datetime]] = []
while segment_index < len(segments) and segments[segment_index][1] <= slot:
segment_index += 1
for index in range(segment_index, len(segments)):
a, b, pa, pb, method, flag = segments[index]
if a >= stop:
break
lo, hi = max(a, slot), min(b, stop)
if hi <= lo:
continue
# Quality follows the endpoints used by the integration rule.
source_time = a
if channel.quantity == "interval_energy" and channel.timestamp_reference == "end":
source_time = b
statuses = {sample_quality(source_time).status}
if channel.quantity == "cumulative_energy" or channel.integration_method == "linear":
statuses.add(sample_quality(b).status)
flags.update(s for s in statuses if s != "measured")
if flag is not None:
flags.add(flag)
continue
if pa is None or pb is None:
raise ValueError("Valid segment requires finite endpoint powers.")
p_lo = pa + (pb - pa) * (lo - a) / (b - a)
p_hi = pa + (pb - pa) * (hi - a) / (b - a)
total += (p_lo + p_hi) / 2 * (hi - lo) / 3600
coverage += hi - lo
if ranges and ranges[-1][1] == _date(lo):
ranges[-1] = (ranges[-1][0], _date(hi))
else:
ranges.append((_date(lo), _date(hi)))
methods.add(method)
if method in ("meter_difference", "interval_energy") and (lo != a or hi != b):
methods.add("allocated_energy")
complete = abs(coverage - (stop - slot)) < 1e-6
status: Literal["complete", "partial", "missing", "invalid"] = (
"complete" if complete else "partial" if coverage else "invalid" if flags else "missing"
)
result.append(
EnergyInterval(
_date(slot),
_date(stop),
total if complete else None,
total if coverage else None,
coverage,
status,
tuple(sorted(methods)),
tuple(sorted(flags)),
tuple(ranges),
)
)
slot = stop
return result
+118
View File
@@ -0,0 +1,118 @@
"""AC household balances on the intersection of actual measurement support."""
from datetime import datetime
from typing import Callable, Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
from akkudoktoreos.measurement.energy import EnergyInterval
class HouseholdInput(BaseModel):
"""A non-overlapping AC branch; polarity normalizes the sensor's sign."""
model_config = ConfigDict(extra="forbid")
key: str
branch: str = Field(min_length=1)
role: Literal["site", "grid", "pv", "battery", "inverter", "ev", "device"]
polarity: Literal[-1, 1] = 1
class HouseholdSettings(BaseModel):
"""Fixed topology, never user-supplied executable balance expressions.
Grid import, PV production and battery/inverter discharge are positive.
EV/device inputs are positive consumption, subtracted only from derived loads.
"""
model_config = ConfigDict(extra="forbid")
topology: Literal["direct", "separate_ac", "hybrid_ac"]
inputs: list[HouseholdInput] = Field(min_length=1)
@model_validator(mode="after")
def validate_topology(self) -> "HouseholdSettings":
keys = [item.key for item in self.inputs]
branches = [item.branch for item in self.inputs]
if len(keys) != len(set(keys)) or len(branches) != len(set(branches)):
raise ValueError("Duplicate measurement key or physical branch in household balance.")
roles = {item.role for item in self.inputs} - {"ev", "device"}
allowed = {
"direct": {"site"},
"separate_ac": {"grid", "pv", "battery"},
# Additional AC PV branches are independent of the hybrid's net output.
# Its DC PV/battery must not be added again as separate inputs.
"hybrid_ac": {"grid", "inverter", "pv"},
}[self.topology]
required = {"site"} if self.topology == "direct" else {"grid"}
if self.topology == "hybrid_ac":
required.add("inverter")
if not required <= roles or not roles <= allowed:
raise ValueError("Inputs do not match the selected AC topology.")
return self
def household_intervals(
settings: HouseholdSettings,
convert: Callable[[str, datetime, datetime, int], list[EnergyInterval]],
start: datetime,
end: datetime,
interval_seconds: int = 900,
) -> dict[str, list[EnergyInterval]]:
"""Integrate each source again over shared support, never prorate partial sums."""
series = {item.key: convert(item.key, start, end, interval_seconds) for item in settings.inputs}
output: dict[str, list[EnergyInterval]] = {}
for name, excluded in (
("site", {"ev", "device"}),
("household", {"device"}),
("base", set()),
):
inputs = [item for item in settings.inputs if item.role not in excluded]
result = []
for index, template in enumerate(series[inputs[0].key]):
rows = [series[item.key][index] for item in inputs]
# Partition at every coverage boundary; keep only the intersection.
boundaries = sorted({t for row in rows for pair in row.coverage_ranges for t in pair})
support = [
(a, b)
for a, b in zip(boundaries, boundaries[1:])
if all(any(lo <= a and b <= hi for lo, hi in row.coverage_ranges) for row in rows)
]
total = 0.0
methods = {method for row in rows for method in row.methods}
flags = {flag for row in rows for flag in row.flags}
for a, b in support:
for item in inputs:
# The range is at most one target interval, but may be fractional seconds.
parts = convert(item.key, a, b, interval_seconds)
value = parts[0].energy_wh
if value is None:
raise ValueError("Inconsistent measurement support during balance.")
sign = -1 if item.role in ("ev", "device") else 1
total += sign * item.polarity * value
methods.update(parts[0].methods)
flags.update(parts[0].flags)
coverage = sum((b - a).total_seconds() for a, b in support)
complete = abs(coverage - (template.end - template.start).total_seconds()) < 1e-6
if coverage and total < -1e-6:
flags.add("negative_balance")
result.append(
EnergyInterval(
template.start,
template.end,
total if complete else None,
total if coverage else None,
coverage,
"complete"
if complete
else "partial"
if coverage
else "invalid"
if flags
else "missing",
tuple(sorted(methods | {"ac_balance"})),
tuple(sorted(flags)),
tuple(support),
)
)
output[name] = result
return output
+273 -10
View File
@@ -6,17 +6,29 @@ data records for measurements.
The measurements can be added programmatically or imported from a file or JSON string.
"""
import json
from bisect import bisect_left, bisect_right
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Optional
from typing import Any, Callable, Literal, Optional
import numpy as np
from loguru import logger
from numpydantic import NDArray, Shape
from pydantic import Field, computed_field
from pydantic import Field, computed_field, model_validator
from akkudoktoreos.config.configabc import SettingsBaseModel
from akkudoktoreos.core.coreabc import SingletonMixin
from akkudoktoreos.core.dataabc import DataImportMixin, DataRecord, DataSequence
from akkudoktoreos.core.databaseabc import DatabaseTimestamp
from akkudoktoreos.measurement.energy import EnergyInterval
from akkudoktoreos.measurement.batterycapacity import (
BatteryCapacityEstimate,
BatteryCapacityRequest,
estimate_capacity,
)
from akkudoktoreos.measurement.household import HouseholdSettings, household_intervals
from akkudoktoreos.measurement.quality import MeasurementSample, SampleQuality
from akkudoktoreos.utils.datetimeutil import (
DateTime,
Duration,
@@ -25,6 +37,41 @@ from akkudoktoreos.utils.datetimeutil import (
)
class MeasurementChannelSettings(SettingsBaseModel):
"""Meaning of a raw measurement channel; no conversion is performed on storage."""
quantity: Literal["power", "cumulative_energy", "interval_energy"]
unit: Literal["W", "kW", "Wh", "kWh"]
integration_method: Optional[Literal["hold", "linear"]] = None
max_gap_seconds: Optional[int] = Field(default=None, gt=0, strict=True)
interval_seconds: Optional[int] = Field(default=None, gt=0, strict=True)
timestamp_reference: Optional[Literal["start", "end"]] = None
@model_validator(mode="after")
def validate_semantics(self) -> "MeasurementChannelSettings":
"""Reject ambiguous units and time semantics before accepting a channel."""
if self.quantity == "power":
if self.unit not in ("W", "kW"):
raise ValueError("Power channels require W or kW.")
if self.integration_method is None or self.max_gap_seconds is None:
raise ValueError("Power channels require integration_method and max_gap_seconds.")
else:
if self.unit not in ("Wh", "kWh"):
raise ValueError("Energy channels require Wh or kWh.")
if self.integration_method is not None:
raise ValueError("integration_method is only applicable to power channels.")
if self.quantity == "interval_energy":
if self.interval_seconds is None or self.timestamp_reference is None:
raise ValueError(
"Interval energy requires interval_seconds and timestamp_reference."
)
if self.max_gap_seconds is not None:
raise ValueError("Interval energy uses explicit intervals, not max_gap_seconds.")
elif self.interval_seconds is not None or self.timestamp_reference is not None:
raise ValueError("Interval metadata is only applicable to interval_energy channels.")
return self
class MeasurementCommonSettings(SettingsBaseModel):
"""Measurement Configuration."""
@@ -37,6 +84,41 @@ class MeasurementCommonSettings(SettingsBaseModel):
},
)
channels: dict[str, MeasurementChannelSettings] = Field(
default_factory=dict,
json_schema_extra={
"description": "Typed raw measurement channels keyed by measurement key."
},
)
household: Optional[HouseholdSettings] = None
energy_context_seconds: int = Field(default=86400, gt=0, le=604800, strict=True)
@model_validator(mode="after")
def validate_channels(self) -> "MeasurementCommonSettings":
"""Preserve the kWh meter contract of legacy keys and avoid record collisions."""
for key, channel in self.channels.items():
if (
not key
or key != key.strip()
or key.startswith("_")
or hasattr(DataRecord, key)
or key in DataRecord.model_fields
or key == "sample_quality"
):
raise ValueError(f"Invalid or reserved measurement channel key: {key!r}")
for name in type(self).model_fields:
if name.endswith("_emr_keys") and key in (getattr(self, name) or []):
if channel.quantity != "cumulative_energy" or channel.unit != "kWh":
raise ValueError(
f"Legacy meter key {key!r} must remain cumulative_energy in kWh."
)
if self.household is not None:
for item in self.household.inputs:
if item.key not in self.keys:
raise ValueError(f"Unknown household measurement key: {item.key!r}")
return self
load_emr_keys: Optional[list[str]] = Field(
default=None,
json_schema_extra={
@@ -74,7 +156,7 @@ class MeasurementCommonSettings(SettingsBaseModel):
@property
def keys(self) -> list[str]:
"""The keys of the measurements that can be stored."""
key_list = []
key_list = list(self.channels)
for key in self.__class__.model_fields.keys():
if key.endswith("_keys") and (value := getattr(self, key)):
key_list.extend(value)
@@ -84,6 +166,18 @@ class MeasurementCommonSettings(SettingsBaseModel):
class MeasurementDataRecord(DataRecord):
"""Represents a measurement data record containing various measurements at a specific datetime."""
sample_quality: dict[str, SampleQuality] = Field(default_factory=dict)
@classmethod
def record_keys(cls) -> list[str]:
"""Quality is stored alongside values, not exposed as a numeric channel."""
return [key for key in super().record_keys() if key != "sample_quality"]
@classmethod
def record_keys_writable(cls) -> list[str]:
"""Only the typed sample path writes quality, never numeric import paths."""
return [key for key in super().record_keys_writable() if key != "sample_quality"]
@classmethod
def configured_data_keys(cls) -> Optional[list[str]]:
"""Return the keys for the configured field like data."""
@@ -149,6 +243,176 @@ class Measurement(SingletonMixin, DataImportMixin, DataSequence[MeasurementDataR
# Return ceiling of division to include partial intervals
return int(np.ceil(diff_seconds / interval_seconds))
async def import_samples(self, samples: list[MeasurementSample]) -> None:
"""Validate the whole batch before replacing samples; omitted keys stay untouched."""
for sample in samples:
self._energy_channel(sample.key)
for sample in samples:
dt = to_datetime(sample.date_time)
await self.update_value(dt, sample.key, sample.value)
record = await self.db_get_record(DatabaseTimestamp.from_datetime(dt))
if not isinstance(record, MeasurementDataRecord):
raise RuntimeError("Measurement sample was not stored.")
record.sample_quality[sample.key] = sample.quality.model_copy(deep=True)
await self.db_mark_dirty_record(record)
async def insert_by_datetime(self, record: DataRecord) -> None:
"""Merge quality by channel as well as the ordinary measurement fields."""
await super().insert_by_datetime(record)
if (
isinstance(record, MeasurementDataRecord)
and record.sample_quality
and record.date_time is not None
):
stored = await self.db_get_record(DatabaseTimestamp.from_datetime(record.date_time))
if not isinstance(stored, MeasurementDataRecord):
raise RuntimeError("Measurement record was not stored.")
stored.sample_quality = stored.sample_quality | record.sample_quality
await self.db_mark_dirty_record(stored)
def _energy_channel(self, key: str) -> MeasurementChannelSettings:
channel = self.config.measurement.channels.get(key)
if channel is not None:
return channel
for name in type(self.config.measurement).model_fields:
if name.endswith("_emr_keys") and key in (getattr(self.config.measurement, name) or []):
return MeasurementChannelSettings(quantity="cumulative_energy", unit="kWh")
raise ValueError(f"No energy channel definition for {key!r}.")
async def _energy_converter(
self,
keys: list[str],
start: datetime,
end: datetime,
) -> Callable[[str, datetime, datetime, int], list[EnergyInterval]]:
"""Load a bounded window once, including configured boundary context."""
from akkudoktoreos.measurement.energy import energy_intervals
if any(dt.tzinfo is None or dt.utcoffset() is None for dt in (start, end)):
raise ValueError("Explicit timezone required.")
seconds = end.timestamp() - start.timestamp()
if not 0 < seconds <= 31 * 86400:
raise ValueError("Energy queries require a positive range of at most 31 days.")
channels = {key: self._energy_channel(key) for key in keys}
context = self.config.measurement.energy_context_seconds
records = [record async for record in self.db_iterate_records(
DatabaseTimestamp.from_datetime(to_datetime(start) - timedelta(seconds=context)),
DatabaseTimestamp.from_datetime(
to_datetime(end) + timedelta(seconds=context, microseconds=1)
),
)
]
samples = {
key: [
(record.date_time, record.configured_data[key])
for record in records
if key in record.configured_data and record.date_time is not None
]
for key in keys
}
quality = {
key: {
record.date_time.timestamp(): record.sample_quality[key]
for record in records
if key in record.sample_quality and record.date_time is not None
}
for key in keys
}
timestamps = {key: [dt.timestamp() for dt, _ in values] for key, values in samples.items()}
for key, channel in channels.items():
duration = channel.interval_seconds
if (
channel.quantity == "interval_energy"
and duration is not None
and any(b - a < duration for a, b in zip(timestamps[key], timestamps[key][1:]))
):
raise ValueError(f"Overlapping source energy intervals for {key!r}.")
def convert(
key: str, left: datetime, right: datetime, interval_seconds: int
) -> list[EnergyInterval]:
if type(interval_seconds) is not int or interval_seconds <= 0:
raise ValueError("Positive integer interval_seconds required.")
if (right.timestamp() - left.timestamp()) / interval_seconds > 10000:
raise ValueError("At most 10000 output intervals per query.")
lo = max(0, bisect_left(timestamps[key], left.timestamp()) - 1)
hi = bisect_right(timestamps[key], right.timestamp()) + 1
return energy_intervals(
samples[key][lo:hi], channels[key], left, right, interval_seconds, quality[key]
)
return convert
async def household_intervals(
self, start_datetime: datetime, end_datetime: datetime, interval_seconds: int = 900
) -> dict[str, list[EnergyInterval]]:
"""Return site, household without EV, and base without configured devices."""
settings = self.config.measurement.household
if settings is None:
raise ValueError("No household balance configured.")
# Revalidate references even after an in-place configuration mutation.
settings = HouseholdSettings.model_validate(settings.model_dump())
convert = await self._energy_converter(
[item.key for item in settings.inputs], start_datetime, end_datetime
)
return household_intervals(
settings, convert, start_datetime, end_datetime, interval_seconds
)
async def energy_intervals(
self,
key: str,
start_datetime: datetime,
end_datetime: datetime,
interval_seconds: int = 900,
) -> list[EnergyInterval]:
"""Convert a typed channel without changing the legacy kWh calculation.
Read only explicitly stored values for this key: another channel's timestamp
must not introduce a synthetic outage. Explicit null values remain barriers.
"""
convert = await self._energy_converter([key], start_datetime, end_datetime)
return convert(key, start_datetime, end_datetime, interval_seconds)
async def estimate_battery_capacity(
self, battery_id: str, request: BatteryCapacityRequest
) -> BatteryCapacityEstimate:
"""Read signed DC samples without mutating raw data or the active capacity."""
batteries = [b for b in (self.config.devices.batteries or {}).values() if b.device_id == battery_id]
if len(batteries) != 1:
raise ValueError("Require exactly one configured battery with this device_id.")
battery = batteries[0]
settings = battery.capacity_estimation
if settings is None:
raise ValueError("Configure devices.batteries[device_id].capacity_estimation first.")
if request.end.timestamp() - request.start.timestamp() > settings.max_duration_hours * 3600:
raise ValueError("Requested period exceeds capacity_estimation.max_duration_hours.")
channel = self._energy_channel(settings.power_key)
if channel.quantity != "power":
raise ValueError("Capacity estimation requires a battery DC power channel.")
context = channel.max_gap_seconds
samples = []
async for record in self.db_iterate_records(
DatabaseTimestamp.from_datetime(to_datetime(request.start) - timedelta(seconds=context)),
DatabaseTimestamp.from_datetime(
to_datetime(request.end) + timedelta(seconds=context, microseconds=1)
),
):
if record.date_time is not None and settings.power_key in record.configured_data:
samples.append((
record.date_time,
record.configured_data[settings.power_key],
record.sample_quality.get(settings.power_key, SampleQuality()),
))
if len(samples) > 250000:
raise ValueError("More than 250000 power samples; request a shorter period.")
return estimate_capacity(
request, settings, channel, samples, battery_id=battery.device_id,
capacity_wh=battery.capacity_wh,
charging_efficiency=battery.charging_efficiency,
discharging_efficiency=battery.discharging_efficiency,
)
async def _energy_from_meter_readings(
self,
key: str,
@@ -313,13 +577,12 @@ class Measurement(SingletonMixin, DataImportMixin, DataSequence[MeasurementDataR
if not measurement_file_path.exists():
return False
try:
# Validate into a temporary instance
loaded = self.__class__.model_validate_json(
measurement_file_path.read_text(encoding="utf-8")
)
# Explicitly add data records to the existing singleton
for record in loaded.records:
# Measurement is a singleton: validating a temporary Measurement
# returns the existing instance and discards serialized records.
payload = json.loads(measurement_file_path.read_text(encoding="utf-8"))
records = [MeasurementDataRecord.model_validate(data)
for data in payload.get("records", [])]
for record in records:
await self.insert_by_datetime(record)
except Exception as e:
logger.exception("Cannot load measurements")
+24
View File
@@ -0,0 +1,24 @@
"""Optional quality information on raw measurement samples."""
from typing import Literal
from pydantic import AwareDatetime, BaseModel, ConfigDict, Field
class SampleQuality(BaseModel):
"""Reset marks the first reading after a reset; generation identifies a meter."""
model_config = ConfigDict(extra="forbid")
status: Literal["measured", "estimated", "invalid", "unavailable"] = "measured"
reset: bool = False
generation: str | None = Field(default=None, max_length=128)
class MeasurementSample(BaseModel):
"""Complete replacement of one key/timestamp, including its quality."""
model_config = ConfigDict(extra="forbid", allow_inf_nan=False)
date_time: AwareDatetime
key: str
value: float | None = Field(strict=True)
quality: SampleQuality = Field(default_factory=SampleQuality)
+3
View File
@@ -75,6 +75,7 @@ from akkudoktoreos.prediction.loadakkudoktor import LoadAkkudoktorCommonSettings
from akkudoktoreos.prediction.pvforecast import PVForecastCommonSettings
from akkudoktoreos.prediction.pvforecastpvlib import _cec_inverters, _cec_modules
from akkudoktoreos.server.container_healthcheck import publish_port
from akkudoktoreos.server.rest.measurement import router as measurement_router
from akkudoktoreos.server.rest.error import (
EOSProblem,
create_error_page,
@@ -261,6 +262,8 @@ The genetic optimization API fields were renamed from German to English. For bac
# Application generic exception handling
# ----------------------
app.include_router(measurement_router)
register_problem_handlers(app)
@@ -0,0 +1,103 @@
"""Typed samples and derived energy within the existing measurement API."""
from typing import Annotated
from fastapi import APIRouter, Body, HTTPException, Query
from pydantic import AwareDatetime
from akkudoktoreos.core.coreabc import get_config, get_measurement
from akkudoktoreos.measurement.batterycapacity import BatteryCapacityEstimate, BatteryCapacityRequest
from akkudoktoreos.core.databaseabc import DatabaseTimestamp
from akkudoktoreos.measurement.energy import EnergyInterval
from akkudoktoreos.measurement.quality import MeasurementSample, SampleQuality
from akkudoktoreos.utils.datetimeutil import to_datetime
router = APIRouter(prefix="/v1/measurement", tags=["measurement"])
@router.post("/battery-capacity/{battery_id}", response_model=BatteryCapacityEstimate)
async def post_battery_capacity(battery_id: str, request: BatteryCapacityRequest) -> BatteryCapacityEstimate:
"""Estimate capacity from independent SoC anchors and configured DC power.
store_estimate writes the separate capacity_estimate config field in memory.
Persistence follows the regular EOS configuration save mechanism.
The active capacity_wh and raw measurements are never changed here.
"""
try:
estimate = await get_measurement().estimate_battery_capacity(battery_id, request)
if request.store_estimate:
batteries = [b for b in (get_config().devices.batteries or {}).values() if b.device_id == battery_id]
if len(batteries) != 1:
raise ValueError("Battery configuration changed during estimation; retry.")
get_config().set_nested_value(
f"devices/batteries/{battery_id}/capacity_estimate", estimate
)
return estimate
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc
@router.put("/samples")
async def put_samples(
samples: Annotated[list[MeasurementSample], Body(max_length=10000)],
) -> dict[str, int]:
"""Upsert raw values and their quality; legacy value/series payloads remain valid."""
try:
await get_measurement().import_samples(samples)
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc
return {"updated": len(samples)}
@router.get("/samples", response_model=list[MeasurementSample])
async def get_samples(key: str, start: AwareDatetime, end: AwareDatetime) -> list[MeasurementSample]:
"""Read raw samples including quality, in a bounded half-open range."""
measurement = get_measurement()
try:
measurement._energy_channel(key)
if not 0 < end.timestamp() - start.timestamp() <= 31 * 86400:
raise ValueError("Require a positive range of at most 31 days.")
result = []
async for record in measurement.db_iterate_records(
DatabaseTimestamp.from_datetime(to_datetime(start)),
DatabaseTimestamp.from_datetime(to_datetime(end)),
):
if key in record.configured_data and record.date_time is not None:
result.append(
MeasurementSample(
date_time=record.date_time,
key=key,
value=record.configured_data[key],
quality=record.sample_quality.get(key, SampleQuality()),
)
)
if len(result) > 10000:
raise ValueError("More than 10000 samples; request a shorter range.")
return result
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc
@router.get("/energy", response_model=list[EnergyInterval])
async def get_energy(
key: str,
start: AwareDatetime,
end: AwareDatetime,
interval_seconds: Annotated[int, Query(gt=0)] = 900,
) -> list[EnergyInterval]:
"""Energy in Wh, with temporal coverage and quality; no missing-to-zero filling."""
try:
return await get_measurement().energy_intervals(key, start, end, interval_seconds)
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc
@router.get("/household", response_model=dict[str, list[EnergyInterval]])
async def get_household(
start: AwareDatetime, end: AwareDatetime, interval_seconds: Annotated[int, Query(gt=0)] = 900
) -> dict[str, list[EnergyInterval]]:
"""Site, household without EV, and base without separately measured devices."""
try:
return await get_measurement().household_intervals(start, end, interval_seconds)
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc