refactor

DistributionCost.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from datetime import datetime
+import zoneinfo
+
+from utils.time_helpers import WARSAW_TZ
+
+@dataclass
+class DistributionCostBase:
+    tz: zoneinfo.ZoneInfo = WARSAW_TZ
+
+    def to_local_dt(self, ts: datetime) -> datetime:
+        if ts.tzinfo is None:
+            return ts.replace(tzinfo=self.tz)
+        return ts.astimezone(self.tz)
+
+    def rate(self, ts: datetime) -> float:
+        """Return PLN/kWh for given timestamp (override in subclasses)."""
+        raise NotImplementedError
+
+    def cost(self, ts: datetime, consumption_kwh: float) -> float:
+        return self.rate(ts) * float(consumption_kwh)
+

DistributionCostFactory.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+import importlib
+from typing import Any, Type, cast
+
+from DistributionCost import DistributionCostBase
+
+def create(name: str, /, **kwargs: Any) -> DistributionCostBase:
+    """
+    Instantiate provider by name using a simple convention:
+      module: DistributionProvider.<Name>Provider
+      class:  <Name>Provider
+    Example: create("TauronG13", rates=...)
+             -> DistributionCostProvider.TauronG13Provider.TauronG13Provider(...)
+    """
+    safe = "".join(ch for ch in name if ch.isalnum() or ch == "_")
+    module_name = f"DistributionCostProvider.{safe}Provider"
+    class_name  = f"{safe}Provider"
+
+    try:
+        mod = importlib.import_module(module_name)
+    except ModuleNotFoundError as e:
+        raise ValueError(f"Provider module '{module_name}' not found for name '{name}'.") from e
+
+    try:
+        cls: Type = getattr(mod, class_name)
+    except AttributeError as e:
+        raise ValueError(f"Provider class '{class_name}' not found in '{module_name}'.") from e
+
+    # be sure that this is a subclass of DistributionCostBase
+    if not issubclass(cls, DistributionCostBase):
+        raise TypeError(f"{class_name} does not derive from DistributionCostBase")
+
+    ProviderCls = cast(type[DistributionCostBase], cls)
+    return ProviderCls(**kwargs)  # type: ignore[call-arg]

DistributionCostProvider/TauronG12Provider.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+from datetime import time, datetime
+from DistributionCost import DistributionCostBase
+from utils.time_helpers import in_range_local, TimeRange
+
+
+# ---------- TAURON G12 ----------
+class TauronG12Provider(DistributionCostBase):
+    """
+    Dzień 0,3310 zł/kWh netto
+    Noc 0,0994 zł/kWh netto
+    """
+    low_1: TimeRange = (time(22,0), time(6,0))   # over midnight
+    low_2: TimeRange = (time(13,0), time(15,0))
+    high_1: TimeRange = (time(6,0), time(13,0))
+    high_2: TimeRange = (time(15,0), time(22,0))
+
+    def __init__(self):
+        self.rates={
+            "high" : 0.3310,
+            "low" : 0.0994
+        }
+
+    def rate(self, ts: datetime) -> float:
+        dt = self.to_local_dt(ts)
+        t = dt.time()
+
+        if in_range_local(t, self.low_1) or in_range_local(t, self.low_2):
+            return self.rates["low"]
+        if in_range_local(t, self.high_1) or in_range_local(t, self.high_2):
+            return self.rates["high"]
+        return self.rates["high"]
+

DistributionCostProvider/TauronG12WProvider.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+from DistributionCostProvider.TauronG12Provider import TauronG12Provider
+from utils.calendar_pl import is_weekend_or_holiday
+
+
+# ---------- TAURON G12w ----------
+class TauronG12WProvider(TauronG12Provider):
+    """
+    Like G12 on weekdays; whole weekends & holidays are 'low'.
+    rates={'low':..., 'high':...}
+    """
+    def rate(self, ts):
+        dt = self.to_local_dt(ts)
+        if is_weekend_or_holiday(dt.date()):
+            return self.rates["low"]
+        return super().rate(ts)

