File: control.py

package info (click to toggle)
python-bsblan 3.1.6-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,032 kB
  • sloc: python: 4,738; makefile: 3
file content (381 lines) | stat: -rw-r--r-- 13,589 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
# pylint: disable=W0621
"""Asynchronous Python client for BSBLan.

This example demonstrates the optimized hot water functionality:
- HotWaterState: Essential parameters for frequent polling (5 fields)
- HotWaterConfig: Configuration parameters checked less frequently (16 fields)
- HotWaterSchedule: Time program schedules checked occasionally (8 fields)

This three-tier approach reduces API calls by 79% for regular monitoring.
"""

from __future__ import annotations

import asyncio
import os
from datetime import datetime
from typing import Any

from bsblan import (
    BSBLAN,
    BSBLANConfig,
    Device,
    DeviceTime,
    HeatingCircuitStatus,
    HotWaterConfig,
    HotWaterSchedule,
    HotWaterState,
    Info,
    Sensor,
    SetHotWaterParam,
    State,
    StaticState,
    get_hvac_action_category,
)
from bsblan.models import DHWTimeSwitchPrograms


async def get_attribute(
    attribute: Any, attr_type: str = "value", default: str = "N/A"
) -> str:
    """Safely retrieve the desired property ('value' or 'desc') of an attribute.

    Args:
        attribute: The attribute object which may have 'value' or 'desc'.
        attr_type (str): The type of attribute to retrieve ('value' or 'desc').
        default (str): The default value to return if the attribute is None.

    Returns:
        str: The retrieved attribute value or the default.

    """
    if attribute is None:
        return default
    return getattr(attribute, attr_type, default)


def print_attributes(title: str, attributes: dict[str, str]) -> None:
    """Print a set of attributes with their labels under a given title.

    Args:
        title (str): The title for the group of attributes.
        attributes (dict): A dictionary where keys are labels and values are
            attribute values.

    """
    print(f"\n{title}:")
    for label, value in attributes.items():
        print(f"{label}: {value}")


def get_hvac_action_name(status_code: int) -> str:
    """Map BSB-LAN parameter 8000 status code to a human-readable HVAC action.

    BSB-LAN parameter 8000 ("Status heating circuit 1") returns vendor-specific
    status codes. This function maps those codes to simplified HVAC action states
    compatible with Home Assistant and other automation systems.

    Args:
        status_code: The raw status code from parameter 8000 (hvac_action.value).

    Returns:
        str: Human-readable HVAC action name. Returns "idle" for unmapped codes.

    Example:
        >>> state = await bsblan.state()
        >>> if state.hvac_action is not None:
        ...     action = get_hvac_action_name(state.hvac_action.value)
        ...     print(f"Current HVAC action: {action}")

    """
    # Use the new enum-based approach
    category = get_hvac_action_category(status_code)
    return category.name.lower()


async def print_state(state: State) -> None:
    """Print the current state of the BSBLan device.

    Args:
        state (State): The current state of the BSBLan device.

    """
    # Get the HVAC action - both the raw value and mapped action name
    hvac_action_desc = await get_attribute(state.hvac_action, "desc", "Unknown Action")
    hvac_action_value = await get_attribute(state.hvac_action, "value", "N/A")

    # Map the raw status code to a simplified action name using the new enum approach
    hvac_action_mapped = "N/A"
    status_name = "N/A"
    if hvac_action_value != "N/A":
        try:
            status_code = int(hvac_action_value)
            # Get the category (heating, cooling, etc.)
            hvac_action_mapped = get_hvac_action_name(status_code)
            # Get the specific status name from the enum (if known)
            status = HeatingCircuitStatus.from_value(status_code)
            status_name = status.name if status else "UNKNOWN"
        except (ValueError, TypeError):
            hvac_action_mapped = "unknown"

    attributes = {
        "HVAC Action (raw value)": str(hvac_action_value),
        "HVAC Action (device desc)": hvac_action_desc,
        "HVAC Action (status name)": status_name,
        "HVAC Action (category)": hvac_action_mapped,
        "HVAC Mode": await get_attribute(state.hvac_mode, "desc", "Unknown Mode"),
        "Current Temperature": await get_attribute(
            state.current_temperature, "value", "N/A"
        ),
    }
    print_attributes("Device State", attributes)


