Merge pull request #1 from deKibi/feature/backup-manual-power-mode-swtich

Feature/backup manual power mode swtich

Author: deKibi

.env.example

@@ -0,0 +1,36 @@
+# CONFIGURATION
+MUST_PORT=COM3
+
+# MYSQL DATABASE
+MYSQL_HOST='db_ip'
+MYSQL_DATABASE='db_name'
+MYSQL_USER='db_user'
+MYSQL_PASSWORD='user_password'
+
+# DATA GATHERING
+# Interval between inverter data requests, in seconds.
+# Minimum: 10 seconds
+# Maximum: 3600 seconds
+# Invalid or missing values will use the default of 60 seconds.
+DATA_GATHER_INTERVAL_SECONDS=60
+
+# POWER MODE AUTO SWITCH
+# Allowed values: true / false
+ENABLE_AUTO_SWITCH=false
+
+# Format: HH:MM
+# 19:59 - False, 20:00 - True, 20:01 - True, 20:02 - True
+AUTO_SWITCH_TARGET_TIME=20:00
+
+# Target inverter energy mode
+# Allowed values: SBU, SUB, UTI, SOL
+AUTO_SWITCH_TARGET_MODE=SUB
+
+# POWER MODE AUTO SWITCH — GRID OUTAGE
+# Automatically switch the inverter mode when the electrical grid is unavailable.
+# Allowed values: true / false
+ENABLE_GRID_OUTAGE_AUTO_SWITCH=false
+
+# Target inverter mode when the electrical grid is unavailable.
+# Allowed values: SBU, SUB, UTI, SOL
+GRID_OUTAGE_TARGET_MODE=SUB

config.py

