mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-10-11 00:36:38 +00:00
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:
@@ -11,6 +11,7 @@ from numpydantic import NDArray, Shape
|
|||||||
from pydantic import Field, computed_field, field_validator, model_validator
|
from pydantic import Field, computed_field, field_validator, model_validator
|
||||||
|
|
||||||
from akkudoktoreos.devices.settings.devicebasesettings import DevicesBaseSettings
|
from akkudoktoreos.devices.settings.devicebasesettings import DevicesBaseSettings
|
||||||
|
from akkudoktoreos.measurement.batterycapacity import BatteryCapacityEstimate, BatteryCapacityEstimationSettings
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from akkudoktoreos.devices.genetic0.genetic0battery import (
|
from akkudoktoreos.devices.genetic0.genetic0battery import (
|
||||||
@@ -38,6 +39,12 @@ class BatteriesCommonSettings(DevicesBaseSettings):
|
|||||||
Note: Used for the GENETIC and GENETIC0 algorithm.
|
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(
|
capacity_wh: int = Field(
|
||||||
default=8000,
|
default=8000,
|
||||||
gt=0,
|
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,
|
||||||
|
)
|
||||||
@@ -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
|
||||||
@@ -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
|
||||||
@@ -6,17 +6,29 @@ data records for measurements.
|
|||||||
The measurements can be added programmatically or imported from a file or JSON string.
|
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 pathlib import Path
|
||||||
from typing import Any, Optional
|
from typing import Any, Callable, Literal, Optional
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
from numpydantic import NDArray, Shape
|
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.config.configabc import SettingsBaseModel
|
||||||
from akkudoktoreos.core.coreabc import SingletonMixin
|
from akkudoktoreos.core.coreabc import SingletonMixin
|
||||||
from akkudoktoreos.core.dataabc import DataImportMixin, DataRecord, DataSequence
|
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 (
|
from akkudoktoreos.utils.datetimeutil import (
|
||||||
DateTime,
|
DateTime,
|
||||||
Duration,
|
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):
|
class MeasurementCommonSettings(SettingsBaseModel):
|
||||||
"""Measurement Configuration."""
|
"""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(
|
load_emr_keys: Optional[list[str]] = Field(
|
||||||
default=None,
|
default=None,
|
||||||
json_schema_extra={
|
json_schema_extra={
|
||||||
@@ -74,7 +156,7 @@ class MeasurementCommonSettings(SettingsBaseModel):
|
|||||||
@property
|
@property
|
||||||
def keys(self) -> list[str]:
|
def keys(self) -> list[str]:
|
||||||
"""The keys of the measurements that can be stored."""
|
"""The keys of the measurements that can be stored."""
|
||||||
key_list = []
|
key_list = list(self.channels)
|
||||||
for key in self.__class__.model_fields.keys():
|
for key in self.__class__.model_fields.keys():
|
||||||
if key.endswith("_keys") and (value := getattr(self, key)):
|
if key.endswith("_keys") and (value := getattr(self, key)):
|
||||||
key_list.extend(value)
|
key_list.extend(value)
|
||||||
@@ -84,6 +166,18 @@ class MeasurementCommonSettings(SettingsBaseModel):
|
|||||||
class MeasurementDataRecord(DataRecord):
|
class MeasurementDataRecord(DataRecord):
|
||||||
"""Represents a measurement data record containing various measurements at a specific datetime."""
|
"""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
|
@classmethod
|
||||||
def configured_data_keys(cls) -> Optional[list[str]]:
|
def configured_data_keys(cls) -> Optional[list[str]]:
|
||||||
"""Return the keys for the configured field like data."""
|
"""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 ceiling of division to include partial intervals
|
||||||
return int(np.ceil(diff_seconds / interval_seconds))
|
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(
|
async def _energy_from_meter_readings(
|
||||||
self,
|
self,
|
||||||
key: str,
|
key: str,
|
||||||
@@ -313,13 +577,12 @@ class Measurement(SingletonMixin, DataImportMixin, DataSequence[MeasurementDataR
|
|||||||
if not measurement_file_path.exists():
|
if not measurement_file_path.exists():
|
||||||
return False
|
return False
|
||||||
try:
|
try:
|
||||||
# Validate into a temporary instance
|
# Measurement is a singleton: validating a temporary Measurement
|
||||||
loaded = self.__class__.model_validate_json(
|
# returns the existing instance and discards serialized records.
|
||||||
measurement_file_path.read_text(encoding="utf-8")
|
payload = json.loads(measurement_file_path.read_text(encoding="utf-8"))
|
||||||
)
|
records = [MeasurementDataRecord.model_validate(data)
|
||||||
|
for data in payload.get("records", [])]
|
||||||
# Explicitly add data records to the existing singleton
|
for record in records:
|
||||||
for record in loaded.records:
|
|
||||||
await self.insert_by_datetime(record)
|
await self.insert_by_datetime(record)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.exception("Cannot load measurements")
|
logger.exception("Cannot load measurements")
|
||||||
|
|||||||
@@ -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)
|
||||||
@@ -75,6 +75,7 @@ from akkudoktoreos.prediction.loadakkudoktor import LoadAkkudoktorCommonSettings
|
|||||||
from akkudoktoreos.prediction.pvforecast import PVForecastCommonSettings
|
from akkudoktoreos.prediction.pvforecast import PVForecastCommonSettings
|
||||||
from akkudoktoreos.prediction.pvforecastpvlib import _cec_inverters, _cec_modules
|
from akkudoktoreos.prediction.pvforecastpvlib import _cec_inverters, _cec_modules
|
||||||
from akkudoktoreos.server.container_healthcheck import publish_port
|
from akkudoktoreos.server.container_healthcheck import publish_port
|
||||||
|
from akkudoktoreos.server.rest.measurement import router as measurement_router
|
||||||
from akkudoktoreos.server.rest.error import (
|
from akkudoktoreos.server.rest.error import (
|
||||||
EOSProblem,
|
EOSProblem,
|
||||||
create_error_page,
|
create_error_page,
|
||||||
@@ -261,6 +262,8 @@ The genetic optimization API fields were renamed from German to English. For bac
|
|||||||
# Application generic exception handling
|
# Application generic exception handling
|
||||||
# ----------------------
|
# ----------------------
|
||||||
|
|
||||||
|
app.include_router(measurement_router)
|
||||||
|
|
||||||
register_problem_handlers(app)
|
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
|
||||||
@@ -0,0 +1,237 @@
|
|||||||
|
import pytest_asyncio
|
||||||
|
"""Capacity fits must preserve energy direction, coverage and independent anchors."""
|
||||||
|
|
||||||
|
# ruff: noqa: S101
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from pydantic import ValidationError
|
||||||
|
|
||||||
|
from akkudoktoreos.measurement.batterycapacity import (
|
||||||
|
BatteryCapacityEstimationSettings,
|
||||||
|
BatteryCapacityRequest,
|
||||||
|
estimate_capacity,
|
||||||
|
)
|
||||||
|
from akkudoktoreos.measurement.measurement import MeasurementChannelSettings
|
||||||
|
from akkudoktoreos.measurement.quality import SampleQuality
|
||||||
|
|
||||||
|
START = datetime(2026, 9, 10, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
|
||||||
|
def fit(
|
||||||
|
points,
|
||||||
|
*,
|
||||||
|
start_soc=20,
|
||||||
|
end_soc=100,
|
||||||
|
efficiency=1,
|
||||||
|
method="hold",
|
||||||
|
polarity="charging",
|
||||||
|
unit="W",
|
||||||
|
end_seconds=3600,
|
||||||
|
quality=None,
|
||||||
|
max_gap=3600,
|
||||||
|
):
|
||||||
|
request = BatteryCapacityRequest(
|
||||||
|
start=START,
|
||||||
|
end=START + timedelta(seconds=end_seconds),
|
||||||
|
start_soc_percentage=start_soc,
|
||||||
|
end_soc_percentage=end_soc,
|
||||||
|
soc_reference="voltage_current_anchor",
|
||||||
|
)
|
||||||
|
settings = BatteryCapacityEstimationSettings(power_key="dc", positive_power=polarity)
|
||||||
|
channel = MeasurementChannelSettings(
|
||||||
|
quantity="power",
|
||||||
|
unit=unit,
|
||||||
|
integration_method=method,
|
||||||
|
max_gap_seconds=max_gap,
|
||||||
|
)
|
||||||
|
return estimate_capacity(
|
||||||
|
request,
|
||||||
|
settings,
|
||||||
|
channel,
|
||||||
|
[
|
||||||
|
(START + timedelta(seconds=t), v, (quality or {}).get(t, SampleQuality()))
|
||||||
|
for t, v in points
|
||||||
|
],
|
||||||
|
battery_id="battery",
|
||||||
|
capacity_wh=12000,
|
||||||
|
charging_efficiency=efficiency,
|
||||||
|
discharging_efficiency=efficiency,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_charge_fit_and_unclipped_model_error():
|
||||||
|
result = fit([(0, 8000), (3600, 8000)])
|
||||||
|
assert result.estimated_capacity_wh == pytest.approx(10000)
|
||||||
|
assert result.configured_capacity_wh == 12000
|
||||||
|
assert result.model_end_soc_percentage_unclipped == pytest.approx(86.6666667)
|
||||||
|
assert result.model_soc_error_percentage_points == pytest.approx(-13.3333333)
|
||||||
|
assert result.charge_energy_wh == 8000
|
||||||
|
assert result.coverage_seconds == 3600
|
||||||
|
|
||||||
|
|
||||||
|
def test_efficiency_is_applied_once_on_dc_boundary():
|
||||||
|
result = fit([(0, 10000), (3600, 10000)], efficiency=0.8)
|
||||||
|
assert result.estimated_capacity_wh == pytest.approx(10000)
|
||||||
|
assert result.stored_energy_change_wh == 8000
|
||||||
|
|
||||||
|
|
||||||
|
def test_discharge_fit_and_reversed_sensor_sign():
|
||||||
|
result = fit(
|
||||||
|
[(0, 6400), (3600, 6400)], start_soc=100, end_soc=20, polarity="discharging", efficiency=0.8
|
||||||
|
)
|
||||||
|
assert result.discharge_energy_wh == 6400
|
||||||
|
assert result.estimated_capacity_wh == pytest.approx(10000)
|
||||||
|
|
||||||
|
|
||||||
|
def test_linear_zero_crossing_is_split_before_losses():
|
||||||
|
result = fit([(0, -4000), (3600, 12000)], method="linear", efficiency=0.8)
|
||||||
|
assert result.charge_energy_wh == pytest.approx(4500)
|
||||||
|
assert result.discharge_energy_wh == pytest.approx(500)
|
||||||
|
assert result.stored_energy_change_wh == pytest.approx(4500 * 0.8 - 500 / 0.8)
|
||||||
|
|
||||||
|
|
||||||
|
def test_kw_and_clipped_boundary_interpolation():
|
||||||
|
result = fit([(-3600, 0), (3600, 16)], method="linear", unit="kW", max_gap=7200)
|
||||||
|
assert result.charge_energy_wh == pytest.approx(12000)
|
||||||
|
# Deliberately do not saturate the old model at 100%.
|
||||||
|
assert result.model_end_soc_percentage_unclipped == pytest.approx(120)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"points", [[], [(0, 8000)], [(60, 8000), (3600, 8000)], [(0, 8000), (3500, 8000)]]
|
||||||
|
)
|
||||||
|
def test_missing_coverage_is_not_extrapolated(points):
|
||||||
|
with pytest.raises(ValueError, match="coverage"):
|
||||||
|
fit(points)
|
||||||
|
|
||||||
|
|
||||||
|
def test_gap_is_rejected():
|
||||||
|
with pytest.raises(ValueError, match="gap"):
|
||||||
|
fit([(0, 8000), (3600, 8000)], max_gap=300)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("value", [None, float("nan"), float("inf"), True])
|
||||||
|
def test_bad_power_is_rejected(value):
|
||||||
|
with pytest.raises(ValueError, match="finite measured"):
|
||||||
|
fit([(0, value), (3600, 8000)])
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("status", ["estimated", "invalid", "unavailable"])
|
||||||
|
def test_nonmeasured_quality_is_rejected(status):
|
||||||
|
with pytest.raises(ValueError, match="finite measured"):
|
||||||
|
fit([(0, 8000), (3600, 8000)], quality={0: SampleQuality(status=status)})
|
||||||
|
|
||||||
|
|
||||||
|
def test_reset_and_sensor_change_are_rejected():
|
||||||
|
for quality in (SampleQuality(reset=True), SampleQuality(generation="replacement")):
|
||||||
|
with pytest.raises(ValueError, match="reset or generation"):
|
||||||
|
fit([(0, 8000), (3600, 8000)], quality={3600: quality})
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("start_soc", [100, 99, 85])
|
||||||
|
def test_full_to_full_and_small_soc_span_do_not_produce_estimates(start_soc):
|
||||||
|
with pytest.raises(ValueError, match="SoC change is too small"):
|
||||||
|
fit([(0, 8000), (3600, 8000)], start_soc=start_soc)
|
||||||
|
|
||||||
|
|
||||||
|
def test_wrong_polarity_is_rejected():
|
||||||
|
with pytest.raises(ValueError, match="disagrees"):
|
||||||
|
fit([(0, -8000), (3600, -8000)])
|
||||||
|
|
||||||
|
|
||||||
|
def test_hidden_saturation_cannot_be_fixed_by_end_point_fitting():
|
||||||
|
with pytest.raises(ValueError, match="Fitted SoC leaves"):
|
||||||
|
fit([(0, 24000), (1800, -8000), (3600, -8000)])
|
||||||
|
|
||||||
|
|
||||||
|
def test_model_soc_is_not_an_accepted_reference():
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
BatteryCapacityRequest(
|
||||||
|
start=START,
|
||||||
|
end=START + timedelta(hours=1),
|
||||||
|
start_soc_percentage=20,
|
||||||
|
soc_reference="calculated_soc",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest_asyncio.fixture
|
||||||
|
async def database_case(config_eos):
|
||||||
|
from akkudoktoreos.core.coreabc import get_measurement
|
||||||
|
from akkudoktoreos.measurement.quality import MeasurementSample
|
||||||
|
|
||||||
|
config_eos.merge_settings_from_dict(
|
||||||
|
{
|
||||||
|
"devices": {
|
||||||
|
"batteries": {
|
||||||
|
"battery": {
|
||||||
|
"device_id": "battery",
|
||||||
|
"capacity_wh": 12000,
|
||||||
|
"charging_efficiency": 1,
|
||||||
|
"discharging_efficiency": 1,
|
||||||
|
"capacity_estimation": {
|
||||||
|
"power_key": "battery_dc",
|
||||||
|
"positive_power": "charging",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"measurement": {
|
||||||
|
"channels": {
|
||||||
|
"battery_dc": {
|
||||||
|
"quantity": "power",
|
||||||
|
"unit": "W",
|
||||||
|
"integration_method": "hold",
|
||||||
|
"max_gap_seconds": 3600,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
}
|
||||||
|
)
|
||||||
|
(await get_measurement().import_samples(
|
||||||
|
[
|
||||||
|
MeasurementSample(
|
||||||
|
date_time=START + timedelta(seconds=t), key="battery_dc", value=8000.0
|
||||||
|
)
|
||||||
|
for t in (0, 1800, 3600)
|
||||||
|
]
|
||||||
|
))
|
||||||
|
return config_eos
|
||||||
|
|
||||||
|
|
||||||
|
def test_http_reads_database_and_stores_only_explicit_estimate(database_case):
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
from akkudoktoreos.server.eos import app
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
body = {
|
||||||
|
"start": START.isoformat(),
|
||||||
|
"end": (START + timedelta(hours=1)).isoformat(),
|
||||||
|
"start_soc_percentage": 20,
|
||||||
|
"soc_reference": "voltage_current_anchor",
|
||||||
|
}
|
||||||
|
response = client.post("/v1/measurement/battery-capacity/battery", json=body)
|
||||||
|
assert response.status_code == 200, response.text
|
||||||
|
assert response.json()["estimated_capacity_wh"] == pytest.approx(10000)
|
||||||
|
battery = database_case.devices.batteries["battery"]
|
||||||
|
assert battery.capacity_estimate is None
|
||||||
|
assert battery.capacity_wh == 12000
|
||||||
|
body["store_estimate"] = True
|
||||||
|
response = client.post("/v1/measurement/battery-capacity/battery", json=body)
|
||||||
|
assert response.status_code == 200, response.text
|
||||||
|
assert battery.capacity_estimate.estimated_capacity_wh == pytest.approx(10000)
|
||||||
|
assert battery.capacity_wh == 12000
|
||||||
|
database_case.merge_settings_from_dict({"optimization": {"genetic": {"individuals": 100}}})
|
||||||
|
battery = database_case.devices.batteries["battery"]
|
||||||
|
assert battery.capacity_estimate.estimated_capacity_wh == pytest.approx(10000)
|
||||||
|
assert battery.capacity_wh == 12000
|
||||||
|
|
||||||
|
# The estimate survives normal config serialization without becoming capacity_wh.
|
||||||
|
data = database_case.to_config_json()
|
||||||
|
assert '"estimated_capacity_wh": 10000.0' in data
|
||||||
|
assert '"capacity_wh": 12000' in data
|
||||||
|
previous = battery.capacity_estimate
|
||||||
|
body["start_soc_percentage"] = 100
|
||||||
|
assert client.post("/v1/measurement/battery-capacity/battery", json=body).status_code == 422
|
||||||
|
assert battery.capacity_estimate is previous
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
"""Verify fallback JSON loading does not lose records through the singleton."""
|
||||||
|
from unittest.mock import AsyncMock
|
||||||
|
import pytest
|
||||||
|
from akkudoktoreos.core.coreabc import get_measurement
|
||||||
|
from akkudoktoreos.core.dataabc import DataSequence
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_measurement_json_roundtrip(config_eos, tmp_path, monkeypatch):
|
||||||
|
m = get_measurement()
|
||||||
|
config_eos.measurement.load_emr_keys = ["meter"]
|
||||||
|
config_eos.general.data_folder_path = tmp_path
|
||||||
|
config_eos.database.provider = None
|
||||||
|
m._db_reset_state()
|
||||||
|
try:
|
||||||
|
await m.update_value("2026-09-16T08:00:00Z", "meter", 123.5)
|
||||||
|
monkeypatch.setattr(DataSequence, "save", AsyncMock(return_value=False))
|
||||||
|
monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=False))
|
||||||
|
assert await m.save()
|
||||||
|
m._db_reset_state()
|
||||||
|
assert await m.load()
|
||||||
|
assert len(m.records) == 1
|
||||||
|
assert m.records[0]["meter"] == 123.5
|
||||||
|
finally:
|
||||||
|
m._db_reset_state()
|
||||||
@@ -0,0 +1,105 @@
|
|||||||
|
from unittest.mock import AsyncMock
|
||||||
|
"""Contracts for typed channels sharing the existing measurement storage."""
|
||||||
|
|
||||||
|
# ruff: noqa: S101
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from pydantic import ValidationError
|
||||||
|
|
||||||
|
from akkudoktoreos.core.coreabc import get_measurement
|
||||||
|
from akkudoktoreos.measurement.measurement import (
|
||||||
|
MeasurementChannelSettings,
|
||||||
|
MeasurementCommonSettings,
|
||||||
|
MeasurementDataRecord,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
CHANNELS = {
|
||||||
|
"house_power": dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=120),
|
||||||
|
"house_meter": dict(quantity="cumulative_energy", unit="kWh"),
|
||||||
|
"house_interval": dict(
|
||||||
|
quantity="interval_energy", unit="Wh", interval_seconds=900, timestamp_reference="start"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"definition",
|
||||||
|
[
|
||||||
|
dict(quantity="power", unit="kWh", integration_method="hold", max_gap_seconds=60),
|
||||||
|
dict(quantity="power", unit="W"),
|
||||||
|
dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=0),
|
||||||
|
dict(quantity="cumulative_energy", unit="W"),
|
||||||
|
dict(quantity="cumulative_energy", unit="kWh", timestamp_reference="end"),
|
||||||
|
dict(quantity="interval_energy", unit="Wh", interval_seconds=900),
|
||||||
|
dict(
|
||||||
|
quantity="interval_energy", unit="Wh", interval_seconds=True, timestamp_reference="end"
|
||||||
|
),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_reject_ambiguous_channel(definition):
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
MeasurementChannelSettings(**definition)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"legacy_field",
|
||||||
|
["load_emr_keys", "grid_import_emr_keys", "grid_export_emr_keys", "pv_production_emr_keys"],
|
||||||
|
)
|
||||||
|
def test_legacy_keys_keep_meter_semantics(legacy_field):
|
||||||
|
settings = MeasurementCommonSettings(**{legacy_field: ["legacy"]}, channels=CHANNELS)
|
||||||
|
assert settings.keys == sorted(["legacy", *CHANNELS])
|
||||||
|
compatible = MeasurementCommonSettings(
|
||||||
|
**{legacy_field: ["legacy"]}, channels={"legacy": CHANNELS["house_meter"]}
|
||||||
|
)
|
||||||
|
assert compatible.keys == ["legacy"]
|
||||||
|
with pytest.raises(ValidationError, match="must remain"):
|
||||||
|
MeasurementCommonSettings(
|
||||||
|
**{legacy_field: ["legacy"]}, channels={"legacy": CHANNELS["house_power"]}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("key", ["", " x", "date_time", "configured_data", "keys", "_private"])
|
||||||
|
def test_reject_reserved_key(key):
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
MeasurementCommonSettings(channels={key: CHANNELS["house_meter"]})
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_existing_import_and_file_reload(config_eos, tmp_path, monkeypatch):
|
||||||
|
"""All three quantities retain their raw values through the existing file path."""
|
||||||
|
from akkudoktoreos.core.dataabc import DataSequence
|
||||||
|
|
||||||
|
measurement = get_measurement()
|
||||||
|
previous_settings = config_eos.measurement
|
||||||
|
previous_records = measurement.records
|
||||||
|
previous_folder = config_eos.general.data_folder_path
|
||||||
|
try:
|
||||||
|
config_eos.measurement = MeasurementCommonSettings(channels=CHANNELS)
|
||||||
|
config_eos.general.data_folder_path = tmp_path
|
||||||
|
measurement._db_reset_state()
|
||||||
|
values = dict(house_power=800.0, house_meter=12345.6, house_interval=200.0)
|
||||||
|
for key, value in values.items():
|
||||||
|
(await measurement.update_value("2026-09-10T18:00:00Z", key, value))
|
||||||
|
assert set(values).issubset(measurement.record_keys)
|
||||||
|
monkeypatch.setattr(DataSequence, "save", AsyncMock(return_value=False))
|
||||||
|
monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=False))
|
||||||
|
assert (await measurement.save())
|
||||||
|
measurement._db_reset_state()
|
||||||
|
assert (await measurement.load())
|
||||||
|
assert len(measurement.records) == 1
|
||||||
|
for key, value in values.items():
|
||||||
|
assert measurement.records[0][key] == value
|
||||||
|
restored = MeasurementDataRecord.model_validate_json(
|
||||||
|
measurement.records[0].model_dump_json()
|
||||||
|
)
|
||||||
|
assert restored.configured_data == values
|
||||||
|
settings = MeasurementCommonSettings.model_validate_json(
|
||||||
|
config_eos.measurement.model_dump_json()
|
||||||
|
)
|
||||||
|
assert settings.model_dump() == config_eos.measurement.model_dump()
|
||||||
|
finally:
|
||||||
|
measurement._db_reset_state()
|
||||||
|
measurement.records = previous_records
|
||||||
|
config_eos.measurement = previous_settings
|
||||||
|
config_eos.general.data_folder_path = previous_folder
|
||||||
@@ -0,0 +1,132 @@
|
|||||||
|
"""Physical and temporal contracts for interval energy."""
|
||||||
|
|
||||||
|
# ruff: noqa: S101
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from zoneinfo import ZoneInfo
|
||||||
|
|
||||||
|
from akkudoktoreos.measurement.energy import energy_intervals
|
||||||
|
from akkudoktoreos.measurement.measurement import MeasurementChannelSettings
|
||||||
|
|
||||||
|
|
||||||
|
START = datetime(2026, 9, 10, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
|
||||||
|
def channel(quantity="power", **kwargs):
|
||||||
|
defaults = {
|
||||||
|
"power": dict(unit="W", integration_method="hold", max_gap_seconds=900),
|
||||||
|
"cumulative_energy": dict(unit="kWh"),
|
||||||
|
"interval_energy": dict(unit="Wh", interval_seconds=900, timestamp_reference="start"),
|
||||||
|
}
|
||||||
|
return MeasurementChannelSettings(quantity=quantity, **(defaults[quantity] | kwargs))
|
||||||
|
|
||||||
|
|
||||||
|
def convert(points, config=None, seconds=900):
|
||||||
|
return energy_intervals(
|
||||||
|
[(START + timedelta(seconds=t), v) for t, v in points],
|
||||||
|
config or channel(),
|
||||||
|
START,
|
||||||
|
START + timedelta(seconds=seconds),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"config,points",
|
||||||
|
[
|
||||||
|
(channel(), [(0, 800), (900, 800)]),
|
||||||
|
(channel(unit="kW"), [(0, 0.8), (900, 0.8)]),
|
||||||
|
(channel("cumulative_energy"), [(0, 10), (900, 10.2)]),
|
||||||
|
(channel("interval_energy"), [(0, 200)]),
|
||||||
|
(channel("interval_energy", timestamp_reference="end"), [(900, 200)]),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_equivalent_measurements(config, points):
|
||||||
|
result = convert(points, config)[0]
|
||||||
|
assert result.energy_wh == pytest.approx(200)
|
||||||
|
assert result.coverage_seconds == 900
|
||||||
|
assert result.coverage_status == "complete"
|
||||||
|
|
||||||
|
|
||||||
|
def test_time_weighting_and_linear_interpolation():
|
||||||
|
assert convert([(0, 0), (600, 1200), (900, 1200)])[0].energy_wh == 100
|
||||||
|
assert convert([(0, 0), (900, 1600)], channel(integration_method="linear"))[0].energy_wh == 200
|
||||||
|
|
||||||
|
|
||||||
|
def test_gap_and_no_extrapolation():
|
||||||
|
result = convert([(0, 800), (300, 800)])[0]
|
||||||
|
assert result.energy_wh is None
|
||||||
|
assert result.observed_energy_wh == pytest.approx(800 / 12)
|
||||||
|
assert result.coverage_status == "partial"
|
||||||
|
assert convert([(0, 800), (900, 800)], channel(max_gap_seconds=60))[0].energy_wh is None
|
||||||
|
assert convert([])[0].coverage_status == "missing"
|
||||||
|
|
||||||
|
|
||||||
|
def test_null_breaks_hold_at_outage():
|
||||||
|
result = convert([(0, 800), (300, None), (600, 800), (900, 800)])[0]
|
||||||
|
assert result.coverage_seconds == 600
|
||||||
|
assert result.coverage_status == "partial"
|
||||||
|
assert len(result.coverage_ranges) == 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_reset_does_not_create_negative_consumption():
|
||||||
|
result = convert([(0, 10), (450, 0), (900, 0.1)], channel("cumulative_energy"))[0]
|
||||||
|
assert result.energy_wh is None
|
||||||
|
assert result.observed_energy_wh == pytest.approx(100)
|
||||||
|
assert "meter_reset" in result.flags
|
||||||
|
|
||||||
|
|
||||||
|
def test_hour_allocation_conserves_energy_and_is_labelled():
|
||||||
|
result = convert([(0, 1000)], channel("interval_energy", interval_seconds=3600), 3600)
|
||||||
|
assert sum(r.energy_wh for r in result) == 1000
|
||||||
|
assert all("allocated_energy" in r.methods for r in result)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"quantity,points",
|
||||||
|
[
|
||||||
|
("power", [(0, 1), (0, 2)]),
|
||||||
|
("interval_energy", [(0, 100), (450, 100)]),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_ambiguous_time_support_rejected(quantity, points):
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
convert(points, channel(quantity))
|
||||||
|
|
||||||
|
|
||||||
|
def test_nan_and_signed_power():
|
||||||
|
assert convert([(0, float("nan")), (900, 800)])[0].energy_wh is None
|
||||||
|
assert convert([(0, -800), (900, -800)])[0].energy_wh == -200
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("month,day,hours", [(3, 29, 23), (10, 25, 25)])
|
||||||
|
def test_dst_calendar_day(month, day, hours):
|
||||||
|
start = datetime(2026, month, day, tzinfo=ZoneInfo("Europe/Berlin"))
|
||||||
|
end = start + timedelta(days=1)
|
||||||
|
result = energy_intervals([(start, 0), (end, hours)], channel("cumulative_energy"), start, end)
|
||||||
|
assert len(result) == hours * 4
|
||||||
|
assert sum(r.energy_wh for r in result) == pytest.approx(hours * 1000)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_measurement_wrapper_preserves_asynchronous_channels(config_eos):
|
||||||
|
from akkudoktoreos.core.coreabc import get_measurement
|
||||||
|
from akkudoktoreos.measurement.measurement import MeasurementCommonSettings
|
||||||
|
|
||||||
|
measurement = get_measurement()
|
||||||
|
previous, records = config_eos.measurement, measurement.records
|
||||||
|
try:
|
||||||
|
config_eos.measurement = MeasurementCommonSettings(
|
||||||
|
channels={"p": channel(), "other": channel()}
|
||||||
|
)
|
||||||
|
measurement._db_reset_state()
|
||||||
|
(await measurement.update_value(START, "p", 800))
|
||||||
|
(await measurement.update_value(START + timedelta(seconds=450), "other", 1))
|
||||||
|
(await measurement.update_value(START + timedelta(seconds=900), "p", 800))
|
||||||
|
result = (await measurement.energy_intervals("p", START, START + timedelta(seconds=900)))
|
||||||
|
assert result[0].energy_wh == 200
|
||||||
|
finally:
|
||||||
|
measurement._db_reset_state()
|
||||||
|
measurement.records = records
|
||||||
|
config_eos.measurement = previous
|
||||||
@@ -0,0 +1,366 @@
|
|||||||
|
"""API, persistence and physical balance contracts for typed measurements."""
|
||||||
|
|
||||||
|
# ruff: noqa: S101
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi import FastAPI
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
from pydantic import ValidationError
|
||||||
|
|
||||||
|
from akkudoktoreos.core.coreabc import get_database, get_measurement
|
||||||
|
from akkudoktoreos.measurement.measurement import MeasurementCommonSettings
|
||||||
|
from akkudoktoreos.measurement.quality import MeasurementSample
|
||||||
|
from akkudoktoreos.server.rest.measurement import router
|
||||||
|
|
||||||
|
START = datetime(2026, 9, 10, tzinfo=timezone.utc)
|
||||||
|
END = START + timedelta(seconds=900)
|
||||||
|
POWER = dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=900)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def setup_measurement(config_eos):
|
||||||
|
measurement = get_measurement()
|
||||||
|
settings, records = config_eos.measurement, measurement.records
|
||||||
|
measurement._db_reset_state()
|
||||||
|
|
||||||
|
def configure(topology="direct", inputs=None, channels=None):
|
||||||
|
inputs = inputs or [dict(key="site", branch="house", role="site")]
|
||||||
|
config_eos.measurement = MeasurementCommonSettings(
|
||||||
|
channels=channels or {item["key"]: POWER for item in inputs},
|
||||||
|
household=dict(topology=topology, inputs=inputs),
|
||||||
|
)
|
||||||
|
return measurement
|
||||||
|
|
||||||
|
yield configure
|
||||||
|
measurement._db_reset_state()
|
||||||
|
measurement.records = records
|
||||||
|
config_eos.measurement = settings
|
||||||
|
|
||||||
|
|
||||||
|
async def write(measurement, key, points):
|
||||||
|
(await measurement.import_samples(
|
||||||
|
[
|
||||||
|
MeasurementSample(
|
||||||
|
date_time=START + timedelta(seconds=t),
|
||||||
|
key=key,
|
||||||
|
value=value,
|
||||||
|
quality=quality if quality else {},
|
||||||
|
)
|
||||||
|
for t, value, quality in points
|
||||||
|
]
|
||||||
|
))
|
||||||
|
|
||||||
|
|
||||||
|
async def constant(measurement, key, watts):
|
||||||
|
(await write(measurement, key, [(0, watts, None), (900, watts, None)]))
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"topology,values",
|
||||||
|
[
|
||||||
|
("direct", {"site": 4000}),
|
||||||
|
("separate_ac", {"grid": 1000, "pv": 2000, "battery": 1000}),
|
||||||
|
("hybrid_ac", {"grid": 1000, "inverter": 3000}),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
async def test_topologies_and_device_subtraction(setup_measurement, topology, values):
|
||||||
|
values = values | {"ev": 2000, "device": 800}
|
||||||
|
m = setup_measurement(topology, [dict(key=k, branch=k, role=k) for k in values])
|
||||||
|
for key, value in values.items():
|
||||||
|
(await constant(m, key, value))
|
||||||
|
rows = (await m.household_intervals(START, END))
|
||||||
|
assert rows["site"][0].energy_wh == 1000
|
||||||
|
assert rows["household"][0].energy_wh == 500
|
||||||
|
assert rows["base"][0].energy_wh == 300
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_export_charge_and_polarity(setup_measurement):
|
||||||
|
m = setup_measurement(
|
||||||
|
"separate_ac",
|
||||||
|
[
|
||||||
|
dict(key="grid", branch="grid", role="grid"),
|
||||||
|
dict(key="solar", branch="solar", role="pv"),
|
||||||
|
dict(key="charge", branch="battery", role="battery", polarity=-1),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
for key, watts in dict(grid=-1000, solar=4000, charge=1000).items():
|
||||||
|
(await constant(m, key, watts))
|
||||||
|
assert (await m.household_intervals(START, END))["site"][0].energy_wh == 500
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
@pytest.mark.parametrize("grid,inverter", [(-6000, 7200), (4200, -3000)])
|
||||||
|
async def test_hybrid_with_independent_ac_pv_and_ev(setup_measurement, grid, inverter):
|
||||||
|
inputs = [dict(key=k, branch=k, role=k) for k in ("grid", "inverter", "pv", "ev")]
|
||||||
|
m = setup_measurement("hybrid_ac", inputs)
|
||||||
|
for key, watts in dict(grid=grid, inverter=inverter, pv=400, ev=600).items():
|
||||||
|
(await constant(m, key, watts))
|
||||||
|
rows = (await m.household_intervals(START, END))
|
||||||
|
assert rows["site"][0].energy_wh == 400
|
||||||
|
assert rows["household"][0].energy_wh == 250
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_hybrid_missing_ac_pv_is_not_zero(setup_measurement):
|
||||||
|
m = setup_measurement(
|
||||||
|
"hybrid_ac", [dict(key=k, branch=k, role=k) for k in ("grid", "inverter", "pv")]
|
||||||
|
)
|
||||||
|
(await constant(m, "grid", -6000))
|
||||||
|
(await constant(m, "inverter", 7200))
|
||||||
|
row = (await m.household_intervals(START, END))["site"][0]
|
||||||
|
assert row.energy_wh is None
|
||||||
|
assert row.coverage_seconds == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_hybrid_rejects_separate_battery_to_avoid_double_counting(setup_measurement):
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
setup_measurement(
|
||||||
|
"hybrid_ac",
|
||||||
|
[dict(key=k, branch=k, role=k) for k in ("grid", "inverter", "battery")],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_missing_ev_does_not_destroy_site(setup_measurement):
|
||||||
|
m = setup_measurement(inputs=[dict(key=k, branch=k, role=k) for k in ("site", "ev")])
|
||||||
|
(await constant(m, "site", 800))
|
||||||
|
rows = (await m.household_intervals(START, END))
|
||||||
|
assert rows["site"][0].energy_wh == 200
|
||||||
|
assert rows["household"][0].energy_wh is None
|
||||||
|
assert rows["household"][0].coverage_seconds == 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_intersection_integrates_actual_shape(setup_measurement):
|
||||||
|
m = setup_measurement(inputs=[dict(key=k, branch=k, role=k) for k in ("site", "ev")])
|
||||||
|
(await write(m, "site", [(0, 1000, None), (300, 2000, None), (600, None, None)]))
|
||||||
|
(await write(m, "ev", [(300, 500, None), (900, 500, None)]))
|
||||||
|
row = (await m.household_intervals(START, END))["household"][0]
|
||||||
|
assert row.coverage_seconds == 300
|
||||||
|
assert row.energy_wh is None
|
||||||
|
assert row.observed_energy_wh == pytest.approx(125)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_nonoverlapping_coverage_is_missing(setup_measurement):
|
||||||
|
m = setup_measurement(inputs=[dict(key=k, branch=k, role=k) for k in ("site", "ev")])
|
||||||
|
(await write(m, "site", [(0, 1000, None), (300, 1000, None)]))
|
||||||
|
(await write(m, "ev", [(600, 500, None), (900, 500, None)]))
|
||||||
|
row = (await m.household_intervals(START, END))["household"][0]
|
||||||
|
assert row.observed_energy_wh is None
|
||||||
|
assert row.coverage_seconds == 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"inputs",
|
||||||
|
[
|
||||||
|
[dict(key="p", branch="x", role="grid"), dict(key="p", branch="y", role="pv")],
|
||||||
|
[dict(key="p", branch="x", role="grid"), dict(key="q", branch="x", role="pv")],
|
||||||
|
[dict(key="p", branch="x", role="site"), dict(key="q", branch="y", role="pv")],
|
||||||
|
[dict(key="unknown", branch="x", role="grid")],
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_invalid_balance_configuration(setup_measurement, inputs):
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
setup_measurement("separate_ac", inputs, {"p": POWER, "q": POWER})
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_quality_reset_even_with_increasing_meter(setup_measurement):
|
||||||
|
m = setup_measurement(channels={"site": dict(quantity="cumulative_energy", unit="Wh")})
|
||||||
|
(await write(
|
||||||
|
m,
|
||||||
|
"site",
|
||||||
|
[
|
||||||
|
(0, 0, {"generation": "old"}),
|
||||||
|
(450, 1000, {"generation": "new", "reset": True}),
|
||||||
|
(900, 1100, {"generation": "new", "status": "estimated"}),
|
||||||
|
],
|
||||||
|
))
|
||||||
|
row = (await m.energy_intervals("site", START, END))[0]
|
||||||
|
assert row.energy_wh is None
|
||||||
|
assert row.observed_energy_wh == 100
|
||||||
|
assert set(row.flags) == {"meter_reset", "estimated"}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_api_upsert_outage_validation_and_readback(setup_measurement):
|
||||||
|
m = setup_measurement()
|
||||||
|
app = FastAPI()
|
||||||
|
app.include_router(router)
|
||||||
|
with TestClient(app) as client:
|
||||||
|
payload = [
|
||||||
|
dict(
|
||||||
|
date_time=START.isoformat(), key="site", value=800, quality={"status": "estimated"}
|
||||||
|
),
|
||||||
|
dict(date_time=END.isoformat(), key="site", value=800),
|
||||||
|
]
|
||||||
|
assert client.put("/v1/measurement/samples", json=payload).status_code == 200
|
||||||
|
params = dict(key="site", start=START.isoformat(), end=END.isoformat())
|
||||||
|
row = client.get("/v1/measurement/energy", params=params).json()[0]
|
||||||
|
assert row["energy_wh"] == 200
|
||||||
|
assert "estimated" in row["flags"]
|
||||||
|
raw = client.get("/v1/measurement/samples", params=params).json()
|
||||||
|
assert raw[0]["quality"]["status"] == "estimated"
|
||||||
|
payload[0]["value"] = None
|
||||||
|
payload[0]["quality"] = {"status": "unavailable"}
|
||||||
|
assert client.put("/v1/measurement/samples", json=payload[:1]).status_code == 200
|
||||||
|
row = client.get("/v1/measurement/energy", params=params).json()[0]
|
||||||
|
assert row["energy_wh"] is None
|
||||||
|
assert "unavailable" in row["flags"]
|
||||||
|
assert len(m.records) == 2
|
||||||
|
payload[0]["value"] = 0
|
||||||
|
payload[0]["quality"] = {}
|
||||||
|
assert client.put("/v1/measurement/samples", json=payload[:1]).status_code == 200
|
||||||
|
assert client.get("/v1/measurement/energy", params=params).json()[0]["energy_wh"] == 0
|
||||||
|
assert (
|
||||||
|
client.get("/v1/measurement/household", params=params).json()["site"][0]["energy_wh"]
|
||||||
|
== 0
|
||||||
|
)
|
||||||
|
# Validate a whole batch before changing any value.
|
||||||
|
bad = [payload[0] | {"value": 999}, payload[0] | {"key": "unknown"}]
|
||||||
|
assert client.put("/v1/measurement/samples", json=bad).status_code == 422
|
||||||
|
assert (await m.energy_intervals("site", START, END))[0].energy_wh == 0
|
||||||
|
assert (
|
||||||
|
client.get(
|
||||||
|
"/v1/measurement/energy", params=params | {"interval_seconds": 0}
|
||||||
|
).status_code
|
||||||
|
== 422
|
||||||
|
)
|
||||||
|
assert (
|
||||||
|
client.get(
|
||||||
|
"/v1/measurement/energy", params=params | {"start": "2026-09-10T00:00:00"}
|
||||||
|
).status_code
|
||||||
|
== 422
|
||||||
|
)
|
||||||
|
assert (
|
||||||
|
client.get(
|
||||||
|
"/v1/measurement/energy",
|
||||||
|
params=params | {"end": (START + timedelta(days=32)).isoformat()},
|
||||||
|
).status_code
|
||||||
|
== 422
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
@pytest.mark.parametrize("provider", [None, "SQLite", "LMDB"])
|
||||||
|
async def test_quality_persists_through_storage_restart(
|
||||||
|
setup_measurement, config_eos, tmp_path, provider
|
||||||
|
):
|
||||||
|
m = setup_measurement()
|
||||||
|
folder, previous_db = config_eos.general.data_folder_path, config_eos.database.provider
|
||||||
|
try:
|
||||||
|
config_eos.general.data_folder_path = tmp_path
|
||||||
|
config_eos.database.provider = provider
|
||||||
|
(await write(
|
||||||
|
m,
|
||||||
|
"site",
|
||||||
|
[
|
||||||
|
(0, 800, {"status": "estimated", "generation": "meter1"}),
|
||||||
|
(450, None, {"status": "unavailable"}),
|
||||||
|
(900, 800, None),
|
||||||
|
],
|
||||||
|
))
|
||||||
|
before = (await m.energy_intervals("site", START, END))
|
||||||
|
assert (await m.save())
|
||||||
|
if provider:
|
||||||
|
(await get_database().close())
|
||||||
|
m._db_reset_state()
|
||||||
|
if not provider:
|
||||||
|
assert (await m.load())
|
||||||
|
# DB loads on demand via the bounded query, without a full-history load.
|
||||||
|
assert (await m.energy_intervals("site", START, END)) == before
|
||||||
|
assert m.records[0].sample_quality["site"].generation == "meter1"
|
||||||
|
finally:
|
||||||
|
if provider:
|
||||||
|
(await get_database().close())
|
||||||
|
m._db_reset_state()
|
||||||
|
config_eos.database.provider = previous_db
|
||||||
|
config_eos.general.data_folder_path = folder
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_query_passes_bounded_storage_window(setup_measurement, monkeypatch):
|
||||||
|
m = setup_measurement()
|
||||||
|
calls = []
|
||||||
|
original = type(m).db_iterate_records
|
||||||
|
|
||||||
|
def record_window(self, start_timestamp=None, end_timestamp=None):
|
||||||
|
calls.append((start_timestamp, end_timestamp))
|
||||||
|
return original(self, start_timestamp, end_timestamp)
|
||||||
|
|
||||||
|
monkeypatch.setattr(type(m), "db_iterate_records", record_window)
|
||||||
|
(await m.energy_intervals("site", START, END))
|
||||||
|
assert calls and all(a is not None and b is not None for a, b in calls)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_actual_server_routes_and_legacy_value_api(setup_measurement):
|
||||||
|
setup_measurement()
|
||||||
|
from akkudoktoreos.server.eos import app
|
||||||
|
|
||||||
|
# No lifespan: this test must not start schedulers or write application state.
|
||||||
|
client = TestClient(app)
|
||||||
|
try:
|
||||||
|
for time in (START, END):
|
||||||
|
response = client.put(
|
||||||
|
"/v1/measurement/value",
|
||||||
|
params={
|
||||||
|
"datetime": time.isoformat(),
|
||||||
|
"key": "site",
|
||||||
|
"value": 800,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert response.status_code == 200, response.text
|
||||||
|
response = client.get(
|
||||||
|
"/v1/measurement/energy",
|
||||||
|
params={
|
||||||
|
"key": "site",
|
||||||
|
"start": START.isoformat(),
|
||||||
|
"end": END.isoformat(),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert response.json()[0]["energy_wh"] == 200
|
||||||
|
assert "sample_quality" not in get_measurement().record_keys
|
||||||
|
finally:
|
||||||
|
client.close()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_quality_merge_keeps_other_channels(setup_measurement):
|
||||||
|
from akkudoktoreos.measurement.measurement import MeasurementDataRecord
|
||||||
|
|
||||||
|
m = setup_measurement(inputs=[dict(key=k, branch=k, role=k) for k in ("site", "ev")])
|
||||||
|
(await write(m, "site", [(0, 800, {"status": "estimated"})]))
|
||||||
|
(await write(m, "ev", [(0, None, {"status": "unavailable"})]))
|
||||||
|
(await m.insert_by_datetime(
|
||||||
|
MeasurementDataRecord(date_time=START, sample_quality={"site": {"status": "measured"}})
|
||||||
|
))
|
||||||
|
assert m.records[0].sample_quality["site"].status == "measured"
|
||||||
|
assert m.records[0].sample_quality["ev"].status == "unavailable"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_hour_energy_allocation_and_short_last_interval(setup_measurement):
|
||||||
|
m = setup_measurement(
|
||||||
|
channels={
|
||||||
|
"site": dict(
|
||||||
|
quantity="interval_energy",
|
||||||
|
unit="Wh",
|
||||||
|
interval_seconds=3600,
|
||||||
|
timestamp_reference="start",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
)
|
||||||
|
(await write(m, "site", [(0, 1000, None)]))
|
||||||
|
rows = (await m.household_intervals(START, START + timedelta(seconds=1800)))["site"]
|
||||||
|
assert [row.energy_wh for row in rows] == [250, 250]
|
||||||
|
assert all("allocated_energy" in row.methods for row in rows)
|
||||||
|
row = (await m.energy_intervals("site", START, START + timedelta(seconds=60)))[0]
|
||||||
|
assert row.coverage_seconds == 60
|
||||||
|
assert row.energy_wh == pytest.approx(1000 / 60)
|
||||||
Reference in New Issue
Block a user