diff --git a/src/bsblan/bsblan.py b/src/bsblan/bsblan.py index b0b8c7fe..249b8ac5 100644 --- a/src/bsblan/bsblan.py +++ b/src/bsblan/bsblan.py @@ -19,23 +19,11 @@ from .constants import ( API_VERSIONS, - CIRCUIT_HEATING_SECTIONS, - CIRCUIT_PROBE_PARAMS, - CIRCUIT_STATIC_SECTIONS, - CIRCUIT_STATUS_PARAMS, - CIRCUIT_THERMOSTAT_PARAMS, - DHW_TIME_PROGRAM_PARAMS, - HOT_WATER_CONFIG_PARAMS, - HOT_WATER_ESSENTIAL_PARAMS, - HOT_WATER_SCHEDULE_PARAMS, - INACTIVE_CIRCUIT_MARKER, - MAX_VALID_YEAR, - MIN_VALID_YEAR, - SETTABLE_HOT_WATER_PARAMS, - VALID_CIRCUITS, - VALID_HVAC_MODES, APIConfig, + CircuitConfig, ErrorMsg, + HotWaterParams, + Validation, ) from .exceptions import ( BSBLANAuthError, @@ -186,7 +174,7 @@ async def get_available_circuits(self) -> list[int]: """ available: list[int] = [] - for circuit, param_id in CIRCUIT_PROBE_PARAMS.items(): + for circuit, param_id in CircuitConfig.PROBE_PARAMS.items(): try: response = await self._request( params={"Parameter": param_id}, @@ -200,7 +188,7 @@ async def get_available_circuits(self) -> list[int]: # Inactive circuits either: # - return value="0" and desc="---" # - return an empty dict {} (param not supported) - status_id = CIRCUIT_STATUS_PARAMS[circuit] + status_id = CircuitConfig.STATUS_PARAMS[circuit] status_resp = await self._request( params={"Parameter": status_id}, ) @@ -216,7 +204,7 @@ async def get_available_circuits(self) -> list[int]: # value="0" + desc="---" means inactive if ( - status_data.get("desc") == INACTIVE_CIRCUIT_MARKER + status_data.get("desc") == CircuitConfig.INACTIVE_MARKER and str(status_data.get("value", "")) == "0" ): logger.debug( @@ -674,7 +662,7 @@ def _validate_circuit(self, circuit: int) -> None: BSBLANInvalidParameterError: If the circuit number is invalid. """ - if circuit not in VALID_CIRCUITS: + if circuit not in CircuitConfig.VALID: msg = ErrorMsg.INVALID_CIRCUIT.format(circuit) raise BSBLANInvalidParameterError(msg) @@ -1012,7 +1000,7 @@ async def state( """ self._validate_circuit(circuit) section: SectionLiteral = cast( - "SectionLiteral", CIRCUIT_HEATING_SECTIONS[circuit] + "SectionLiteral", CircuitConfig.HEATING_SECTIONS[circuit] ) return await self._fetch_section_data(section, State, include) @@ -1060,7 +1048,7 @@ async def static_values( """ self._validate_circuit(circuit) section: SectionLiteral = cast( - "SectionLiteral", CIRCUIT_STATIC_SECTIONS[circuit] + "SectionLiteral", CircuitConfig.STATIC_SECTIONS[circuit] ) return await self._fetch_section_data(section, StaticState, include) @@ -1180,7 +1168,7 @@ async def _prepare_thermostat_state( dict[str, Any]: The prepared state for the thermostat. """ - param_ids = CIRCUIT_THERMOSTAT_PARAMS[circuit] + param_ids = CircuitConfig.THERMOSTAT_PARAMS[circuit] state: dict[str, Any] = {} if target_temperature is not None: await self._validate_target_temperature( @@ -1262,7 +1250,7 @@ def _validate_hvac_mode(self, hvac_mode: int) -> None: BSBLANInvalidParameterError: If the HVAC mode is invalid. """ - if hvac_mode not in VALID_HVAC_MODES: + if hvac_mode not in Validation.HVAC_MODES: raise BSBLANInvalidParameterError(str(hvac_mode)) def _validate_time_format(self, time_value: str) -> None: @@ -1276,7 +1264,7 @@ def _validate_time_format(self, time_value: str) -> None: """ try: - validate_time_format(time_value, MIN_VALID_YEAR, MAX_VALID_YEAR) + validate_time_format(time_value, Validation.MIN_YEAR, Validation.MAX_YEAR) except ValueError as err: raise BSBLANInvalidParameterError(str(err)) from err @@ -1379,7 +1367,7 @@ async def hot_water_state(self, include: list[str] | None = None) -> HotWaterSta """ return await self._fetch_hot_water_data( - param_filter=HOT_WATER_ESSENTIAL_PARAMS, + param_filter=HotWaterParams.ESSENTIAL, model_class=HotWaterState, error_msg="No essential hot water parameters available", group_name="essential", @@ -1419,7 +1407,7 @@ async def hot_water_config( """ return await self._fetch_hot_water_data( - param_filter=HOT_WATER_CONFIG_PARAMS, + param_filter=HotWaterParams.CONFIG, model_class=HotWaterConfig, error_msg="No hot water configuration parameters available", group_name="config", @@ -1455,7 +1443,7 @@ async def hot_water_schedule( """ return await self._fetch_hot_water_data( - param_filter=HOT_WATER_SCHEDULE_PARAMS, + param_filter=HotWaterParams.SCHEDULE, model_class=HotWaterSchedule, error_msg="No hot water schedule parameters available", group_name="schedule", @@ -1548,7 +1536,9 @@ async def set_hot_water_schedule(self, schedule: DHWSchedule) -> None: # Invert DHW_TIME_PROGRAM_PARAMS to get day_name -> param_id mapping # Exclude standard_values as it's not a day of the week day_param_map = { - v: k for k, v in DHW_TIME_PROGRAM_PARAMS.items() if v != "standard_values" + v: k + for k, v in HotWaterParams.TIME_PROGRAMS.items() + if v != "standard_values" } for day_name, param_id in day_param_map.items(): @@ -1580,14 +1570,14 @@ def _prepare_hot_water_state( state: dict[str, Any] = {} # Process all mapped parameters using constants - for param_id, attr_name in SETTABLE_HOT_WATER_PARAMS.items(): + for param_id, attr_name in HotWaterParams.SETTABLE.items(): value = getattr(params, attr_name) if value is not None: state.update({"Parameter": param_id, "Value": str(value), "Type": "1"}) # Process time programs if provided using constants if params.dhw_time_programs: - for param_id, attr_name in DHW_TIME_PROGRAM_PARAMS.items(): + for param_id, attr_name in HotWaterParams.TIME_PROGRAMS.items(): value = getattr(params.dhw_time_programs, attr_name) if value is not None: state.update({"Parameter": param_id, "Value": value, "Type": "1"}) diff --git a/src/bsblan/constants.py b/src/bsblan/constants.py index 731574fb..91a85597 100644 --- a/src/bsblan/constants.py +++ b/src/bsblan/constants.py @@ -8,7 +8,6 @@ # Supported heating circuits (1-based) MIN_CIRCUIT: Final[int] = 1 MAX_CIRCUIT: Final[int] = 2 -VALID_CIRCUITS: Final[set[int]] = {1, 2} # API Versions @@ -127,39 +126,33 @@ class APIConfig(TypedDict): "1016": "max_temp", } -# Mapping from circuit number to section names -CIRCUIT_HEATING_SECTIONS: Final[dict[int, str]] = { - 1: "heating", - 2: "heating_circuit2", -} -CIRCUIT_STATIC_SECTIONS: Final[dict[int, str]] = { - 1: "staticValues", - 2: "staticValues_circuit2", -} +# Circuit configuration +class CircuitConfig: + """Circuit-related constants for BSBLAN.""" -# Mapping from circuit number to thermostat parameter IDs -CIRCUIT_THERMOSTAT_PARAMS: Final[dict[int, dict[str, str]]] = { - 1: {"target_temperature": "710", "hvac_mode": "700"}, - 2: {"target_temperature": "1010", "hvac_mode": "1000"}, -} - -# Parameter IDs used to probe whether a heating circuit exists on the device. -# We query the operating mode (hvac_mode) for each circuit. -CIRCUIT_PROBE_PARAMS: Final[dict[int, str]] = { - 1: "700", - 2: "1000", -} - -# Status parameter IDs used as a secondary check for circuit availability. -# Inactive circuits return value="0" and desc="---" for these parameters. -CIRCUIT_STATUS_PARAMS: Final[dict[int, str]] = { - 1: "8000", - 2: "8001", -} - -# Marker value returned by BSB-LAN for parameters on inactive circuits -INACTIVE_CIRCUIT_MARKER: Final[str] = "---" + VALID: Final[set[int]] = {1, 2} + HEATING_SECTIONS: Final[dict[int, str]] = { + 1: "heating", + 2: "heating_circuit2", + } + STATIC_SECTIONS: Final[dict[int, str]] = { + 1: "staticValues", + 2: "staticValues_circuit2", + } + THERMOSTAT_PARAMS: Final[dict[int, dict[str, str]]] = { + 1: {"target_temperature": "710", "hvac_mode": "700"}, + 2: {"target_temperature": "1010", "hvac_mode": "1000"}, + } + PROBE_PARAMS: Final[dict[int, str]] = { + 1: "700", + 2: "1000", + } + STATUS_PARAMS: Final[dict[int, str]] = { + 1: "8000", + 2: "8001", + } + INACTIVE_MARKER: Final[str] = "---" def build_api_config(version: str) -> APIConfig: @@ -208,8 +201,14 @@ def build_api_config(version: str) -> APIConfig: "v3": API_V3, } -# Valid HVAC mode values for validation -VALID_HVAC_MODES: Final[set[int]] = {0, 1, 2, 3} + +# Validation constants +class Validation: + """Validation-related constants for BSBLAN.""" + + HVAC_MODES: Final[set[int]] = {0, 1, 2, 3} + MIN_YEAR: Final[int] = 1900 + MAX_YEAR: Final[int] = 2100 class HVACActionCategory(IntEnum): @@ -494,10 +493,6 @@ class ErrorMsg: ) -# Time validation constants -MIN_VALID_YEAR: Final[int] = 1900 # Reasonable minimum year for BSB-LAN devices -MAX_VALID_YEAR: Final[int] = 2100 # Reasonable maximum year for BSB-LAN devices - # Handle both ASCII and Unicode degree symbols TEMPERATURE_UNITS = {"°C", "°F", "°C", "°F", "°C", "°F"} @@ -572,88 +567,91 @@ class ErrorMsg: "s": "measurement", } + # Hot Water Parameter Groups -# Essential parameters for frequent monitoring -HOT_WATER_ESSENTIAL_PARAMS: Final[set[str]] = { - param_id - for param_id, name in BASE_HOT_WATER_PARAMS.items() - if name - in { - "operating_mode", - "nominal_setpoint", - "release", - "dhw_actual_value_top_temperature", - "state_dhw_pump", +class HotWaterParams: + """Hot water parameter group constants for BSBLAN.""" + + # Essential parameters for frequent monitoring + ESSENTIAL: Final[set[str]] = { + param_id + for param_id, name in BASE_HOT_WATER_PARAMS.items() + if name + in { + "operating_mode", + "nominal_setpoint", + "release", + "dhw_actual_value_top_temperature", + "state_dhw_pump", + } } -} -# Configuration parameters checked less frequently -HOT_WATER_CONFIG_PARAMS: Final[set[str]] = { - param_id - for param_id, name in BASE_HOT_WATER_PARAMS.items() - if name - in { - "eco_mode_selection", - "nominal_setpoint_max", - "reduced_setpoint", - "dhw_charging_priority", - "operating_mode_changeover", - "legionella_function", - "legionella_function_setpoint", - "legionella_function_periodicity", - "legionella_function_day", - "legionella_function_time", - "legionella_function_dwelling_time", - "legionella_circulation_pump", - "legionella_circulation_temp_diff", - "dhw_circulation_pump_release", - "dhw_circulation_pump_cycling", - "dhw_circulation_setpoint", + # Configuration parameters checked less frequently + CONFIG: Final[set[str]] = { + param_id + for param_id, name in BASE_HOT_WATER_PARAMS.items() + if name + in { + "eco_mode_selection", + "nominal_setpoint_max", + "reduced_setpoint", + "dhw_charging_priority", + "operating_mode_changeover", + "legionella_function", + "legionella_function_setpoint", + "legionella_function_periodicity", + "legionella_function_day", + "legionella_function_time", + "legionella_function_dwelling_time", + "legionella_circulation_pump", + "legionella_circulation_temp_diff", + "dhw_circulation_pump_release", + "dhw_circulation_pump_cycling", + "dhw_circulation_setpoint", + } } -} -# Schedule parameters (time programs) -HOT_WATER_SCHEDULE_PARAMS: Final[set[str]] = { - param_id - for param_id, name in BASE_HOT_WATER_PARAMS.items() - if name - in { - "dhw_time_program_monday", - "dhw_time_program_tuesday", - "dhw_time_program_wednesday", - "dhw_time_program_thursday", - "dhw_time_program_friday", - "dhw_time_program_saturday", - "dhw_time_program_sunday", - "dhw_time_program_standard_values", + # Schedule parameters (time programs) + SCHEDULE: Final[set[str]] = { + param_id + for param_id, name in BASE_HOT_WATER_PARAMS.items() + if name + in { + "dhw_time_program_monday", + "dhw_time_program_tuesday", + "dhw_time_program_wednesday", + "dhw_time_program_thursday", + "dhw_time_program_friday", + "dhw_time_program_saturday", + "dhw_time_program_sunday", + "dhw_time_program_standard_values", + } } -} -# Settable hot water parameters mapping (param_id -> attribute name) -# Used by set_hot_water to map SetHotWaterParam attributes to BSB-LAN parameter IDs -SETTABLE_HOT_WATER_PARAMS: Final[dict[str, str]] = { - "1610": "nominal_setpoint", - "1612": "reduced_setpoint", - "1614": "nominal_setpoint_max", - "1600": "operating_mode", - "1601": "eco_mode_selection", - "1630": "dhw_charging_priority", - "1645": "legionella_function_setpoint", - "1641": "legionella_function_periodicity", - "1642": "legionella_function_day", - "1644": "legionella_function_time", - "1646": "legionella_function_dwelling_time", - "1680": "operating_mode_changeover", -} + # Settable parameters mapping (param_id -> attribute name) + SETTABLE: Final[dict[str, str]] = { + "1610": "nominal_setpoint", + "1612": "reduced_setpoint", + "1614": "nominal_setpoint_max", + "1600": "operating_mode", + "1601": "eco_mode_selection", + "1630": "dhw_charging_priority", + "1645": "legionella_function_setpoint", + "1641": "legionella_function_periodicity", + "1642": "legionella_function_day", + "1644": "legionella_function_time", + "1646": "legionella_function_dwelling_time", + "1680": "operating_mode_changeover", + } -# DHW time program parameter mappings -DHW_TIME_PROGRAM_PARAMS: Final[dict[str, str]] = { - "561": "monday", - "562": "tuesday", - "563": "wednesday", - "564": "thursday", - "565": "friday", - "566": "saturday", - "567": "sunday", - "576": "standard_values", -} + # DHW time program parameter mappings + TIME_PROGRAMS: Final[dict[str, str]] = { + "561": "monday", + "562": "tuesday", + "563": "wednesday", + "564": "thursday", + "565": "friday", + "566": "saturday", + "567": "sunday", + "576": "standard_values", + } diff --git a/tests/test_constants.py b/tests/test_constants.py index 076b68e0..337b0ec0 100644 --- a/tests/test_constants.py +++ b/tests/test_constants.py @@ -8,10 +8,8 @@ API_V1, API_V3, BASE_HOT_WATER_PARAMS, - HOT_WATER_CONFIG_PARAMS, - HOT_WATER_ESSENTIAL_PARAMS, - HOT_WATER_SCHEDULE_PARAMS, HeatingCircuitStatus, + HotWaterParams, HVACActionCategory, build_api_config, get_hvac_action_category, @@ -90,7 +88,7 @@ def test_pre_built_api_configurations( def test_hot_water_parameter_groups_completeness() -> None: """Test that hot water parameter groups cover all parameters.""" all_grouped_params = ( - HOT_WATER_ESSENTIAL_PARAMS | HOT_WATER_CONFIG_PARAMS | HOT_WATER_SCHEDULE_PARAMS + HotWaterParams.ESSENTIAL | HotWaterParams.CONFIG | HotWaterParams.SCHEDULE ) # All BASE_HOT_WATER_PARAMS should be categorized into one of the groups @@ -101,9 +99,9 @@ def test_hot_water_parameter_groups_completeness() -> None: @pytest.mark.parametrize( ("group1", "group2"), [ - (HOT_WATER_ESSENTIAL_PARAMS, HOT_WATER_CONFIG_PARAMS), - (HOT_WATER_ESSENTIAL_PARAMS, HOT_WATER_SCHEDULE_PARAMS), - (HOT_WATER_CONFIG_PARAMS, HOT_WATER_SCHEDULE_PARAMS), + (HotWaterParams.ESSENTIAL, HotWaterParams.CONFIG), + (HotWaterParams.ESSENTIAL, HotWaterParams.SCHEDULE), + (HotWaterParams.CONFIG, HotWaterParams.SCHEDULE), ], ) def test_hot_water_parameter_groups_no_overlap( @@ -117,9 +115,9 @@ def test_hot_water_parameter_groups_no_overlap( @pytest.mark.parametrize( ("group", "expected_count"), [ - (HOT_WATER_ESSENTIAL_PARAMS, 5), # Current optimized count - (HOT_WATER_CONFIG_PARAMS, 16), # Configuration parameters - (HOT_WATER_SCHEDULE_PARAMS, 8), # Time program parameters + (HotWaterParams.ESSENTIAL, 5), # Current optimized count + (HotWaterParams.CONFIG, 16), # Configuration parameters + (HotWaterParams.SCHEDULE, 8), # Time program parameters ], ) def test_hot_water_parameter_groups_expected_counts( @@ -133,9 +131,9 @@ def test_hot_water_parameter_groups_expected_counts( def test_hot_water_parameter_groups_total_count() -> None: """Test that total grouped parameters match base parameters.""" total_grouped = ( - len(HOT_WATER_ESSENTIAL_PARAMS) - + len(HOT_WATER_CONFIG_PARAMS) - + len(HOT_WATER_SCHEDULE_PARAMS) + len(HotWaterParams.ESSENTIAL) + + len(HotWaterParams.CONFIG) + + len(HotWaterParams.SCHEDULE) ) assert total_grouped == len(BASE_HOT_WATER_PARAMS)