Encode battery charge source in action states

custom_components/wattplan/coordinator_logic/projection.py

@@ -41,8 +41,6 @@ def planner_output_from_result(
         batteries: dict[str, dict[str, Any]] = {}
         comforts: dict[str, dict[str, Any]] = {}
         optionals: dict[str, dict[str, Any]] = {}
-        battery_charge_source: dict[str, str] = {}
-
         name_maps = request["name_to_subentry"]
         for entity in result.get("entities", []):
             entity_name = str(entity.get("name"))
@@ -56,8 +54,6 @@ def planner_output_from_result(
                 if subentry_id is None:
                     continue
                 current = schedule[0]
-                current_source = self._map_charge_source(int(current.get("charge_source", 0)))
-                battery_charge_source[subentry_id] = current_source
                 next_change = self._next_change_point(
                     schedule,
                     key="state",
@@ -71,22 +67,14 @@ def planner_output_from_result(
                     if isinstance(next_point, dict)
                     else None
                 )
-                next_action_source = (
-                    self._map_charge_source(int(next_point.get("charge_source", 0)))
-                    if isinstance(next_point, dict)
-                    and str(next_point.get("state", "hold")) == "charge"
-                    else None
-                )
                 batteries[subentry_id] = {
                     "action": str(current.get("state", "hold")),
-                    "charge_source": current_source,
                     "next_action_timestamp": (
                         next_action_timestamp.isoformat()
                         if next_action_timestamp is not None
                         else None
                     ),
                     "next_action": next_action,
-                    "next_charge_source": next_action_source,
                 }
                 continue
 
@@ -155,7 +143,6 @@ def planner_output_from_result(
         return {
             "status": status,
             "message": message,
-            "battery_charge_source": battery_charge_source,
             "diagnostics": {
                 "batteries": batteries,
                 "comforts": comforts,
@@ -192,7 +179,6 @@ def project_snapshot(self, planner_output: dict[str, Any]) -> CoordinatorSnapsho
                 if planner_output.get("message") is not None
                 else None
             ),
-            battery_charge_source=planner_output.get("battery_charge_source"),
             diagnostics=planner_output.get("diagnostics"),
         )
 
@@ -286,12 +272,6 @@ def _build_plan_details_payload(
                         schedule, horizon_slots, "level", default=0.0
                     )
                 )
-                plan_details[f"{key_base}_charge_source"] = [
-                    self._map_charge_source(int(value))
-                    for value in self._series_from_schedule(
-                        schedule, horizon_slots, "charge_source", default=0.0
-                    )
-                ]
                 continue
 
             if entity_type == "comfort":
@@ -388,14 +368,10 @@ def _aggregate_plan_details_series(
                     aggregated.append(round(sum(float(value) for value in chunk) / len(chunk), 2))
                 else:
                     aggregated.append(round(sum(float(value) for value in chunk), 2))
-            elif key.endswith("_charge_source"):
-                sources = {str(value) for value in chunk if isinstance(value, str)}
-                active_sources = sorted(source for source in sources if source != "n")
-                aggregated.append("".join(active_sources) if active_sources else "n")
             elif key.endswith("_action"):
                 actions = {str(value) for value in chunk if isinstance(value, str)}
                 active_actions = sorted(action for action in actions if action != "h")
-                aggregated.append("".join(active_actions) if active_actions else "h")
+                aggregated.append("+".join(active_actions) if active_actions else "h")
             else:
                 aggregated.append(chunk[-1])
         return aggregated
@@ -479,18 +455,11 @@ def _next_change_point(
                 )
         return None
 
-    def _map_charge_source(self, charge_source: int) -> str:
-        if charge_source == 1:
-            return "g"
-        if charge_source == 2:
-            return "p"
-        if charge_source == 3:
-            return "gp"
-        return "n"
-
     def _map_action_code(self, action: str) -> str:
         return {
-            "charge": "c",
+            "charge_grid": "c_g",
+            "charge_pv": "c_p",
+            "charge_grid_pv": "c_gp",
             "discharge": "d",
             "hold": "h",
         }.get(action, "h")

custom_components/wattplan/coordinator_parts/snapshot.py

@@ -26,7 +26,6 @@ class CoordinatorSnapshot:
     created_at: datetime
     planner_status: str
     planner_message: str | None = None
-    battery_charge_source: dict[str, str] | None = None
     diagnostics: dict[str, Any] | None = None
 
     def to_dict(self) -> dict[str, Any]:
@@ -35,7 +34,6 @@ def to_dict(self) -> dict[str, Any]:
             "created_at": self.created_at.isoformat(),
             "planner_status": self.planner_status,
             "planner_message": self.planner_message,
-            "battery_charge_source": self.battery_charge_source,
             "diagnostics": self.diagnostics,
         }
 
@@ -51,10 +49,6 @@ def from_dict(cls, payload: dict[str, Any]) -> CoordinatorSnapshot | None:
         if not isinstance(planner_message, str | type(None)):
             return None
 
-        battery_charge_source = payload.get("battery_charge_source")
-        if not isinstance(battery_charge_source, dict | type(None)):
-            return None
-
         diagnostics = payload.get("diagnostics")
         if not isinstance(diagnostics, dict | type(None)):
             return None
@@ -63,7 +57,6 @@ def from_dict(cls, payload: dict[str, Any]) -> CoordinatorSnapshot | None:
             created_at=created_at,
             planner_status=planner_status,
             planner_message=planner_message,
-            battery_charge_source=battery_charge_source,
             diagnostics=diagnostics,
         )
 

custom_components/wattplan/optimizer/mpc_power_optimizer.py

@@ -91,7 +91,7 @@ def _battery_target_bounds_kwh(entity):
     }
 
 
-def _charge_source_permissions(entity):
+def _charge_ingress_permissions(entity):
     flags = int(entity.can_charge_from)
     return (
         bool(flags & int(ChargeSource.GRID)),
@@ -404,7 +404,7 @@ def _solve_mpc_step(
         capacity = float(entity.capacity_kwh)
         charge_eff = float(entity.charge_efficiency)
         discharge_eff = float(entity.discharge_efficiency)
-        can_charge_from_grid, can_charge_from_pv = _charge_source_permissions(entity)
+        can_charge_from_grid, can_charge_from_pv = _charge_ingress_permissions(entity)
         throughput_penalty = float(entity.throughput_cost_per_kwh)
         mode_switch_cost = float(entity.mode_switch_cost)
         previous_state = int(battery_states_now[b]) if battery_states_now.size else 0
@@ -676,7 +676,7 @@ def _apply_controls_step(
         charge_eff = float(entity.charge_efficiency)
         discharge_eff = float(entity.discharge_efficiency)
         action_deadband = float(entity.action_deadband_kwh)
-        can_charge_from_grid, can_charge_from_pv = _charge_source_permissions(entity)
+        can_charge_from_grid, can_charge_from_pv = _charge_ingress_permissions(entity)
 
         requested_grid = max(
             float(controls.get("charge_grid", np.zeros(num_battery))[i]), 0.0
@@ -1219,16 +1219,28 @@ def _optional_entity_options(entity, grid_import_prices, baseline_net_import):
     ]
 
 
-def _battery_schedule_charge_source(result, battery_index: int, timeslot: int) -> int:
-    """Return a normalized charge source bitmask for one battery schedule slot."""
+def _battery_schedule_state(result, battery_index: int, timeslot: int) -> str:
+    """Return the serialized battery action state for one schedule slot."""
     battery_state = int(result["battery_states"][battery_index, timeslot])
-    if battery_state != 1:
-        return 0
-
-    return int(
-        (1 if result["battery_charge_grid"][battery_index, timeslot] > EPSILON else 0)
-        | (2 if result["battery_charge_pv"][battery_index, timeslot] > EPSILON else 0)
-    )
+    if battery_state == 0:
+        return "hold"
+    if battery_state == 1:
+        charge_ingress = int(
+            (1 if result["battery_charge_grid"][battery_index, timeslot] > EPSILON else 0)
+            | (2 if result["battery_charge_pv"][battery_index, timeslot] > EPSILON else 0)
+        )
+        if charge_ingress == 1:
+            return "charge_grid"
+        if charge_ingress == 2:
+            return "charge_pv"
+        if charge_ingress == 3:
+            return "charge_grid_pv"
+        raise ValueError(
+            "battery schedule serialization encountered charging state without charge ingress"
+        )
+    if battery_state == 2:
+        return "discharge"
+    raise ValueError(f"battery schedule serialization encountered unknown state {battery_state}")
 
 
 def optimize_internal(normalized: CalculationInput):
@@ -1317,20 +1329,14 @@ def optimize_internal(normalized: CalculationInput):
     )
 
     entities = []
-    battery_state_name = {0: "hold", 1: "charge", 2: "discharge"}
     for i, entity in enumerate(battery_entities):
         entities.append(
             {
                 "name": entity.name,
                 "type": "battery",
                 "schedule": [
                     {
-                        "state": battery_state_name[
-                            int(result["battery_states"][i, t])
-                        ],
-                        "charge_source": _battery_schedule_charge_source(
-                            result, i, t
-                        ),
+                        "state": _battery_schedule_state(result, i, t),
                         "level": float(result["battery_levels"][i, t + 1]),
                     }
                     for t in range(total_steps)

custom_components/wattplan/sensors/actions.py

@@ -4,14 +4,22 @@
 
 from typing import Any
 
-from homeassistant.components.sensor import SensorEntity
+from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import UnitOfEnergy
 
 from ..coordinator import WattPlanCoordinator
 from ..target_runtime import get_active_battery_target
 from .base import WattPlanCoordinatorSensor
-from .common import as_datetime, entry_device_info, friendly_charge_source_label
+from .common import as_datetime, entry_device_info
+
+BATTERY_ACTION_STATES = [
+    "hold",
+    "discharge",
+    "charge_grid",
+    "charge_pv",
+    "charge_grid_pv",
+]
 
 
 class SubentryActionSensor(WattPlanCoordinatorSensor):
@@ -32,6 +40,9 @@ def __init__(
         super().__init__(config_entry, coordinator, **kwargs)
         self._subentry_id = subentry_id
         self._group = group
+        if group == "batteries":
+            self._attr_device_class = SensorDeviceClass.ENUM
+            self._attr_options = BATTERY_ACTION_STATES
 
     def _action_data(self) -> dict[str, Any]:
         """Return action data for this subentry from snapshot diagnostics."""
@@ -56,15 +67,7 @@ def native_value(self) -> str | None:
     @property
     def extra_state_attributes(self) -> dict[str, str] | None:
         """Return action metadata."""
-        data = self._action_data()
-        attrs: dict[str, str] = {}
-        if self._group == "batteries" and (charge_source := data.get("charge_source")):
-            charge_source_code = str(charge_source)
-            attrs["charge_source"] = charge_source_code
-            attrs["charge_source_friendly"] = friendly_charge_source_label(
-                charge_source_code
-            )
-        return attrs or None
+        return None
 
 
 class NextActionSensor(SubentryActionSensor):
@@ -86,13 +89,6 @@ def extra_state_attributes(self) -> dict[str, str] | None:
         timestamp = as_datetime(data.get("next_action_timestamp"))
         if timestamp is not None:
             attrs["timestamp"] = timestamp.isoformat()
-
-        if self._group == "batteries" and (charge_source := data.get("next_charge_source")):
-            charge_source_code = str(charge_source)
-            attrs["charge_source"] = charge_source_code
-            attrs["charge_source_friendly"] = friendly_charge_source_label(
-                charge_source_code
-            )
         return attrs or None
 
 

custom_components/wattplan/sensors/common.py

@@ -12,13 +12,6 @@
 from ..const import DOMAIN
 from ..datetime_utils import parse_datetime_like
 
-BATTERY_CHARGE_SOURCE_LABELS: dict[str, str] = {
-    "n": "(N)one",
-    "g": "(G)rid",
-    "p": "(P)V",
-    "gp": "(G)rid and (P)V",
-}
-
 MAX_EXPOSED_PROJECTED_SAVINGS_PCT = 200.0
 TIMESTAMP_DEVICE_CLASS = SensorDeviceClass.TIMESTAMP
 
@@ -36,8 +29,3 @@ def entry_device_info(config_entry: ConfigEntry) -> DeviceInfo:
 def as_datetime(value: Any) -> datetime | None:
     """Convert a dynamic value to datetime when possible."""
     return parse_datetime_like(value)
-
-
-def friendly_charge_source_label(charge_source: str) -> str:
-    """Return a user-facing charge source label for compact planner codes."""
-    return BATTERY_CHARGE_SOURCE_LABELS.get(charge_source, charge_source)

custom_components/wattplan/strings.json

@@ -1120,7 +1120,7 @@
         },
         "data_description": {
           "planning_enabled": "When enabled, optimization runs every {slot_minutes} minutes and publishes the new plan to WattPlan entities.",
-          "action_emission_enabled": "When enabled, action states like charge/discharge/hold are published every {slot_minutes} minutes."
+          "action_emission_enabled": "When enabled, action states like charge_grid/charge_pv/charge_grid_pv/discharge/hold are published every {slot_minutes} minutes."
         }
       },
       "battery_entities": {

custom_components/wattplan/test_plan_invariants.py

@@ -26,17 +26,12 @@ def assert_plan_invariants(result: dict[str, Any]) -> dict[str, Any]:
                 continue
 
             state = str(point.get("state", "hold"))
-            charge_source = int(point.get("charge_source", 0))
-
-            if state == "charge":
-                assert charge_source in {1, 2, 3}, (
-                    f"battery {entity.get('name')} schedule[{index}] is charge "
-                    f"but has invalid charge_source={charge_source}"
-                )
-            elif state in {"hold", "discharge"}:
-                assert charge_source == 0, (
-                    f"battery {entity.get('name')} schedule[{index}] is {state} "
-                    f"but has non-empty charge_source={charge_source}"
-                )
+            assert state in {
+                "hold",
+                "discharge",
+                "charge_grid",
+                "charge_pv",
+                "charge_grid_pv",
+            }, f"battery {entity.get('name')} schedule[{index}] has invalid state={state}"
 
     return result

custom_components/wattplan/translations/en.json

@@ -1120,7 +1120,7 @@
         },
         "data_description": {
           "planning_enabled": "When enabled, optimization runs every {slot_minutes} minutes and publishes the new plan to WattPlan entities.",
-          "action_emission_enabled": "When enabled, action states like charge/discharge/hold are published every {slot_minutes} minutes."
+          "action_emission_enabled": "When enabled, action states like charge_grid/charge_pv/charge_grid_pv/discharge/hold are published every {slot_minutes} minutes."
         }
       },
       "battery_entities": {

docs/entities-and-services.md

@@ -40,7 +40,7 @@ These exist once per configured battery:
 
 | Entity | Purpose |
 | --- | --- |
-| `sensor.<setup_slug>_<battery_name>_action` | Current planned action: `charge`, `discharge`, or `hold`. WattPlan updates this entity on its planning schedule so **your own automation can translate the planned action into a real inverter or battery command**. Includes attributes such as `next_action` and `next_action_timestamp`. |
+| `sensor.<setup_slug>_<battery_name>_action` | Current planned action: `hold`, `discharge`, `charge_grid`, `charge_pv`, or `charge_grid_pv`. WattPlan updates this entity on its planning schedule so **your own automation can translate the planned action into a real inverter or battery command**. The state itself now encodes the planned charging ingress, so there is no separate `charge_source` attribute to inspect. |
 | `sensor.<setup_slug>_<battery_name>_target` | User-supplied target SoC in kWh. Includes a `by` attribute with the requested deadline and returns `unknown` when no active target is set. |
 
 ## Comfort Load Entities