DistributionCostProvider/TauronG13Provider.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+from datetime import time, date, datetime
+from DistributionCost import DistributionCostBase
+from utils.time_helpers import in_range_local, TimeRange
+from utils.calendar_pl import is_weekend_or_holiday
+
+
+# ---------- TAURON G13 ----------
+class TauronG13Provider(DistributionCostBase):
+    """
+Szczyt przedpołudniowy
+    0,2826 zł/kWh brutto
+    0,2298 zł/kWh netto
+Szczyt popołudniowy
+    0,4645 zł/kWh brutto
+    0,3777 zł/kWh netto
+Pozostałe godziny
+ 	0,0911 zł/kWh brutto
+ 	0,0741 zł/kWh netto
+    """
+    winter_med: TimeRange = (time(7,0), time(13,0))
+    winter_high: TimeRange = (time(16,0), time(21,0))
+    winter_low_1: TimeRange = (time(13,0), time(16,0))
+    winter_low_2: TimeRange = (time(21,0), time(7,0))  # over midnight
+
+    summer_med: TimeRange = (time(7,0), time(13,0))
+    summer_high: TimeRange = (time(19,0), time(22,0))
+    summer_low_1: TimeRange = (time(13,0), time(19,0))
+    summer_low_2: TimeRange = (time(22,0), time(7,0))  # over midnight
+
+    def __init__(self):
+        self.rates={
+            "high" : 0.3777,
+            "med" : 0.2298,
+            "low" : 0.0741
+        }
+
+    def _is_winter(self, d: date) -> bool:
+        return d.month in (10, 11, 12, 1, 2, 3)
+
+    def rate(self, ts: datetime) -> float:
+        dt = self.to_local_dt(ts)
+        d, t = dt.date(), dt.time()
+
+        # weekend/holiday → always 'low'
+        if is_weekend_or_holiday(d):
+            return self.rates["low"]
+
+        if self._is_winter(d):
+            if in_range_local(t, self.winter_high):
+                key = "high"
+            elif in_range_local(t, self.winter_med):
+                key = "med"
+            elif in_range_local(t, self.winter_low_1) or in_range_local(t, self.winter_low_2):
+                key = "low"
+            else:
+                key = "low"
+        else:
+            if in_range_local(t, self.summer_high):
+                key = "high"
+            elif in_range_local(t, self.summer_med):
+                key = "med"
+            elif in_range_local(t, self.summer_low_1) or in_range_local(t, self.summer_low_2):
+                key = "low"
+            else:
+                key = "low"
+
+        return self.rates[key]

DistributionCostProvider/__init__.py

@@ -0,0 +1 @@
+# empty (kept for package import); discovery is done by the factory via importlib

DistributionCosts.py

