diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index c5e9daf..cb72f0d 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -19,7 +19,7 @@ jobs: - name: Install test dependencies run: | python -m pip install --upgrade pip - pip install pytest + pip install pytest pytest-asyncio - name: Run tests run: pytest -q diff --git a/custom_components/flipper_rc/__init__.py b/custom_components/flipper_rc/__init__.py index ff81790..1934bf1 100644 --- a/custom_components/flipper_rc/__init__.py +++ b/custom_components/flipper_rc/__init__.py @@ -1,4 +1,4 @@ -"""LocalTuyaIR Remote Control integration.""" +"""Flipper Zero Remote Control integration.""" import logging import voluptuous as vol import homeassistant.helpers.config_validation as cv @@ -12,7 +12,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry): """Set up Flipper Zero Remote Control from a config entry.""" - _LOGGER.debug("Setting up entry") + _LOGGER.info("Setting up Flipper RC integration") # Add entities await hass.config_entries.async_forward_entry_setups(entry, [Platform.REMOTE, Platform.BUTTON]) @@ -20,5 +20,5 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry): async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry): """Unload a config entry.""" - _LOGGER.debug("Unloading") + _LOGGER.info("Unloading Flipper RC integration") return await hass.config_entries.async_unload_platforms(entry, [Platform.REMOTE, Platform.BUTTON]) diff --git a/custom_components/flipper_rc/button.py b/custom_components/flipper_rc/button.py index e81ec8d..6463a19 100644 --- a/custom_components/flipper_rc/button.py +++ b/custom_components/flipper_rc/button.py @@ -5,6 +5,7 @@ import os from homeassistant.components.button import ButtonEntity +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import entity_registry as er from homeassistant.helpers.entity import DeviceInfo from homeassistant.util import slugify @@ -51,8 +52,8 @@ async def async_setup_entry(hass, entry, async_add_entities): "/ext/subghz/Saved", "/ext/subghz_playlist", "/ext/apps_data/subghz", + "/ext", ] - fallback_root = "/ext" for root in search_roots: try: @@ -64,16 +65,6 @@ async def async_setup_entry(hass, entry, async_add_entities): _LOGGER.info("Discovered %d Sub-GHz files in %s for %s", len(discovered), root, remote_entity.port) files.extend(discovered) - if not files: - try: - discovered = await remote_entity.async_list_subghz_files(fallback_root) - except Exception as e: - _LOGGER.debug("Cannot discover Sub-GHz files in %s on %s: %s", fallback_root, remote_entity.port, e) - else: - if discovered: - _LOGGER.info("Discovered %d Sub-GHz files in %s for %s", len(discovered), fallback_root, remote_entity.port) - files.extend(discovered) - files = sorted(set(files)) if not files: @@ -116,9 +107,35 @@ def extra_state_attributes(self): async def async_press(self): """Replay file when button is pressed.""" - _LOGGER.info("Sending Sub-GHz saved file: %s", self._file_path) + _LOGGER.info("Button press triggered for Sub-GHz file: %s", self._file_path) + + file_path = self._file_path + + if not self._remote_entity.available: + _LOGGER.warning("Button press for %s rejected: remote entity is not available", file_path) + raise HomeAssistantError( + f"Cannot send Sub-GHz file '{file_path}': Flipper Zero is not connected. " + "Please check the USB connection and wait for the device to become available." + ) + try: await self._remote_entity.async_send_subghz_from_file(self._file_path, repeat=1, antenna=0) + _LOGGER.info("Button press for %s completed successfully", file_path) + except TimeoutError as e: + _LOGGER.error("Timeout sending Sub-GHz file %s: %s", file_path, e) + raise HomeAssistantError( + f"Timed out sending Sub-GHz file '{file_path}': {e}. " + "The Flipper Zero may be busy transmitting. Please try again in a moment." + ) from e + except ConnectionError as e: + _LOGGER.error("Connection lost while sending Sub-GHz file %s: %s", file_path, e) + raise HomeAssistantError( + f"Connection to Flipper Zero lost while sending '{file_path}': {e}. " + "Please check the USB connection." + ) from e except Exception as e: - _LOGGER.error("Failed to send Sub-GHz saved file %s: %s", self._file_path, e, exc_info=True) - raise + _LOGGER.error("Failed to send Sub-GHz saved file %s: %s", file_path, e, exc_info=True) + raise HomeAssistantError( + f"Failed to send Sub-GHz file '{file_path}': {e}. " + "Check device connectivity and try again." + ) from e diff --git a/custom_components/flipper_rc/flipper_ir.py b/custom_components/flipper_rc/flipper_ir.py index f21c30b..f02ed82 100644 --- a/custom_components/flipper_rc/flipper_ir.py +++ b/custom_components/flipper_rc/flipper_ir.py @@ -1,9 +1,6 @@ -import sys import asyncio -import serial_asyncio_fast as serial_asyncio import logging import time -from collections import deque from posixpath import normpath import re @@ -13,14 +10,6 @@ def _is_supported_subghz_path(path): return isinstance(path, str) and path.startswith("/ext/") - -def _has_forbidden_subghz_path_chars(path): - return any(ch.isspace() for ch in path) or "\x00" in path - - -def _is_sendable_subghz_path(path): - return _is_supported_subghz_path(path) and not _has_forbidden_subghz_path_chars(path) - class FlipperIR: def __init__(self, port, default_timeout=10): """ @@ -38,7 +27,8 @@ def __init__(self, port, default_timeout=10): self._lock = asyncio.Lock() self._on_connection_lost = None - def __del__(self): + async def async_close(self): + """Safely close the connection""" self.close() async def open(self): @@ -51,7 +41,7 @@ async def open(self): return loop = asyncio.get_running_loop() self._transport, self._protocol = await serial_asyncio.create_serial_connection( - loop, lambda: FlipperProtocol(), self.port, baudrate=115200 # boudrate is ignored for VCP + loop, lambda: FlipperProtocol(), self.port, baudrate=115200 # baudrate is ignored for VCP ) self._protocol.set_on_connection_lost(self.close) # Waiting for connection @@ -67,10 +57,15 @@ async def open(self): raise TimeoutError("Timeout while waiting for Flipper Zero to connect") _LOGGER.debug(f"Serial port {self.port} opened") try: - await self._protocol.wait_for_prompt() + await self._protocol.wait_for_prompt(timeout=10) except asyncio.TimeoutError as e: - self.close() - raise TimeoutError("Timeout while waiting for Flipper Zero prompt") from e + _LOGGER.warning("Initial prompt wait timed out, attempting recovery") + self._send_ctrl_c() + try: + await self._protocol.wait_for_prompt(timeout=5) + except asyncio.TimeoutError: + self.close() + raise TimeoutError("Timeout while waiting for Flipper Zero prompt") from e except asyncio.CancelledError: self.close() raise @@ -102,7 +97,14 @@ def connected(self): Returns: bool: True if connected, False otherwise. """ - return self._transport is not None + if self._transport is None: + return False + # Detect dead connections: transport exists but is closed or broken + if self._transport.is_closing(): + return False + if self._protocol is None or not self._protocol.connected: + return False + return True @property def busy(self): @@ -114,34 +116,72 @@ def busy(self): return self._lock.locked() async def ensure_open(self): + """Ensure the serial connection is open, detecting dead connections and reconnecting.""" if not self.connected: + _LOGGER.info("Connection not established (port %s), opening...", self.port) await self.open() + _LOGGER.info("Connection to Flipper Zero on %s established successfully", self.port) + else: + _LOGGER.debug("Connection to Flipper Zero on %s already open", self.port) def _validate_cli_response(self, lines, expected_prefixes, command_name): - """Validate command response while tolerating blank/noisy lines.""" - non_empty = [line.strip() for line in lines if isinstance(line, str) and line.strip()] + """Validate response: reject errors, require tx indicators for subghz, accept prefix.""" + non_empty = [l.strip() for l in lines if isinstance(l, str) and l.strip()] + # 1. Reject if any error indicator found for line in non_empty: - for prefix in expected_prefixes: - if line.startswith(prefix): + low = line.lower() + if any(ind in low for ind in ("error", "failed", "invalid", "unknown", + "file not found", "cannot", "refused", "denied", + "not supported", "no such")): + raise ValueError(f"{command_name} failed: {line!r}") + + # 2. Subghz tx commands require transmission confirmation + low_name = command_name.lower() + if "subghz" in low_name and "tx" in low_name: + for line in non_empty: + if any(ind in line.lower() for ind in ("transmitting", "frequency", + "transmission", "sending", "done", "success", "complete")): return + raise ValueError( + f"{command_name} did not confirm transmission. " + f"Response: {'; '.join(non_empty)}. Check firmware/command syntax." + ) - # Some firmware builds may not echo command line consistently. - # Only fail when there is an explicit error in response. + # 3. Accept if expected prefix found (non-tx commands) for line in non_empty: - low = line.lower() - if "error" in low or "failed" in low or "invalid" in low or "unknown" in low: - raise ValueError(f"{command_name} failed: {line!r}") + if any(line.startswith(p) for p in expected_prefixes): + return - _LOGGER.debug( - "No expected echo found for %s; accepting response. Lines: %s", - command_name, - lines, - ) + # 4. Tolerant fallback for non-tx commands + _LOGGER.debug("No expected echo for %s; accepting (no errors). Lines: %s", + command_name, [str(l) for l in lines]) def _send_ctrl_c(self): - if self._transport: + """Send Ctrl-C (0x03) to break the Flipper out of its current operation.""" + if self._transport and not self._transport.is_closing(): self._transport.write(b'\x03') + _LOGGER.debug("Sent Ctrl-C to Flipper Zero to break current operation") + + async def recover_from_timeout(self): + """Send Ctrl-C and drain buffer to recover from a stuck Flipper.""" + _LOGGER.warning("Recovery: sending Ctrl-C and draining buffer") + self._send_ctrl_c() + await asyncio.sleep(0.2) + if self._protocol: + try: + while self._protocol.lines_available > 0: + try: + line = await self._protocol.readline(timeout=0.5) + _LOGGER.debug("Drained during recovery: %s", line) + except TimeoutError: + break + if self._protocol.buffer: + _LOGGER.debug("Drained partial buffer: %s", self._protocol.buffer) + self._protocol.buffer = b'' + except (RuntimeError, ConnectionError) as e: + _LOGGER.warning("Error during recovery drain: %s", e) + await asyncio.sleep(0.3) async def command(self, cmd, timeout=None): """ @@ -160,19 +200,34 @@ async def command(self, cmd, timeout=None): if "\n" in cmd or "\r" in cmd or "\x00" in cmd: raise ValueError("CLI command contains forbidden control characters") - _LOGGER.debug(f"Sending command: {cmd.strip()}") + _LOGGER.debug("Sending command: %s", cmd.strip()) await self.ensure_open() async with self._lock: if timeout is None: timeout = self.default_timeout - await self._protocol.wait_for_prompt() - self._transport.write((cmd.strip() + "\r\n").encode()) + try: + await self._protocol.wait_for_prompt() + except TimeoutError as e: + _LOGGER.warning("Prompt wait failed, recovering: %s", e) + await self.recover_from_timeout() + await self._protocol.wait_for_prompt(timeout=5) + if self._transport and not self._transport.is_closing(): + self._transport.write((cmd.strip() + "\r\n").encode()) + else: + raise ConnectionError("Serial transport closed after recovery") await asyncio.sleep(0.1) try: lines = await self._protocol.wait_for_prompt(timeout=timeout) + _LOGGER.debug("Command %s completed: %d lines", cmd.strip(), len(lines)) except asyncio.TimeoutError as e: - raise TimeoutError("Timeout reached while waiting for Flipper Zero response") from e + _LOGGER.warning("Timeout for '%s', recovering", cmd.strip()) + await self.recover_from_timeout() + try: + lines = await self._protocol.wait_for_prompt(timeout=5) + _LOGGER.debug("Recovered from timeout for '%s': %d lines", cmd.strip(), len(lines)) + except TimeoutError as e2: + raise TimeoutError(f"Timeout waiting for response to '{cmd.strip()}'") from e except asyncio.CancelledError: self.close() raise @@ -249,7 +304,13 @@ async def send_subghz(self, key, frequency, te=350, repeat=1, antenna=0): if int(repeat) <= 0: raise ValueError("Sub-GHz repeat must be positive") - cmd = f"subghz tx {int(key):06X} {int(frequency)} {int(te)} {int(repeat)} {int(antenna)}" + key_int = int(key) + freq_int = int(frequency) + te_int = int(te) + repeat_int = int(repeat) + antenna_int = int(antenna) + + cmd = f"subghz tx {key_int:06X} {freq_int} {te_int} {repeat_int} {antenna_int}" lines = await self.command(cmd) self._validate_cli_response(lines, [">: subghz tx"], "subghz tx") @@ -257,8 +318,8 @@ async def send_subghz_from_file(self, path, repeat=1, antenna=0): """Send Sub-GHz transmission from saved Flipper SD card file.""" if not _is_supported_subghz_path(path): raise ValueError('Sub-GHz file path must start with "/ext/"') - if _has_forbidden_subghz_path_chars(path): - raise ValueError("Sub-GHz file path must not contain whitespace or control characters") + if "\n" in path or "\r" in path or "\x00" in path: + raise ValueError("Sub-GHz file path contains forbidden control characters") if int(repeat) <= 0: raise ValueError("Sub-GHz repeat must be positive") if int(antenna) not in (0, 1): @@ -336,18 +397,18 @@ async def list_subghz_files(self, root="/ext/subghz"): """Recursively list Sub-GHz .sub files on Flipper storage.""" try: tree_files = await self._storage_tree_sub_files(root) - tree_files = [p for p in tree_files if _is_sendable_subghz_path(p) and p.lower().endswith(".sub")] + tree_files = [p for p in tree_files if _is_supported_subghz_path(p) and p.lower().endswith(".sub")] if tree_files: return sorted(set(tree_files)) except Exception as e: _LOGGER.debug("Cannot read storage tree for %s: %s", root, e) discovered = [] - queue = deque([root.rstrip("/")]) + queue = [root.rstrip("/")] visited = set() while queue: - current = queue.popleft() + current = queue.pop(0) if current in visited: continue visited.add(current) @@ -359,7 +420,7 @@ async def list_subghz_files(self, root="/ext/subghz"): continue for file_path in files: - if _is_sendable_subghz_path(file_path) and file_path.lower().endswith(".sub"): + if _is_supported_subghz_path(file_path) and file_path.lower().endswith(".sub"): discovered.append(file_path) for dir_path in dirs: if _is_supported_subghz_path(dir_path) and dir_path not in visited: @@ -421,10 +482,13 @@ def connection_made(self, transport): self._connected = True def data_received(self, data): + """Handle data received from the serial port.""" + _LOGGER.debug("Data received from Flipper (%d bytes): %s", len(data), data) self.buffer += data while b'\n' in self.buffer: line, self.buffer = self.buffer.split(b'\n', 1) line_str = line.strip().decode(errors="ignore") + _LOGGER.debug("Parsed line from Flipper: %s", line_str) self.lines.append(line_str) if self._line_futures: future = self._line_futures.pop(0) @@ -459,16 +523,16 @@ async def readline(self, timeout=10): """ async with self._readline_lock: - # Если уже есть готовая строка — сразу отдаём + # If line is already available, return immediately if self.lines: return self.lines.pop(0) - # Ждём! + # Wait for data future = self._loop.create_future() self._line_futures.append(future) try: return await asyncio.wait_for(future, timeout=timeout) except asyncio.TimeoutError as e: - # Если таймаут, то надо убрать future из списка ожидания + # On timeout, remove future from pending list if not future.done(): self._line_futures.remove(future) raise TimeoutError("Timeout while waiting for Flipper Zero response") from e @@ -478,24 +542,68 @@ async def readline(self, timeout=10): async def wait_for_prompt(self, timeout=3): """ Wait for the Flipper Zero prompt to appear. + Args: timeout (int or float, optional): Timeout for waiting for the prompt in seconds, default is 3. Returns: list: List of lines received before the prompt. - """ + Raises: + TimeoutError: If the prompt is not found within the timeout period. + """ + _LOGGER.debug("Waiting for Flipper Zero prompt (timeout=%ss)...", timeout) plines = [] start_time = time.time() - while self.lines_available or not self.has_prompt: + + while True: + # Drain all available lines. Use remaining timeout so readline waits + # long enough for slow responses (e.g., subghz transmission completion). + remaining = max(0.1, timeout - (time.time() - start_time)) while self.lines_available > 0: - line = await self.readline(timeout=timeout) - plines.append(line) - if self.has_prompt: - break + try: + line = await self.readline(timeout=remaining) + plines.append(line) + except TimeoutError: + break + + # Check for prompt using multiple strategies: + # 1. Check remaining buffer for partial prompt data + # 2. Check collected lines for prompt (race-condition resistant) + prompt_found = self._check_prompt_in_lines(plines) or self.has_prompt + + if prompt_found: + _LOGGER.debug("Flipper Zero prompt found after %.2fs, collected %d lines", + time.time() - start_time, len(plines)) + return plines + + elapsed = time.time() - start_time + if elapsed > timeout: + _LOGGER.warning("Timeout (%.1fs) waiting for Flipper Zero prompt. " + "Buffer: %s, Lines collected: %d, Last lines: %s", + timeout, self.buffer, len(plines), plines[-3:] if plines else []) + raise TimeoutError( + f"Timeout while waiting for Flipper Zero prompt after {timeout:.1f}s. " + f"Collected {len(plines)} lines. Remaining buffer: {self.buffer!r}" + ) + await asyncio.sleep(0.1) - if time.time() - start_time > timeout: - raise TimeoutError("Timeout while waiting for Flipper Zero prompt") - return plines + + def _check_prompt_in_lines(self, lines): + """ + Check if the prompt marker (': ') appears at the end of any collected line. + + This is a race-condition-resistant check: when data_received splits on '\n', + the prompt '>: ' may end up as the last line in self.lines rather than in + self.buffer. By checking all collected lines, we avoid missing the prompt. + """ + prompt_markers = [b'>: ', b' >:', b'>:\r'] + for line in lines: + line_bytes = line.encode() if isinstance(line, str) else line + for marker in prompt_markers: + if line_bytes.endswith(marker): + _LOGGER.debug("Prompt found in collected line: %s", line) + return True + return False # def reset(self): # self.buffer = b'' @@ -517,17 +625,23 @@ def connected(self): @property def has_prompt(self): """ - Check if the prompt is present in the buffer. + Check if the prompt is present in the remaining buffer. + + Note: This only checks the unprocessed buffer (data not yet split into lines). + For robust prompt detection, also use _check_prompt_in_lines() to scan + already-collected lines, as the prompt may appear there due to race conditions + between data_received() and wait_for_prompt(). + Returns: - bool: True if the prompt is present, False otherwise. + bool: True if the prompt is present in the buffer, False otherwise. """ return self.buffer.endswith(b'>: ') -# Пример использования: if __name__ == "__main__": - + async def main(): + import sys logging.basicConfig(level=logging.DEBUG) port = sys.argv[1] if len(sys.argv) > 1 else '/dev/ttyACM0_' ir = FlipperIR(port) @@ -535,25 +649,24 @@ async def main(): try: await ir.open() info = await ir.get_device_info() - print(f"Информация о устройстве: {info}") + print(f"Device info: {info}") uptime = await ir.get_uptime() print(f"Uptime: {uptime}") - print("🌸 Отправляю сигнал...") + print("Sending IR signal...") await ir.send_ir(frequency=38000, duty_cycle=50, samples=[9010, 4495, 559, 555, 588, 526, 556, 559, 564, 550, 563, 553, 560, 555, 558, 557, 556, 559, 564, 1669, 608, 1635, 611, 1632, 583, 1660, 586, 529, 584, 1659, 587, 1656, 590, 1653, 614, 1630, 616, 1627, 589, 526, 607, 507, 616, 499, 583, 532, 611, 503, 610, 506, 586, 528, 615, 499, 614, 1630, 616, 1626, 589, 1654, 612, 1631, 615, 1628, 587, 1656, 611]) - print("🌸 Сигнал отправлен!") - - print("🌸 Готова принимать сигналы! Нажми Ctrl+C для выхода.") + print("Signal sent!") + + print("Ready to receive signals. Press Ctrl+C to exit.") signals = await ir.receive_ir(timeout=10) - print(f"Получено {len(signals)} сигналов:") + print(f"Received {len(signals)} signals:") print(signals) except asyncio.exceptions.CancelledError: pass except KeyboardInterrupt: - print("Приёмчик остановлен~") + print("Receiver stopped") except Exception as e: - print(f"Ошибка {e.__class__.__name__}: {e}") + print(f"Error {e.__class__.__name__}: {e}") finally: ir.close() - pass asyncio.run(main()) diff --git a/custom_components/flipper_rc/remote.py b/custom_components/flipper_rc/remote.py index fe9820d..fd4a5bd 100644 --- a/custom_components/flipper_rc/remote.py +++ b/custom_components/flipper_rc/remote.py @@ -82,6 +82,8 @@ def __init__(self, name, port, device_info_storage, device_info, codes_storage, self._codes_storage = codes_storage self._codes = codes self._available = False + self._last_error = None + self._last_operation = None self._device = FlipperIR(self._port) self._device.set_on_connection_lost(self._on_connection_lost) @@ -129,7 +131,12 @@ def device_info(self): @property def extra_state_attributes(self): - return self._device_info + attrs = dict(self._device_info) + if self._last_operation is not None: + attrs["last_operation"] = self._last_operation + if self._last_error is not None: + attrs["last_error"] = self._last_error + return attrs @property def supported_features(self): @@ -157,14 +164,16 @@ async def async_update(self): self._last_device_info_update = time.time() try: device_info = await self._device.get_device_info() - # compare with the previous device info if self._device_info != device_info: _LOGGER.info("Device info changed: %s", device_info) self._device_info = device_info await self._device_info_storage.async_save(self._device_info) self._available = True + except (TimeoutError, ConnectionError, OSError) as e: + _LOGGER.warning("Failed to update Flipper device info: %s", e) + self._available = False except Exception as e: - _LOGGER.error("Failed to update Flipper device info, exception %s: %s", type(e), e, exc_info=True) + _LOGGER.error("Unexpected error updating device info: %s", e, exc_info=True) self._available = False async def async_turn_on(self, **kwargs): @@ -181,7 +190,18 @@ async def async_list_subghz_files(self, root): async def async_send_subghz_from_file(self, path, repeat=1, antenna=0): """Public API for replaying Sub-GHz capture files from storage.""" - await self._device.send_subghz_from_file(path, repeat=repeat, antenna=antenna) + _LOGGER.info("async_send_subghz_from_file called: path=%s, repeat=%d, antenna=%d", path, repeat, antenna) + self._last_operation = f"Sending Sub-GHz file: {path}" + self._last_error = None + try: + await self._device.send_subghz_from_file(path, repeat=repeat, antenna=antenna) + _LOGGER.info("async_send_subghz_from_file succeeded for %s", path) + except Exception as e: + self._last_error = str(e) + _LOGGER.error("async_send_subghz_from_file failed for %s: %s", path, e) + raise HomeAssistantError(f"Failed to send Sub-GHz file '{path}' from Flipper Zero: {e}. " + "Check that the Flipper is connected, the file exists on the SD card, " + "and the Sub-GHz radio is not busy.") async def async_send_command(self, command, **kwargs): """Send a list of commands to a device.""" @@ -189,27 +209,33 @@ async def async_send_command(self, command, **kwargs): repeat = kwargs.get(ATTR_NUM_REPEATS, 1) repeat_delay = kwargs.get(ATTR_DELAY_SECS, 0) hold = kwargs.get(ATTR_HOLD_SECS, 0) - + if hold != 0: raise NotImplementedError("Hold time is not supported.") - + + _LOGGER.info("async_send_command called: commands=%s, device=%s, repeat=%d", command, device, repeat) + self._last_error = None + try: for n in range(repeat): for cmd in command: if device: - if not device in self._codes: + if device not in self._codes: raise KeyError(f"Device '{device}' not found in the codes storage.") - if not cmd in self._codes[device]: + if cmd not in self._codes[device]: raise KeyError(f"Command '{cmd}' not found in the codes storage for device '{device}'.") code = self._codes[device][cmd] - _LOGGER.debug("Sending command '%s' for device '%s', code: %s", cmd, device, code) + self._last_operation = f"Sending IR command '{cmd}' for device '{device}'" + _LOGGER.info("Sending IR command '%s' for device '%s', code: %s", cmd, device, code) else: code = cmd - _LOGGER.debug("Sending command, code: '%s'", code) + self._last_operation = f"Sending command: {code}" + _LOGGER.info("Sending command, code: '%s'", code) if isinstance(code, str) and code.startswith("subghz-file:"): tx = parse_subghz_file_command(code) - _LOGGER.debug("Sub-GHz file command parsed: %s", tx) + self._last_operation = f"Sending Sub-GHz file: {tx['path']}" + _LOGGER.info("Sub-GHz file command parsed: %s", tx) await self._device.send_subghz_from_file( path=tx["path"], repeat=tx["repeat"], @@ -217,7 +243,8 @@ async def async_send_command(self, command, **kwargs): ) elif isinstance(code, str) and code.startswith("subghz:"): tx = parse_subghz_command(code) - _LOGGER.debug("Sub-GHz command parsed: %s", tx) + self._last_operation = f"Sending Sub-GHz: key=0x{tx['key']:06X}, freq={tx['frequency']}" + _LOGGER.info("Sub-GHz command parsed: %s", tx) await self._device.send_subghz( key=tx["key"], frequency=tx["frequency"], @@ -227,16 +254,46 @@ async def async_send_command(self, command, **kwargs): ) else: pulses = rc_auto_encode(code) - _LOGGER.debug("Command pulses: %s", pulses) + self._last_operation = f"Sending IR signal ({len(pulses)} pulses)" + _LOGGER.info("Encoded IR command: %s pulses", len(pulses)) await self._device.send_ir(pulses) if n < repeat - 1 and repeat_delay > 0: await asyncio.sleep(repeat_delay) if not self._available: self._available = True self.schedule_update_ha_state() + _LOGGER.info("async_send_command completed successfully") + except HomeAssistantError: + # Re-raise HomeAssistantError as-is (already user-friendly) + raise + except TimeoutError as e: + self._last_error = str(e) + _LOGGER.error("Timeout sending command: %s", e) + raise HomeAssistantError( + f"Command timed out: {e}. " + "The Flipper Zero may be busy or unresponsive. " + "Try again in a moment, or check the device connection." + ) from e + except (ValueError, KeyError) as e: + self._last_error = str(e) + _LOGGER.error("Invalid command parameters: %s", e) + raise HomeAssistantError(f"Invalid command: {e}") from e + except ConnectionError as e: + self._last_error = str(e) + self._available = False + self.schedule_update_ha_state() + _LOGGER.error("Connection lost while sending command: %s", e) + raise HomeAssistantError( + f"Connection to Flipper Zero lost: {e}. " + "Please check the USB connection and try again." + ) from e except Exception as e: - _LOGGER.error("Failed to send command, exception %s: %s", type(e), e, exc_info=True) - raise HomeAssistantError(str(e)) + self._last_error = str(e) + _LOGGER.error("Failed to send command, exception %s: %s", type(e).__name__, e, exc_info=True) + raise HomeAssistantError( + f"Failed to send command to Flipper Zero: {e}. " + "Check device connectivity and try again." + ) from e async def async_learn_command(self, **kwargs): """Learn a command to a device, or just show the received command code.""" @@ -332,12 +389,12 @@ async def async_delete_command(self, **kwargs): if not device: raise HomeAssistantError("You need to specify a device.") - if not device in self._codes: + if device not in self._codes: raise HomeAssistantError(f"Device '{device}' not found in the codes storage.") deleted = False for command in commands: - if device in self._codes and command in self._codes[device]: + if command in self._codes.get(device, {}): del self._codes[device][command] deleted = True async_create( diff --git a/tests/test_subghz.py b/tests/test_subghz.py new file mode 100644 index 0000000..49d4a22 --- /dev/null +++ b/tests/test_subghz.py @@ -0,0 +1,382 @@ +"""Tests for FlipperIR Sub-GHz functionality.""" +import asyncio +import pytest +from unittest.mock import AsyncMock, MagicMock, patch +from pathlib import Path +import importlib.util + + +# Load flipper_ir module directly +_flipper_path = Path(__file__).resolve().parents[1] / "custom_components" / "flipper_rc" / "flipper_ir.py" +_spec = importlib.util.spec_from_file_location("flipper_ir", _flipper_path) +_module = importlib.util.module_from_spec(_spec) +assert _spec is not None and _spec.loader is not None +_spec.loader.exec_module(_module) + +FlipperIR = _module.FlipperIR +_is_supported_subghz_path = _module._is_supported_subghz_path + + +@pytest.fixture +def event_loop(): + """Create an instance of the default event loop for the test module.""" + loop = asyncio.new_event_loop() + yield loop + loop.close() + + +@pytest.fixture +def mock_protocol(): + """Create a mock FlipperProtocol.""" + protocol = AsyncMock() + protocol.connected = True + protocol.lines_available = 0 + protocol.buffer = b'' + protocol.has_prompt = False + protocol.wait_for_prompt = AsyncMock(return_value=[]) + protocol.readline = AsyncMock() + return protocol + + +@pytest.fixture +def mock_transport(): + """Create a mock transport.""" + transport = MagicMock() + transport.is_closing = MagicMock(return_value=False) + transport.write = MagicMock() + return transport + + +class TestIsSupportedSubghzPath: + """Tests for _is_supported_subghz_path helper.""" + + def test_valid_ext_path(self): + assert _is_supported_subghz_path("/ext/subghz/test.sub") is True + + def test_valid_ext_root(self): + assert _is_supported_subghz_path("/ext/") is True + + def test_invalid_int_path(self): + assert _is_supported_subghz_path("/int/subghz/test.sub") is False + + def test_invalid_no_slash(self): + assert _is_supported_subghz_path("ext/subghz/test.sub") is False + + def test_non_string_returns_false(self): + assert _is_supported_subghz_path(123) is False + assert _is_supported_subghz_path(None) is False + + +class TestSendSubghzValidation: + """Tests for send_subghz parameter validation.""" + + @pytest.mark.asyncio + async def test_valid_parameters(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + mock_protocol.wait_for_prompt.return_value = [ + ">: subghz tx 123456 433920000 350 1 0", + "Transmitting...", + ">: " + ] + + await ir.send_subghz(key=0x123456, frequency=433920000, te=350, repeat=1, antenna=0) + assert mock_transport.write.called + + @pytest.mark.asyncio + async def test_rejects_negative_key(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="key must be in range"): + await ir.send_subghz(key=-1, frequency=433920000) + + @pytest.mark.asyncio + async def test_rejects_key_over_max(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="key must be in range"): + await ir.send_subghz(key=0x1000000, frequency=433920000) + + @pytest.mark.asyncio + async def test_rejects_zero_frequency(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="frequency must be positive"): + await ir.send_subghz(key=0x123456, frequency=0) + + @pytest.mark.asyncio + async def test_rejects_negative_frequency(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="frequency must be positive"): + await ir.send_subghz(key=0x123456, frequency=-100) + + @pytest.mark.asyncio + async def test_rejects_invalid_antenna(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="antenna must be 0 or 1"): + await ir.send_subghz(key=0x123456, frequency=433920000, antenna=2) + + @pytest.mark.asyncio + async def test_rejects_zero_te(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="te must be positive"): + await ir.send_subghz(key=0x123456, frequency=433920000, te=0) + + @pytest.mark.asyncio + async def test_rejects_zero_repeat(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="repeat must be positive"): + await ir.send_subghz(key=0x123456, frequency=433920000, repeat=0) + + +class TestSendSubghzFromFileValidation: + """Tests for send_subghz_from_file parameter validation.""" + + @pytest.mark.asyncio + async def test_valid_parameters(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + mock_protocol.wait_for_prompt.return_value = [ + ">: subghz tx_from_file /ext/subghz/test.sub 1 0", + "Frequency=433920000", + ">: " + ] + + await ir.send_subghz_from_file(path="/ext/subghz/test.sub", repeat=1, antenna=0) + assert mock_transport.write.called + + @pytest.mark.asyncio + async def test_rejects_non_ext_path(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match='must start with "/ext/"'): + await ir.send_subghz_from_file(path="/int/subghz/test.sub") + + @pytest.mark.asyncio + async def test_rejects_path_with_newline(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="forbidden control characters"): + await ir.send_subghz_from_file(path="/ext/subghz/test\n.sub") + + @pytest.mark.asyncio + async def test_rejects_path_with_null(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="forbidden control characters"): + await ir.send_subghz_from_file(path="/ext/subghz/test\x00.sub") + + @pytest.mark.asyncio + async def test_rejects_zero_repeat(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="repeat must be positive"): + await ir.send_subghz_from_file(path="/ext/subghz/test.sub", repeat=0) + + @pytest.mark.asyncio + async def test_rejects_invalid_antenna(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + with pytest.raises(ValueError, match="antenna must be 0 or 1"): + await ir.send_subghz_from_file(path="/ext/subghz/test.sub", antenna=2) + + +class TestValidateCliResponse: + """Tests for _validate_cli_response validation logic.""" + + def test_accepts_valid_subghz_tx_with_frequency(self): + ir = FlipperIR("/dev/ttyACM0") + lines = [">: subghz tx 123456 433920000", "Frequency=433920000", ">: "] + # Should not raise + ir._validate_cli_response(lines, [">: subghz tx"], "subghz tx") + + def test_accepts_valid_subghz_tx_with_transmitting(self): + ir = FlipperIR("/dev/ttyACM0") + lines = [">: subghz tx 123456 433920000", "Transmitting...", ">: "] + ir._validate_cli_response(lines, [">: subghz tx"], "subghz tx") + + def test_rejects_subghz_tx_without_indicator(self): + ir = FlipperIR("/dev/ttyACM0") + lines = [">: subghz tx 123456 433920000", ">: "] + with pytest.raises(ValueError, match="did not confirm transmission"): + ir._validate_cli_response(lines, [">: subghz tx"], "subghz tx") + + def test_rejects_subghz_tx_with_error(self): + ir = FlipperIR("/dev/ttyACM0") + lines = [">: subghz tx 123456 433920000", "Error: invalid key", ">: "] + with pytest.raises(ValueError, match="failed"): + ir._validate_cli_response(lines, [">: subghz tx"], "subghz tx") + + def test_accepts_non_tx_with_prefix(self): + ir = FlipperIR("/dev/ttyACM0") + lines = [">: info device", "Some info", ">: "] + # Should not raise + ir._validate_cli_response(lines, [">: info device"], "info device") + + def test_accepts_non_tx_without_prefix(self): + ir = FlipperIR("/dev/ttyACM0") + lines = ["Some response", ">: "] + # Should not raise (tolerant fallback) + ir._validate_cli_response(lines, [">: expected"], "command") + + def test_rejects_any_command_with_error(self): + ir = FlipperIR("/dev/ttyACM0") + lines = ["Error: something failed", ">: "] + with pytest.raises(ValueError, match="failed"): + ir._validate_cli_response(lines, [">: cmd"], "cmd") + + def test_handles_empty_lines(self): + ir = FlipperIR("/dev/ttyACM0") + lines = ["", " ", ">: "] + ir._validate_cli_response(lines, [">: cmd"], "cmd") + + def test_subghz_tx_from_file_with_frequency_indicator(self): + ir = FlipperIR("/dev/ttyACM0") + lines = [ + ">: subghz tx_from_file /ext/subghz/test.sub 1 0", + "Listening at /ext/subghz/test.sub. Frequency=434176948, Protocol=RAW", + ">: " + ] + ir._validate_cli_response(lines, [">: subghz tx_from_file"], "subghz tx_from_file") + + +class TestSendCtrlC: + """Tests for _send_ctrl_c method.""" + + def test_sends_ctrl_c_when_connected(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + mock_transport.is_closing.return_value = False + + ir._send_ctrl_c() + mock_transport.write.assert_called_with(b'\x03') + + def test_noop_when_no_transport(self, mock_protocol): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = None + + # Should not raise + ir._send_ctrl_c() + + def test_noop_when_transport_closing(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + mock_transport.is_closing.return_value = True + + ir._send_ctrl_c() + mock_transport.write.assert_not_called() + + +class TestCommandBuilding: + """Tests for command string generation.""" + + @pytest.mark.asyncio + async def test_subghz_tx_command_format(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + mock_protocol.wait_for_prompt.return_value = [ + ">: subghz tx 123456 433920000 350 1 0", + "Transmitting...", + ">: " + ] + + await ir.send_subghz(key=0x123456, frequency=433920000, te=350, repeat=1, antenna=0) + + # Verify the command was sent correctly + write_calls = mock_transport.write.call_args_list + assert len(write_calls) > 0 + cmd_sent = write_calls[0][0][0].decode() + assert "subghz tx 123456 433920000 350 1 0" in cmd_sent + + @pytest.mark.asyncio + async def test_subghz_tx_from_file_command_format(self, mock_protocol, mock_transport): + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + mock_protocol.wait_for_prompt.return_value = [ + ">: subghz tx_from_file /ext/subghz/test.sub 2 1", + "Frequency=433920000", + ">: " + ] + + await ir.send_subghz_from_file(path="/ext/subghz/test.sub", repeat=2, antenna=1) + + # Verify the command was sent correctly + write_calls = mock_transport.write.call_args_list + assert len(write_calls) > 0 + cmd_sent = write_calls[0][0][0].decode() + assert "subghz tx_from_file /ext/subghz/test.sub 2 1" in cmd_sent + + @pytest.mark.asyncio + async def test_subghz_tx_key_zero_padded(self, mock_protocol, mock_transport): + """Verify key is zero-padded to 6 hex digits.""" + ir = FlipperIR("/dev/ttyACM0") + ir._protocol = mock_protocol + ir._transport = mock_transport + ir._connected = True + + mock_protocol.wait_for_prompt.return_value = [ + ">: subghz tx 001234 433920000 350 1 0", + "Transmitting...", + ">: " + ] + + await ir.send_subghz(key=0x1234, frequency=433920000, te=350, repeat=1, antenna=0) + + write_calls = mock_transport.write.call_args_list + cmd_sent = write_calls[0][0][0].decode() + assert "001234" in cmd_sent # Zero-padded