diff --git a/pykumo/__init__.py b/pykumo/__init__.py index 3b022ee..d4634b1 100644 --- a/pykumo/__init__.py +++ b/pykumo/__init__.py @@ -6,6 +6,12 @@ from .py_kumo import PyKumo from .py_kumo_base import PyKumoBase from .py_kumo_station import PyKumoStation +from .cn105 import ( + DEFAULT_INFO_CODES, + InfoCode, + decode_info_reply, + valid_cn105_reply, +) __version__ = "0.5.3" name = "pykumo" @@ -17,4 +23,8 @@ "PyKumo", "PyKumoBase", "PyKumoStation", + "DEFAULT_INFO_CODES", + "InfoCode", + "decode_info_reply", + "valid_cn105_reply", ] diff --git a/pykumo/cn105.py b/pykumo/cn105.py new file mode 100644 index 0000000..f45c79b --- /dev/null +++ b/pykumo/cn105.py @@ -0,0 +1,252 @@ +"""Build and decode raw CN105/ITP serial frames. + +Frames reach the indoor unit through the Kumo adapter's +``indoorUnit.settings.rawITPFrame`` node (see :class:`pykumo.cn105_bus.Cn105Bus`). +The wire format is:: + + FC | type | 01 30 | payloadLen | payload | checksum + +The checksum is ``(0xFC - sum(preceding_bytes)) & 0xFF``. An info request is type +``0x42`` with the info code in payload byte 0, and the reply is type ``0x62`` +echoing that code. + +Byte numbers below count from the start of the frame, so ``frame[5]`` is the info +code and "byte N" means ``payload[N - 5]``. +""" + +from enum import IntEnum + +PACKET_HEADER = 0xFC +# Standard CN105 sub-header that follows the type byte on every frame. +PACKET_SUBHEADER = bytes([0x01, 0x30]) +# Bytes of framing around the payload: header, type, sub-header, len, checksum. +FRAME_OVERHEAD = 6 +# Info request/response type bytes. +INFO_REQUEST_TYPE = 0x42 +INFO_RESPONSE_TYPE = 0x62 +# Info request payloads are always padded to 16 bytes on the wire. +PAYLOAD_SIZE = 16 +# A signed byte is used for the frame length in firmware, so 1..127 is usable. +MAX_FRAME_LEN = 127 +# The runtime counter has 1-minute resolution, so samples must be at least this +# far apart before a flat counter means "not running". +RUNTIME_SAMPLE_MIN_INTERVAL_SECONDS = 70.0 + + +class InfoCode(IntEnum): + """The info codes you can ask a unit for, sent as the first payload byte.""" + + # Room temperature, outdoor temperature, compressor runtime counter. + TEMPERATURES = 0x03 + # Operating flag and compressor frequency. Not every unit has it, and asking + # one that does not wastes half a minute and disturbs the reads that follow. + COMPRESSOR = 0x06 + # Sub mode, indoor fan stage, auto sub mode. + SUB_MODE = 0x09 + + +# Codes every unit we have tried answers. COMPRESSOR is left out on purpose. +DEFAULT_INFO_CODES = (InfoCode.TEMPERATURES, InfoCode.SUB_MODE) + +# Sub mode names for the 0x09 response, byte 8. +SUB_MODE_NAMES = { + 0x00: "NORMAL", + 0x01: "WARMUP", + 0x02: "DEFROST", + 0x04: "PREHEAT", + 0x08: "STANDBY", + 0x10: "OFF", +} +# Indoor fan stage names for the 0x09 response, byte 9. This describes the fan, +# not whether the compressor is running. +STAGE_NAMES = { + 0x00: "IDLE", + 0x01: "LOW", + 0x02: "GENTLE", + 0x03: "MEDIUM", + 0x04: "MODERATE", + 0x05: "HIGH", + 0x06: "DIFFUSE", +} +# Auto sub mode names for the 0x09 response, byte 10. Two generations of units +# share this byte: older ones use 0x00..0x03, newer MFZ ones use 0x40/0x41/0x43. +AUTO_SUB_MODE_NAMES = { + 0x00: "AUTO_OFF", + 0x01: "AUTO_COOL", + 0x02: "AUTO_HEAT", + 0x03: "AUTO_LEADER", + 0x40: "AUTO_INACTIVE", + 0x41: "AUTO_IDLE", + 0x43: "AUTO_ACTIVE", +} + + +def cn105_checksum(data: bytes) -> int: + """Return the CN105 frame checksum for ``data`` (all preceding bytes).""" + return (0xFC - sum(data)) & 0xFF + + +def build_cn105_frame( + type_byte: int, payload: bytes, pad_to: int = PAYLOAD_SIZE +) -> bytes: + """Build a whole frame: header, payload, then the checksum. + + ``payload`` gets padded with zeros out to ``pad_to`` bytes, which is how info + requests (``0x42``) and set commands (``0x41``) look on the wire. Pass + ``pad_to=0`` for frames with a short payload, like connect. + """ + if not 0 <= type_byte <= 0xFF: + raise ValueError("type_byte must be 0..255") + if not 0 <= pad_to <= PAYLOAD_SIZE: + raise ValueError(f"pad_to must be 0..{PAYLOAD_SIZE}") + payload = bytes(payload).ljust(pad_to, b"\x00") + if len(payload) > PAYLOAD_SIZE: + raise ValueError(f"payload must be <= {PAYLOAD_SIZE} bytes") + body = bytes([PACKET_HEADER, type_byte]) + PACKET_SUBHEADER + body += bytes([len(payload)]) + payload + return body + bytes([cn105_checksum(body)]) + + +def build_info_request(code: int) -> bytes: + """Build an info-request frame (type ``0x42``) for the given info ``code``.""" + if not 0 <= code <= 0xFF: + raise ValueError("code must be 0..255") + return build_cn105_frame(INFO_REQUEST_TYPE, bytes([code])) + + +def valid_cn105_reply(frame) -> bool: + """True if ``frame`` begins with a well-formed frame whose checksum matches. + + Trailing bytes are ignored, since the adapter sometimes hands back more than + one frame at once. + """ + if not frame or len(frame) < FRAME_OVERHEAD: + return False + if frame[0] != PACKET_HEADER or frame[2:4] != PACKET_SUBHEADER: + return False + end = frame[4] + FRAME_OVERHEAD + if len(frame) < end: + return False + return cn105_checksum(frame[: end - 1]) == frame[end - 1] + + +def is_info_reply(frame, code: int) -> bool: + """True if ``frame`` is a valid ``0x62`` reply to a request for ``code``.""" + if not valid_cn105_reply(frame): + return False + return frame[1] == INFO_RESPONSE_TYPE and frame[5] == code + + +def _decode_temperatures(frame) -> dict: + """Read the temperatures and the runtime counter out of a ``0x03`` reply.""" + room = outdoor = runtime = None + # Byte 10, (b - 128) / 2. Anything <= 1 means the unit has no reading, which + # many report while the compressor is idle. + if len(frame) > 10 and frame[10] > 1: + outdoor = (frame[10] - 128) / 2 + # Byte 11 if it has a value, otherwise the older byte 8 scale (0x00..0x1F is + # 10..41 C). + if len(frame) > 11 and frame[11]: + room = (frame[11] - 128) / 2 + elif len(frame) > 8 and frame[8] <= 0x1F: + room = 10 + frame[8] + # 24-bit big-endian across bytes 16..18. It only ticks up while the compressor + # runs, about once a minute, so it measures compressor time rather than how + # long the unit has been powered on. + if len(frame) > 18: + runtime = (frame[16] << 16) | (frame[17] << 8) | frame[18] + return { + "room_temperature": room, + "outdoor_temperature": outdoor, + "compressor_runtime_minutes": runtime, + } + + +def _decode_compressor(frame) -> dict: + """Read the operating flag and compressor frequency out of a ``0x06`` reply.""" + operating = frequency = None + if len(frame) > 9: + # Byte 9: 1 = compressor running, 0 = standby. + operating = frame[9] == 1 + # Byte 8, but report 0 while idle since some units put noise here. + frequency = frame[8] if operating else 0 + return {"operating": operating, "compressor_frequency": frequency} + + +def _decode_sub_mode(frame) -> dict: + """Read the sub mode, fan stage, and auto sub mode out of a ``0x09`` reply.""" + return { + "sub_mode": SUB_MODE_NAMES.get(frame[8]) if len(frame) > 8 else None, + "stage": STAGE_NAMES.get(frame[9]) if len(frame) > 9 else None, + "auto_sub_mode": ( + AUTO_SUB_MODE_NAMES.get(frame[10]) if len(frame) > 10 else None + ), + } + + +_DECODERS = { + InfoCode.TEMPERATURES: _decode_temperatures, + InfoCode.COMPRESSOR: _decode_compressor, + InfoCode.SUB_MODE: _decode_sub_mode, +} + +# Which fields each info code reports, for callers that need the names before +# reading anything. Built by decoding an empty frame, so it always matches what +# the decoders actually return. +TELEMETRY_KEYS = {code: tuple(decoder(b"")) for code, decoder in _DECODERS.items()} + + +def decode_info_reply(frame, code: int) -> dict: + """Decode a reply into a ``{field: value}`` dict. + + Every field for ``code`` is there, ``None`` where the frame has no usable + value. ``code`` is required because a stale reply in the adapter's buffer would + otherwise look fresh. Raises ``ValueError`` for a code with no decoder. + """ + decoder = _DECODERS.get(code) + if decoder is None: + raise ValueError(f"no CN105 decoder for info code 0x{code:02x}") + return decoder(frame if is_info_reply(frame, code) else b"") + + +class CompressorActivityEstimator: + """Work out whether the compressor is running from the ``0x03`` runtime counter. + + The counter only moves while the compressor runs, so if it changed between two + readings at least ``min_interval`` apart, the compressor was running. Used + where info code ``0x06``, the real operating flag, is unsafe to ask for. + + This is inherently slow, with one to two minutes of delay from the actual + state, because of how the detection works. + """ + + def __init__(self, min_interval: float = RUNTIME_SAMPLE_MIN_INTERVAL_SECONDS): + self._min_interval = min_interval + # The reading we compare against, as (time, runtime minutes). + self._sample = None + self._running = None + + @property + def running(self) -> bool | None: + """Whether the compressor is running, or ``None`` if we cannot tell yet.""" + return self._running + + def update(self, runtime_minutes, now: float) -> bool | None: + """Add a counter reading and return whether the compressor is running.""" + if runtime_minutes is None: + return self._running + if self._sample is None: + self._sample = (now, runtime_minutes) + return self._running + sampled_at, sampled_minutes = self._sample + if runtime_minutes == sampled_minutes and now - sampled_at < self._min_interval: + # Keep the older reading so the gap between them keeps growing. + return self._running + self._sample = (now, runtime_minutes) + self._running = runtime_minutes > sampled_minutes + return self._running + + def reset(self) -> None: + """Throw away the stored reading and start over.""" + self._sample = None + self._running = None diff --git a/pykumo/cn105_bus.py b/pykumo/cn105_bus.py new file mode 100644 index 0000000..4f021b5 --- /dev/null +++ b/pykumo/cn105_bus.py @@ -0,0 +1,179 @@ +"""Send and receive raw CN105 frames through a Kumo adapter.""" + +import logging +import threading +import time + +from .cn105 import ( + DEFAULT_INFO_CODES, + INFO_RESPONSE_TYPE, + MAX_FRAME_LEN, + build_info_request, + valid_cn105_reply, +) + +_LOGGER = logging.getLogger(__name__) + +# How often and how long to check for a reply after sending a frame. The adapter +# only holds a reply briefly, so we check repeatedly. The same unit answers some +# codes much faster than others (0x03 in about 3 s, 0x09 in about 11 s) and never +# answers a few, so the window is generous and giving up just returns None. +POLL_INTERVAL_SECONDS = 0.5 +REPLY_TIMEOUT_SECONDS = 20.0 + + +class Cn105Bus: + """Talks to one adapter's ``rawITPFrame`` node, one exchange at a time. + + The adapter has a single buffer for replies, so sending a frame and reading + its answer happen together under one lock. That only covers threads in this + process: another program on the network can still overwrite the buffer. + """ + + def __init__(self, unit): + self._unit = unit + self._lock = threading.RLock() + self._answered = set() + self._unsupported = set() + + @property + def unsupported_codes(self) -> frozenset: + """Codes that never answered, which :meth:`read_info` no longer sends.""" + return frozenset(self._unsupported) + + def forget_unsupported(self) -> None: + """Allow dropped codes to be sent again, e.g. after an adapter reboot.""" + self._unsupported.clear() + + def send(self, frame: bytes, id_byte: int = 1) -> bool: + """Send a raw frame to the indoor unit. + + ``frame`` is the whole wire frame (``FC | type | 01 30 | len | payload | + checksum``), 1 to 127 bytes. True if the adapter accepted it. + """ + frame = bytes(frame) + length = len(frame) + if not 1 <= length <= MAX_FRAME_LEN: + _LOGGER.warning( + "%s: raw CN105 frame length %d out of range 1..%d", + self._unit.get_name(), + length, + MAX_FRAME_LEN, + ) + return False + if not 0 <= id_byte <= 0xFF: + _LOGGER.warning( + "%s: CN105 id byte %r out of range", self._unit.get_name(), id_byte + ) + return False + command = ( + '{"c":{"indoorUnit":{"settings":{"rawITPFrame":' + '{"frame":"%s","len":%d,"id":%d}}}}}' % (frame.hex(), length, id_byte) + ).encode("utf-8") + with self._lock: + response = self._unit._request(command) # pylint: disable=protected-access + if not response or "_api_error" in response: + _LOGGER.warning( + "%s: failed to send raw CN105 frame: %s", + self._unit.get_name(), + response, + ) + return False + return True + + def read(self) -> bytes | None: + """Read back the last reply the adapter is holding. + + Returns None if the buffer is empty or the response cannot be parsed. + """ + query = b'{"c":{"indoorUnit":{"settings":{"rawITPFrame":{}}}}}' + with self._lock: + response = self._unit._request(query) # pylint: disable=protected-access + try: + node = response["r"]["indoorUnit"]["settings"]["rawITPFrame"] + except (KeyError, TypeError): + return None + hexstr = node.get("frame") if isinstance(node, dict) else None + if not hexstr: + return None + try: + return bytes.fromhex(hexstr) + except ValueError: + _LOGGER.warning( + "%s: raw CN105 readback is not valid hex: %r", + self._unit.get_name(), + hexstr, + ) + return None + + def transceive( + self, + frame: bytes, + id_byte: int = 1, + expect_type: int | None = None, + expect_code: int | None = None, + timeout: float = REPLY_TIMEOUT_SECONDS, + ) -> bytes | None: + """Send a frame once and wait for the unit's reply. + + Polls the reply buffer, checking the checksum of whatever comes back. Pass + ``expect_type`` and ``expect_code`` to skip replies that do not match. + Without them, what is left in the buffer from the previous exchange looks + like an answer to this one. Returns None if nothing matching arrives. + + ``timeout`` bounds how many times we poll rather than wall-clock time. + Each poll costs half a second of sleep plus a request, so the real wait + runs longer than ``timeout``. The whole wait uses one connection. + """ + with self._lock, self._unit.request_cycle(): + if not self.send(frame, id_byte): + return None + poll_count = max(1, int(timeout / POLL_INTERVAL_SECONDS)) + for _ in range(poll_count): + time.sleep(POLL_INTERVAL_SECONDS) + reply = self.read() + if not valid_cn105_reply(reply): + continue + if expect_type is not None and reply[1] != expect_type: + continue + if expect_code is not None and reply[5] != expect_code: + continue + return reply + _LOGGER.debug( + "%s: no valid CN105 reply within %.1fs", self._unit.get_name(), timeout + ) + return None + + def read_info( + self, code: int, timeout: float = REPLY_TIMEOUT_SECONDS + ) -> bytes | None: + """Ask for info ``code`` and return the reply, or None. + + If a code outside :data:`~pykumo.cn105.DEFAULT_INFO_CODES` misses and has + never answered, the bus stops sending it. Asking a unit for a code it does + not implement wastes half a minute and disturbs the reads that follow, so + one miss is enough to give up. A code that has answered before is kept. + """ + if code in self._unsupported: + return None + reply = self.transceive( + build_info_request(code), + expect_type=INFO_RESPONSE_TYPE, + expect_code=code, + timeout=timeout, + ) + if reply is not None: + self._answered.add(code) + return reply + if code not in DEFAULT_INFO_CODES and code not in self._answered: + self._unsupported.add(code) + _LOGGER.warning( + "%s: info code 0x%02x did not answer in %.1fs and never has, so " + "it will not be sent again. This unit probably does not support " + "it, and CN105 reads may stay stuck until the adapter reboots. " + "Call forget_unsupported() to try again.", + self._unit.get_name(), + code, + timeout, + ) + return None diff --git a/pykumo/py_kumo.py b/pykumo/py_kumo.py index e2c82a1..2773b30 100644 --- a/pykumo/py_kumo.py +++ b/pykumo/py_kumo.py @@ -8,6 +8,13 @@ from .schedule import UnitSchedule from .const import CACHE_INTERVAL_SECONDS, POSSIBLE_SENSORS, SETTABLE_TEMP_SOURCES +from .cn105 import ( + DEFAULT_INFO_CODES, + TELEMETRY_KEYS, + CompressorActivityEstimator, + decode_info_reply, +) +from .cn105_bus import REPLY_TIMEOUT_SECONDS, Cn105Bus from .py_kumo_base import PyKumoBase _LOGGER = logging.getLogger(__name__) @@ -50,6 +57,11 @@ def __init__( """Constructor""" self._last_reboot = None self._unit_schedule = UnitSchedule(self) if use_schedule else None + self._cn105 = Cn105Bus(self) + # What the last successful update_cn105_telemetry() read. + self._cn105_telemetry = {} + self._cn105_telemetry_at = None + self._compressor_activity = CompressorActivityEstimator() super().__init__(name, addr, cfg_json, timeouts, serial) def _rebootable_response(self, response): @@ -732,6 +744,120 @@ def set_hold(self, end_time): self._last_status_update = time.monotonic() - 2 * CACHE_INTERVAL_SECONDS return response + def get_cn105_bus(self) -> Cn105Bus: + """Return the raw frame interface, for other info codes and experiments.""" + return self._cn105 + + def update_cn105_telemetry( + self, codes=None, timeout: float = REPLY_TIMEOUT_SECONDS + ) -> bool: + """Read fresh CN105 data. True if at least one code answered. + + Asks for one info code at a time and caches what comes back, which is what + the ``get_*`` methods below return. This never raises. A code that fails + leaves its own fields None and the others alone. + + ``codes`` defaults to :data:`~pykumo.cn105.DEFAULT_INFO_CODES` (``0x03`` + and ``0x09``). Ask for ``InfoCode.COMPRESSOR`` (``0x06``) only on units you + know support it. On a unit that does not, it costs about 30 s to give up + and usually takes the codes after it in the same refresh down with it. + + Waits for each reply, a few seconds per code, so call it from an executor + rather than an event loop. + """ + answered = False + runtime_sample = None + for code in DEFAULT_INFO_CODES if codes is None else codes: + if code not in TELEMETRY_KEYS: + _LOGGER.warning( + "%s: no CN105 fields known for info code 0x%02x", self._name, code + ) + continue + self._cn105_telemetry.update(dict.fromkeys(TELEMETRY_KEYS[code])) + try: + reply = self._cn105.read_info(code, timeout=timeout) + except Exception: # pylint: disable=broad-except + _LOGGER.warning( + "%s: CN105 read for info code 0x%02x failed", + self._name, + code, + exc_info=True, + ) + continue + if reply is None: + continue + fields = decode_info_reply(reply, code) + self._cn105_telemetry.update(fields) + if "compressor_runtime_minutes" in fields: + # Time the counter when it was read, not when the whole refresh + # ends. A code that times out adds tens of seconds, which would + # otherwise make a fresh reading look far enough from the last + # one to be compared against it. + runtime_sample = ( + fields["compressor_runtime_minutes"], + time.monotonic(), + ) + answered = True + now = time.monotonic() + if answered: + self._cn105_telemetry_at = now + if runtime_sample is not None: + self._compressor_activity.update(*runtime_sample) + return answered + + def get_cn105_telemetry(self) -> dict: + """Everything read on the last refresh. + + ``operating`` holds whatever :meth:`is_compressor_running` reports, so it + is there even when info code ``0x06`` was not asked for. Empty until the + first refresh. + """ + if not self._cn105_telemetry: + return {} + return {**self._cn105_telemetry, "operating": self.is_compressor_running()} + + def get_cn105_telemetry_age(self) -> float | None: + """How many seconds ago the data came back, or None if it never has.""" + if self._cn105_telemetry_at is None: + return None + return time.monotonic() - self._cn105_telemetry_at + + def get_outdoor_temperature(self) -> float | None: + """Outdoor temperature in C, as of the last refresh. + + The adapter reports this as null in its own status. + """ + return self._cn105_telemetry.get("outdoor_temperature") + + def get_raw_room_temperature(self) -> float | None: + """Room temperature in C as the unit itself measures it. + + Not :meth:`get_current_temperature`, which uses whatever sensor the + adapter was told to use. + """ + return self._cn105_telemetry.get("room_temperature") + + def get_compressor_runtime_minutes(self) -> int | None: + """How many minutes the compressor has run, as of the last refresh. + + This is compressor time, not power-on time. + """ + return self._cn105_telemetry.get("compressor_runtime_minutes") + + def is_compressor_running(self) -> bool | None: + """Whether the compressor is running, or None if we cannot tell. + + Uses the ``0x06`` operating flag if you asked for that code. Otherwise it + infers the answer from the runtime counter, which needs two refreshes at + least 70 s apart and lags the real state by a minute or two. + """ + operating = self._cn105_telemetry.get("operating") + if operating is not None: + return operating + if self.get_mode() in ("off", "idle"): + return False + return self._compressor_activity.running + def do_reboot(self): """Issue a reboot command to the indoor unit's adapter.""" command = ('{"c":{"adapter":{"status":{"runState":"reboot"}}}}').encode("utf-8") diff --git a/pykumo/py_kumo_base.py b/pykumo/py_kumo_base.py index 5ab0a05..0dc8f4f 100644 --- a/pykumo/py_kumo_base.py +++ b/pykumo/py_kumo_base.py @@ -6,6 +6,7 @@ import time import logging import threading +from contextlib import contextmanager import requests from requests.adapters import HTTPAdapter from requests.exceptions import Timeout @@ -171,6 +172,26 @@ def close(self): """ self.end_cycle() + def _in_cycle(self) -> bool: + """True if this thread already has a connection open to this unit.""" + return self._address in getattr(_tl, "cycles", set()) + + @contextmanager + def request_cycle(self): + """Keep one connection open for every request inside this block. + + Safe to nest: an inner block joins the one its caller opened and leaves + closing it to that caller. + """ + nested = self._in_cycle() + if not nested: + self.begin_cycle() + try: + yield + finally: + if not nested: + self.end_cycle() + def _request(self, post_data): """Send request to configured unit and return response dict. diff --git a/tests/cn105_frames.py b/tests/cn105_frames.py new file mode 100644 index 0000000..1185439 --- /dev/null +++ b/tests/cn105_frames.py @@ -0,0 +1,49 @@ +"""Shared CN105 frame fixtures for the tests.""" + +from pykumo.cn105 import build_cn105_frame, cn105_checksum + +# Full 0x62 reply frames with valid checksums. +FRAME_03_HEX = "fc620130100300000c00b6acadfe4200021abe000025" +FRAME_03_ALT_HEX = "fc620130100300000d00b0aeaefe4200021aa0000045" +FRAME_09_OFF_HEX = "fc620130100900001000400000000000000000000004" +FRAME_09_ACTIVE_HEX = "fc620130100900000002400000000000000000000012" + +# Known-good info request frames. +INFO_03_REQUEST_HEX = "fc42013010030000000000000000000000000000007a" +INFO_06_REQUEST_HEX = "fc420130100600000000000000000000000000000077" +INFO_09_REQUEST_HEX = "fc420130100900000000000000000000000000000074" + + +def make_0x03_reply(outdoor_byte=0x01, room_b_byte=0x00, legacy_byte=0x00, code=0x03): + """Build a 0x62 reply to a 0x03 request carrying the given temperatures.""" + payload = bytearray(16) + payload[0] = code # echoed info code -> raw[5] + payload[3] = legacy_byte # raw[8] + payload[5] = outdoor_byte # raw[10] + payload[6] = room_b_byte # raw[11] + return build_cn105_frame(0x62, bytes(payload)) + + +def make_0x06_reply(freq_byte=0x00, operating_byte=0x00, code=0x06): + """Build a valid 0x62 reply to a 0x06 request.""" + payload = bytearray(16) + payload[0] = code # echoed info code -> raw[5] + payload[3] = freq_byte # raw[8], compressor Hz + payload[4] = operating_byte # raw[9], operating flag + return build_cn105_frame(0x62, bytes(payload)) + + +def make_0x09_reply(sub_byte=0x00, stage_byte=0x00, auto_byte=0x00, code=0x09): + """Build a valid 0x62 reply to a 0x09 request.""" + payload = bytearray(16) + payload[0] = code # echoed info code -> raw[5] + payload[3] = sub_byte # raw[8], sub mode + payload[4] = stage_byte # raw[9], stage + payload[5] = auto_byte # raw[10], auto sub mode + return build_cn105_frame(0x62, bytes(payload)) + + +def make_short_reply(code): + """Build a 0x62 reply for ``code`` that is too short to hold any field.""" + body = bytes([0xFC, 0x62, 0x01, 0x30, 0x03, code, 0x00, 0x00]) + return body + bytes([cn105_checksum(body)]) diff --git a/tests/test_cn105.py b/tests/test_cn105.py new file mode 100644 index 0000000..e795191 --- /dev/null +++ b/tests/test_cn105.py @@ -0,0 +1,358 @@ +"""Tests for CN105/ITP frame building, decoding, and the compressor estimator.""" + +import unittest + +from pykumo.cn105 import ( + AUTO_SUB_MODE_NAMES, + DEFAULT_INFO_CODES, + PAYLOAD_SIZE, + TELEMETRY_KEYS, + CompressorActivityEstimator, + InfoCode, + build_cn105_frame, + build_info_request, + cn105_checksum, + decode_info_reply, + is_info_reply, + valid_cn105_reply, +) +from tests.cn105_frames import ( + FRAME_03_ALT_HEX, + FRAME_03_HEX, + FRAME_09_ACTIVE_HEX, + FRAME_09_OFF_HEX, + INFO_03_REQUEST_HEX, + INFO_06_REQUEST_HEX, + INFO_09_REQUEST_HEX, + make_0x03_reply, + make_0x06_reply, + make_0x09_reply, + make_short_reply, +) + + +class TestChecksum(unittest.TestCase): + def test_checksum_formula(self): + self.assertEqual(cn105_checksum(bytes([0xFC, 0x42])), (0xFC - 0x13E) & 0xFF) + + def test_checksum_of_known_frame(self): + frame = bytes.fromhex(INFO_03_REQUEST_HEX) + self.assertEqual(cn105_checksum(frame[:-1]), frame[-1]) + + +class TestFrameBuilding(unittest.TestCase): + def test_info_requests_match_known_frames(self): + self.assertEqual(build_info_request(0x03).hex(), INFO_03_REQUEST_HEX) + self.assertEqual(build_info_request(0x06).hex(), INFO_06_REQUEST_HEX) + self.assertEqual(build_info_request(0x09).hex(), INFO_09_REQUEST_HEX) + + def test_info_code_enum_builds_the_same_frame(self): + self.assertEqual( + build_info_request(InfoCode.TEMPERATURES), build_info_request(0x03) + ) + + def test_payload_is_padded_to_16(self): + frame = build_cn105_frame(0x42, b"\x03") + self.assertEqual(frame[4], PAYLOAD_SIZE) + self.assertEqual(len(frame), PAYLOAD_SIZE + 6) + + def test_pad_to_zero_keeps_short_payload(self): + frame = build_cn105_frame(0x5A, b"\xca", pad_to=0) + self.assertEqual(frame[4], 1) + self.assertEqual(len(frame), 7) + + def test_checksum_is_appended(self): + frame = build_cn105_frame(0x42, b"\x03") + self.assertEqual(frame[-1], cn105_checksum(frame[:-1])) + + def test_pad_to_beyond_payload_size_is_rejected(self): + with self.assertRaises(ValueError): + build_cn105_frame(0x42, b"\x03", pad_to=32) + + def test_oversized_payload_is_rejected(self): + with self.assertRaises(ValueError): + build_cn105_frame(0x42, b"\x00" * 17) + + def test_out_of_range_type_byte_is_rejected(self): + with self.assertRaises(ValueError): + build_cn105_frame(0x100, b"\x03") + + def test_out_of_range_info_code_is_rejected(self): + with self.assertRaises(ValueError): + build_info_request(0x100) + + +class TestReplyValidation(unittest.TestCase): + def test_known_replies_are_valid(self): + for hexstr in ( + FRAME_03_HEX, + FRAME_03_ALT_HEX, + FRAME_09_OFF_HEX, + FRAME_09_ACTIVE_HEX, + ): + self.assertTrue(valid_cn105_reply(bytes.fromhex(hexstr)), hexstr) + + def test_empty_and_none_are_invalid(self): + self.assertFalse(valid_cn105_reply(None)) + self.assertFalse(valid_cn105_reply(b"")) + + def test_too_short_for_a_header_is_invalid(self): + self.assertFalse(valid_cn105_reply(b"\xfc\x62\x01\x30\x10")) + + def test_wrong_header_is_invalid(self): + frame = bytearray(bytes.fromhex(FRAME_03_HEX)) + frame[0] = 0xFB + self.assertFalse(valid_cn105_reply(bytes(frame))) + + def test_wrong_subheader_is_invalid(self): + frame = bytearray(bytes.fromhex(FRAME_03_HEX)) + frame[2] = 0x02 + self.assertFalse(valid_cn105_reply(bytes(frame))) + + def test_truncated_payload_is_invalid(self): + self.assertFalse(valid_cn105_reply(bytes.fromhex(FRAME_03_HEX)[:-4])) + + def test_bad_checksum_is_invalid(self): + frame = bytearray(bytes.fromhex(FRAME_03_HEX)) + frame[-1] ^= 0xFF + self.assertFalse(valid_cn105_reply(bytes(frame))) + + def test_trailing_bytes_are_tolerated(self): + # The readback buffer sometimes returns more than one frame. + frame = bytes.fromhex(FRAME_03_HEX) + bytes.fromhex(FRAME_09_OFF_HEX) + self.assertTrue(valid_cn105_reply(frame)) + + +class TestIsInfoReply(unittest.TestCase): + def test_matching_code_is_a_reply(self): + self.assertTrue(is_info_reply(bytes.fromhex(FRAME_03_HEX), 0x03)) + + def test_other_code_is_not_a_reply(self): + self.assertFalse(is_info_reply(bytes.fromhex(FRAME_03_HEX), 0x09)) + + def test_request_frame_is_not_a_reply(self): + # A 0x42 request carrying code 0x03 must not pass as its own answer. + self.assertFalse(is_info_reply(build_info_request(0x03), 0x03)) + + def test_invalid_frame_is_not_a_reply(self): + self.assertFalse(is_info_reply(b"\xfc\x62", 0x03)) + + +class TestDecodeTemperatures(unittest.TestCase): + def _decode(self, frame): + return decode_info_reply(frame, InfoCode.TEMPERATURES) + + def test_known_frame(self): + self.assertEqual( + self._decode(bytes.fromhex(FRAME_03_HEX)), + { + "room_temperature": 22.0, + "outdoor_temperature": 27.0, + "compressor_runtime_minutes": 137918, + }, + ) + + def test_second_known_frame(self): + result = self._decode(bytes.fromhex(FRAME_03_ALT_HEX)) + self.assertEqual(result["room_temperature"], 23.0) + self.assertEqual(result["outdoor_temperature"], 24.0) + + def test_outdoor_encoding(self): + result = self._decode(make_0x03_reply(outdoor_byte=0xB6)) + self.assertEqual(result["outdoor_temperature"], 27.0) + + def test_outdoor_unavailable_bytes(self): + # Many outdoor units send 0x00 or 0x01 while the compressor is idle. + for byte in (0x00, 0x01): + self.assertIsNone( + self._decode(make_0x03_reply(outdoor_byte=byte))["outdoor_temperature"] + ) + + def test_room_prefers_encoding_b(self): + result = self._decode(make_0x03_reply(room_b_byte=0xAC, legacy_byte=0x05)) + self.assertEqual(result["room_temperature"], 22.0) + + def test_room_falls_back_to_legacy_map(self): + result = self._decode(make_0x03_reply(room_b_byte=0x00, legacy_byte=0x0C)) + self.assertEqual(result["room_temperature"], 22) + + def test_room_out_of_legacy_range(self): + result = self._decode(make_0x03_reply(room_b_byte=0x00, legacy_byte=0x40)) + self.assertIsNone(result["room_temperature"]) + + def test_runtime_counter_is_24_bit_big_endian(self): + self.assertEqual( + self._decode(bytes.fromhex(FRAME_03_HEX))["compressor_runtime_minutes"], + (0x02 << 16) | (0x1A << 8) | 0xBE, + ) + + def test_short_frame_degrades_every_field(self): + # Valid 0x03 reply but too short to hold anything, so every value is None. + self.assertEqual( + self._decode(make_short_reply(0x03)), + { + "room_temperature": None, + "outdoor_temperature": None, + "compressor_runtime_minutes": None, + }, + ) + + def test_reply_for_another_code_yields_all_none(self): + result = self._decode(bytes.fromhex(FRAME_09_ACTIVE_HEX)) + self.assertEqual(set(result), set(TELEMETRY_KEYS[InfoCode.TEMPERATURES])) + self.assertTrue(all(value is None for value in result.values())) + + +class TestDecodeCompressor(unittest.TestCase): + def _decode(self, frame): + return decode_info_reply(frame, InfoCode.COMPRESSOR) + + def test_operating_and_frequency(self): + self.assertEqual( + self._decode(make_0x06_reply(freq_byte=0x2A, operating_byte=0x01)), + {"operating": True, "compressor_frequency": 0x2A}, + ) + + def test_standby_zeroes_the_frequency(self): + # Some units report noise on the frequency byte while idle. + self.assertEqual( + self._decode(make_0x06_reply(freq_byte=0x2A, operating_byte=0x00)), + {"operating": False, "compressor_frequency": 0}, + ) + + def test_short_frame_degrades_every_field(self): + self.assertEqual( + self._decode(make_short_reply(0x06)), + {"operating": None, "compressor_frequency": None}, + ) + + def test_reply_for_another_code_yields_all_none(self): + result = self._decode(make_0x09_reply()) + self.assertTrue(all(value is None for value in result.values())) + + +class TestDecodeSubMode(unittest.TestCase): + def _decode(self, frame): + return decode_info_reply(frame, InfoCode.SUB_MODE) + + def test_off_frame(self): + self.assertEqual( + self._decode(bytes.fromhex(FRAME_09_OFF_HEX)), + {"sub_mode": "OFF", "stage": "IDLE", "auto_sub_mode": "AUTO_INACTIVE"}, + ) + + def test_active_frame(self): + self.assertEqual( + self._decode(bytes.fromhex(FRAME_09_ACTIVE_HEX)), + {"sub_mode": "NORMAL", "stage": "GENTLE", "auto_sub_mode": "AUTO_INACTIVE"}, + ) + + def test_sub_mode_names(self): + for byte, name in ((0x00, "NORMAL"), (0x02, "DEFROST"), (0x10, "OFF")): + self.assertEqual( + self._decode(make_0x09_reply(sub_byte=byte))["sub_mode"], name + ) + + def test_stage_names(self): + for byte, name in ((0x00, "IDLE"), (0x03, "MEDIUM"), (0x06, "DIFFUSE")): + self.assertEqual( + self._decode(make_0x09_reply(stage_byte=byte))["stage"], name + ) + + def test_auto_sub_mode_covers_both_protocol_generations(self): + for byte, name in AUTO_SUB_MODE_NAMES.items(): + self.assertEqual( + self._decode(make_0x09_reply(auto_byte=byte))["auto_sub_mode"], name + ) + + def test_unrecognized_enum_byte_is_none(self): + # Nothing is remembered between calls, so there is no old value to reuse. + result = self._decode( + make_0x09_reply(sub_byte=0x7F, stage_byte=0x7F, auto_byte=0x7F) + ) + self.assertEqual( + result, {"sub_mode": None, "stage": None, "auto_sub_mode": None} + ) + + def test_short_frame_degrades_every_field(self): + self.assertEqual( + self._decode(make_short_reply(0x09)), + {"sub_mode": None, "stage": None, "auto_sub_mode": None}, + ) + + +class TestDecodeInfoReplyContract(unittest.TestCase): + def test_absent_and_invalid_frames_still_return_the_key_set(self): + for frame in (None, b"", b"\xfc\x62", build_info_request(0x03)): + result = decode_info_reply(frame, InfoCode.TEMPERATURES) + self.assertEqual(set(result), set(TELEMETRY_KEYS[InfoCode.TEMPERATURES])) + self.assertTrue(all(value is None for value in result.values()), frame) + + def test_unknown_code_raises(self): + with self.assertRaises(ValueError): + decode_info_reply(bytes.fromhex(FRAME_03_HEX), 0x11) + + def test_telemetry_keys_match_what_decoders_return(self): + for code, keys in TELEMETRY_KEYS.items(): + self.assertEqual(set(decode_info_reply(None, code)), set(keys)) + + def test_default_codes_exclude_the_compressor_code(self): + self.assertEqual(DEFAULT_INFO_CODES, (InfoCode.TEMPERATURES, InfoCode.SUB_MODE)) + self.assertNotIn(InfoCode.COMPRESSOR, DEFAULT_INFO_CODES) + + +class TestCompressorActivityEstimator(unittest.TestCase): + def setUp(self): + self.est = CompressorActivityEstimator(min_interval=70.0) + + def test_starts_undetermined(self): + self.assertIsNone(self.est.running) + + def test_first_sample_is_undetermined(self): + self.assertIsNone(self.est.update(100, now=1000.0)) + + def test_missing_sample_is_a_noop(self): + self.est.update(100, now=1000.0) + self.assertIsNone(self.est.update(None, now=2000.0)) + + def test_counter_advanced_means_running(self): + self.est.update(100, now=1000.0) + self.assertIs(self.est.update(101, now=1010.0), True) + self.assertIs(self.est.running, True) + + def test_flat_counter_inside_the_window_stays_undetermined(self): + self.est.update(100, now=1000.0) + self.assertIsNone(self.est.update(100, now=1030.0)) + + def test_flat_counter_past_the_window_means_not_running(self): + self.est.update(100, now=1000.0) + self.assertIs(self.est.update(100, now=1071.0), False) + + def test_baseline_is_kept_so_the_window_grows(self): + self.est.update(100, now=1000.0) + # Three unchanged readings in a row. The 1000.0 reading is kept, so by + # 1071.0 enough time has passed to answer. + self.est.update(100, now=1030.0) + self.est.update(100, now=1060.0) + self.assertIs(self.est.update(100, now=1071.0), False) + + def test_previous_answer_is_held_while_undetermined(self): + self.est.update(100, now=1000.0) + self.est.update(101, now=1010.0) + self.assertIs(self.est.update(101, now=1020.0), True) + + def test_reset_forgets_everything(self): + self.est.update(100, now=1000.0) + self.est.update(101, now=1010.0) + self.est.reset() + self.assertIsNone(self.est.running) + self.assertIsNone(self.est.update(101, now=1020.0)) + + def test_custom_min_interval(self): + est = CompressorActivityEstimator(min_interval=10.0) + est.update(100, now=1000.0) + self.assertIs(est.update(100, now=1011.0), False) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_cn105_bus.py b/tests/test_cn105_bus.py new file mode 100644 index 0000000..30fa3a0 --- /dev/null +++ b/tests/test_cn105_bus.py @@ -0,0 +1,501 @@ +"""Tests for the CN105 bus transport and PyKumo's cached telemetry.""" + +import unittest +from contextlib import contextmanager +from unittest.mock import patch + +from pykumo.cn105 import CompressorActivityEstimator, InfoCode +from pykumo.cn105_bus import Cn105Bus +from pykumo.py_kumo import PyKumo +from tests.cn105_frames import ( + FRAME_03_HEX, + FRAME_09_ACTIVE_HEX, + INFO_03_REQUEST_HEX, + make_0x06_reply, +) + + +def readback(frame: bytes | None) -> dict: + """Build the response the adapter gives back when you read the buffer.""" + node = {"frame": frame.hex()} if frame else {} + return {"r": {"indoorUnit": {"settings": {"rawITPFrame": node}}}} + + +class FakeUnit: + """A fake PyKumo: just a name, a request hook, and cycle counting.""" + + def __init__(self, responses=None): + self.responses = list(responses or []) + self.requests = [] + # Requests made outside a cycle, meaning they had to reconnect. + self.unbracketed = 0 + self.cycles_opened = 0 + self._depth = 0 + + def get_name(self): + return "Fake Unit" + + def _request(self, post_data): + self.requests.append(post_data) + if self._depth == 0: + self.unbracketed += 1 + return self.responses.pop(0) if self.responses else {} + + @contextmanager + def request_cycle(self): + self.cycles_opened += 1 + self._depth += 1 + try: + yield + finally: + self._depth -= 1 + + +def make_bus(responses=None): + unit = FakeUnit(responses) + return Cn105Bus(unit), unit + + +def make_unit(replies=None): + """Build a PyKumo that reaches no network and answers with ``replies``.""" + with patch.object(PyKumo, "__init__", lambda self, *a, **kw: None): + unit = PyKumo.__new__(PyKumo) + unit._name = "Test Unit" + unit._address = "192.0.2.1" + unit._status = {} + unit._profile = {} + unit._cn105 = Cn105Bus(unit) + unit._cn105_telemetry = {} + unit._cn105_telemetry_at = None + unit._compressor_activity = CompressorActivityEstimator() + unit.sent_codes = [] + + replies = replies or {} + + def fake_send(frame, id_byte=1): + unit.sent_codes.append(frame[5]) + return True + + def fake_read(): + return replies.get(unit.sent_codes[-1]) if unit.sent_codes else None + + unit._cn105.send = fake_send + unit._cn105.read = fake_read + return unit + + +class TestSend(unittest.TestCase): + def test_command_shape(self): + bus, unit = make_bus([{"r": {}}]) + frame = bytes.fromhex(INFO_03_REQUEST_HEX) + self.assertTrue(bus.send(frame)) + expected = ( + '{"c":{"indoorUnit":{"settings":{"rawITPFrame":' + '{"frame":"%s","len":%d,"id":1}}}}}' % (frame.hex(), len(frame)) + ).encode("utf-8") + self.assertEqual(unit.requests, [expected]) + + def test_custom_id_byte(self): + bus, unit = make_bus([{"r": {}}]) + bus.send(bytes.fromhex(INFO_03_REQUEST_HEX), id_byte=2) + self.assertIn(b'"id":2', unit.requests[0]) + + def test_empty_frame_is_rejected_without_a_request(self): + bus, unit = make_bus() + self.assertFalse(bus.send(b"")) + self.assertEqual(unit.requests, []) + + def test_overlong_frame_is_rejected_without_a_request(self): + bus, unit = make_bus() + self.assertFalse(bus.send(b"\x00" * 128)) + self.assertEqual(unit.requests, []) + + def test_out_of_range_id_byte_is_rejected_without_a_request(self): + bus, unit = make_bus() + self.assertFalse(bus.send(bytes.fromhex(INFO_03_REQUEST_HEX), id_byte=256)) + self.assertEqual(unit.requests, []) + + def test_api_error_is_a_failure(self): + bus, _ = make_bus([{"_api_error": "device_authentication_error"}]) + self.assertFalse(bus.send(bytes.fromhex(INFO_03_REQUEST_HEX))) + + def test_empty_response_is_a_failure(self): + bus, _ = make_bus([None]) + self.assertFalse(bus.send(bytes.fromhex(INFO_03_REQUEST_HEX))) + + +class TestRead(unittest.TestCase): + def test_reads_back_a_frame(self): + reply = bytes.fromhex(FRAME_03_HEX) + bus, _ = make_bus([readback(reply)]) + self.assertEqual(bus.read(), reply) + + def test_empty_buffer(self): + bus, _ = make_bus([readback(None)]) + self.assertIsNone(bus.read()) + + def test_malformed_response(self): + bus, _ = make_bus([{"r": {}}]) + self.assertIsNone(bus.read()) + + def test_none_response(self): + bus, _ = make_bus([None]) + self.assertIsNone(bus.read()) + + def test_non_hex_payload(self): + bus, _ = make_bus( + [{"r": {"indoorUnit": {"settings": {"rawITPFrame": {"frame": "zz"}}}}}] + ) + self.assertIsNone(bus.read()) + + +class TestTransceive(unittest.TestCase): + def _bus(self, reads): + """A bus whose send works and whose reads hand back ``reads`` in order.""" + bus, unit = make_bus([{"r": {}}] + [readback(r) for r in reads]) + return bus, unit + + def test_returns_the_matching_reply(self): + reply = bytes.fromhex(FRAME_03_HEX) + bus, _ = self._bus([reply]) + with patch("pykumo.cn105_bus.time.sleep"): + got = bus.transceive( + bytes.fromhex(INFO_03_REQUEST_HEX), expect_type=0x62, expect_code=0x03 + ) + self.assertEqual(got, reply) + + def test_skips_a_stale_reply_for_another_code(self): + stale = bytes.fromhex(FRAME_09_ACTIVE_HEX) + wanted = bytes.fromhex(FRAME_03_HEX) + bus, _ = self._bus([stale, None, wanted]) + with patch("pykumo.cn105_bus.time.sleep"): + got = bus.transceive( + bytes.fromhex(INFO_03_REQUEST_HEX), expect_type=0x62, expect_code=0x03 + ) + self.assertEqual(got, wanted) + + def test_timeout_returns_none(self): + bus, _ = self._bus([None, None]) + with patch("pykumo.cn105_bus.time.sleep"): + self.assertIsNone( + bus.transceive(bytes.fromhex(INFO_03_REQUEST_HEX), timeout=1.0) + ) + + def test_failed_send_skips_polling(self): + bus, unit = make_bus([{"_api_error": "x"}]) + with patch("pykumo.cn105_bus.time.sleep") as sleep: + self.assertIsNone(bus.transceive(bytes.fromhex(INFO_03_REQUEST_HEX))) + sleep.assert_not_called() + self.assertEqual(len(unit.requests), 1) + + def test_every_request_reuses_one_connection(self): + # Without a cycle, _request closes the session after every call, so a + # 20 s wait would mean dozens of reconnects. + bus, unit = self._bus([None, None, bytes.fromhex(FRAME_03_HEX)]) + with patch("pykumo.cn105_bus.time.sleep"): + bus.transceive( + bytes.fromhex(INFO_03_REQUEST_HEX), expect_type=0x62, expect_code=0x03 + ) + self.assertEqual(unit.unbracketed, 0) + self.assertEqual(unit.cycles_opened, 1) + self.assertEqual(len(unit.requests), 4) + + def test_poll_count_follows_the_timeout(self): + bus, _ = self._bus([None] * 10) + with patch("pykumo.cn105_bus.time.sleep") as sleep: + bus.transceive(bytes.fromhex(INFO_03_REQUEST_HEX), timeout=2.0) + self.assertEqual(sleep.call_count, 4) + + +class TestUnsupportedCodeLatch(unittest.TestCase): + def _bus_that_never_answers(self): + bus, unit = make_bus() + unit.responses = [{"r": {}}] * 200 + return bus, unit + + def test_compressor_code_latches_after_one_miss(self): + bus, unit = self._bus_that_never_answers() + with patch("pykumo.cn105_bus.time.sleep"): + self.assertIsNone(bus.read_info(InfoCode.COMPRESSOR, timeout=1.0)) + self.assertEqual(bus.unsupported_codes, frozenset({InfoCode.COMPRESSOR})) + before = len(unit.requests) + with patch("pykumo.cn105_bus.time.sleep"): + self.assertIsNone(bus.read_info(InfoCode.COMPRESSOR, timeout=1.0)) + # Given up on, so the second call must not touch the adapter at all. + self.assertEqual(len(unit.requests), before) + + def test_default_codes_never_latch(self): + # One miss on a code every unit supports must not disable it. + bus, _ = self._bus_that_never_answers() + with patch("pykumo.cn105_bus.time.sleep"): + bus.read_info(InfoCode.TEMPERATURES, timeout=1.0) + bus.read_info(InfoCode.SUB_MODE, timeout=1.0) + self.assertEqual(bus.unsupported_codes, frozenset()) + + def test_a_code_that_answered_once_never_latches(self): + reply = make_0x06_reply(operating_byte=0x01) + bus, unit = make_bus([{"r": {}}, readback(reply)]) + with patch("pykumo.cn105_bus.time.sleep"): + self.assertEqual(bus.read_info(InfoCode.COMPRESSOR, timeout=1.0), reply) + unit.responses = [{"r": {}}] * 200 + with patch("pykumo.cn105_bus.time.sleep"): + self.assertIsNone(bus.read_info(InfoCode.COMPRESSOR, timeout=1.0)) + self.assertEqual(bus.unsupported_codes, frozenset()) + + def test_forget_unsupported_allows_a_retry(self): + bus, unit = self._bus_that_never_answers() + with patch("pykumo.cn105_bus.time.sleep"): + bus.read_info(InfoCode.COMPRESSOR, timeout=1.0) + bus.forget_unsupported() + self.assertEqual(bus.unsupported_codes, frozenset()) + before = len(unit.requests) + with patch("pykumo.cn105_bus.time.sleep"): + bus.read_info(InfoCode.COMPRESSOR, timeout=1.0) + self.assertGreater(len(unit.requests), before) + + +class TestUpdateCn105Telemetry(unittest.TestCase): + def test_default_codes_are_temperatures_and_sub_mode(self): + unit = make_unit( + { + 0x03: bytes.fromhex(FRAME_03_HEX), + 0x09: bytes.fromhex(FRAME_09_ACTIVE_HEX), + } + ) + with patch("pykumo.cn105_bus.time.sleep"): + self.assertTrue(unit.update_cn105_telemetry()) + self.assertEqual(unit.sent_codes, [0x03, 0x09]) + self.assertEqual( + unit.get_cn105_telemetry(), + { + "room_temperature": 22.0, + "outdoor_temperature": 27.0, + "compressor_runtime_minutes": 137918, + "sub_mode": "NORMAL", + "stage": "GENTLE", + "auto_sub_mode": "AUTO_INACTIVE", + "operating": None, + }, + ) + + def test_keys_present_and_false_when_nothing_answers(self): + unit = make_unit({}) + with patch("pykumo.cn105_bus.time.sleep"): + self.assertFalse(unit.update_cn105_telemetry()) + self.assertEqual( + sorted(unit.get_cn105_telemetry()), + [ + "auto_sub_mode", + "compressor_runtime_minutes", + "operating", + "outdoor_temperature", + "room_temperature", + "stage", + "sub_mode", + ], + ) + self.assertTrue(all(v is None for v in unit.get_cn105_telemetry().values())) + + def test_compressor_code_is_opt_in(self): + unit = make_unit({0x06: make_0x06_reply(freq_byte=0x2A, operating_byte=0x01)}) + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.COMPRESSOR]) + self.assertEqual(unit.sent_codes, [0x06]) + self.assertEqual( + unit.get_cn105_telemetry(), + {"operating": True, "compressor_frequency": 0x2A}, + ) + + def test_one_code_failing_keeps_the_other(self): + unit = make_unit({0x09: bytes.fromhex(FRAME_09_ACTIVE_HEX)}) + with patch("pykumo.cn105_bus.time.sleep"): + self.assertTrue(unit.update_cn105_telemetry()) + telemetry = unit.get_cn105_telemetry() + self.assertEqual(telemetry["sub_mode"], "NORMAL") + self.assertIsNone(telemetry["outdoor_temperature"]) + + def test_unknown_code_is_skipped_without_a_send(self): + unit = make_unit({}) + with patch("pykumo.cn105_bus.time.sleep"): + self.assertFalse(unit.update_cn105_telemetry(codes=[0x11])) + self.assertEqual(unit.sent_codes, []) + self.assertEqual(unit.get_cn105_telemetry(), {}) + + def test_a_raising_read_is_contained(self): + unit = make_unit({}) + unit._cn105.read_info = lambda code, timeout=None: (_ for _ in ()).throw( + RuntimeError("boom") + ) + self.assertFalse(unit.update_cn105_telemetry()) + self.assertTrue(all(v is None for v in unit.get_cn105_telemetry().values())) + + def test_a_refresh_replaces_the_previous_snapshot(self): + unit = make_unit({0x03: bytes.fromhex(FRAME_03_HEX)}) + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.TEMPERATURES]) + self.assertEqual(unit.get_outdoor_temperature(), 27.0) + # The unit stops answering, so the old values must not stick around. + unit._cn105.read = lambda: None + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.TEMPERATURES]) + self.assertIsNone(unit.get_outdoor_temperature()) + + +class TestCachedAccessors(unittest.TestCase): + def test_none_before_any_refresh(self): + unit = make_unit() + self.assertIsNone(unit.get_outdoor_temperature()) + self.assertIsNone(unit.get_raw_room_temperature()) + self.assertIsNone(unit.get_compressor_runtime_minutes()) + self.assertEqual(unit.get_cn105_telemetry(), {}) + self.assertIsNone(unit.get_cn105_telemetry_age()) + + def test_values_after_a_refresh(self): + unit = make_unit({0x03: bytes.fromhex(FRAME_03_HEX)}) + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.TEMPERATURES]) + self.assertEqual(unit.get_outdoor_temperature(), 27.0) + self.assertEqual(unit.get_raw_room_temperature(), 22.0) + self.assertEqual(unit.get_compressor_runtime_minutes(), 137918) + + def test_accessors_do_not_touch_the_adapter(self): + unit = make_unit({0x03: bytes.fromhex(FRAME_03_HEX)}) + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.TEMPERATURES]) + unit.sent_codes.clear() + unit.get_outdoor_temperature() + unit.get_raw_room_temperature() + unit.get_compressor_runtime_minutes() + unit.get_cn105_telemetry() + self.assertEqual(unit.sent_codes, []) + + def test_telemetry_snapshot_is_a_copy(self): + unit = make_unit({0x03: bytes.fromhex(FRAME_03_HEX)}) + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.TEMPERATURES]) + unit.get_cn105_telemetry()["outdoor_temperature"] = 999 + self.assertEqual(unit.get_outdoor_temperature(), 27.0) + + def test_age_tracks_the_last_answer(self): + unit = make_unit({0x03: bytes.fromhex(FRAME_03_HEX)}) + with ( + patch("pykumo.cn105_bus.time.sleep"), + patch( + "pykumo.py_kumo.time.monotonic", side_effect=[1000.0, 1000.0, 1042.0] + ), + ): + unit.update_cn105_telemetry(codes=[InfoCode.TEMPERATURES]) + self.assertEqual(unit.get_cn105_telemetry_age(), 42.0) + + def test_age_stays_none_when_nothing_answers(self): + unit = make_unit({}) + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.TEMPERATURES]) + self.assertIsNone(unit.get_cn105_telemetry_age()) + + +class TestIsCompressorRunning(unittest.TestCase): + def _refresh(self, unit, now): + with ( + patch("pykumo.cn105_bus.time.sleep"), + patch("pykumo.py_kumo.time.monotonic", return_value=now), + ): + unit.update_cn105_telemetry(codes=[InfoCode.TEMPERATURES]) + + def test_operating_flag_wins_when_the_compressor_code_answered(self): + unit = make_unit({0x06: make_0x06_reply(operating_byte=0x01)}) + unit._status = {"mode": "cool"} + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.COMPRESSOR]) + self.assertIs(unit.is_compressor_running(), True) + + def test_operating_flag_false_is_respected(self): + unit = make_unit({0x06: make_0x06_reply(operating_byte=0x00)}) + unit._status = {"mode": "cool"} + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.COMPRESSOR]) + self.assertIs(unit.is_compressor_running(), False) + + def test_off_mode_is_false_without_the_compressor_code(self): + unit = make_unit({0x03: bytes.fromhex(FRAME_03_HEX)}) + unit._status = {"mode": "off"} + self.assertIs(unit.is_compressor_running(), False) + + def test_undetermined_until_two_samples_exist(self): + unit = make_unit({0x03: bytes.fromhex(FRAME_03_HEX)}) + unit._status = {"mode": "cool"} + self._refresh(unit, 1000.0) + self.assertIsNone(unit.is_compressor_running()) + + def test_falls_back_to_the_runtime_estimate(self): + replies = {0x03: bytes.fromhex(FRAME_03_HEX)} + unit = make_unit(replies) + unit._status = {"mode": "cool"} + self._refresh(unit, 1000.0) + # Same counter 80 s later, so the compressor did not run. + self._refresh(unit, 1080.0) + self.assertIs(unit.is_compressor_running(), False) + # A frame whose counter advanced by one minute. + advanced = bytearray(bytes.fromhex(FRAME_03_HEX)) + advanced[18] += 1 + advanced[-1] = (advanced[-1] - 1) & 0xFF + replies[0x03] = bytes(advanced) + self._refresh(unit, 1090.0) + self.assertIs(unit.is_compressor_running(), True) + + def test_a_slow_code_does_not_age_a_fresh_counter_reading(self): + unit = make_unit({0x03: bytes.fromhex(FRAME_03_HEX)}) + unit._status = {"mode": "cool"} + with ( + patch("pykumo.cn105_bus.time.sleep"), + patch("pykumo.py_kumo.time.monotonic", side_effect=[1000.0, 1000.0]), + ): + unit.update_cn105_telemetry(codes=[InfoCode.TEMPERATURES]) + # 0x03 is read first, then 0x06 times out and burns 70 s. The counter + # reading is only 10 s newer than the last one, so there is still + # nothing to compare. + with ( + patch("pykumo.cn105_bus.time.sleep"), + patch("pykumo.py_kumo.time.monotonic", side_effect=[1010.0, 1080.0]), + ): + unit.update_cn105_telemetry( + codes=[InfoCode.TEMPERATURES, InfoCode.COMPRESSOR] + ) + self.assertIsNone(unit.is_compressor_running()) + + def test_a_refresh_without_the_counter_does_not_resample_it(self): + replies = { + 0x03: bytes.fromhex(FRAME_03_HEX), + 0x09: bytes.fromhex(FRAME_09_ACTIVE_HEX), + } + unit = make_unit(replies) + unit._status = {"mode": "cool"} + self._refresh(unit, 1000.0) + # No 0x03 this time, so the cached counter must not stand in for a + # second reading. + with ( + patch("pykumo.cn105_bus.time.sleep"), + patch("pykumo.py_kumo.time.monotonic", return_value=1200.0), + ): + unit.update_cn105_telemetry(codes=[InfoCode.SUB_MODE]) + self.assertIsNone(unit.is_compressor_running()) + + def test_telemetry_operating_comes_from_the_estimate(self): + replies = {0x03: bytes.fromhex(FRAME_03_HEX)} + unit = make_unit(replies) + unit._status = {"mode": "cool"} + self._refresh(unit, 1000.0) + self._refresh(unit, 1080.0) + # 0x06 was never asked for, so the snapshot carries the estimate. + self.assertIs(unit.get_cn105_telemetry()["operating"], False) + + def test_telemetry_operating_uses_the_flag_when_present(self): + unit = make_unit({0x06: make_0x06_reply(operating_byte=0x01)}) + unit._status = {"mode": "cool"} + with patch("pykumo.cn105_bus.time.sleep"): + unit.update_cn105_telemetry(codes=[InfoCode.COMPRESSOR]) + self.assertIs(unit.get_cn105_telemetry()["operating"], True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_py_kumo_base.py b/tests/test_py_kumo_base.py index f12ef11..25f0be6 100644 --- a/tests/test_py_kumo_base.py +++ b/tests/test_py_kumo_base.py @@ -38,5 +38,40 @@ def test_false_when_profile_reset_to_empty(self): self.assertFalse(unit.has_profile()) +class TestRequestCycle(unittest.TestCase): + """PyKumoBase.request_cycle() connection reuse.""" + + def _make_unit(self): + return PyKumoBase("Test Unit", "192.168.1.1", _CFG) + + def test_opens_and_closes_a_cycle(self): + unit = self._make_unit() + self.assertFalse(unit._in_cycle()) + with unit.request_cycle(): + self.assertTrue(unit._in_cycle()) + self.assertFalse(unit._in_cycle()) + + def test_nested_block_leaves_closing_to_the_outer_one(self): + unit = self._make_unit() + with unit.request_cycle(): + with unit.request_cycle(): + self.assertTrue(unit._in_cycle()) + # The inner block must not have ended the outer cycle. + self.assertTrue(unit._in_cycle()) + self.assertFalse(unit._in_cycle()) + + def test_cycle_closes_on_exception(self): + unit = self._make_unit() + with self.assertRaises(RuntimeError), unit.request_cycle(): + raise RuntimeError("boom") + self.assertFalse(unit._in_cycle()) + + def test_cycles_are_per_unit(self): + unit = self._make_unit() + other = PyKumoBase("Other Unit", "192.168.1.2", _CFG) + with unit.request_cycle(): + self.assertFalse(other._in_cycle()) + + if __name__ == "__main__": unittest.main()