@@ -0,0 +1,201 @@
+# config.py
+
+# Standard Libraries
+import os
+from datetime import datetime, time
+from enum import IntEnum
+from typing import Final
+
+# Third-Party Libraries
+from dotenv import load_dotenv
+
+
+load_dotenv()
+
+
+class EnergyMode(IntEnum):
+    """
+    Inverter energy mode codes.
+
+    Replace the codes below with the actual codes used by the inverter.
+    """
+
+    # 1 = SBU, 2 = SUB, 3 = UTI, 4 = SOL
+    SBU = 1
+    SUB = 2
+    UTI = 3
+    SOL = 4
+
+def get_env_bool(variable_name: str, default: bool = False) -> bool:
+    """
+    Read and convert a boolean environment variable.
+    """
+    value = os.getenv(variable_name)
+
+    if value is None:
+        return default
+
+    normalized_value = value.strip().lower()
+
+    if normalized_value in {"true", "1", "yes", "on"}:
+        return True
+
+    if normalized_value in {"false", "0", "no", "off"}:
+        return False
+
+    raise ValueError(
+        f"Environment variable {variable_name} must contain true or false"
+    )
+
+
+def get_env_int(
+    variable_name: str,
+    default: int,
+    min_value: int | None = None,
+    max_value: int | None = None,
+) -> int:
+    """
+    Read an integer environment variable.
+
+    Returns the default value if the variable is missing
+    or contains an invalid value.
+
+    Optionally limits the value to a minimum and maximum.
+    """
+    raw_value = os.getenv(variable_name)
+
+    if raw_value is None:
+        return default
+
+    try:
+        value = int(raw_value.strip())
+    except (TypeError, ValueError):
+        print(
+            f"Invalid value for {variable_name}: {raw_value!r}. "
+            f"Using default value: {default}."
+        )
+        return default
+
+    if min_value is not None and value < min_value:
+        print(
+            f"{variable_name} is below the minimum value "
+            f"of {min_value}. Using {min_value}."
+        )
+        return min_value
+
+    if max_value is not None and value > max_value:
+        print(
+            f"{variable_name} is above the maximum value "
+            f"of {max_value}. Using {max_value}."
+        )
+        return max_value
+
+    return value
+
+
+def get_env_time(variable_name: str, default: str) -> time:
+    """
+    Read an environment variable in HH:MM format and convert it to time.
+    """
+    value = os.getenv(variable_name, default)
+
+    try:
+        return datetime.strptime(value, "%H:%M").time()
+    except ValueError as error:
+        raise ValueError(
+            f"Environment variable {variable_name} "
+            f"must use the HH:MM format"
+        ) from error
+
+
+def get_env_energy_mode(
+    variable_name: str,
+    default: EnergyMode,
+) -> EnergyMode:
+    """
+    Read an energy mode name and convert it to EnergyMode.
+    """
+    value = os.getenv(variable_name, default.name).strip().upper()
+
+    try:
+        return EnergyMode[value]
+    except KeyError as error:
+        allowed_modes = ", ".join(mode.name for mode in EnergyMode)
+
+        raise ValueError(
+            f"Environment variable {variable_name} contains "
+            f"unsupported mode {value}. Allowed modes: {allowed_modes}"
+        ) from error
+
+
+MUST_PORT: Final[str] = os.getenv("MUST_PORT", "COM3")
+
+DATA_GATHER_INTERVAL_SECONDS: Final[int] = get_env_int(
+    variable_name="DATA_GATHER_INTERVAL_SECONDS",
+    default=60,
+    min_value=10,
+    max_value=3600,
+)
+
+ENABLE_AUTO_SWITCH: Final[bool] = get_env_bool(
+    variable_name="ENABLE_AUTO_SWITCH",
+    default=False,
+)
+
+AUTO_SWITCH_TARGET_TIME: Final[time] = get_env_time(
+    variable_name="AUTO_SWITCH_TARGET_TIME",
+    default="20:00",
+)
+
+AUTO_SWITCH_TARGET_MODE: Final[EnergyMode] = get_env_energy_mode(
+    variable_name="AUTO_SWITCH_TARGET_MODE",
+    default=EnergyMode.SUB,
+)
+
+# Grid outage automatic mode switch
+ENABLE_GRID_OUTAGE_AUTO_SWITCH: Final[bool] = (
+    get_env_bool(
+        variable_name="ENABLE_GRID_OUTAGE_AUTO_SWITCH",
+        default=False,
+    )
+)
+
+GRID_OUTAGE_TARGET_MODE: Final[EnergyMode] = (
+    get_env_energy_mode(
+        variable_name="GRID_OUTAGE_TARGET_MODE",
+        default=EnergyMode.SUB,
+    )
+)
+
+
+if __name__ == '__main__':
+    print("debugging config.py")
+
+    # if (
+    #         ENABLE_AUTO_SWITCH
+    #         and is_time_reached(AUTO_SWITCH_TARGET_TIME)
+    #         and current_energy_mode != AUTO_SWITCH_TARGET_MODE.value
+    # ):
+    #     switch_energy_mode(AUTO_SWITCH_TARGET_MODE.value)
+
+    print("Enable auto switch:", ENABLE_AUTO_SWITCH)
+    print("Object Type:", type(ENABLE_AUTO_SWITCH))
+
+    auto_switch_target_mode_name = AUTO_SWITCH_TARGET_MODE.name
+    print("Auto switch target mode name:", auto_switch_target_mode_name)
+    print("Object Type:", type(auto_switch_target_mode_name))
+
+    auto_switch_target_mode_code = AUTO_SWITCH_TARGET_MODE.value
+    print("Auto switch target mode code:", auto_switch_target_mode_code)
+    print("Object Type:", type(auto_switch_target_mode_code))
+
+    grid_outage_auto_switch = ENABLE_GRID_OUTAGE_AUTO_SWITCH
+    print("Grid outage auto switch:", grid_outage_auto_switch)
+    print("Object Type:", type(grid_outage_auto_switch))
+
+    grid_outage_target_mode_name = GRID_OUTAGE_TARGET_MODE.name
+    grid_outage_target_mode_code = GRID_OUTAGE_TARGET_MODE.value
+    print("Grid outage target mode:", grid_outage_target_mode_name)
+    print("Object Type:", type(grid_outage_target_mode_name))
+    print("Grid outage target mode code:", grid_outage_target_mode_code)
+    print("Object Type:", type(grid_outage_target_mode_code))

