consumer rfid

packages/control/chargepoint/rfid.py

@@ -71,7 +71,7 @@ def _validate_rfid(self) -> None:
                     else:
                         msg = "Nach Ladestart wird kein anderer ID-Tag akzeptiert."
                 else:
-                    msg = f"Der ID-Tag {rfid} ist an diesem Ladepunkt nicht gültig."
+                    msg = f"Der ID-Tag {rfid} ist hier keinem Ladepunkt-Profil oder Fahrzeug zugeordnet."
             else:
                 msg = "Identifikation von Fahrzeugen ist nicht aktiviert."
             self.data.get.rfid = None

packages/control/consumer/consumer.py

@@ -4,6 +4,7 @@
 
 from control import data
 from control.consumer.usage import NOT_CONTROLLED, ON_OFF
+from control.consumer.rfid import ConsumerRfidMixin
 from control.algorithm.utils import get_medium_charging_current
 from control.chargemode import Chargemode
 from control.chargepoint.chargepoint_state import CHARGING_STATES, ChargepointState
@@ -18,7 +19,7 @@
 log = logging.getLogger(__name__)
 
 
-class Consumer(Load):
+class Consumer(ConsumerRfidMixin, Load):
     PAUSE_BETWEEN_WAIT_FOR_START_TEST_RUNS = 3600
     STANDBY_THRESHOLD = 0.15  # 35 Watt pro Phase
 
@@ -59,6 +60,7 @@ def update(self):
             self.data.set.phases_to_use = self.data.config.connected_phases
             self.data.get.charge_state = max(self.data.get.currents) > self.STANDBY_THRESHOLD
             self.process_on_time()
+            self.update_rfid()
 
             if self.data.usage.type in NOT_CONTROLLED:
                 return
@@ -121,6 +123,9 @@ def is_switch_interval_elapsed(self):
                                        "Verbraucher nicht abgeschaltet werden darf.")
 
     def get_parameter(self) -> Tuple[float, Optional[str], Optional[Chargemode], Chargemode]:
+        if self.rfid_required() and self.data.set.rfid is None:
+            return (0, "Verbraucher gesperrt. Bitte per RFID freischalten.",
+                    self.data.usage.chargemode, Chargemode.STOP)
         if self.data.set.switch_interval_elapsed is False:
             log.debug("Intervall für neuen Schaltbefehl nicht abgelaufen.")
             return (max(self.data.get.currents),

packages/control/consumer/consumer_data.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import List, Optional
+from typing import List, Optional, Protocol
 from control.chargemode import Chargemode
 from control.chargepoint.control_parameter import ControlParameter, control_parameter_factory
 from control.consumer.usage import ConsumerUsage
@@ -33,6 +33,11 @@ class WaitForStartStates(Enum):
     START_SIGNAL_RECEIVED = "start_signal_received"
 
 
+class RfidReset(Enum):
+    AFTER_TIME = "after_time"
+    AFTER_POWER_END = "after_power_end"
+
+
 @dataclass
 class ResetChargemode(TimepointPlan):
     active: bool = False
@@ -58,10 +63,16 @@ class ConsumerConfig:
     min_power: float = 115
     min_interval: int = 60
     is_home_consumption_consumer: str = "auto_home_consumption"
+    rfid_enabled: bool = False
+    valid_tags: List[str] = field(default_factory=empty_list_factory)
+    rfid_reset: RfidReset = RfidReset.AFTER_TIME
+    rfid_timeout: int = 300
 
 
 @dataclass
 class Get:
+    rfid: Optional[str] = field(default=None, metadata={"topic": "get/rfid"})
+    rfid_timestamp: Optional[float] = field(default=None, metadata={"topic": "get/rfid_timestamp"})
     charge_state: bool = field(default=False, metadata={"topic": "get/charge_state"})
     currents: List[float] = field(default_factory=currents_list_factory)
     daily_imported: float = field(default=0, metadata={"topic": "get/daily_imported"})
@@ -82,6 +93,10 @@ class Get:
 
 @dataclass
 class Set:
+    rfid: Optional[str] = field(default=None, metadata={"topic": "set/rfid"})
+    rfid_timestamp: Optional[float] = field(default=None, metadata={"topic": "set/rfid_timestamp"})
+    rfid_power_end: Optional[float] = field(default=None, metadata={"topic": "set/rfid_power_end"})
+    rfid_power_seen: bool = field(default=False, metadata={"topic": "set/rfid_power_seen"})
     current: float = field(default=0, metadata={"topic": "set/current"})
     phases_to_use: int = field(default=1, metadata={"topic": "set/phases_to_use"})
     plug_time: Optional[float] = field(default=None, metadata={"topic": "set/plug_time"})
@@ -109,3 +124,16 @@ class ConsumerData:
     config: ConsumerConfig = field(default_factory=lambda: ConsumerConfig())
     extra_meter: Optional[int] = None
     usage: Usage = field(default_factory=lambda: Usage(), metadata={"topic": "usage"})
+
+
+class ConsumerRfidProtocol(Protocol):
+    @property
+    def data(self) -> ConsumerData: ...
+    @property
+    def module(self) -> ConsumerSetup: ...
+
+    def set_state_and_log(self, message: Optional[str]) -> None: ...
+    def rfid_required(self) -> bool: ...
+    def update_rfid(self) -> None: ...
+    def clear_rfid(self) -> None: ...
+    def reset_rfid(self) -> None: ...

packages/control/consumer/rfid.py

@@ -0,0 +1,61 @@
+from fnmatch import fnmatch
+
+from control import data
+from control.consumer.consumer_data import ConsumerRfidProtocol, RfidReset
+from control.consumer.usage import NOT_CONTROLLED
+from helpermodules import timecheck
+
+
+def rfid_supported(module) -> bool:
+    return getattr(module, "type", None) == "mqtt"
+
+
+class ConsumerRfidMixin:
+    def rfid_required(self: ConsumerRfidProtocol) -> bool:
+        return (self.data.usage.type not in NOT_CONTROLLED and self.data.extra_meter is None and
+                rfid_supported(self.data.module) and self.data.config.rfid_enabled and
+                data.data.optional_data.data.rfid.active)
+
+    def update_rfid(self: ConsumerRfidProtocol) -> None:
+        state = self.data.set
+        tag = self.data.get.rfid
+        if tag is not None:
+            if self.rfid_required():
+                if any(fnmatch(tag.lower(), pattern.lower()) for pattern in self.data.config.valid_tags):
+                    state.rfid = tag
+                    state.rfid_timestamp = (self.data.get.rfid_timestamp if self.data.get.rfid_timestamp is not None
+                                            else timecheck.create_timestamp())
+                    state.rfid_power_seen = False
+                    state.rfid_power_end = None
+                else:
+                    self.set_state_and_log(f"Der ID-Tag {tag} ist diesem Verbraucher nicht zugeordnet.")
+            self.clear_rfid()
+        if not self.rfid_required():
+            self.reset_rfid()
+            return
+        if state.rfid is None:
+            return
+        now = timecheck.create_timestamp()
+        if self.data.config.rfid_reset == RfidReset.AFTER_POWER_END:
+            if self.data.get.charge_state:
+                state.rfid_power_seen = True
+                state.rfid_power_end = None
+            elif state.rfid_power_seen and state.rfid_power_end is None:
+                state.rfid_power_end = now
+            timestamp = state.rfid_power_end
+        else:
+            timestamp = state.rfid_timestamp
+        if timestamp is not None and now - timestamp >= self.data.config.rfid_timeout:
+            self.reset_rfid()
+
+    def clear_rfid(self: ConsumerRfidProtocol) -> None:
+        """Verarbeiteten Eingangstag löschen."""
+        self.data.get.rfid = None
+        self.data.get.rfid_timestamp = None
+        self.module.clear_rfid()
+
+    def reset_rfid(self: ConsumerRfidProtocol) -> None:
+        self.data.set.rfid = None
+        self.data.set.rfid_timestamp = None
+        self.data.set.rfid_power_seen = False
+        self.data.set.rfid_power_end = None