| Number of Cars Charging |
diff --git a/lib/TWCManager/Policy/Policy.py b/lib/TWCManager/Policy/Policy.py
index 5d2b3254..3f207633 100644
--- a/lib/TWCManager/Policy/Policy.py
+++ b/lib/TWCManager/Policy/Policy.py
@@ -237,6 +237,10 @@ def enforcePolicy(self, policy, updateLatch=False):
limit = -1
self.master.queue_background_task({"cmd": "applyChargeLimit", "limit": limit})
+ # If at least one pricing module is active, fetch current pricing
+ if len(self.master.getModulesByType("Pricing")) > 0:
+ self.master.queue_background_task({"cmd": "getPricing"})
+
# Report current policy via Status modules
for module in self.master.getModulesByType("Status"):
module["ref"].setStatus(
@@ -308,6 +312,10 @@ def policyValue(self, value):
return self.master.getMaxAmpsForTargetGridUsage()
elif value == "checkScheduledCharging()":
return self.master.checkScheduledCharging()
+ elif value == "getImportPrice()":
+ return self.master.getImportPrice()
+ elif value == "getExportPrice()":
+ return self.master.getExportPrice()
# If value is tiered, split it up
if value.find(".") != -1:
diff --git a/lib/TWCManager/Pricing/PVPCesPricing.py b/lib/TWCManager/Pricing/PVPCesPricing.py
new file mode 100644
index 00000000..49429b23
--- /dev/null
+++ b/lib/TWCManager/Pricing/PVPCesPricing.py
@@ -0,0 +1,210 @@
+from datetime import datetime
+from datetime import timedelta
+
+
+class PVPCesPricing:
+
+ import requests
+ import time
+
+ # https://www.esios.ree.es/es/pvpc publishes at 20:30CET eveyday the prices for next day
+ # Prices are fetched once per day; the cache expires at midnight (detected via lastFetch.hour)
+ capabilities = {
+ "AdvancePricing": True
+ }
+ config = None
+ configConfig = None
+ configPvpc = None
+ exportPrice = 0
+ fetchFailed = False
+ importPrice = 0
+ lastFetch = 0
+ status = False
+ timeout = 10
+ headers = {}
+ todayImportPrice = {}
+
+ def __init__(self, master):
+
+ self.master = master
+ self.config = master.config
+ try:
+ self.configConfig = master.config["config"]
+ except KeyError:
+ self.configConfig = {}
+
+ try:
+ self.configPvpc = master.config["pricing"]["PVPCes"]
+ except KeyError:
+ self.configPvpc = {}
+
+ self.status = self.configPvpc.get("enabled", self.status)
+ self.debugLevel = self.configConfig.get("debugLevel", 0)
+
+ token = self.configPvpc.get("token")
+ if self.status:
+ self.headers = {
+ "Accept": "application/json; application/vnd.esios-api-v1+json",
+ "Content-Type": "application/json",
+ "Host": "api.esios.ree.es",
+ "Cookie": "",
+ }
+ self.headers["Authorization"] = "Token token=" + token
+
+ # Unload if this module is disabled or misconfigured
+ if not self.status:
+ self.master.releaseModule("lib.TWCManager.Pricing", self.__class__.__name__)
+ return None
+
+ def getCapabilities(self, capability):
+ # Allows query of module capabilities
+ return self.capabilities.get(capability, False)
+
+ def getExportPrice(self):
+
+ if not self.status:
+ self.master.debugLog(
+ 10,
+ "$PVPCes",
+ "PVPCes Pricing Module Disabled. Skipping getExportPrice",
+ )
+ return 0
+
+ # Perform updates if necessary
+ self.update()
+
+ # Return current export price
+ return float(self.exportPrice)
+
+ def getImportPrice(self):
+
+ if not self.status:
+ self.master.debugLog(
+ 10,
+ "$PVPCes",
+ "PVPCes Pricing Module Disabled. Skipping getImportPrice",
+ )
+ return 0
+
+ # Perform updates if necessary
+ self.update()
+
+ # Return current import price
+ return float(self.importPrice)
+
+ def update(self):
+
+ # Fetch the current pricing data from the https://www.esios.ree.es/es/pvpc API
+ self.fetchFailed = False
+ now = datetime.now()
+ tomorrow = datetime.now() + timedelta(days=1)
+ if self.lastFetch == 0 or (now.hour < self.lastFetch.hour):
+ # Cache not fetched or was fetched yesterday. Fetch values from API.
+ ini = (
+ str(now.year)
+ + "-"
+ + str(now.month)
+ + "-"
+ + str(now.day)
+ + "T"
+ + "00:00:00"
+ )
+ end = (
+ str(tomorrow.year)
+ + "-"
+ + str(tomorrow.month)
+ + "-"
+ + str(tomorrow.day)
+ + "T"
+ + "23:00:00"
+ )
+
+ url = (
+ "https://api.esios.ree.es/indicators/1014?start_date="
+ + ini
+ + "&end_date="
+ + end
+ )
+
+ try:
+ r = self.requests.get(url, headers=self.headers, timeout=self.timeout)
+ except self.requests.exceptions.ConnectionError as e:
+ self.master.debugLog(
+ 4,
+ "$PVPCes",
+ "Error connecting to PVPCes API to fetch market pricing",
+ )
+ self.fetchFailed = True
+ return False
+
+ self.lastFetch = now
+
+ try:
+ r.raise_for_status()
+ except self.requests.exceptions.HTTPError as e:
+ self.master.debugLog(
+ 4,
+ "$PVPCes",
+ "HTTP status "
+ + str(e.response.status_code)
+ + " connecting to PVPCes API to fetch market pricing",
+ )
+ return False
+
+ if r.json():
+ self.todayImportPrice = r.json()
+
+ if self.todayImportPrice:
+ try:
+ self.importPrice = float(
+ self.todayImportPrice["indicator"]["values"][now.hour]["value"]
+ )
+ # Convert MWh price to KWh
+ self.importPrice = round(self.importPrice / 1000, 5)
+
+ except (KeyError, TypeError) as e:
+ self.master.debugLog(
+ 4,
+ "$PVPCes",
+ "Exception during parsing PVPCes pricing",
+ )
+
+ def getCheapestStartHour(self, numHours, ini, end):
+ # Perform updates if necessary
+ self.update()
+
+ minPriceHstart = ini
+ if self.todayImportPrice:
+ try:
+ if end < ini:
+ # If the scheduled hours are between days we consider hours going from 0 to 47
+ # tomorrow 1am will be 25
+ end = 24 + end
+
+ i = ini
+ minPrice = 999999999
+ while i <= (end - numHours):
+ j = 0
+ priceH = 0
+ while j < numHours:
+ price = float(
+ self.todayImportPrice["indicator"]["values"][i + j]["value"]
+ )
+ priceH = priceH + price
+ j = j + 1
+ if priceH < minPrice:
+ minPrice = priceH
+ minPriceHstart = i
+ i = i + 1
+
+ except (KeyError, TypeError) as e:
+ self.master.debugLog(
+ 4,
+ "$PVPCes",
+ "Exception during cheaper pricing analysis",
+ )
+
+ if minPriceHstart > 23:
+ minPriceHstart = minPriceHstart - 24
+
+ return minPriceHstart
diff --git a/lib/TWCManager/Pricing/StaticPricing.py b/lib/TWCManager/Pricing/StaticPricing.py
new file mode 100644
index 00000000..eee1afa9
--- /dev/null
+++ b/lib/TWCManager/Pricing/StaticPricing.py
@@ -0,0 +1,81 @@
+class StaticPricing:
+
+ # For environments where dynamic pricing information is not available via an
+ # API, the pricing information can be configured statically within the config
+ # file
+
+ import time
+
+ capabilities = {
+ "AdvancePricing": True
+ }
+ config = None
+ configConfig = None
+ configStatic = None
+ exportPrice = 0
+ importPrice = 0
+ status = False
+
+ def __init__(self, master):
+
+ self.master = master
+ self.config = master.config
+ try:
+ self.configConfig = master.config["config"]
+ except KeyError:
+ self.configConfig = {}
+
+ try:
+ self.configStatic = master.config["pricing"]["Static"]
+ except KeyError:
+ self.configStatic = {}
+
+ self.status = self.configStatic.get("enabled", self.status)
+ self.debugLevel = self.configConfig.get("debugLevel", 0)
+
+ # Unload if this module is disabled or misconfigured
+ if not self.status:
+ self.master.releaseModule("lib.TWCManager.Pricing", self.__class__.__name__)
+ return None
+
+ def getCapabilities(self, capability):
+ # Allows query of module capabilities
+ return self.capabilities.get(capability, False)
+
+ def getExportPrice(self):
+
+ if not self.status:
+ self.master.debugLog(
+ 10,
+ "$Static",
+ "Static Pricing Module Disabled. Skipping getExportPrice",
+ )
+ return 0
+
+ # For now, we just use the peak price
+ try:
+ self.exportPrice = self.configStatic["peak"]["export"]
+ except ValueError:
+ self.exportPrice = 0
+
+ # Return current export price
+ return float(self.exportPrice)
+
+ def getImportPrice(self):
+
+ if not self.status:
+ self.master.debugLog(
+ 10,
+ "$Static",
+ "Static Pricing Module Disabled. Skipping getImportPrice",
+ )
+ return 0
+
+ # For now, we just use the peak price
+ try:
+ self.importPrice = self.configStatic["peak"]["import"]
+ except ValueError:
+ self.exportPrice = 0
+
+ # Return current import price
+ return float(self.importPrice)
diff --git a/lib/TWCManager/Pricing/__init__.py b/lib/TWCManager/Pricing/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/lib/TWCManager/Pricing/aWATTarPricing.py b/lib/TWCManager/Pricing/aWATTarPricing.py
new file mode 100644
index 00000000..b671aa68
--- /dev/null
+++ b/lib/TWCManager/Pricing/aWATTarPricing.py
@@ -0,0 +1,130 @@
+class aWATTarPricing:
+
+ import requests
+ import time
+
+ # aWATTar asks us to limit queries to once every 15 minutes
+ cacheTime = 900
+ capabilities = {
+ "AdvancePricing": True
+ }
+ config = None
+ configConfig = None
+ configAwattar = None
+ exportPrice = 0
+ fetchFailed = False
+ importPrice = 0
+ lastFetch = 0
+ status = False
+ timeout = 10
+
+ def __init__(self, master):
+
+ self.master = master
+ self.config = master.config
+ try:
+ self.configConfig = master.config["config"]
+ except KeyError:
+ self.configConfig = {}
+
+ try:
+ self.configAwattar = master.config["pricing"]["aWATTar"]
+ except KeyError:
+ self.configAwattar = {}
+
+ self.status = self.configAwattar.get("enabled", self.status)
+ self.debugLevel = self.configConfig.get("debugLevel", 0)
+
+ # Unload if this module is disabled or misconfigured
+ if not self.status:
+ self.master.releaseModule("lib.TWCManager.Pricing", self.__class__.__name__)
+ return None
+
+ def getCapabilities(self, capability):
+ # Allows query of module capabilities
+ return self.capabilities.get(capability, False)
+
+ def getExportPrice(self):
+
+ if not self.status:
+ self.master.debugLog(
+ 10,
+ "$aWATTar",
+ "aWATTar Pricing Module Disabled. Skipping getExportPrice",
+ )
+ return 0
+
+ # Perform updates if necessary
+ self.update()
+
+ # Return current export price
+ return float(self.exportPrice)
+
+ def getImportPrice(self):
+
+ if not self.status:
+ self.master.debugLog(
+ 10,
+ "$aWATTar",
+ "aWATTar Pricing Module Disabled. Skipping getImportPrice",
+ )
+ return 0
+
+ # Perform updates if necessary
+ self.update()
+
+ # Return current import price
+ return float(self.importPrice)
+
+ def update(self):
+
+ # Fetch the current pricing data from the Awattar API
+ self.fetchFailed = False
+
+ if (int(self.time.time()) - self.lastFetch) > self.cacheTime:
+ # Cache has expired. Fetch values from API.
+
+ url = "https://api.awattar.at/v1/marketdata"
+
+ try:
+ r = self.requests.get(url, timeout=self.timeout)
+ except self.requests.exceptions.ConnectionError as e:
+ self.master.debugLog(
+ 4,
+ "$aWATTar",
+ "Error connecting to aWATTar API to fetch market pricing",
+ )
+ self.master.debugLog(10, "$aWATTar", str(e))
+ self.fetchFailed = True
+ return False
+
+ # Update lastFetch regardless of parse outcome so we respect the
+ # 15-minute rate limit even when the response is unexpected.
+ self.lastFetch = int(self.time.time())
+
+ try:
+ r.raise_for_status()
+ except self.requests.exceptions.HTTPError as e:
+ self.master.debugLog(
+ 4,
+ "$aWATTar",
+ "HTTP status "
+ + str(e.response.status_code)
+ + " connecting to aWATTar API to fetch market pricing",
+ )
+
+ if r.json():
+ try:
+ self.importPrice = float(
+ r.json()["data"][0]["marketprice"]
+ )
+ if r.json()["data"][0]["unit"] == "Eur/MWh":
+ # Convert MWh price to KWh
+ self.importPrice = self.importPrice / 1000
+
+ except (KeyError, TypeError) as e:
+ self.master.debugLog(
+ 4,
+ "$aWATTar",
+ "Exception during parsing aWATTar pricing",
+ )
diff --git a/lib/TWCManager/TWCManager.py b/lib/TWCManager/TWCManager.py
index dc2e7ad9..b9a494b6 100755
--- a/lib/TWCManager/TWCManager.py
+++ b/lib/TWCManager/TWCManager.py
@@ -114,6 +114,9 @@
"EMS.URL",
"Status.HASSStatus",
"Status.MQTTStatus",
+ "Pricing.aWATTarPricing",
+ "Pricing.PVPCesPricing",
+ "Pricing.StaticPricing",
]
# Enable support for Python Visual Studio Debugger
@@ -337,6 +340,8 @@ def background_tasks_thread(master):
elif task["cmd"] == "getLifetimekWh":
master.getSlaveLifetimekWh()
+ elif task["cmd"] == "getPricing":
+ master.getPricing()
elif task["cmd"] == "getVehicleVIN":
master.getVehicleVIN(task["slaveTWC"], task["vinPart"])
elif task["cmd"] == "snapHistoryData":
diff --git a/lib/TWCManager/TWCMaster.py b/lib/TWCManager/TWCMaster.py
index 1a45257b..7ee8937a 100644
--- a/lib/TWCManager/TWCMaster.py
+++ b/lib/TWCManager/TWCMaster.py
@@ -28,7 +28,9 @@ class TWCMaster:
consumptionValues = {}
consumptionAmpsValues = {}
debugOutputToFile = False
+ exportPricingValues = {}
generationValues = {}
+ importPricingValues = {}
lastkWhMessage = time.time()
lastkWhPoll = 0
lastSaveFailed = 0
@@ -416,6 +418,55 @@ def getModulesByType(self, type):
def getInterfaceModule(self):
return self.getModulesByType("Interface")[0]["ref"]
+ def getPricing(self):
+ for module in self.getModulesByType("Pricing"):
+ self.exportPricingValues[module["name"]] = module["ref"].getExportPrice()
+ self.importPricingValues[module["name"]] = module["ref"].getImportPrice()
+
+ def getExportPrice(self):
+ price = 0
+ multiPrice = ""
+
+ # The policy for how we should deal with multiple export
+ # prices (multiple concurrent modules) is defined in the config
+ # file in config->pricing->policy->multiPrice
+ try:
+ multiPrice = self.config["config"]["pricing"]["policy"]["multiPrice"]
+ except KeyError:
+ multiPrice = "first"
+
+ # Iterate through values and apply multiPrice policy
+ for key in self.exportPricingValues:
+ if multiPrice == "add":
+ price += float(self.exportPricingValues[key])
+ elif multiPrice == "first":
+ if price == 0 and self.exportPricingValues[key] > 0:
+ price = float(self.exportPricingValues[key])
+
+ return float(price)
+
+ def getImportPrice(self):
+ price = 0
+ multiPrice = ""
+
+ # The policy for how we should deal with multiple import
+ # prices (multiple concurrent modules) is defined in the config
+ # file in config->pricing->policy->multiPrice
+ try:
+ multiPrice = self.config["config"]["pricing"]["policy"]["multiPrice"]
+ except KeyError:
+ multiPrice = "first"
+
+ # Iterate through values and apply multiPrice policy
+ for key in self.importPricingValues:
+ if multiPrice == "add":
+ price += float(self.importPricingValues[key])
+ elif multiPrice == "first":
+ if price == 0 and self.importPricingValues[key] > 0:
+ price = float(self.importPricingValues[key])
+
+ return float(price)
+
def getScheduledAmpsDaysBitmap(self):
return self.settings.get("scheduledAmpsDaysBitmap", 0x7F)
diff --git a/tests/scenario_runner.py b/tests/scenario_runner.py
index 835589e7..ea0149e7 100644
--- a/tests/scenario_runner.py
+++ b/tests/scenario_runner.py
@@ -24,7 +24,7 @@ def test_green_energy_basic(run_scenario):
import time
from dataclasses import dataclass, field
from pathlib import Path
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Tuple
import pytest
import requests
@@ -158,7 +158,7 @@ def _resolve_path(data: Any, path: str) -> Any:
return current
-def _evaluate_check(check: dict, api_responses: dict, api_get_fn) -> tuple[bool, str]:
+def _evaluate_check(check: dict, api_responses: dict, api_get_fn) -> Tuple[bool, str]:
"""
Evaluate a single check dict. Returns (passed, detail_string).
@@ -213,7 +213,7 @@ def _evaluate_check(check: dict, api_responses: dict, api_get_fn) -> tuple[bool,
return passed, detail
-def _wait_for_assertion(assertion: dict, api_get_fn, timeout: float) -> tuple[bool, str, dict]:
+def _wait_for_assertion(assertion: dict, api_get_fn, timeout: float) -> Tuple[bool, str, dict]:
"""
Poll the API until all checks in the assertion pass or timeout expires.
Returns (passed, detail, api_responses_snapshot).
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
new file mode 100644
index 00000000..20169f58
--- /dev/null
+++ b/tests/unit/conftest.py
@@ -0,0 +1,33 @@
+"""
+Unit test configuration.
+
+The project uses namespace packages (find_namespace_packages in setup.py), so
+lib/TWCManager/ has no __init__.py. When running tests from the project root
+without an installed package, Python finds TWCManager.py (the entry-point
+launcher) before the lib/TWCManager/ namespace package, causing the launcher
+to execute and fail (it tries to drop privileges to a 'twcmanager' OS user).
+
+Fix: pre-register TWCManager in sys.modules as a package pointing at
+lib/TWCManager/ so that subsequent `from TWCManager.*` imports resolve
+correctly without ever executing the launcher script.
+"""
+
+import os
+import sys
+import types
+
+_lib_path = os.path.abspath(
+ os.path.join(os.path.dirname(__file__), "..", "..", "lib")
+)
+
+# Insert lib/ so sub-package imports are resolvable.
+if _lib_path not in sys.path:
+ sys.path.insert(0, _lib_path)
+
+# Pre-register the top-level namespace so Python never falls back to
+# executing TWCManager.py (the root launcher script).
+if "TWCManager" not in sys.modules:
+ _pkg = types.ModuleType("TWCManager")
+ _pkg.__path__ = [os.path.join(_lib_path, "TWCManager")]
+ _pkg.__package__ = "TWCManager"
+ sys.modules["TWCManager"] = _pkg
diff --git a/tests/unit/test_pricing.py b/tests/unit/test_pricing.py
new file mode 100644
index 00000000..c8001352
--- /dev/null
+++ b/tests/unit/test_pricing.py
@@ -0,0 +1,395 @@
+"""
+Unit tests for TWCManager Pricing modules.
+
+Tests StaticPricing, aWATTarPricing, and PVPCesPricing modules.
+"""
+
+import os
+import sys
+
+# Ensure lib/ takes precedence over the root TWCManager.py script so that
+# `TWCManager.*` imports resolve to the package in lib/TWCManager/ rather
+# than the top-level entry-point script.
+_lib_path = os.path.join(os.path.dirname(__file__), "..", "..", "lib")
+if _lib_path not in sys.path:
+ sys.path.insert(0, _lib_path)
+
+import pytest
+import time
+from unittest.mock import Mock, MagicMock, patch
+
+
+# ---------------------------------------------------------------------------
+# Shared fixtures
+# ---------------------------------------------------------------------------
+
+
+def make_master(pricing_config=None):
+ """Return a mock master with a minimal config."""
+ master = Mock()
+ master.config = {
+ "config": {"debugLevel": 0},
+ "pricing": pricing_config or {},
+ }
+ master.releaseModule = Mock()
+ master.debugLog = Mock()
+ return master
+
+
+# ---------------------------------------------------------------------------
+# StaticPricing
+# ---------------------------------------------------------------------------
+
+
+class TestStaticPricing:
+ """Tests for the StaticPricing module."""
+
+ @pytest.fixture
+ def master_enabled(self):
+ return make_master(
+ {
+ "Static": {
+ "enabled": True,
+ "peak": {"import": 0.25, "export": 0.10},
+ }
+ }
+ )
+
+ @pytest.fixture
+ def master_disabled(self):
+ return make_master({"Static": {"enabled": False}})
+
+ @pytest.fixture
+ def master_no_config(self):
+ return make_master({})
+
+ def test_import_price_returned(self, master_enabled):
+ from TWCManager.Pricing.StaticPricing import StaticPricing
+
+ module = StaticPricing(master_enabled)
+ assert module.getImportPrice() == pytest.approx(0.25)
+
+ def test_export_price_returned(self, master_enabled):
+ from TWCManager.Pricing.StaticPricing import StaticPricing
+
+ module = StaticPricing(master_enabled)
+ assert module.getExportPrice() == pytest.approx(0.10)
+
+ def test_disabled_module_unloads(self, master_disabled):
+ from TWCManager.Pricing.StaticPricing import StaticPricing
+
+ StaticPricing(master_disabled)
+ master_disabled.releaseModule.assert_called_once()
+
+ def test_disabled_import_returns_zero(self, master_disabled):
+ from TWCManager.Pricing.StaticPricing import StaticPricing
+
+ module = StaticPricing(master_disabled)
+ assert module.getImportPrice() == 0
+
+ def test_disabled_export_returns_zero(self, master_disabled):
+ from TWCManager.Pricing.StaticPricing import StaticPricing
+
+ module = StaticPricing(master_disabled)
+ assert module.getExportPrice() == 0
+
+ def test_missing_config_unloads(self, master_no_config):
+ from TWCManager.Pricing.StaticPricing import StaticPricing
+
+ StaticPricing(master_no_config)
+ master_no_config.releaseModule.assert_called_once()
+
+ def test_get_capabilities(self, master_enabled):
+ from TWCManager.Pricing.StaticPricing import StaticPricing
+
+ module = StaticPricing(master_enabled)
+ assert module.getCapabilities("AdvancePricing") is True
+ assert module.getCapabilities("NonExistent") is False
+
+
+# ---------------------------------------------------------------------------
+# aWATTarPricing
+# ---------------------------------------------------------------------------
+
+
+class TestAWATTarPricing:
+ """Tests for the aWATTarPricing module."""
+
+ @pytest.fixture
+ def master_enabled(self):
+ return make_master({"aWATTar": {"enabled": True}})
+
+ @pytest.fixture
+ def master_disabled(self):
+ return make_master({"aWATTar": {"enabled": False}})
+
+ def test_disabled_module_unloads(self, master_disabled):
+ from TWCManager.Pricing.aWATTarPricing import aWATTarPricing
+
+ aWATTarPricing(master_disabled)
+ master_disabled.releaseModule.assert_called_once()
+
+ def test_disabled_import_returns_zero(self, master_disabled):
+ from TWCManager.Pricing.aWATTarPricing import aWATTarPricing
+
+ module = aWATTarPricing(master_disabled)
+ assert module.getImportPrice() == 0
+
+ def test_disabled_export_returns_zero(self, master_disabled):
+ from TWCManager.Pricing.aWATTarPricing import aWATTarPricing
+
+ module = aWATTarPricing(master_disabled)
+ assert module.getExportPrice() == 0
+
+ def test_cache_respected(self, master_enabled):
+ """After a successful fetch, lastFetch should be set so the cache is active."""
+ from TWCManager.Pricing.aWATTarPricing import aWATTarPricing
+
+ module = aWATTarPricing(master_enabled)
+
+ mock_response = Mock()
+ mock_response.json.return_value = {
+ "data": [{"marketprice": 150.0, "unit": "Eur/MWh"}]
+ }
+
+ with patch.object(module.requests, "get", return_value=mock_response) as mock_get:
+ module.update()
+ first_fetch_time = module.lastFetch
+ assert first_fetch_time > 0
+
+ # Second call should not hit the network (cache still valid)
+ module.update()
+ assert mock_get.call_count == 1
+
+ def test_mwh_to_kwh_conversion(self, master_enabled):
+ """Prices returned in Eur/MWh must be divided by 1000."""
+ from TWCManager.Pricing.aWATTarPricing import aWATTarPricing
+
+ module = aWATTarPricing(master_enabled)
+
+ mock_response = Mock()
+ mock_response.json.return_value = {
+ "data": [{"marketprice": 200.0, "unit": "Eur/MWh"}]
+ }
+
+ with patch.object(module.requests, "get", return_value=mock_response):
+ module.update()
+
+ assert module.importPrice == pytest.approx(0.20)
+
+ def test_connection_error_sets_fetch_failed(self, master_enabled):
+ from TWCManager.Pricing.aWATTarPricing import aWATTarPricing
+ import requests as req
+
+ module = aWATTarPricing(master_enabled)
+
+ with patch.object(
+ module.requests,
+ "get",
+ side_effect=module.requests.exceptions.ConnectionError,
+ ):
+ module.update()
+
+ assert module.fetchFailed is True
+
+ def test_connection_error_does_not_update_last_fetch(self, master_enabled):
+ """A connection error should not update lastFetch (allow retry sooner)."""
+ from TWCManager.Pricing.aWATTarPricing import aWATTarPricing
+
+ module = aWATTarPricing(master_enabled)
+
+ with patch.object(
+ module.requests,
+ "get",
+ side_effect=module.requests.exceptions.ConnectionError,
+ ):
+ module.update()
+
+ assert module.lastFetch == 0
+
+ def test_get_capabilities(self, master_enabled):
+ from TWCManager.Pricing.aWATTarPricing import aWATTarPricing
+
+ module = aWATTarPricing(master_enabled)
+ assert module.getCapabilities("AdvancePricing") is True
+
+
+# ---------------------------------------------------------------------------
+# PVPCesPricing
+# ---------------------------------------------------------------------------
+
+
+class TestPVPCesPricing:
+ """Tests for the PVPCesPricing module."""
+
+ @pytest.fixture
+ def master_enabled(self):
+ return make_master(
+ {"PVPCes": {"enabled": True, "token": "test-token-123"}}
+ )
+
+ @pytest.fixture
+ def master_disabled(self):
+ return make_master({"PVPCes": {"enabled": False, "token": "x"}})
+
+ def _make_price_response(self, prices_by_hour):
+ """Build a minimal API response with 24 hourly values."""
+ values = [{"value": prices_by_hour.get(h, 0)} for h in range(24)]
+ return {"indicator": {"values": values}}
+
+ def test_disabled_module_unloads(self, master_disabled):
+ from TWCManager.Pricing.PVPCesPricing import PVPCesPricing
+
+ PVPCesPricing(master_disabled)
+ master_disabled.releaseModule.assert_called_once()
+
+ def test_disabled_import_returns_zero(self, master_disabled):
+ from TWCManager.Pricing.PVPCesPricing import PVPCesPricing
+
+ module = PVPCesPricing(master_disabled)
+ assert module.getImportPrice() == 0
+
+ def test_import_price_current_hour(self, master_enabled):
+ """Import price should reflect the current hour's value from the API."""
+ from TWCManager.Pricing.PVPCesPricing import PVPCesPricing
+ from datetime import datetime
+
+ module = PVPCesPricing(master_enabled)
+ now = datetime.now()
+ price_mwh = 80000.0 # 80 Eur/MWh → 0.08 Eur/kWh
+ response_data = self._make_price_response({now.hour: price_mwh})
+
+ mock_response = Mock()
+ mock_response.json.return_value = response_data
+
+ with patch.object(module.requests, "get", return_value=mock_response):
+ price = module.getImportPrice()
+
+ assert price == pytest.approx(round(price_mwh / 1000, 5))
+
+ def test_cache_not_refetched_same_hour(self, master_enabled):
+ """Within the same hour, the API should only be called once."""
+ from TWCManager.Pricing.PVPCesPricing import PVPCesPricing
+ from datetime import datetime
+
+ module = PVPCesPricing(master_enabled)
+ response_data = self._make_price_response({datetime.now().hour: 50000.0})
+ mock_response = Mock()
+ mock_response.json.return_value = response_data
+
+ with patch.object(module.requests, "get", return_value=mock_response) as mock_get:
+ module.update()
+ module.update()
+ assert mock_get.call_count == 1
+
+ def test_token_included_in_headers(self, master_enabled):
+ """API token from config must appear in Authorization header."""
+ from TWCManager.Pricing.PVPCesPricing import PVPCesPricing
+
+ module = PVPCesPricing(master_enabled)
+ assert "Token token=test-token-123" in module.headers.get("Authorization", "")
+
+ def test_get_cheapest_start_hour_simple(self, master_enabled):
+ """getCheapestStartHour should return the cheapest contiguous window start."""
+ from TWCManager.Pricing.PVPCesPricing import PVPCesPricing
+
+ module = PVPCesPricing(master_enabled)
+ # Hours 2 and 3 are cheapest
+ prices = {h: 100000.0 for h in range(24)}
+ prices[2] = 10000.0
+ prices[3] = 10000.0
+ module.todayImportPrice = self._make_price_response(prices)
+
+ result = module.getCheapestStartHour(numHours=2, ini=0, end=10)
+ assert result == 2
+
+ def test_get_cheapest_start_hour_cross_midnight(self, master_enabled):
+ """getCheapestStartHour should handle end < ini (cross-midnight window)."""
+ from TWCManager.Pricing.PVPCesPricing import PVPCesPricing
+
+ module = PVPCesPricing(master_enabled)
+ # Build 48-slot response (hours 0-47 for cross-midnight support)
+ prices = {h: 100000.0 for h in range(48)}
+ prices[25] = 5000.0 # cheapest slot is hour 1 of next day (slot 25)
+ prices[26] = 5000.0
+ response = {"indicator": {"values": [{"value": prices.get(h, 100000.0)} for h in range(48)]}}
+ module.todayImportPrice = response
+
+ # Window: 22:00 to 04:00 (ini=22, end=4), needing 2 hours
+ result = module.getCheapestStartHour(numHours=2, ini=22, end=4)
+ # end becomes 28 (4+24), cheapest 2-hour window starting at slot 25 → hour 1
+ assert result == 1
+
+ def test_get_capabilities(self, master_enabled):
+ from TWCManager.Pricing.PVPCesPricing import PVPCesPricing
+
+ module = PVPCesPricing(master_enabled)
+ assert module.getCapabilities("AdvancePricing") is True
+
+
+# ---------------------------------------------------------------------------
+# TWCMaster pricing aggregation
+# ---------------------------------------------------------------------------
+
+
+class TestTWCMasterPricingAggregation:
+ """Tests for TWCMaster's getImportPrice/getExportPrice aggregation logic."""
+
+ @pytest.fixture
+ def master(self):
+ import logging
+ from TWCManager.TWCMaster import TWCMaster
+
+ # Register the custom log levels that TWCManager.py normally installs.
+ for name, level in [
+ ("INFO2", 19), ("INFO3", 18), ("INFO4", 17), ("INFO5", 16),
+ ("INFO6", 15), ("INFO7", 14), ("INFO8", 13), ("INFO9", 12),
+ ("DEBUG2", 9),
+ ]:
+ if not hasattr(logging, name):
+ logging.addLevelName(level, name)
+ setattr(logging, name, level)
+
+ config = {
+ "config": {
+ "wiringMaxAmpsAllTWCs": 48,
+ "wiringMaxAmpsPerTWC": 48,
+ "minAmpsPerTWC": 6,
+ "debugLevel": 0,
+ "displayMilliseconds": False,
+ }
+ }
+ m = TWCMaster(b"\x77\x78", config)
+ return m
+
+ def test_get_import_price_first_policy(self, master):
+ """With 'first' policy, only the first non-zero price is returned."""
+ master.importPricingValues = {"ModuleA": 0.25, "ModuleB": 0.15}
+ master.config["config"]["pricing"] = {"policy": {"multiPrice": "first"}}
+ # dict ordering in Python 3.7+ is insertion order
+ price = master.getImportPrice()
+ assert price == pytest.approx(0.25)
+
+ def test_get_import_price_add_policy(self, master):
+ """With 'add' policy, prices from all modules are summed."""
+ master.importPricingValues = {"ModuleA": 0.10, "ModuleB": 0.05}
+ master.config["config"]["pricing"] = {"policy": {"multiPrice": "add"}}
+ assert master.getImportPrice() == pytest.approx(0.15)
+
+ def test_get_export_price_first_policy(self, master):
+ master.exportPricingValues = {"ModuleA": 0.08, "ModuleB": 0.04}
+ master.config["config"]["pricing"] = {"policy": {"multiPrice": "first"}}
+ assert master.getExportPrice() == pytest.approx(0.08)
+
+ def test_get_import_price_default_is_first(self, master):
+ """Default multiPrice policy (no config key) must be 'first'."""
+ master.importPricingValues = {"ModuleA": 0.20, "ModuleB": 0.10}
+ # No pricing key in config
+ price = master.getImportPrice()
+ assert price == pytest.approx(0.20)
+
+ def test_empty_pricing_values_returns_zero(self, master):
+ master.importPricingValues = {}
+ master.exportPricingValues = {}
+ assert master.getImportPrice() == 0.0
+ assert master.getExportPrice() == 0.0
|