energy_mode_control/__init__.py

@@ -0,0 +1,115 @@
+# energy_mode_control/energy_mode_controller.py
+
+# Standard Libraries
+from typing import Any, Final
+
+# Custom Modules
+from config import (
+    ENABLE_AUTO_SWITCH,
+    AUTO_SWITCH_TARGET_MODE,
+    AUTO_SWITCH_TARGET_TIME,
+    ENABLE_GRID_OUTAGE_AUTO_SWITCH,
+    GRID_OUTAGE_TARGET_MODE,
+    EnergyMode
+)
+from energy_mode_control.time_utils import is_time_reached
+from energy_mode_control.energy_mode_switcher import (
+    switch_energy_mode,
+)
+
+
+# Grid availability
+GRID_OUTAGE_VOLTAGE_THRESHOLD: Final[float] = 10.0
+
+
+def handle_energy_mode_control(must_data: dict[str, Any] | None) -> bool:
+    try:
+        return _handle_energy_mode_control(must_data)
+    except Exception as e:
+        print("Failed to handle energy mode control:", e)
+        return False
+
+
+def _handle_energy_mode_control(must_data: dict[str, Any] | None) -> bool:
+    if not ENABLE_AUTO_SWITCH and not ENABLE_GRID_OUTAGE_AUTO_SWITCH:
+        print("All automatic energy mode control features are disabled, exiting.")
+        return False
+
+    if must_data is None:
+        print("must_data is None - cannot control energy mode.")
+        return False
+
+    # Further business logic will be here.
+    if not isinstance(must_data, dict):
+        print("must_data has invalid type.")
+        return False
+
+    current_energy_mode_raw = must_data.get("EnergyUseMode")
+    grid_voltage_raw = must_data.get("GridVoltage")
+
+    if current_energy_mode_raw is None:
+        print("EnergyUseMode is missing from must_data.")
+        return False
+
+    if grid_voltage_raw is None:
+        print("GridVoltage is missing from must_data.")
+        return False
+
+    try:
+        current_energy_mode = EnergyMode(
+            int(current_energy_mode_raw)
+        )
+        grid_voltage = float(grid_voltage_raw)
+    except (TypeError, ValueError) as error:
+        print(
+            "EnergyUseMode or GridVoltage contains "
+            f"an invalid value: {error}"
+        )
+        return False
+
+    is_grid_available = (
+        grid_voltage >= GRID_OUTAGE_VOLTAGE_THRESHOLD
+    )
+
+    target_mode: EnergyMode | None = None
+    switch_reason: str | None = None
+
+    # Grid outage rule has the highest priority.
+    if (
+            ENABLE_GRID_OUTAGE_AUTO_SWITCH
+            and not is_grid_available
+    ):
+        target_mode = GRID_OUTAGE_TARGET_MODE
+        switch_reason = "electrical grid is unavailable"
+
+    # Time-based rule is checked when the grid is available.
+    elif (
+        ENABLE_AUTO_SWITCH
+        and is_grid_available
+        and is_time_reached(AUTO_SWITCH_TARGET_TIME)
+    ):
+        target_mode = AUTO_SWITCH_TARGET_MODE
+        switch_reason = "auto-switch target time has been reached"
+
+    if target_mode is None:
+        print("No energy mode switch is currently required.")
+        return False
+
+    if current_energy_mode == target_mode:
+        print(
+            f"Inverter is already operating in "
+            f"{target_mode.name} mode."
+        )
+        return False
+
+    print(
+        f"Energy mode switch required: "
+        f"{current_energy_mode.name} -> {target_mode.name}. "
+        f"Reason: {switch_reason}."
+    )
+
+    # The real inverter command will be added here:
+    switch_energy_mode(
+        target_mode=target_mode,
+    )
+    return True