@@ -1,191 +0,0 @@
-from __future__ import annotations
-from dataclasses import dataclass
-from datetime import datetime, date, timedelta, time
-from typing import Dict, Iterable, Tuple, Optional
-import zoneinfo
-
-WARSAW_TZ = zoneinfo.ZoneInfo("Europe/Warsaw")
-
-# ---------- proste święta PL (bez zależności zewnętrznych) ----------
-def _easter_sunday(year: int) -> date:
-    # algorytm Meeusa/Jonesa/Butchera (kalendarz gregoriański)
-    a = year % 19
-    b = year // 100
-    c = year % 100
-    d = b // 4
-    e = b % 4
-    f = (b + 8) // 25
-    g = (b - f + 1) // 3
-    h = (19*a + b - d - g + 15) % 30
-    i = c // 4
-    k = c % 4
-    l = (32 + 2*e + 2*i - h - k) % 7
-    m = (a + 11*h + 22*l) // 451
-    month = (h + l - 7*m + 114) // 31
-    day = 1 + ((h + l - 7*m + 114) % 31)
-    return date(year, month, day)
-
-def polish_holidays(year: int) -> set[date]:
-    easter = _easter_sunday(year)
-    holidays = {
-        date(year, 1, 1),   # Nowy Rok
-        date(year, 1, 6),   # Trzech Króli
-        date(year, 5, 1),   # Święto Pracy
-        date(year, 5, 3),   # Święto Konstytucji
-        date(year, 8, 15),  # Wniebowzięcie
-        date(year, 11, 1),  # Wszystkich Świętych
-        date(year, 11, 11), # Niepodległości
-        date(year, 12, 25), # Boże Narodzenie (1)
-        date(year, 12, 26), # Boże Narodzenie (2)
-        easter,                             # Niedziela Wielkanocna
-        easter + timedelta(days=1),         # Poniedziałek Wielkanocny
-        easter + timedelta(days=49),        # Zesłanie Ducha Św. (niedziela)
-        easter + timedelta(days=60),        # Boże Ciało (czwartek)
-    }
-    return holidays
-
-def is_weekend_or_holiday(d: date) -> bool:
-    if d.weekday() >= 5:  # 5=Saturday, 6=Sunday
-        return True
-    return d in polish_holidays(d.year)
-
-TimeRange = Tuple[time, time]
-
-def in_range_local(t_local: time, rng: TimeRange) -> bool:
-    start, end = rng
-    if start <= end:
-        return start <= t_local < end
-    # zakres przez północ
-    return (t_local >= start) or (t_local < end)
-
-@dataclass
-class DistributionCostBase:
-    tz: zoneinfo.ZoneInfo = WARSAW_TZ
-
-    def to_local_dt(self, ts: datetime) -> datetime:
-        if ts.tzinfo is None:
-            return ts.replace(tzinfo=self.tz)
-        return ts.astimezone(self.tz)
-
-    def rate(self, ts: datetime) -> float:
-        """Zwraca stawkę PLN/kWh dla wskazanego czasu (implementuje klasa potomna)."""
-        raise NotImplementedError
-
-    def cost(self, ts: datetime, consumption_kwh: float) -> float:
-        """Koszt = stawka(ts) * zużycie."""
-        return self.rate(ts) * float(consumption_kwh)
-
-# ---------- TAURON G13 ----------
-class TauronG13DistributionCost(DistributionCostBase):
-    """
-Szczyt przedpołudniowy
-    0,2826 zł/kWh brutto
-    0,2298 zł/kWh netto
-Szczyt popołudniowy
-    0,4645 zł/kWh brutto
-    0,3777 zł/kWh netto
-Pozostałe godziny
- 	0,0911 zł/kWh brutto
- 	0,0741 zł/kWh netto
-    """
-    winter_med: TimeRange = (time(7,0), time(13,0))
-    winter_high: TimeRange = (time(16,0), time(21,0))
-    winter_low_1: TimeRange = (time(13,0), time(16,0))
-    winter_low_2: TimeRange = (time(21,0), time(7,0))  # przez północ
-
-    summer_med: TimeRange = (time(7,0), time(13,0))
-    summer_high: TimeRange = (time(19,0), time(22,0))
-    summer_low_1: TimeRange = (time(13,0), time(19,0))
-    summer_low_2: TimeRange = (time(22,0), time(7,0))  # przez północ
-
-    def __init__(self):
-        self.rates={
-            "high" : 0.3777,
-            "med" : 0.2298,
-            "low" : 0.0741
-        }
-
-    def _is_winter(self, d: date) -> bool:
-        return d.month in (10, 11, 12, 1, 2, 3)
-
-    def rate(self, ts: datetime) -> float:
-        dt = self.to_local_dt(ts)
-        d, t = dt.date(), dt.time()
-
-        # weekend/święto → zawsze 'low'
-        if is_weekend_or_holiday(d):
-            return self.rates["low"]
-
-        if self._is_winter(d):
-            if in_range_local(t, self.winter_high):
-                key = "high"
-            elif in_range_local(t, self.winter_med):
-                key = "med"
-            elif in_range_local(t, self.winter_low_1) or in_range_local(t, self.winter_low_2):
-                key = "low"
-            else:
-                key = "low"
-        else:
-            if in_range_local(t, self.summer_high):
-                key = "high"
-            elif in_range_local(t, self.summer_med):
-                key = "med"
-            elif in_range_local(t, self.summer_low_1) or in_range_local(t, self.summer_low_2):
-                key = "low"
-            else:
-                key = "low"
-
-        return self.rates[key]
-
-# ---------- TAURON G12 ----------
-class TauronG12DistributionCost(DistributionCostBase):
-    """
-    Dzień 0,3310 zł/kWh netto
-    Noc 0,0994 zł/kWh netto
-    """
-    low_1: TimeRange = (time(22,0), time(6,0))   # przez północ
-    low_2: TimeRange = (time(13,0), time(15,0))
-    high_1: TimeRange = (time(6,0), time(13,0))
-    high_2: TimeRange = (time(15,0), time(22,0))
-
-    def __init__(self):
-        self.rates={
-            "high" : 0.3310,
-            "low" : 0.0994
-        }
-
-    def rate(self, ts: datetime) -> float:
-        dt = self.to_local_dt(ts)
-        t = dt.time()
-
-        if in_range_local(t, self.low_1) or in_range_local(t, self.low_2):
-            return self.rates["low"]
-        if in_range_local(t, self.high_1) or in_range_local(t, self.high_2):
-            return self.rates["high"]
-        return self.rates["high"]
-
-# ---------- TAURON G12w ----------
-class TauronG12wDistributionCost(TauronG12DistributionCost):
-    """
-    Dzień 0,3690 zł/kWh netto
-    Noc/święta/weekendy 0,0903  zł/kWh netto
-    """
-    def rate(self, ts: datetime) -> float:
-        dt = self.to_local_dt(ts)
-        if is_weekend_or_holiday(dt.date()):
-            return self.rates["low"]
-        return super().rate(ts)
-
-# ---------- przykładowe użycie ----------
-if __name__ == "__main__":
-    g13 = TauronG13DistributionCost()
-    g12 = TauronG12DistributionCost()
-    g12w = TauronG12wDistributionCost()
-
-    ts = datetime(2025, 8, 27, 20, 30)
-    cena_kWh = 0.95
-    koszt_pstryk = 0.08
-
-    print(f"G13 rate:{g12w.rate(ts)} PLN/kWh + cena/kWh:{cena_kWh} + {koszt_pstryk} = {(g12w.rate(ts) + cena_kWh + koszt_pstryk) * 1.23}")
-    # print("G12 rate:", g12.rate(ts), "PLN/kWh")
-    # print("G12w rate:", g12w.rate(ts), "PLN/kWh")

