Три года назад у меня возникла идея создать систему управления аудио-видео потоками в доме. Хотелось, чтобы любой источник сигнала — кабельное ТВ, медиаплеер, игровая консоль или ПК — мог быть мгновенно переключен на любой телевизор в квартире. Достаточно нажать кнопку в смартфоне, и просмотр фильма продолжается в другой комнате без потери качества и необходимости бегать с пультами.
Тогда проект отложили: не хватало знаний в программировании, а стоимость подходящей HDMI-матрицы с возможностью внешнего управления казалась неподъемной.
Что изменилось
Спустя время ситуация поменялась. Во-первых, на Хабре появилась подробная статья от @metalstiv, в которой автор пошагово описал процесс реверс-инжиниринга протокола управления бюджетной матрицей 4×4 через RS-232. Это дало понимание, что задача решаема даже без глубоких технических знаний, если есть желание и немного помощи от современных инструментов.
Во-вторых, цены на оборудование стали доступнее. Была заказана матрица 4×4 с поддержкой 4K@60 Гц и управлением по RS-232. Параллельно приобрел USB-to-RS232 конвертер для подключения к управляющему устройству.
В-третьих, появился мощный ассистент в виде нейросети Qwen, которая помогла разобраться с протоколом обмена, написать код интеграции и адаптировать решения под Home Assistant.
Что понадобилось для реализации
Home Assistant — центральная платформа умного дома. Выбрана за гибкость, открытость и огромное сообщество.
USB-to-RS232 конвертер — для физического подключения матрицы к управляющему устройству (в моем случае — Raspberry Pi).
HDMI Matrix 4×4 с поддержкой RS-232 — аппаратная основа системы. Коммутирует любой из четырех входов на любой из четырех выходов.
Нейросеть Qwen — использовалась как интеллектуальный помощник: для анализа протокола, генерации кода, отладки команд и оптимизации интеграции.
Как это работает
Интеграция построена по модульному принципу и использует паттерн Coordinator для безопасной работы с последовательным портом. Вся логика разделена на три уровня: управление соединением, обработка команд и представление сущностей в Home Assistant.
Архитектура компонента
В основе лежит класс HDMISwitchCoordinator, который управляет подключением к матрице через RS-232. Поскольку работа с последовательным портом — блокирующая операция, вся коммуникация вынесена в отдельные потоки (threading), чтобы не замедлять основной цикл событий Home Assistant.
Координатор выполняет три ключевые задачи:
Инициализация порта с параметрами по умолчанию: /dev/ttyACM0, скорость 115200 бод, таймаут 5 секунд;
Отправка команд с потокобезопасной блокировкой (threading.Lock), что исключает конфликты при одновременных запросах;
Фоновое чтение ответов от устройства в цикле listenloop и передача статусов зарегистрированным слушателям.
Протокол обмена данными
Матрица принимает команды в текстовом формате. Для переключения источника используется команда cir, за которой следует шестнадцатеричное значение, рассчитываемое по формуле:
значение = 8 × (номер_зоны − 1) + (номер_входа − 1)
Например:
Переключить вход 1 на зону 1:
cir 00Переключить вход 3 на зону 2:
cir 11(8×1 + 2 = 10 → 0x0A, но в коде используется форматирование :02x, поэтому отправляетсяcir 0a)
Команда завершается символами перевода строки \r\n, что обеспечивает корректную обработку на стороне устройства.
Устройство может отправлять уведомления о текущем состоянии в формате s{zone}{input}, например s13 означает, что в зоне 1 активен вход 3. Координатор парсит такие сообщения и через механизм callback обновляет состояние соответствующей сущности в Home Assistant.
Сущности в Home Assistant
Для каждой зоны матрицы создаётся отдельная сущность типа media_player. Это позволяет:
Отображать текущий источник в интерфейсе;
Переключать входы через выпадающий список;
Использовать стандартные сервисы media_player.select_source;
Задавать кастомные названия источников через параметр sources в конфигурации.
Каждая сущность поддерживает:
Включение (переключение на первый источник в списке);
Выключение (переключение на второй источник, условно «заглушка»);
Отображение иконки входа через свойство entity_picture — если в папке /config/www/hdmi_
switch/ размещены файлы img_1.png, img_2.png и т.д., они автоматически подставляются в карточку.
Структура файлов и установка
Для работы интеграции необходимо создать специальную структуру папок в директории конфигурации Home Assistant.
Структура папок
Все файлы интеграции должны находиться в папке custom_components/hdmi_switch/. Внутри также потребуется папка images для хранения иконок источников.
config/ ├── custom_components/ │ └── hdmi_switch/ │ ├── __init__.py # Основной файл инициализации и координатор │ ├── manifest.json # Манифест интеграции (версия, зависимости) │ ├── const.py # Константы (домен, настройки по умолчанию) │ ├── media_player.py # Логика сущностей медиаплеера │ └── images/ # Папка для иконок источников (img_1.png и т.д.) ├── www/ │ └── hdmi_switch/ # Сюда копируются изображения автоматически └── configuration.yaml # Файл конфигурации Home Assistant
Установка через Samba
Самый удобный способ загрузить файлы интеграции — использовать сетевой доступ Samba, который предоставляется в Home Assistant OS.
Включите Samba в Home Assistant:
Зайдите в Настройки -> Дополнения -> Магазин дополнений.
Найдите и установите дополнение "Samba share".
В конфигурации дополнения укажите имя пользователя и пароль.
Запустите дополнение.
Подключитесь к файловой системе:
На компьютере с Windows откройте Проводник.
В адресной строке введите
\\homeassistant\config(или IP-адрес вашего сервера).Введите логин и пароль, заданные в дополнении Samba.
Создайте структуру папок:
Перейдите в папку
custom_components. Если её нет, создайте.Внутри создайте папку
hdmi_switch.Внутри
hdmi_switchсоздайте папкуimages.
Загрузите файлы:
Создайте текстовые файлы с именами __
init __.py,manifest.json,const.py,media_player.py.Скопируйте в них соответствующий код (приведен ниже).
Поместите изображения источников (
img_1.png,img_2.png, img_3.png,img_4.png) в папкуimages.
Ссылки на изобращения:а)img_1.pngб)img_2.pngв)img_3.pngг)img_4.pngПерезагрузите Home Assistant:
Зайдите в Настройки -> Система -> Перезагрузка.
После перезагрузки интеграция будет готова к настройке через
configuration.yaml.
Код интеграции с комментариями
Ниже приведены основные файлы интеграции с подробными комментариями, объясняющими назначение каждого блока кода.
manifest.json
Файл манифеста сообщает Home Assistant метаданные об интеграции: имя, версию, необходимые библиотеки и тип подключения.
{ "domain": "hdmi_switch", "name": "HDMI Switch", "version": "1.1.0", "requirements": ["pyserial>=3.5"], "iot_class": "local_push", "config_flow": false }
const.py
Файл констант хранит настройки по умолчанию и идентификатор домена. Это упрощает поддержку кода, так как все магические числа и строки собраны в одном месте.
"""Константы для HDMI Switch.""" DOMAIN = "hdmi_switch" DEFAULT_NAME = "HDMI Switch" DEFAULT_PORT = "/dev/ttyACM0" DEFAULT_COMMAND_BAUDRATE = 115200 DEFAULT_TIMEOUT = 5.0 DEFAULT_SOURCES = ["Вход 1", "Вход 2", "Вход 3", "Вход 4"] """ Если в configuration.yaml не указать hdmi_switch: port: /dev/ttyACM1 command_baudrate: 9600 timeout: 10 То эти значения встанут по умолчанию DEFAULT_PORT = "/dev/ttyACM0" DEFAULT_COMMAND_BAUDRATE = 115200 DEFAULT_TIMEOUT = 5.0""""
__init __.py
Основной файл инициализации. Здесь создается Координатор, который управляет потоками и соединением с портом.
Обратите внимание на имя файла инициализации: __init__.py. Оно должно содержать двойное подчеркивание слева и справа. Без этого Python не распознает папку как пакет, и интеграция не загрузится.
"""HDMI Switch Integration.""" import logging import os import shutil import threading import time import serial import asyncio from homeassistant.core import HomeAssistant from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.discovery import load_platform from .const import DOMAIN, DEFAULT_PORT, DEFAULT_COMMAND_BAUDRATE, DEFAULT_TIMEOUT _LOGGER = logging.getLogger(__name__) class HDMISwitchCoordinator: """Координатор для управления общим последовательным портом.""" def __init__(self, hass, port, baudrate, timeout): self.hass = hass self.port = port self.baudrate = baudrate self.timeout = timeout self._serial = None self._lock = threading.Lock() self._stop_listener = False self._listener_thread = None self._listeners = {} # zone_id -> callback function self._initialized = False def register_listener(self, zone_id, callback): """Зарегистрировать callback для зоны.""" self._listeners[zone_id] = callback _LOGGER.debug("Зона %d: зарегистрирована в координаторе", zone_id) def unregister_listener(self, zone_id): """Удалить callback для зоны.""" if zone_id in self._listeners: del self._listeners[zone_id] _LOGGER.debug("Зона %d: удалена из координатора", zone_id) def start(self): """Запустить поток инициализации и слушателя.""" # Запускаем инициализацию в отдельном потоке, чтобы не блокировать event loop thread = threading.Thread(target=self._initialize_and_listen, daemon=True) thread.start() return True def _initialize_and_listen(self): """Инициализация порта и запуск слушателя (в потоке).""" if not self._open_serial(): _LOGGER.error("Не удалось открыть порт в потоке инициализации") return self._initialized = True self._stop_listener = False self._listener_thread = threading.Thread(target=self._listen_loop, daemon=True) self._listener_thread.start() _LOGGER.info("Координатор запущен, порт %s открыт", self.port) def stop(self): """Остановить поток и закрыть порт.""" self._stop_listener = True if self._listener_thread and self._listener_thread.is_alive(): self._listener_thread.join(timeout=2) self._close_serial() _LOGGER.info("Координатор остановлен, порт закрыт") def _open_serial(self): """Открыть последовательный порт (вызывается в потоке).""" try: self._serial = serial.Serial( port=self.port, baudrate=self.baudrate, timeout=0.5, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE ) # time.sleep здесь безопасен, т.к. мы в отдельном потоке time.sleep(0.2) _LOGGER.debug("Порт %s открыт успешно", self.port) return True except Exception as e: _LOGGER.error("Ошибка открытия порта %s: %s", self.port, e) return False def _close_serial(self): """Закрыть последовательный порт.""" if self._serial: try: if self._serial.is_open: self._serial.close() _LOGGER.debug("Порт %s закрыт", self.port) except Exception as e: _LOGGER.error("Ошибка закрытия порта: %s", e) self._serial = None def send_command(self, command): """Отправить команду через порт (с блокировкой).""" # Ждем инициализации порта retries = 0 while not self._initialized and retries < 10: time.sleep(0.1) retries += 1 if not self._initialized: _LOGGER.error("Порт еще не инициализирован") return False with self._lock: try: if not self._serial or not self._serial.is_open: if not self._open_serial(): return False _LOGGER.debug("Отправка команды: %s", command.strip()) self._serial.write(command.encode('utf-8')) self._serial.flush() time.sleep(0.1) if self._serial.in_waiting > 0: response = self._serial.readline().decode('utf-8', errors='ignore').strip() _LOGGER.debug("Ответ устройства: %s", response) return True except Exception as e: _LOGGER.error("Ошибка отправки команды: %s", e) return False def _listen_loop(self): """Цикл чтения данных из порта в фоновом потоке.""" _LOGGER.info("Поток слушателя запущен") while not self._stop_listener: try: with self._lock: if self._serial and self._serial.is_open: if self._serial.in_waiting > 0: line = self._serial.readline().decode('utf-8', errors='ignore').strip() if line: _LOGGER.debug("Получено из порта: %s", line) self._handle_incoming_status(line) else: pass else: self._open_serial() time.sleep(0.1) except Exception as e: _LOGGER.error("Ошибка в цикле слушателя: %s", e) time.sleep(1) _LOGGER.info("Поток слушателя остановлен") def _handle_incoming_status(self, status_line): """Обработка входящего статуса (например, 's11').""" if not status_line.startswith('s') or len(status_line) < 3: return try: zone = int(status_line[1]) input_num = int(status_line[2]) if zone in self._listeners: callback = self._listeners[zone] self.hass.add_job(callback, zone, input_num) _LOGGER.debug("Зона %d: статус передан в callback (вход %d)", zone, input_num) except ValueError as e: _LOGGER.debug("Не удалось распарсить статус %s: %s", status_line, e) async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Настройка из configuration.yaml.""" _LOGGER.info("HDMI Switch: async_setup вызван") # Копирование изображений (в executor, чтобы не блокировать поток) await hass.async_add_executor_job(setup_images, hass) if "hdmi_switch" not in config: _LOGGER.info("HDMI Switch: нет конфигурации в YAML") return True # Создаем координатор и сохраняем в hass.data hdmi_config = config["hdmi_switch"] port = hdmi_config.get("port", DEFAULT_PORT) baudrate = hdmi_config.get("command_baudrate", DEFAULT_COMMAND_BAUDRATE) timeout = hdmi_config.get("timeout", DEFAULT_TIMEOUT) coordinator = HDMISwitchCoordinator(hass, port, baudrate, timeout) hass.data[DOMAIN] = coordinator # Запускаем координатор (теперь это не блокирует event loop) coordinator.start() # Даем время на инициализацию порта перед загрузкой платформы await asyncio.sleep(0.5) _LOGGER.info("HDMI Switch: загрузка media_player платформы") load_platform( hass, "media_player", "hdmi_switch", hdmi_config, config ) _LOGGER.info("HDMI Switch: платформа загружена") return True async def async_unload_platforms(hass, config): """Выгрузка интеграции.""" _LOGGER.info("HDMI Switch: выгрузка интеграции") if DOMAIN in hass.data: coordinator = hass.data[DOMAIN] coordinator.stop() del hass.data[DOMAIN] return True def setup_images(hass): """Копирование изображений в папку www.""" component_dir = os.path.dirname(__file__) images_dir = os.path.join(component_dir, "images") www_dir = os.path.join(hass.config.config_dir, "www", "hdmi_switch") os.makedirs(www_dir, exist_ok=True) if os.path.exists(images_dir): for filename in os.listdir(images_dir): if filename.endswith(('.png', '.jpg', '.jpeg', '.gif', '.webp')): src = os.path.join(images_dir, filename) dst = os.path.join(www_dir, filename) shutil.copy2(src, dst) _LOGGER.info("HDMI Switch: Скопировано изображение %s", filename) else: _LOGGER.debug("HDMI Switch: Папка images не найдена")
media_player.py
Файл описывает сущности медиаплеера. Каждая зона матрицы представляется как отдельное устройство в Home Assistant.
"""HDMI Switch Media Player.""" import logging import os from homeassistant.components.media_player import MediaPlayerEntity, MediaPlayerEntityFeature from homeassistant.const import STATE_ON, STATE_OFF from homeassistant.helpers.entity import DeviceInfo from .const import DOMAIN, DEFAULT_NAME, DEFAULT_SOURCES _LOGGER = logging.getLogger(__name__) SUPPORTED_FEATURES = ( MediaPlayerEntityFeature.TURN_ON | MediaPlayerEntityFeature.TURN_OFF | MediaPlayerEntityFeature.SELECT_SOURCE ) class HDMISwitchPlayer(MediaPlayerEntity): """HDMI Switch Media Player.""" def __init__(self, coordinator, name, unique_id, zone, sources): self._coordinator = coordinator self._name = name self._unique_id = unique_id self._zone = zone self._sources = sources if sources else DEFAULT_SOURCES self._source_list = self._sources self._state = STATE_OFF self._source = None self._available = False _LOGGER.info("HDMI Switch: %s (Зона %d) инициализирован", name, zone) @property def unique_id(self): return self._unique_id @property def name(self): return self._name @property def state(self): return self._state @property def available(self): return self._available @property def supported_features(self): return SUPPORTED_FEATURES @property def source(self): return self._source @property def source_list(self): return self._source_list @property def icon(self): return "mdi:hdmi-port" @property def entity_picture(self): """Вернуть путь к изображению контента.""" if self._state == STATE_OFF or not self._source: return None if self._source in self._sources: index = self._sources.index(self._source) input_number = index + 1 image_file = f"img_{input_number}.png" image_path = f"/config/www/hdmi_switch/{image_file}" if os.path.exists(image_path): return f"/local/hdmi_switch/{image_file}" return None @property def device_info(self): return DeviceInfo( identifiers={(DOMAIN, self._unique_id)}, name=self._name, manufacturer="HDMI Matrix", model=f"Switch Zone {self._zone}", ) def _get_command(self, input_index): """Получить команду cir для зоны и входа.""" val = 8 * (self._zone - 1) + input_index return f"cir {val:02x}\r\n" def _switch_input(self, source): """Переключить вход.""" if source not in self._sources: _LOGGER.error("Неизвестный источник: %s", source) return False try: index = self._sources.index(source) except ValueError: _LOGGER.error("Источник %s не найден", source) return False command = self._get_command(index) _LOGGER.info("Зона %d: Переключение на %s (команда %s)", self._zone, source, command.strip()) if self._coordinator.send_command(command): self._source = source self._state = STATE_ON self._available = True _LOGGER.info("Зона %d: Успешно переключено на %s", self._zone, source) self.schedule_update_ha_state() return True self._available = False return False def _status_callback(self, zone, input_num): """Callback для обновления статуса от координатора.""" if zone != self._zone: return index = input_num - 1 if 0 <= index < len(self._sources): new_source = self._sources[index] if new_source != self._source: _LOGGER.info("Зона %d: Обнаружено внешнее переключение на %s", self._zone, new_source) self._source = new_source self._state = STATE_ON self._available = True self.schedule_update_ha_state() def turn_on(self): """Включить (Вход 1).""" if self._sources: self._switch_input(self._sources[0]) def turn_off(self): """Выключить (Вход 2).""" if len(self._sources) > 1: self._switch_input(self._sources[1]) self._state = STATE_OFF self.schedule_update_ha_state() def select_source(self, source): """Переключить источник.""" if source in self._source_list: self._switch_input(source) def update(self): """Нет опроса - состояние обновляется через callback.""" pass async def async_added_to_hass(self): """Запуск при добавлении устройства в HA.""" await super().async_added_to_hass() self._coordinator.register_listener(self._zone, self._status_callback) self._available = True _LOGGER.info("Зона %d: зарегистрирована в координаторе", self._zone) async def async_will_remove_from_hass(self): """Остановка при удалении устройства.""" self._coordinator.unregister_listener(self._zone) _LOGGER.info("Зона %d: удалена из координатора", self._zone) await super().async_will_remove_from_hass() async def async_setup_platform(hass, config, async_add_entities, discovery_info=None): """Настройка платформы.""" _LOGGER.info("HDMI Switch: async_setup_platform вызван") conf = discovery_info if discovery_info else config zones_config = conf.get("zones", []) # Получаем координатор из hass.data coordinator = hass.data.get(DOMAIN) if not coordinator: _LOGGER.error("Координатор HDMI Switch не найден") return _LOGGER.info("HDMI Switch: Найдено зон: %d", len(zones_config)) if not zones_config: zones_config = [{ "zone": 1, "name": conf.get("name", DEFAULT_NAME), "sources": conf.get("sources", None), "unique_id": conf.get("unique_id", None), }] players = [] for zone_config in zones_config: zone = zone_config.get("zone", 1) name = zone_config.get("name", f"{DEFAULT_NAME} Зона {zone}") sources = zone_config.get("sources", None) unique_id = zone_config.get("unique_id", f"hdmi_switch_zone{zone}") _LOGGER.info("HDMI Switch: Зона %d - Источники: %s", zone, sources) player = HDMISwitchPlayer( coordinator=coordinator, name=name, unique_id=unique_id, zone=zone, sources=sources, ) players.append(player) async_add_entities(players, False) _LOGGER.info("HDMI Switch: %d устройств добавлено", len(players))
Конфигурация
После размещения файлов необходимо добавить запись в configuration.yaml. Пример для матрицы 4x4 с двумя телевизорами:
hdmi_switch: port: /dev/ttyACM0 command_baudrate: 115200 timeout: 5 zones: - zone: 1 name: "HDMI Гостиная" unique_id: "hdmi_switch_living" sources: - "ТВ" - "Apple TV" - "PlayStation" - "NVIDIA Shield" - zone: 2 name: "HDMI Спальня" unique_id: "hdmi_switch_bedroom" sources: - "ТВ" - "Apple TV" - "PlayStation" - "NVIDIA Shield" - zone: 3 name: "HDMI Кухня" unique_id: "hdmi_switch_kitchen" sources: - "ТВ" - "Apple TV" - "PlayStation" - "NVIDIA Shield" - zone: 4 name: "HDMI Офис" unique_id: "hdmi_switch_office" sources: - "ТВ" - "Apple TV" - "PlayStation" - "NVIDIA Shield"
После добавления конфигурации обязательно перезагрузите Home Assistant. В интерфейсе появятся новые устройства медиаплеер, которыми можно управлять через стандартные карточки Lovelace.