async def print_sensor(sensor: Sensor) -> None:
    """Print sensor information from the BSBLan device.

    Args:
        sensor (Sensor): The sensor information from the BSBLan device.

    """
    attributes = {
        "Outside Temperature": await get_attribute(
            sensor.outside_temperature, "value", "N/A"
        ),
        "Current Temperature": await get_attribute(
            sensor.current_temperature, "value", "N/A"
        ),
    }
    print_attributes("Sensor Information", attributes)


async def print_device_time(device_time: DeviceTime) -> None:
    """Print device time information.

    Args:
        device_time (DeviceTime): The device time information from the BSBLan device.

    """
    attributes = {
        "Current Time": await get_attribute(device_time.time, "value", "N/A"),
        "Time Unit": await get_attribute(device_time.time, "unit", "N/A"),
        "Time Description": await get_attribute(device_time.time, "desc", "N/A"),
    }
    print_attributes("Device Time", attributes)


async def print_device_info(device: Device, info: Info) -> None:
    """Print device and general information.

    Args:
        device (Device): The device information from the BSBLan device.
        info (Info): The general information from the BSBLan device.

    """
    device_identification = await get_attribute(
        info.device_identification, "value", "N/A"
    )

    attributes = {
        "Device Name": device.name if device.name else "N/A",
        "Version": device.version if device.version else "N/A",
        "Device Identification": device_identification,
    }
    print_attributes("Device Information", attributes)


async def print_static_state(static_state: StaticState) -> None:
    """Print static state information.

    Args:
        static_state (StaticState): The static state information from the BSBLan device.

    """
    min_temp = await get_attribute(static_state.min_temp, "value", "N/A")
    max_temp = await get_attribute(static_state.max_temp, "value", "N/A")
    min_temp_unit = await get_attribute(static_state.min_temp, "unit", "N/A")

    attributes = {
        "Min Temperature": min_temp,
        "Max Temperature": max_temp,
        "Min Temperature Unit": min_temp_unit,
    }
    print_attributes("Static State", attributes)


async def print_hot_water_state(hot_water_state: HotWaterState) -> None:
    """Print essential hot water state information.

    Args:
        hot_water_state (HotWaterState): The essential hot water state information
            from the BSBLan device (optimized for frequent polling).

    """
    attributes = {
        "Operating Mode": await get_attribute(
            hot_water_state.operating_mode, "desc", "Unknown Mode"
        ),
        "Nominal Setpoint": await get_attribute(
            hot_water_state.nominal_setpoint, "value", "N/A"
        ),
        "Release": await get_attribute(hot_water_state.release, "desc", "N/A"),
        "Current Temperature": await get_attribute(
            hot_water_state.dhw_actual_value_top_temperature, "value", "N/A"
        ),
        "DHW Pump State": await get_attribute(
            hot_water_state.state_dhw_pump, "desc", "N/A"
        ),
    }
    print_attributes("Hot Water State (Essential)", attributes)


async def print_hot_water_config(hot_water_config: HotWaterConfig) -> None:
    """Print hot water configuration information.

    Args:
        hot_water_config (HotWaterConfig): The hot water configuration information
            from the BSBLan device (checked less frequently).

    """
    attributes = {
        "Nominal Setpoint Max": await get_attribute(
            hot_water_config.nominal_setpoint_max, "value", "N/A"
        ),
        "Reduced Setpoint": await get_attribute(
            hot_water_config.reduced_setpoint, "value", "N/A"
        ),
        "Legionella Function": await get_attribute(
            hot_water_config.legionella_function, "desc", "N/A"
        ),
        "Legionella Setpoint": await get_attribute(
            hot_water_config.legionella_function_setpoint, "value", "N/A"
        ),
        "Legionella Periodicity": await get_attribute(
            hot_water_config.legionella_function_periodicity, "value", "N/A"
        ),
        "Circulation Pump Release": await get_attribute(
            hot_water_config.dhw_circulation_pump_release, "desc", "N/A"
        ),
        "Circulation Setpoint": await get_attribute(
            hot_water_config.dhw_circulation_setpoint, "value", "N/A"
        ),
    }
    print_attributes("Hot Water Configuration", attributes)


