from enum import Enum
from typing import Any, Dict, List, Optional, TypedDict
from dataclasses import dataclass
from .abstract_zone import AbstractZone, ZoneType
# Define a type for the speed configuration
[docs]
class SpeedConfig(TypedDict):
maximum: int
minimum: int
stepIncrement: int
# Define a type for the flow configuration
[docs]
class FlowConfiguration(TypedDict):
name: Optional[str]
pumps: Optional[List[str]]
speed: SpeedConfig
[docs]
class FlowZoneCapabilities(Enum):
"""Enum for flow zone capabilities"""
SUPPORTS_SPEED_PRESETS = "supports_speed_presets"
SUPPORTS_SPEED_PERCENTAGE = "supports_speed_percentage"
SUPPORTS_TURN_ON = "supports_turn_on"
SUPPORTS_TURN_OFF = "supports_turn_off"
[docs]
class FlowZoneInitiator(Enum):
"""Enum for flow zone initiators"""
USER_DEMAND = "UD"
CHECKFLOW = "CF"
PURGE = "PU"
FILTRATION = "FI"
HEATING = "HT"
COOLDOWN = "CD"
HEAT_PUMP = "HTP"
PRESET_NAMES = ["Low", "Medium", "High", "Max"]
[docs]
@dataclass
class FlowZonePreset:
name: str
speed: float
[docs]
class FlowZoneType(Enum):
FLOW_ZONE = "flow_zone"
WATERFALL_ZONE = "waterfall_zone"
BLOWER_ZONE = "blower_zone"
[docs]
@dataclass
class FlowZoneTypeProperties:
"""Properties for different flow zone types."""
format_name: callable
# Type properties mapping zone types to their characteristics
FLOW_ZONE_TYPE_PROPERTIES: Dict[FlowZoneType, FlowZoneTypeProperties] = {
FlowZoneType.WATERFALL_ZONE: FlowZoneTypeProperties(
format_name=lambda zone_id: "Waterfall",
),
FlowZoneType.BLOWER_ZONE: FlowZoneTypeProperties(
format_name=lambda zone_id: "Blower",
),
FlowZoneType.FLOW_ZONE: FlowZoneTypeProperties(
format_name=lambda zone_id: f"Pump {zone_id}",
),
}
[docs]
@AbstractZone.register_zone_type(ZoneType.FLOW_ZONE)
class FlowZone(AbstractZone):
"""State representation for flow zone v1 with validation"""
[docs]
def __init__(self, zone_id: str, config: FlowConfiguration):
"""Initialize FlowZone with zone_id and config."""
# Set default name if not provided
if "name" not in config or config["name"] is None:
# Determine the flow zone type and get its properties
flow_zone_type = self._determine_flow_zone_type(config)
type_props = FLOW_ZONE_TYPE_PROPERTIES[flow_zone_type]
config["name"] = type_props.format_name(zone_id)
super().__init__(
id=zone_id,
zone_type=ZoneType.FLOW_ZONE,
name=config["name"],
config=config
)
# Initialize flow zone specific attributes with defaults
self.active: Optional[bool] = getattr(self, "active", None)
self.speed: Optional[float] = getattr(self, "speed", None)
self.initiators_: Optional[List[FlowZoneInitiator]] = getattr(
self, "initiators_", None
)
# Validate speed if present
if self.speed is not None:
if not isinstance(self.speed, (int, float)):
raise ValueError(
f"Flow speed must be a number, got {type(self.speed).__name__}: {self.speed}"
)
self._validate_speed(self.speed)
@property
def speed_config(self) -> Optional[SpeedConfig]:
"""Get speed configuration if it exists and is properly structured."""
speed_value = self.config.get("speed")
if isinstance(speed_value, dict):
return speed_value # type: ignore
return None
def _validate_speed(self, speed: float) -> None:
"""Validate speed is within acceptable range."""
if self.speed_config:
if not (self.speed_config["minimum"] <= speed <= self.speed_config["maximum"]):
raise ValueError(f"Flow speed {speed}% must be between {self.speed_config['minimum']} and {self.speed_config['maximum']}")
@property
def initiators(self) -> Optional[List[FlowZoneInitiator]]:
return self.initiators_
@staticmethod
def _determine_flow_zone_type(config: FlowConfiguration) -> FlowZoneType:
"""Determine the flow zone type from configuration."""
if config.get("waterfalls") and len(config.get("waterfalls", [])) > 0:
return FlowZoneType.WATERFALL_ZONE
if config.get("blowers") and len(config.get("blowers", [])) > 0:
return FlowZoneType.BLOWER_ZONE
return FlowZoneType.FLOW_ZONE
@property
def type(self) -> FlowZoneType:
"""Get the type of the flow zone."""
return self._determine_flow_zone_type(self.config)
@property
def capabilities(self) -> List[FlowZoneCapabilities]:
"""Get the capabilities of the flow zone."""
capabilities = [
FlowZoneCapabilities.SUPPORTS_TURN_ON,
FlowZoneCapabilities.SUPPORTS_TURN_OFF,
]
if self.speed_config and self.speed_config["stepIncrement"] != 0:
capabilities.append(FlowZoneCapabilities.SUPPORTS_SPEED_PRESETS)
return capabilities
@property
def presets(self) -> List[FlowZonePreset]:
"""Get the speed presets for the flow zone, if supported."""
presets = []
if FlowZoneCapabilities.SUPPORTS_SPEED_PRESETS in self.capabilities and self.speed_config:
step = self.speed_config["stepIncrement"]
min_speed = self.speed_config["minimum"]
max_speed = self.speed_config["maximum"]
preset_speeds = list(range(min_speed, max_speed + 1, step))
for i, speed in enumerate(preset_speeds):
name = PRESET_NAMES[i] if i < len(PRESET_NAMES) else f"Preset {i+1}"
name = PRESET_NAMES[i] if i < len(PRESET_NAMES) else f"Preset {i + 1}"
return presets
[docs]
def get_flow_state(self) -> Dict[str, Any]:
"""Get the current flow state as a simple dictionary."""
return {
"active": self.active,
"speed": self.speed,
"has_initiators": bool(self.initiators_),
}
def _get_runtime_state_fields(self) -> set:
"""Runtime state fields for flow zones."""
return {"active", "speed"}
def _get_field_mappings(self) -> Dict[str, str]:
"""
Flow zone specific field mappings.
Returns:
Dictionary mapping external field names to internal field names
"""
return {
"isActive": "active",
"flowSpeed": "speed",
"pumpSpeed": "speed",
"running": "active",
"enabled": "active",
}
[docs]
def set_speed(self, speed: float, active: Optional[bool] = True) -> None:
"""Set flow speed with validation and optional active state."""
self._validate_speed(speed)
self.speed = speed
if active is not None:
self.active = active
self._publish_desired_state({"speed": speed, "active": self.active})
[docs]
def activate(self) -> None:
"""Activate this zone."""
self._publish_desired_state({"active": True})
[docs]
def deactivate(self) -> None:
"""Deactivate this zone."""
non_user_initiators = (
self.initiators_ is not None and
any(initiator != FlowZoneInitiator.USER_DEMAND.value for initiator in self.initiators_)
)
if non_user_initiators:
raise RuntimeError("Cannot deactivate flow zone with active non-user initiators.")
self._publish_desired_state({"active": False})