mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-01-01 08:16:18 +00:00
58d70e417b
Adapters for Home Assistant and NodeRED integration are added. Akkudoktor-EOS can now be run as Home Assistant add-on and standalone. As Home Assistant add-on EOS uses ingress to fully integrate the EOSdash dashboard in Home Assistant. The fix includes several bug fixes that are not directly related to the adapter implementation but are necessary to keep EOS running properly and to test and document the changes. * fix: development version scheme The development versioning scheme is adaptet to fit to docker and home assistant expectations. The new scheme is x.y.z and x.y.z.dev<hash>. Hash is only digits as expected by home assistant. Development version is appended by .dev as expected by docker. * fix: use mean value in interval on resampling for array When downsampling data use the mean value of all values within the new sampling interval. * fix: default battery ev soc and appliance wh Make the genetic simulation return default values for the battery SoC, electric vehicle SoC and appliance load if these assets are not used. * fix: import json string Strip outer quotes from JSON strings on import to be compliant to json.loads() expectation. * fix: default interval definition for import data Default interval must be defined in lowercase human definition to be accepted by pendulum. * fix: clearoutside schema change * feat: add adapters for integrations Adapters for Home Assistant and NodeRED integration are added. Akkudoktor-EOS can now be run as Home Assistant add-on and standalone. As Home Assistant add-on EOS uses ingress to fully integrate the EOSdash dashboard in Home Assistant. * feat: allow eos to be started with root permissions and drop priviledges Home assistant starts all add-ons with root permissions. Eos now drops root permissions if an applicable user is defined by paramter --run_as_user. The docker image defines the user eos to be used. * feat: make eos supervise and monitor EOSdash Eos now not only starts EOSdash but also monitors EOSdash during runtime and restarts EOSdash on fault. EOSdash logging is captured by EOS and forwarded to the EOS log to provide better visibility. * feat: add duration to string conversion Make to_duration to also return the duration as string on request. * chore: Use info logging to report missing optimization parameters In parameter preparation for automatic optimization an error was logged for missing paramters. Log is now down using the info level. * chore: make EOSdash use the EOS data directory for file import/ export EOSdash use the EOS data directory for file import/ export by default. This allows to use the configuration import/ export function also within docker images. * chore: improve EOSdash config tab display Improve display of JSON code and add more forms for config value update. * chore: make docker image file system layout similar to home assistant Only use /data directory for persistent data. This is handled as a docker volume. The /data volume is mapped to ~/.local/share/net.akkudoktor.eos if using docker compose. * chore: add home assistant add-on development environment Add VSCode devcontainer and task definition for home assistant add-on development. * chore: improve documentation
366 lines
13 KiB
Python
366 lines
13 KiB
Python
from typing import Optional
|
|
|
|
import pytest
|
|
|
|
from akkudoktoreos.core.emplan import (
|
|
BaseInstruction,
|
|
CommodityQuantity,
|
|
DDBCInstruction,
|
|
EnergyManagementPlan,
|
|
FRBCInstruction,
|
|
OMBCInstruction,
|
|
PEBCInstruction,
|
|
PEBCPowerEnvelope,
|
|
PEBCPowerEnvelopeElement,
|
|
PPBCEndInterruptionInstruction,
|
|
PPBCScheduleInstruction,
|
|
PPBCStartInterruptionInstruction,
|
|
)
|
|
from akkudoktoreos.utils.datetimeutil import Duration, to_datetime, to_duration
|
|
|
|
|
|
@pytest.fixture
|
|
def fixed_now():
|
|
return to_datetime("2025-06-01T12:00:00")
|
|
|
|
|
|
class TestEnergyManagementPlan:
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Helpers (only used inside the class)
|
|
# ----------------------------------------------------------------------
|
|
def _make_instr(self, resource_id, execution_time, duration=None):
|
|
if duration is None:
|
|
instr = OMBCInstruction(
|
|
id=resource_id,
|
|
execution_time=execution_time,
|
|
operation_mode_id="mode",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
else:
|
|
instr = PEBCInstruction(
|
|
id=resource_id,
|
|
execution_time=execution_time,
|
|
power_constraints_id="pc-123",
|
|
power_envelopes=[
|
|
PEBCPowerEnvelope(
|
|
id="pebcpe@1234",
|
|
commodity_quantity=CommodityQuantity.ELECTRIC_POWER_L1,
|
|
power_envelope_elements=[
|
|
PEBCPowerEnvelopeElement(
|
|
duration=to_duration(duration),
|
|
upper_limit=1010.0,
|
|
lower_limit=990.0,
|
|
),
|
|
],
|
|
),
|
|
],
|
|
)
|
|
|
|
return instr
|
|
|
|
def _build_plan(self, instructions, now):
|
|
plan = EnergyManagementPlan(
|
|
id="plan-test",
|
|
generated_at=now,
|
|
instructions=instructions,
|
|
)
|
|
plan._update_time_range()
|
|
return plan
|
|
|
|
def test_add_instruction_and_time_range(self, fixed_now):
|
|
plan = EnergyManagementPlan(
|
|
id="plan-123",
|
|
generated_at=fixed_now,
|
|
instructions=[]
|
|
)
|
|
instr1 = OMBCInstruction(
|
|
resource_id="dev-1",
|
|
execution_time=fixed_now,
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
instr2 = OMBCInstruction(
|
|
resource_id="dev-2",
|
|
execution_time=fixed_now.add(minutes=5),
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
plan.add_instruction(instr1)
|
|
plan.add_instruction(instr2)
|
|
|
|
assert plan.valid_from == fixed_now
|
|
assert plan.valid_until is None
|
|
assert plan.instructions == [instr1, instr2]
|
|
|
|
def test_clear(self, fixed_now):
|
|
plan = EnergyManagementPlan(
|
|
id="plan-123",
|
|
generated_at=fixed_now,
|
|
instructions=[
|
|
OMBCInstruction(
|
|
resource_id="dev-1",
|
|
execution_time=fixed_now,
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
],
|
|
)
|
|
plan.clear()
|
|
assert plan.instructions == []
|
|
assert plan.valid_until is None
|
|
assert plan.valid_from is not None
|
|
|
|
def test_get_next_instruction(self, fixed_now):
|
|
instr1 = OMBCInstruction(
|
|
resource_id="dev-1",
|
|
execution_time=fixed_now.subtract(minutes=1),
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
instr2 = OMBCInstruction(
|
|
resource_id="dev-2",
|
|
execution_time=fixed_now.add(minutes=10),
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
instr3 = OMBCInstruction(
|
|
resource_id="dev-3",
|
|
execution_time=fixed_now.add(minutes=5),
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
plan = EnergyManagementPlan(
|
|
id="plan-123",
|
|
generated_at=fixed_now,
|
|
instructions=[instr1, instr2, instr3],
|
|
)
|
|
plan._update_time_range()
|
|
|
|
next_instr = plan.get_next_instruction(now=fixed_now)
|
|
assert next_instr is not None
|
|
assert next_instr.resource_id == "dev-3"
|
|
|
|
def test_get_instructions_for_resource(self, fixed_now):
|
|
instr1 = OMBCInstruction(
|
|
resource_id="dev-1",
|
|
execution_time=fixed_now,
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
instr2 = OMBCInstruction(
|
|
resource_id="dev-2",
|
|
execution_time=fixed_now,
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
plan = EnergyManagementPlan(
|
|
id="plan-123",
|
|
generated_at=fixed_now,
|
|
instructions=[instr1, instr2],
|
|
)
|
|
dev1_instructions = plan.get_instructions_for_resource("dev-1")
|
|
assert len(dev1_instructions) == 1
|
|
assert dev1_instructions[0].resource_id == "dev-1"
|
|
|
|
def test_add_various_instructions(self, fixed_now):
|
|
plan = EnergyManagementPlan(
|
|
id="plan-123",
|
|
generated_at=fixed_now,
|
|
instructions=[]
|
|
)
|
|
instrs = [
|
|
DDBCInstruction(
|
|
id="actuatorA@123",
|
|
execution_time=fixed_now,
|
|
actuator_id="actuatorA",
|
|
operation_mode_id="mode123",
|
|
operation_mode_factor=0.5,
|
|
),
|
|
FRBCInstruction(
|
|
id="actuatorB@456",
|
|
execution_time=fixed_now.add(minutes=1),
|
|
actuator_id="actuatorB",
|
|
operation_mode_id="FRBC_Mode_1",
|
|
operation_mode_factor=1.0,
|
|
),
|
|
OMBCInstruction(
|
|
id="controller@789",
|
|
execution_time=fixed_now.add(minutes=2),
|
|
operation_mode_id="OMBC_Mode_42",
|
|
operation_mode_factor=0.8,
|
|
),
|
|
PPBCEndInterruptionInstruction(
|
|
id="end_int@001",
|
|
execution_time=fixed_now.add(minutes=3),
|
|
power_profile_id="profile-123",
|
|
sequence_container_id="container-456",
|
|
power_sequence_id="seq-789",
|
|
),
|
|
PPBCStartInterruptionInstruction(
|
|
id="start_int@002",
|
|
execution_time=fixed_now.add(minutes=4),
|
|
power_profile_id="profile-321",
|
|
sequence_container_id="container-654",
|
|
power_sequence_id="seq-987",
|
|
),
|
|
PPBCScheduleInstruction(
|
|
id="schedule@003",
|
|
execution_time=fixed_now.add(minutes=5),
|
|
power_profile_id="profile-999",
|
|
sequence_container_id="container-888",
|
|
power_sequence_id="seq-777",
|
|
),
|
|
PEBCInstruction(
|
|
id="pebc@004",
|
|
execution_time=fixed_now.add(minutes=6),
|
|
power_constraints_id="pc-123",
|
|
power_envelopes=[
|
|
PEBCPowerEnvelope(
|
|
id="pebcpe@1234",
|
|
commodity_quantity=CommodityQuantity.ELECTRIC_POWER_L1,
|
|
power_envelope_elements=[
|
|
PEBCPowerEnvelopeElement(
|
|
duration=to_duration(10),
|
|
upper_limit=1010.0,
|
|
lower_limit=990.0,
|
|
),
|
|
],
|
|
),
|
|
],
|
|
),
|
|
]
|
|
|
|
for instr in instrs:
|
|
plan.add_instruction(instr)
|
|
|
|
assert len(plan.instructions) == len(instrs)
|
|
assert any(
|
|
instr for instr in plan.get_instructions_for_resource("actuatorA")
|
|
if isinstance(instr, DDBCInstruction)
|
|
)
|
|
|
|
# -------------------------------------------
|
|
# Special testing for get_active_instructions
|
|
# -------------------------------------------
|
|
|
|
def test_get_active_instructions(self, fixed_now):
|
|
instr1 = OMBCInstruction(
|
|
resource_id="dev-1",
|
|
execution_time=fixed_now.subtract(minutes=1),
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
instr2 = OMBCInstruction(
|
|
resource_id="dev-2",
|
|
execution_time=fixed_now.add(minutes=1),
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
instr3 = OMBCInstruction(
|
|
resource_id="dev-3",
|
|
execution_time=fixed_now.subtract(minutes=10),
|
|
operation_mode_id="mymode1",
|
|
operation_mode_factor=1.0,
|
|
)
|
|
plan = EnergyManagementPlan(
|
|
id="plan-123",
|
|
generated_at=fixed_now,
|
|
instructions=[instr1, instr2, instr3],
|
|
)
|
|
plan._update_time_range()
|
|
|
|
resource_ids = plan.get_resources()
|
|
assert resource_ids == ["dev-1", "dev-2", "dev-3"]
|
|
|
|
active = plan.get_active_instructions(now=fixed_now)
|
|
ids = {i.resource_id for i in active}
|
|
assert ids == {"dev-1", "dev-3"}
|
|
|
|
|
|
|
|
def test_get_active_instructions_with_duration(self, fixed_now):
|
|
instr = self._make_instr(
|
|
"dev-1",
|
|
fixed_now.subtract(minutes=5),
|
|
duration=Duration(minutes=10),
|
|
)
|
|
plan = self._build_plan([instr], fixed_now)
|
|
active = plan.get_active_instructions(fixed_now)
|
|
assert {i.resource_id for i in active} == {"dev-1"}
|
|
|
|
def test_get_active_instructions_expired_duration(self, fixed_now):
|
|
instr = self._make_instr(
|
|
"dev-1",
|
|
fixed_now.subtract(minutes=20),
|
|
duration=Duration(minutes=10),
|
|
)
|
|
plan = self._build_plan([instr], fixed_now)
|
|
assert plan.get_active_instructions(fixed_now) == []
|
|
|
|
def test_get_active_instructions_end_exactly_now_not_active(self, fixed_now):
|
|
instr = self._make_instr(
|
|
"dev-1",
|
|
fixed_now.subtract(minutes=10),
|
|
duration=Duration(minutes=10),
|
|
)
|
|
plan = self._build_plan([instr], fixed_now)
|
|
assert plan.get_active_instructions(fixed_now) == []
|
|
|
|
def test_get_active_instructions_latest_supersedes(self, fixed_now):
|
|
instr1 = self._make_instr(
|
|
"dev-1",
|
|
fixed_now.subtract(minutes=10),
|
|
duration=Duration(minutes=30),
|
|
)
|
|
instr2 = self._make_instr("dev-1", fixed_now.subtract(minutes=1))
|
|
plan = self._build_plan([instr1, instr2], fixed_now)
|
|
|
|
active = plan.get_active_instructions(fixed_now)
|
|
assert len(active) == 1
|
|
assert active[0] is instr2
|
|
|
|
def test_get_active_instructions_mixed_resources(self, fixed_now):
|
|
instr1 = self._make_instr(
|
|
"r1",
|
|
fixed_now.subtract(minutes=5),
|
|
duration=Duration(minutes=10),
|
|
)
|
|
instr2 = self._make_instr("r2", fixed_now.subtract(minutes=1))
|
|
instr3 = self._make_instr("r3", fixed_now.add(minutes=10))
|
|
plan = self._build_plan([instr1, instr2, instr3], fixed_now)
|
|
|
|
ids = {i.resource_id for i in plan.get_active_instructions(fixed_now)}
|
|
assert ids == {"r1", "r2"}
|
|
|
|
def test_get_active_instructions_start_exactly_now(self, fixed_now):
|
|
instr = self._make_instr("dev-1", fixed_now)
|
|
plan = self._build_plan([instr], fixed_now)
|
|
assert {i.resource_id for i in plan.get_active_instructions(fixed_now)} == {"dev-1"}
|
|
|
|
def test_get_active_instructions_no_active(self, fixed_now):
|
|
instr = self._make_instr("dev-1", fixed_now.add(minutes=1))
|
|
plan = self._build_plan([instr], fixed_now)
|
|
assert plan.get_active_instructions(fixed_now) == []
|
|
|
|
def test_get_active_instructions_future_does_not_override_until_reached(self, fixed_now):
|
|
instr1 = self._make_instr("dev-1", fixed_now.subtract(minutes=5))
|
|
instr2 = self._make_instr("dev-1", fixed_now.add(minutes=5))
|
|
plan = self._build_plan([instr1, instr2], fixed_now)
|
|
|
|
active_before = plan.get_active_instructions(fixed_now)
|
|
assert {i.resource_id for i in active_before} == {"dev-1"}
|
|
|
|
def test_get_active_instructions_future_overrides_once_time_reached(self, fixed_now):
|
|
exec_future = fixed_now.add(minutes=5)
|
|
instr1 = self._make_instr("dev-1", fixed_now.subtract(minutes=5))
|
|
instr2 = self._make_instr("dev-1", exec_future)
|
|
|
|
plan = self._build_plan([instr1, instr2], fixed_now)
|
|
|
|
active_before = plan.get_active_instructions(fixed_now)
|
|
assert {i.resource_id for i in active_before} == {"dev-1"}
|
|
|
|
active_after = plan.get_active_instructions(exec_future)
|
|
assert {i.resource_id for i in active_after} == {"dev-1"}
|