async def print_hot_water_schedule(hot_water_schedule: HotWaterSchedule) -> None:
    """Print hot water schedule information.

    Args:
        hot_water_schedule (HotWaterSchedule): The hot water schedule information
            from the BSBLan device (time programs).

    """
    attributes = {
        "Monday": await get_attribute(
            hot_water_schedule.dhw_time_program_monday, "value", "N/A"
        ),
        "Tuesday": await get_attribute(
            hot_water_schedule.dhw_time_program_tuesday, "value", "N/A"
        ),
        "Wednesday": await get_attribute(
            hot_water_schedule.dhw_time_program_wednesday, "value", "N/A"
        ),
        "Thursday": await get_attribute(
            hot_water_schedule.dhw_time_program_thursday, "value", "N/A"
        ),
        "Friday": await get_attribute(
            hot_water_schedule.dhw_time_program_friday, "value", "N/A"
        ),
        "Saturday": await get_attribute(
            hot_water_schedule.dhw_time_program_saturday, "value", "N/A"
        ),
        "Sunday": await get_attribute(
            hot_water_schedule.dhw_time_program_sunday, "value", "N/A"
        ),
        "Standard Values": await get_attribute(
            hot_water_schedule.dhw_time_program_standard_values, "value", "N/A"
        ),
    }
    print_attributes("Hot Water Schedule", attributes)


async def main() -> None:
    """Show example on controlling your BSBLan device."""
    # Create a configuration object
    config = BSBLANConfig(
        host="10.0.2.60",
        passkey=None,
        username=os.getenv("BSBLAN_USER"),  # Compliant
        password=os.getenv("BSBLAN_PASS"),  # Compliant
    )

    # Initialize BSBLAN with the configuration object
    async with BSBLAN(config) as bsblan:
        # Get and print state
        state: State = await bsblan.state()
        await print_state(state)

        # Set thermostat temperature
        print("\nSetting temperature to 18°C")
        await bsblan.thermostat(target_temperature="18")

        # Set HVAC mode (using raw integer: 0=off, 1=auto, 2=eco, 3=heat)
        print("Setting HVAC mode to heat")
        await bsblan.thermostat(hvac_mode=3)  # 3 = heat

        # Get and print sensor information
        sensor: Sensor = await bsblan.sensor()
        await print_sensor(sensor)

        # Get and print device and general info
        device: Device = await bsblan.device()
        info: Info = await bsblan.info()
        await print_device_info(device, info)

        # Get and print device time
        device_time: DeviceTime = await bsblan.time()
        await print_device_time(device_time)

        # Get and print static state
        static_state: StaticState = await bsblan.static_values()
        await print_static_state(static_state)

        # Get hot water state (essential parameters for frequent polling)
        hot_water_state: HotWaterState = await bsblan.hot_water_state()
        await print_hot_water_state(hot_water_state)

        # Get hot water configuration (checked less frequently)
        try:
            hot_water_config: HotWaterConfig = await bsblan.hot_water_config()
            await print_hot_water_config(hot_water_config)
        except Exception as e:  # noqa: BLE001 - Broad exception for demo purposes
            print(f"Hot water configuration not available: {e}")

        # Get hot water schedule (time programs)
        try:
            hot_water_schedule: HotWaterSchedule = await bsblan.hot_water_schedule()
            await print_hot_water_schedule(hot_water_schedule)
        except Exception as e:  # noqa: BLE001 - Broad exception for demo purposes
            print(f"Hot water schedule not available: {e}")

        # Example: Set DHW time program for Monday
        print("\nSetting DHW time program for Monday to 13:00-14:00")

        dhw_programs = DHWTimeSwitchPrograms(
            monday="13:00-14:00 ##:##-##:## ##:##-##:##"
        )
        await bsblan.set_hot_water(SetHotWaterParam(dhw_time_programs=dhw_programs))

        # Example: Set device time
        print("\nSetting device time to current system time")
        # Get current local system time and format it for BSB-LAN (DD.MM.YYYY HH:MM:SS)
        # Note: Using local time intentionally for this demo to sync BSB-LAN
        current_time = datetime.now().replace(microsecond=0)  # noqa: DTZ005 - Demo uses local time
        formatted_time = current_time.strftime("%d.%m.%Y %H:%M:%S")
        print(f"Current system time: {formatted_time}")
        await bsblan.set_time(formatted_time)


if __name__ == "__main__":
    asyncio.run(main())