main.py

@@ -6,6 +6,7 @@ import zoneinfo
 
 TZ = zoneinfo.ZoneInfo("Europe/Warsaw")
 
+import DistributionCostFactory
 
 def load_instrat_csv(path: str) -> pd.DataFrame:
     """
@@ -47,21 +48,21 @@ if __name__ == "__main__":
 
     conn = es.setup_db()
 
-    s = df["fixing_i_pln_kwh"]
-    rows1 = es.rows_from_series(s,
-                                provider="instrat",
-                                kind="fixing I",
-                                meta={"type":"RDN", "unit":"PLN/kWh","source":"csv_export", "taxes_included":False}
-                                )
-    es.upsert_energy_prices(conn, rows1)
-
-    s = df["fixing_ii_pln_kwh"]
-    rows1 = es.rows_from_series(s,
-                                provider="instrat",
-                                kind="fixing II",
-                                meta={"type":"RDN", "unit":"PLN/kWh","source":"csv_export", "taxes_included":False}
-                                )
-    es.upsert_energy_prices(conn, rows1)
+    # s = df["fixing_i_pln_kwh"]
+    # rows1 = es.rows_from_series(s,
+    #                             provider="instrat",
+    #                             kind="fixing I",
+    #                             meta={"type":"RDN", "unit":"PLN/kWh","source":"csv_export", "taxes_included":False}
+    #                             )
+    # es.upsert_energy_prices(conn, rows1)
+    #
+    # s = df["fixing_ii_pln_kwh"]
+    # rows1 = es.rows_from_series(s,
+    #                             provider="instrat",
+    #                             kind="fixing II",
+    #                             meta={"type":"RDN", "unit":"PLN/kWh","source":"csv_export", "taxes_included":False}
+    #                             )
+    # es.upsert_energy_prices(conn, rows1)
 
     # pstryk_netto = ...      # pd.Series
     # rows2 = rows_from_series(pstryk_netto, provider="PSTRYK", kind="brutto_pstryk",
@@ -73,7 +74,7 @@ if __name__ == "__main__":
     #                          meta={"api":"api.raporty.pse.pl"})
     # upsert_energy_prices(conn, rows3)
 
-### TODO add different distribution
-
+    g13 = DistributionCostFactory.create("TauronG13")
+    print(f"Aktualna cena dystrybucji {g13.rate(datetime.now())}")
 
     conn.close()

utils/calendar_pl.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+from datetime import date, timedelta
+
+def _easter_sunday(year: int) -> date:
+    # Meeus/Jones/Butcher
+    a = year % 19
+    b = year // 100
+    c = year % 100
+    d = b // 4
+    e = b % 4
+    f = (b + 8) // 25
+    g = (b - f + 1) // 3
+    h = (19*a + b - d - g + 15) % 30
+    i = c // 4
+    k = c % 4
+    l = (32 + 2*e + 2*i - h - k) % 7
+    m = (a + 11*h + 22*l) // 451
+    month = (h + l - 7*m + 114) // 31
+    day = 1 + ((h + l - 7*m + 114) % 31)
+    return date(year, month, day)
+
+def polish_holidays(year: int) -> set[date]:
+    easter = _easter_sunday(year)
+    return {
+        date(year, 1, 1), date(year, 1, 6),
+        date(year, 5, 1), date(year, 5, 3),
+        date(year, 8, 15), date(year, 11, 1), date(year, 11, 11),
+        date(year, 12, 25), date(year, 12, 26),
+        easter, easter + timedelta(days=1),
+        easter + timedelta(days=49),  # Pentecost
+        easter + timedelta(days=60),  # Corpus Christi
+    }
+
+def is_weekend_or_holiday(d: date) -> bool:
+    return d.weekday() >= 5 or d in polish_holidays(d.year)

utils/time_helpers.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+from datetime import time
+from typing import Tuple
+import zoneinfo
+
+WARSAW_TZ = zoneinfo.ZoneInfo("Europe/Warsaw")
+TimeRange = Tuple[time, time]
+
+def in_range_local(t_local: time, rng: TimeRange) -> bool:
+    """[start, end) in local time, supports ranges crossing midnight."""
+    start, end = rng
+    if start <= end:
+        return start <= t_local < end
+    return (t_local >= start) or (t_local < end)