Added calculated current sensors (#83)

Author: derolli1976

custom_components/enpal_webparser/tests/test_current_calculation.py

@@ -0,0 +1,168 @@
+"""Test calculated current sensors for PowerSensor."""
+import pytest
+from custom_components.enpal_webparser.utils import add_calculated_current_sensors
+
+
+def test_calculate_current_sensors():
+    """Test that current is calculated correctly from power and voltage."""
+    # Test data based on user's screenshot:
+    # Phase A: -61W / 231.1V = -0.26A
+    # Phase B: -19W / 230.1V = -0.08A
+    # Phase C: 77W / 230.3V = 0.33A
+    
+    sensors = [
+        {
+            "name": "PowerSensor: Power AC Phase (A)",
+            "value": "-61",
+            "unit": "W",
+            "group": "PowerSensor",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.635Z",
+        },
+        {
+            "name": "PowerSensor: Voltage Phase (A)",
+            "value": "231.1",
+            "unit": "V",
+            "group": "PowerSensor",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.634Z",
+        },
+        {
+            "name": "PowerSensor: Power AC Phase (B)",
+            "value": "-19",
+            "unit": "W",
+            "group": "PowerSensor",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.635Z",
+        },
+        {
+            "name": "PowerSensor: Voltage Phase (B)",
+            "value": "230.1",
+            "unit": "V",
+            "group": "PowerSensor",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.634Z",
+        },
+        {
+            "name": "PowerSensor: Power AC Phase (C)",
+            "value": "77",
+            "unit": "W",
+            "group": "PowerSensor",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.636Z",
+        },
+        {
+            "name": "PowerSensor: Voltage Phase (C)",
+            "value": "230.3",
+            "unit": "V",
+            "group": "PowerSensor",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.634Z",
+        },
+    ]
+    
+    result = add_calculated_current_sensors(sensors)
+    
+    # Should have original 6 sensors + 3 calculated current sensors
+    assert len(result) == 9
+    
+    # Find calculated current sensors by name
+    current_a = next((s for s in result if "Current Phase (A)" in s["name"]), None)
+    current_b = next((s for s in result if "Current Phase (B)" in s["name"]), None)
+    current_c = next((s for s in result if "Current Phase (C)" in s["name"]), None)
+    
+    assert current_a is not None, "Current sensor for phase A not found"
+    assert current_b is not None, "Current sensor for phase B not found"
+    assert current_c is not None, "Current sensor for phase C not found"
+    
+    # Check calculated values (I = P / U)
+    # Phase A: -61 / 231.1 = -0.26A
+    assert float(current_a["value"]) == pytest.approx(-0.26, abs=0.01)
+    assert current_a["unit"] == "A"
+    assert current_a["device_class"] == "current"
+    assert current_a["group"] == "PowerSensor"
+    
+    # Phase B: -19 / 230.1 = -0.08A
+    assert float(current_b["value"]) == pytest.approx(-0.08, abs=0.01)
+    assert current_b["unit"] == "A"
+    
+    # Phase C: 77 / 230.3 = 0.33A
+    assert float(current_c["value"]) == pytest.approx(0.33, abs=0.01)
+    assert current_c["unit"] == "A"
+
+
+def test_calculate_current_sensors_missing_voltage():
+    """Test that calculation handles missing voltage gracefully."""
+    sensors = [
+        {
+            "name": "PowerSensor: Power AC Phase (A)",
+            "value": "-61",
+            "unit": "W",
+            "group": "PowerSensor",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.635Z",
+        },
+        # Missing voltage sensor for phase A
+    ]
+    
+    result = add_calculated_current_sensors(sensors)
+    
+    # Should have only the original sensor (no current calculated)
+    assert len(result) == 1
+    assert "Power AC Phase (A)" in result[0]["name"]
+
+
+def test_calculate_current_sensors_zero_voltage():
+    """Test that calculation handles zero voltage (division by zero)."""
+    sensors = [
+        {
+            "name": "PowerSensor: Power AC Phase (A)",
+            "value": "-61",
+            "unit": "W",
+            "group": "PowerSensor",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.635Z",
+        },
+        {
+            "name": "PowerSensor: Voltage Phase (A)",
+            "value": "0",
+            "unit": "V",
+            "group": "PowerSensor",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.634Z",
+        },
+    ]
+    
+    result = add_calculated_current_sensors(sensors)
+    
+    # Should have only the original sensors (no current calculated due to zero voltage)
+    assert len(result) == 2
+    assert not any("Current Phase (A)" in s["name"] for s in result)
+
+
+def test_calculate_current_sensors_non_powersensor_group():
+    """Test that calculation only applies to PowerSensor group."""
+    sensors = [
+        {
+            "name": "Inverter: Power AC Phase (A)",
+            "value": "100",
+            "unit": "W",
+            "group": "Inverter",  # Not PowerSensor
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.635Z",
+        },
+        {
+            "name": "Inverter: Voltage Phase (A)",
+            "value": "230",
+            "unit": "V",
+            "group": "Inverter",
+            "enabled": True,
+            "enpal_last_update": "2025-10-31T09:57:14.634Z",
+        },
+    ]
+    
+    result = add_calculated_current_sensors(sensors)
+    
+    # Should have only the original sensors (no calculation for non-PowerSensor group)
+    assert len(result) == 2
+    assert not any("Current Phase" in s["name"] for s in result)

custom_components/enpal_webparser/utils.py

@@ -229,6 +229,9 @@ def parse_enpal_html_sensors(
 
         sensors.extend(parse_card_rows(card, group, groups))
 
+    # Calculate missing current sensors from power and voltage (I = P / U)
+    sensors = add_calculated_current_sensors(sensors)
+
     return sensors
 
 
@@ -266,6 +269,7 @@ def parse_card_rows(card: Tag, group: str, groups: List[str]) -> List[Dict[str,
             "device_class": device_class,
             "enabled": group in groups,
             "enpal_last_update": timestamp_iso,
+            "group": group,  # Add group for later filtering
         }
 
         sensor_id = make_id(sensor["name"])
@@ -316,3 +320,101 @@ def parse_timestamp(raw: Optional[str]) -> Optional[str]:
         return dt.isoformat()
     except ValueError:
         return raw
+
+
+def add_calculated_current_sensors(sensors: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Calculate missing PowerSensor current sensors from power and voltage.
+    
+    Enpal boxes no longer provide Current.Phase.A/B/C sensors directly.
+    This function calculates them using Ohm's law: I = P / U
+    
+    Args:
+        sensors: List of parsed sensors
+        
+    Returns:
+        Updated sensor list with calculated current sensors added
+    """
+    # Find power and voltage values for each phase
+    power_sensors = {}
+    voltage_sensors = {}
+    
+    for sensor in sensors:
+        name = sensor.get("name", "")
+        group = sensor.get("group", "")
+        
+        # Only process PowerSensor group
+        if group != "PowerSensor":
+            continue
+        
+        # Generate a normalized ID from the name for matching
+        sensor_id = make_id(name)
+        
+        # Collect power values (Power.AC.Phase.A/B/C)
+        if "power_ac_phase_a" in sensor_id:
+            power_sensors["a"] = sensor
+        elif "power_ac_phase_b" in sensor_id:
+            power_sensors["b"] = sensor
+        elif "power_ac_phase_c" in sensor_id:
+            power_sensors["c"] = sensor
+        # Collect voltage values (Voltage.Phase.A/B/C)
+        elif "voltage_phase_a" in sensor_id:
+            voltage_sensors["a"] = sensor
+        elif "voltage_phase_b" in sensor_id:
+            voltage_sensors["b"] = sensor
+        elif "voltage_phase_c" in sensor_id:
+            voltage_sensors["c"] = sensor
+    
+    # Calculate current for each phase where both power and voltage are available
+    calculated_count = 0
+    for phase in ["a", "b", "c"]:
+        power_sensor = power_sensors.get(phase)
+        voltage_sensor = voltage_sensors.get(phase)
+        
+        if not power_sensor or not voltage_sensor:
+            continue
+            
+        try:
+            # Get numeric values as strings, then convert to float
+            power_str = get_numeric_value(power_sensor.get("value", ""))
+            voltage_str = get_numeric_value(voltage_sensor.get("value", ""))
+            
+            if not power_str or not voltage_str:
+                continue
+                
+            power_value = float(power_str)
+            voltage_value = float(voltage_str)
+            
+            if voltage_value == 0:
+                continue
+            
+            # Calculate current: I = P / U
+            current_value = power_value / voltage_value
+            
+            # Create calculated current sensor
+            phase_upper = phase.upper()
+            current_sensor = {
+                "name": f"PowerSensor: Current Phase ({phase_upper})",
+                "value": str(round(current_value, 2)),
+                "unit": "A",
+                "device_class": "current",
+                "enabled": True,
+                "enpal_last_update": power_sensor.get("enpal_last_update") or voltage_sensor.get("enpal_last_update"),
+                "group": "PowerSensor",
+            }
+            
+            sensors.append(current_sensor)
+            calculated_count += 1
+            
+        except (ValueError, ZeroDivisionError, TypeError) as e:
+            _LOGGER.debug(
+                "[Enpal] Could not calculate current for phase %s: %s",
+                phase.upper(), e
+            )
+    
+    if calculated_count > 0:
+        _LOGGER.debug(
+            "[Enpal] Calculated %d missing PowerSensor current sensor(s)",
+            calculated_count
+        )
+    
+    return sensors