Pro: CP Interrupt (#2763)

Author: LKuemmel

packages/modules/chargepoints/openwb_pro/chargepoint_module.py

@@ -24,6 +24,11 @@ class EvseSignaling:
     PWM = "PWM"
 
 
+class CpInterruptionVersion:
+    CP_SIGNAL_0V = "0V"
+    CP_SIGNAL_minus12V = "-12V"
+
+
 class ChargepointModule(AbstractChargepoint):
     WRONG_CHARGE_STATE = "Lade-Status ist nicht aktiv, aber Strom fließt."
     WRONG_PLUG_STATE = "Ladepunkt ist nicht angesteckt, aber es wird geladen."
@@ -40,7 +45,7 @@ def __init__(self, config: OpenWBPro) -> None:
 
         with SingleComponentUpdateContext(self.fault_state, update_always=False):
             self.__session.post(
-                'http://' + self.config.configuration.ip_address + '/connect.php',
+                f'http://{self.config.configuration.ip_address}/connect.php',
                 data={'heartbeatenabled': '1'})
 
     def set_internal_context_handlers(self, hierarchy_id: int, internal_cp: InternalChargepoint):
@@ -58,8 +63,8 @@ def set_current(self, current: float) -> None:
             current = 0
         with SingleComponentUpdateContext(self.fault_state, update_always=False):
             with self.client_error_context:
-                ip_address = self.config.configuration.ip_address
-                self.__session.post('http://'+ip_address+'/connect.php', data={'ampere': current})
+                self.__session.post(
+                    f'http://{self.config.configuration.ip_address}/connect.php', data={'ampere': current})
 
     def get_values(self) -> None:
         with SingleComponentUpdateContext(self.fault_state):
@@ -78,8 +83,7 @@ def get_values(self) -> None:
 
     def request_values(self) -> ChargepointState:
         with self.client_error_context:
-            ip_address = self.config.configuration.ip_address
-            json_rsp = self.__session.get('http://'+ip_address+'/connect.php').json()
+            json_rsp = self.__session.get(f'http://{self.config.configuration.ip_address}/connect.php').json()
 
             chargepoint_state = ChargepointState(
                 power=json_rsp["power_all"],
@@ -131,16 +135,20 @@ def validate_values(self, chargepoint_state: ChargepointState) -> None:
     def switch_phases(self, phases_to_use: int, duration: int) -> None:
         with SingleComponentUpdateContext(self.fault_state, update_always=False):
             with self.client_error_context:
-                ip_address = self.config.configuration.ip_address
-                response = self.__session.get('http://'+ip_address+'/connect.php')
+                response = self.__session.get(f'http://{self.config.configuration.ip_address}/connect.php')
                 if response.json()["phases_target"] != phases_to_use:
-                    ip_address = self.config.configuration.ip_address
-                    self.__session.post('http://'+ip_address+'/connect.php',
+                    self.__session.post(f'http://{self.config.configuration.ip_address}/connect.php',
                                         data={'phasetarget': str(1 if phases_to_use == 1 else 3)})
                     time.sleep(duration)
 
     def clear_rfid(self) -> None:
         pass
 
+    def interrupt_cp(self, duration: int) -> None:
+        self.__session.post(f'http://{self.config.configuration.ip_address}/connect.php',
+                            data={'cp_interrupt': True,
+                                  'cp_interrupt_version': CpInterruptionVersion.CP_SIGNAL_0V,
+                                  'cp_interrupt_duration': duration})
+
 
 chargepoint_descriptor = DeviceDescriptor(configuration_factory=OpenWBPro)