Refactor lock file handling into the main module

Moved lock file management from `InverterChargeController` to `main.py`

Author: felixschndr

source/inverter_charge_controller.py

@@ -1,4 +1,3 @@
-import os
 import socket
 import sys
 from datetime import datetime, timedelta
@@ -21,7 +20,6 @@
 
 
 class InverterChargeController(LoggerMixin):
-    LOCK_FILE_PATH = "/tmp/inverter_charge_controller.lock"  # nosec B108
 
     def __init__(self):
         super().__init__()
@@ -41,23 +39,6 @@ def __init__(self):
         self.iteration_cache = {}
 
     def start(self) -> None:
-        """
-        Starts the inverter charge controller process. Ensures that the process is not already running
-        by checking for the presence of a lock file. If the process is running, logs the error and exits.
-        Upon successful starting, creates and manages a lock file for the process to avoid multiple
-        instances. It Also ensures cleanup of the lock file post execution.
-        """
-        if os.path.exists(self.LOCK_FILE_PATH):
-            self.log.error("Attempted to start the inverter charge controller, but it is already running.")
-            return
-
-        self._lock()
-        try:
-            self._start()
-        finally:
-            self.unlock()
-
-    def _start(self) -> None:
         """
         Starts the continuous running of the program and handles all exceptions possibly raised during execution.
          - Expected exceptions: Wait for some minutes and retry
@@ -433,21 +414,3 @@ def _get_value_from_cache_if_exists(self, cache_key: str) -> Optional[Any]:
 
     def _set_cache_key(self, cache_key: str, value: Any) -> None:
         self.iteration_cache[cache_key] = value
-
-    def _lock(self) -> None:
-        """
-        Writes the current process ID to a lock file to indicate the process is active.
-        """
-        with open(self.LOCK_FILE_PATH, "w") as lock_file:
-            lock_file.write(str(os.getpid()))
-        self.log.debug("Lock file created")
-
-    def unlock(self) -> None:
-        """
-        Removes the lock file if it exists and thus unlocking the process.
-        """
-        if not os.path.exists(self.LOCK_FILE_PATH):
-            return
-
-        os.remove(self.LOCK_FILE_PATH)
-        self.log.debug("Lock file removed")

source/main.py

@@ -1,3 +1,4 @@
+import os
 import signal
 import sys
 import threading
@@ -12,9 +13,31 @@
 from sun_forecast_handler import SunForecastHandler
 from time_handler import TimeHandler
 
+LOCK_FILE_PATH = "/tmp/inverter_charge_controller.lock"  # nosec B108
+
 logger = LoggerMixin("Main")
 
 
+def lock() -> None:
+    """
+    Writes the current process ID to a lock file to indicate the process is active.
+    """
+    with open(LOCK_FILE_PATH, "w") as lock_file:
+        lock_file.write(str(os.getpid()))
+    logger.log.debug("Lock file created")
+
+
+def unlock() -> None:
+    """
+    Removes the lock file if it exists and thus unlocking the process.
+    """
+    if not os.path.exists(LOCK_FILE_PATH):
+        return
+
+    os.remove(LOCK_FILE_PATH)
+    logger.log.debug("Lock file removed")
+
+
 def write_solar_forecast_and_history_to_db() -> None:
     sun_forecast_handler = SunForecastHandler()
     morning_time = time(hour=5, minute=0, second=0, microsecond=0, tzinfo=TimeHandler.get_timezone())
@@ -77,25 +100,35 @@ def handle_stop_signal(signal_number: int, _frame: FrameType) -> None:
     """
     logger.write_newlines_to_log_file()
     logger.log.info(f"Received {signal.Signals(signal_number).name}. Exiting now...")
-    inverter_charge_controller.unlock()
+    unlock()
     sys.exit(0)
 
 
 for signal_to_catch in [signal.SIGINT, signal.SIGTERM]:
     signal.signal(signal_to_catch, handle_stop_signal)
 
+
 if __name__ == "__main__":
-    logger.write_newlines_to_log_file()
-    started_by_systemd = " by systemd" if EnvironmentVariableGetter.get("INVOCATION_ID", "") else ""
-    logger.log.info(f"Starting application{started_by_systemd}")
+    if os.path.exists(LOCK_FILE_PATH):
+        logger.write_newlines_to_log_file()
+        logger.log.error("Attempted to start the inverter charge controller, but it is already running.")
+        sys.exit(1)
+
+    try:
+        lock()
+        logger.write_newlines_to_log_file()
+        started_by_systemd = " by systemd" if EnvironmentVariableGetter.get("INVOCATION_ID", "") else ""
+        logger.log.info(f"Starting application{started_by_systemd}")
 
-    solar_protocol_thread = threading.Thread(target=write_solar_forecast_and_history_to_db, daemon=True)
-    solar_protocol_thread.start()
+        solar_protocol_thread = threading.Thread(target=write_solar_forecast_and_history_to_db, daemon=True)
+        solar_protocol_thread.start()
 
-    # Let the thread calculate and log its next wakeup time before logging all the info of the InverterChargeController
-    pause.seconds(1)
+        # Let the thread calculate and log its next wakeup time before logging all the info of the InverterChargeController
+        pause.seconds(1)
 
-    inverter_charge_controller = InverterChargeController()
-    inverter_charge_controller_thread = threading.Thread(target=inverter_charge_controller.start)
-    inverter_charge_controller_thread.start()
-    inverter_charge_controller_thread.join()
+        inverter_charge_controller = InverterChargeController()
+        inverter_charge_controller_thread = threading.Thread(target=inverter_charge_controller.start)
+        inverter_charge_controller_thread.start()
+        inverter_charge_controller_thread.join()
+    finally:
+        unlock()