siwat
/
benq-rm-homeassistant
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

initial commit

This commit is contained in:
Siwat Sirichai 2025-01-25 19:42:38 +07:00
commit ddec014571
6 changed files with 1030 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
README.md Normal file
View File

@ -0,0 +1,3 @@
# BenQ Smart Board Home Assistant Integration
A custom integration for Home Assistant that allows you to control BenQ Smart Boards using RS232-over-LAN (port 4660). This integration provides Media Player functionality for power, volume, mute, and more. It is implemented using the official BenQ RS232 protocol.

43
__init__.py Normal file
View File

@ -0,0 +1,43 @@
"""BenQ Smart Board integration for Home Assistant."""
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
DOMAIN = "benq_smartboard"
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistant, config: dict):
"""
Legacy setup function for when we might support YAML in the future.
Currently, we do not configure anything via YAML. Just return True.
"""
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""
Set up the BenQ Smart Board integration from a config entry (UI flow).
"""
hass.data.setdefault(DOMAIN, {})
# Store config data (host, port, etc.) so other platforms can access it
hass.data[DOMAIN][entry.entry_id] = entry.data
# Forward entry setup to the media_player platform
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "media_player")
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""
Unload a config entry (if the user removes it).
We must also unload the forwarded platforms.
"""
unload_ok = await hass.config_entries.async_forward_entry_unload(entry, "media_player")
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id, None)
return unload_ok

669
benq_smartboard_lib.py Normal file
View File

@ -0,0 +1,669 @@
"""
A comprehensive Python library for controlling BenQ Smart Board via RS232-over-LAN.
Implements all protocol commands from the official specification.
"""
import socket
import struct
from typing import Optional, Dict, Any
from enum import Enum, unique
# -- Exceptions ----------------------------------------------------------------
class BenQSmartBoardError(Exception):
"""Base exception for BenQ Smart Board errors."""
pass
class ConnectionError(BenQSmartBoardError):
"""Exception for connection-related errors."""
pass
class CommandError(BenQSmartBoardError):
"""Exception for invalid or failed commands."""
pass
# -- Enums for Commands --------------------------------------------------------
@unique
class PowerState(Enum):
OFF = '000'
ON = '001'
STANDBY = '002'
REBOOT = '003'
@unique
class VideoSource(Enum):
VGA = '000'
HDMI = '001'
HDMI1 = '002'
HDMI2 = '021'
DISPLAY_PORT = '007'
TYPE_C = '051'
ANDROID = '101'
OPS = '102'
DEFAULT = 'FF' # used in E0 timer to restore last signal, etc.
@unique
class AspectRatio(Enum):
SIXTEEN_NINE = '000'
PTP = '002'
@unique
class Language(Enum):
ENGLISH = '000'
FRANCAIS = '001'
ESPANOL = '002'
FANTZHONG = '003'
JIANZHONG = '004'
PORTUGUES = '005'
GERMAN = '006'
DUTCH = '007'
POLISH = '008'
RUSSIA = '009'
CZECH = '010'
DANISH = '011'
SWEDISH = '012'
ITALIAN = '013'
ROMANIAN = '014'
NORWEGIAN = '015'
FINNISH = '016'
GREEK = '017'
TURKISH = '018'
ARABIC = '019'
JAPANSE = '020'
THAILAND = '021'
KOREAN = '022'
HUNGARIAN = '023'
PERSIAN = '024'
VIETNAMESE = '025'
INDONESIA = '026'
HEBREW = '027'
@unique
class SoundMode(Enum):
CINEMA = '000'
STANDARD = '001'
MUSIC = '003'
NEWS = '005'
SPORTS = '006'
@unique
class RemoteCommand(Enum):
VOL_PLUS = '000'
VOL_MINUS = '001'
REMOTE_UP = '010'
REMOTE_DOWN = '011'
REMOTE_LEFT = '012'
REMOTE_RIGHT = '013'
REMOTE_OK = '014'
REMOTE_MENU_KEY = '020'
REMOTE_EXIT = '022'
BLANK = '031'
FREEZE = '032'
@unique
class IRControl(Enum):
DISABLE = '000'
ENABLE = '001'
@unique
class ButtonIRControl(Enum):
DISABLE = '000'
ENABLE = '001'
@unique
class ButtonControl(Enum):
DISABLE = '000'
ENABLE = '001'
@unique
class PixelShift(Enum):
OFF = '000'
ON = '001'
@unique
class PictureMode(Enum):
STANDARD = '000'
BRIGHT = '001'
SOFT = '002'
ECO = '003'
CUSTOM_1 = '005'
CUSTOM_2 = '006'
CUSTOM_3 = '007'
@unique
class DCR(Enum):
OFF = '000'
ON = '001'
@unique
class ColorTemp(Enum):
COOL = '000'
NORMAL = '001'
WARM = '002'
@unique
class PowerSaveMode(Enum):
OFF = '000'
LOW = '001'
HIGH = '002'
@unique
class SwitchOnStatus(Enum):
POWER_OFF = '000'
FORCE_ON = '001'
LAST_STATUS = '002'
# -- The Main BenQSmartBoard Class --------------------------------------------
class BenQSmartBoard:
"""
A client for controlling BenQ Smart Boards via LAN (RS232-over-TCP).
Implements all commands from the official protocol specification.
"""
DEFAULT_PORT = 4660
TV_ID = b'01'
CR = b'\x0D'
def __init__(self, ip: str, port: int = DEFAULT_PORT, timeout: float = 5.0):
"""
:param ip: IP address of the BenQ Smart Board
:param port: TCP port, default 4660
:param timeout: socket timeout in seconds
"""
self.ip = ip
self.port = port
self.timeout = timeout
self.sock: Optional[socket.socket] = None
# -- Connection Management -------------------------------------------------
def connect(self):
"""Establish a TCP connection to the Smart Board."""
if self.sock is not None:
# Already connected or trying
return
try:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(self.timeout)
self.sock.connect((self.ip, self.port))
except socket.error as e:
self.sock = None
raise ConnectionError(f"Failed to connect: {e}")
def disconnect(self):
"""Close the TCP connection."""
if self.sock:
try:
self.sock.close()
except socket.error:
pass
self.sock = None
def _ensure_connection(self):
"""Helper to ensure we are connected; reconnect if needed."""
if not self.sock:
self.connect()
# -- Core Send/Receive -----------------------------------------------------
def send_command(self, command_code: str, value: str) -> bool:
"""
Send a set command to the Smart Board.
Returns True if valid (+), False if invalid (-).
Raises ConnectionError on network issues, CommandError on unknown response.
"""
self._ensure_connection()
# Protocol requires length=8 (excluding CR), "TV ID=01", command type='s' (0x73)
# Then command_code (1 byte ASCII) + value (3 bytes ASCII)
if len(command_code) != 2:
raise ValueError("command_code must be 2 ASCII chars (hex).")
if len(value) != 3:
raise ValueError("value must be 3 ASCII chars.")
length = 8 # as per spec
try:
packet = struct.pack(
">B2s1s2s3s1s",
length,
self.TV_ID,
b's', # Set command
command_code.encode(),
value.encode(),
self.CR
)
self.sock.sendall(packet)
response = self._receive_response()
if response == b'+':
return True
elif response == b'-':
return False
else:
# Unknown response type
raise CommandError(f"Unknown response type: {response}")
except socket.error as e:
self.disconnect()
raise ConnectionError(f"Failed to send command: {e}")
def get_command(self, command_code: str) -> str:
"""
Send a get command to the Smart Board.
Returns the 3-byte value as a string if the device replies with data,
or raises CommandError/ConnectionError on issues.
"""
self._ensure_connection()
# For get command, protocol is less documented in your specification,
# but we assume length=6, "TV ID=01", command type='g' (0x67),
# then command_code (2 ASCII) + some placeholder for value?
# We'll assume a 3-byte response for the value after the response header.
length = 6
try:
packet = struct.pack(
">B2s1s2s1s",
length,
self.TV_ID,
b'g',
command_code.encode(),
self.CR
)
self.sock.sendall(packet)
response, value = self._receive_get_response()
if response == b'r':
return value.decode()
raise CommandError(f"Unexpected get response type: {response}")
except socket.error as e:
self.disconnect()
raise ConnectionError(f"Failed to send get command: {e}")
def _receive_response(self):
"""
Receive the standard 5-byte response for a Set command:
[ length(1) | ID(2) | command_type(1) | CR(1) ]
command_type should be '+' or '-'.
"""
try:
resp = self.sock.recv(5)
if len(resp) < 5:
raise CommandError("Incomplete response from Smart Board (set).")
# parse
# >B2s1s1s
# But effectively, we only need the command_type at index 3
length = resp[0]
command_type = resp[3:4]
return command_type
except socket.timeout:
raise ConnectionError("Timed out waiting for response (set).")
def _receive_get_response(self):
"""
Receive the standard 5-byte response header, then a 3-byte value if command_type='r'.
Header structure: [ length(1) | ID(2) | command_type(1) | ???(1)??? Actually length might be 5 bytes total ]
We interpret the protocol to have an additional 3 bytes for the value if command_type='r'.
Return (command_type, value_bytes).
"""
try:
header = self.sock.recv(5)
if len(header) < 5:
raise CommandError("Incomplete response from Smart Board (get header).")
length = header[0]
command_type = header[3:4] # e.g. b'r'
# Now read next 3 bytes if it's a valid get reply
if command_type == b'r':
value = self.sock.recv(3)
if len(value) < 3:
raise CommandError("Incomplete 3-byte value in get response.")
return command_type, value
else:
# No further data expected if it's not 'r'
return command_type, b''
except socket.timeout:
raise ConnectionError("Timed out waiting for response (get).")
# -- Full Set of Commands (from your Protocol) -----------------------------
# Power commands
def set_power(self, state: PowerState) -> bool:
return self.send_command('21', state.value)
def get_power(self) -> str:
return self.get_command('21')
# Video Source
def set_video_source(self, source: VideoSource) -> bool:
return self.send_command('22', source.value)
def get_video_source(self) -> str:
return self.get_command('22')
# Contrast
def set_contrast(self, value: int) -> bool:
if not (0 <= value <= 100):
raise ValueError("Contrast must be 0..100")
return self.send_command('23', f"{value:03}")
def get_contrast(self) -> str:
return self.get_command('23')
# Brightness
def set_brightness(self, value: int) -> bool:
if not (0 <= value <= 100):
raise ValueError("Brightness must be 0..100")
return self.send_command('24', f"{value:03}")
def get_brightness(self) -> str:
return self.get_command('24')
# Sharpness
def set_sharpness(self, value: int) -> bool:
if not (0 <= value <= 100):
raise ValueError("Sharpness must be 0..100")
return self.send_command('25', f"{value:03}")
def get_sharpness(self) -> str:
return self.get_command('25')
# Picture reset
def reset_picture(self) -> bool:
return self.send_command('26', '000')
# Aspect ratio
def set_aspect_ratio(self, ratio: AspectRatio) -> bool:
return self.send_command('31', ratio.value)
def get_aspect_ratio(self) -> str:
return self.get_command('31')
# Language
def set_language(self, lang: Language) -> bool:
return self.send_command('32', lang.value)
def get_language(self) -> str:
return self.get_command('32')
# Sound mode
def set_sound_mode(self, mode: SoundMode) -> bool:
return self.send_command('33', mode.value)
def get_sound_mode(self) -> str:
return self.get_command('33')
# Volume
def set_volume(self, value: int) -> bool:
if not (0 <= value <= 100):
raise ValueError("Volume must be 0..100")
return self.send_command('35', f"{value:03}")
def get_volume(self) -> str:
return self.get_command('35')
# Mute
def set_mute(self, mute: bool) -> bool:
return self.send_command('36', '001' if mute else '000')
def get_mute(self) -> str:
return self.get_command('36')
# Balance
def set_balance(self, value: int) -> bool:
if not (0 <= value <= 100):
raise ValueError("Balance must be 0..100")
return self.send_command('39', f"{value:03}")
def get_balance(self) -> str:
return self.get_command('39')
# Sound reset
def reset_sound(self) -> bool:
return self.send_command('3B', '000')
# Remote command
def send_remote_command(self, command: RemoteCommand) -> bool:
return self.send_command('40', command.value)
# IR control
def set_ir_control(self, control: IRControl) -> bool:
return self.send_command('42', control.value)
def get_ir_control(self) -> str:
return self.get_command('42')
# Button & IR control
def set_button_ir_control(self, control: ButtonIRControl) -> bool:
return self.send_command('43', control.value)
def get_button_ir_control(self) -> str:
return self.get_command('43')
# Button control
def set_button_control(self, control: ButtonControl) -> bool:
return self.send_command('45', control.value)
def get_button_control(self) -> str:
return self.get_command('45')
# Pixel shift
def set_pixel_shift(self, shift: PixelShift) -> bool:
return self.send_command('47', shift.value)
def get_pixel_shift(self) -> str:
return self.get_command('47')
# Screen reset
def reset_screen(self) -> bool:
return self.send_command('7F', '000')
# All reset
def reset_all(self) -> bool:
return self.send_command('7E', '000')
# Picture mode
def set_picture_mode(self, mode: PictureMode) -> bool:
return self.send_command('81', mode.value)
def get_picture_mode(self) -> str:
return self.get_command('81')
# Saturation
def set_saturation(self, value: int) -> bool:
if not (0 <= value <= 100):
raise ValueError("Saturation must be 0..100")
return self.send_command('82', f"{value:03}")
def get_saturation(self) -> str:
return self.get_command('82')
# Hue
def set_hue(self, value: int) -> bool:
if not (0 <= value <= 100):
raise ValueError("Hue must be 0..100")
return self.send_command('83', f"{value:03}")
def get_hue(self) -> str:
return self.get_command('83')
# Backlight
def set_backlight(self, value: int) -> bool:
if not (0 <= value <= 100):
raise ValueError("Backlight must be 0..100")
return self.send_command('84', f"{value:03}")
def get_backlight(self) -> str:
return self.get_command('84')
# DCR
def set_dcr(self, mode: DCR) -> bool:
return self.send_command('85', mode.value)
def get_dcr(self) -> str:
return self.get_command('85')
# Color temperature
def set_color_temp(self, temp: ColorTemp) -> bool:
return self.send_command('86', temp.value)
def get_color_temp(self) -> str:
return self.get_command('86')
# RTC
def set_rtc(self, year: int, month: int, day: int, hour: int, minute: int) -> bool:
if not (0 <= year <= 99):
raise ValueError("Year must be 0..99")
if not (1 <= month <= 12):
raise ValueError("Month must be 1..12")
if not (1 <= day <= 31):
raise ValueError("Day must be 1..31")
if not (0 <= hour <= 23):
raise ValueError("Hour must be 0..23")
if not (0 <= minute <= 59):
raise ValueError("Minute must be 0..59")
ok_y = self.send_command('98', f"{year:03}")
ok_m = self.send_command('99', f"{month:03}")
ok_d = self.send_command('9A', f"{day:03}")
ok_h = self.send_command('9B', f"{hour:03}")
ok_min = self.send_command('9C', f"{minute:03}")
return all([ok_y, ok_m, ok_d, ok_h, ok_min])
def get_rtc(self) -> Dict[str, str]:
return {
"year": self.get_command('98'),
"month": self.get_command('99'),
"day": self.get_command('9A'),
"hour": self.get_command('9B'),
"minute": self.get_command('9C'),
}
# Power save
def set_power_save(self, mode: PowerSaveMode) -> bool:
return self.send_command('A9', mode.value)
def get_power_save(self) -> str:
return self.get_command('A9')
# Switch on status
def set_switch_on_status(self, status: SwitchOnStatus) -> bool:
return self.send_command('AB', status.value)
def get_switch_on_status(self) -> str:
return self.get_command('AB')
# On/Off Timer
def set_on_off_timer(
self,
timer_number: int,
enable: bool,
on_enable: bool,
off_enable: bool,
days: int,
on_hour: int,
on_minute: int,
off_hour: int,
off_minute: int,
video_source: VideoSource
) -> bool:
# Per specification
if not (1 <= timer_number <= 7):
raise ValueError("Timer number must be 1..7")
if not (0 <= days <= 255):
raise ValueError("Days must be 0..255 (bitmask).")
if not (0 <= on_hour <= 23):
raise ValueError("On hour must be 0..23.")
if not (0 <= on_minute <= 59):
raise ValueError("On minute must be 0..59.")
if not (0 <= off_hour <= 23):
raise ValueError("Off hour must be 0..23.")
if not (0 <= off_minute <= 59):
raise ValueError("Off minute must be 0..59.")
byte1 = (timer_number & 0x07)
if enable:
byte1 |= (1 << 6)
if on_enable:
byte1 |= (1 << 5)
if off_enable:
byte1 |= (1 << 4)
byte2 = days & 0xFF
byte3 = on_hour & 0xFF
byte4 = on_minute & 0xFF
byte5 = off_hour & 0xFF
byte6 = off_minute & 0xFF
# video_source is a string hex in the standard commands, e.g. '000' => 0x00, '001' => 0x01
# But for E0, we parse the hex:
vs_hex = int(video_source.value, 16)
byte7 = vs_hex & 0xFF
byte8 = 0x00
byte9 = 0x00
# Convert 9 bytes to hex string
raw = bytes([byte1, byte2, byte3, byte4, byte5, byte6, byte7, byte8, byte9])
value_str = "".join(f"{b:02X}" for b in raw) # E.g. "050018000019000000"
return self.send_command('E0', value_str)
def get_on_off_timer(self, timer_number: int) -> Dict[str, Any]:
"""
Example implementation for reading a timer.
Protocol details are not fully specified, so we demonstrate a potential approach.
"""
if not (1 <= timer_number <= 7):
raise ValueError("Timer number must be 1..7")
self._ensure_connection()
try:
# Attempt a custom get packet for 'E0' with the selected timer in Byte1
length = 6
byte1 = timer_number & 0x07
packet = struct.pack(
">B2s1s2s1s",
length,
self.TV_ID,
b'g',
b'E0',
self.CR
)
self.sock.sendall(packet)
response_header = self.sock.recv(5)
if len(response_header) < 5:
raise CommandError("Incomplete timer response header.")
cmd_type = response_header[3:4]
if cmd_type != b'r':
raise CommandError(f"Unexpected response type for timer: {cmd_type}")
# Read next 9 bytes for the timer data
value_bytes = self.sock.recv(9)
if len(value_bytes) < 9:
raise CommandError("Incomplete timer data.")
byte1, byte2, byte3, byte4, byte5, byte6, byte7, byte8, byte9 = value_bytes
timer_info = {
"timer_number": byte1 & 0x07,
"enabled": bool((byte1 >> 6) & 0x01),
"on_enabled": bool((byte1 >> 5) & 0x01),
"off_enabled": bool((byte1 >> 4) & 0x01),
"days": byte2,
"on_hour": byte3,
"on_minute": byte4,
"off_hour": byte5,
"off_minute": byte6,
"video_source": f"{byte7:02X}",
"reserved": (byte8, byte9),
}
return timer_info
except socket.error as e:
self.disconnect()
raise ConnectionError(f"Failed to get On/Off Timer: {e}")

86
config_flow.py Normal file
View File

@ -0,0 +1,86 @@
"""Config flow for BenQ Smart Board integration."""
import socket
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from . import DOMAIN
from .benq_smartboard_lib import BenQSmartBoard, BenQSmartBoardError, ConnectionError
DEFAULT_PORT = 4660
class BenQSmartBoardConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for BenQ Smart Board."""
VERSION = 1
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Define the flow to handle options."""
return BenQSmartBoardOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input=None):
"""Handle the initial step for user setup."""
errors = {}
if user_input is not None:
host = user_input[CONF_HOST]
port = user_input[CONF_PORT]
# Attempt a short connection test
board = BenQSmartBoard(host, port, timeout=2.0)
try:
await self.hass.async_add_executor_job(board.connect)
board.disconnect()
except (ConnectionError, BenQSmartBoardError, socket.error):
errors["base"] = "cannot_connect"
else:
# If no connection error, proceed
await self.async_set_unique_id(f"benq_smartboard_{host}_{port}")
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"BenQ Smart Board ({host})",
data={
CONF_HOST: host,
CONF_PORT: port,
},
)
data_schema = vol.Schema({
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
})
return self.async_show_form(
step_id="user",
data_schema=data_schema,
errors=errors
)
class BenQSmartBoardOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle an options flow for adjusting settings after setup."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Store the current config entry for reference."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
"""Manage the options for the BenQ Smart Board."""
if user_input is not None:
# In a real scenario, you might re-validate or re-connect
return self.async_create_entry(title="", data=user_input)
data_schema = vol.Schema({
vol.Optional(
CONF_PORT,
default=self.config_entry.data.get(CONF_PORT, DEFAULT_PORT),
): cv.port,
})
return self.async_show_form(step_id="init", data_schema=data_schema)

11
manifest.json Normal file
View File

@ -0,0 +1,11 @@
{
"domain": "benq_smartboard",
"name": "BenQ Smart Board",
"version": "1.0.0",
"documentation": "https://siwatinc.com/benq-hass",
"requirements": [],
"codeowners": ["@siwatsirichai"],
"iot_class": "local_polling",
"config_flow": true
}

218
media_player.py Normal file
View File

@ -0,0 +1,218 @@
"""Media Player platform for BenQ Smart Board integration."""
import logging
from typing import Optional
from homeassistant.components.media_player import (
MediaPlayerEntity,
MediaPlayerEntityFeature,
)
from homeassistant.components.media_player.const import (
MEDIA_TYPE_TVSHOW,
SUPPORT_VOLUME_MUTE,
SUPPORT_VOLUME_SET,
SUPPORT_TURN_ON,
SUPPORT_TURN_OFF,
)
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import DOMAIN
from .benq_smartboard_lib import (
BenQSmartBoard,
BenQSmartBoardError,
ConnectionError,
PowerState,
)
_LOGGER = logging.getLogger(__name__)
SUPPORT_BENQ_SMARTBOARD = (
SUPPORT_VOLUME_SET
| SUPPORT_VOLUME_MUTE
| SUPPORT_TURN_OFF
| SUPPORT_TURN_ON
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
):
"""Set up the BenQ Smart Board media player from a config entry."""
data = hass.data[DOMAIN][entry.entry_id]
host = data["host"]
port = data["port"]
name = f"BenQ Smart Board ({host})"
entity = BenQSmartBoardMediaPlayer(
name=name,
host=host,
port=port,
unique_id=f"benq_smartboard_{host}_{port}",
)
async_add_entities([entity], update_before_add=True)
class BenQSmartBoardMediaPlayer(MediaPlayerEntity):
"""Representation of the BenQ Smart Board as a Media Player."""
_attr_should_poll = True
_attr_supported_features = SUPPORT_BENQ_SMARTBOARD
_attr_media_content_type = MEDIA_TYPE_TVSHOW
def __init__(self, name: str, host: str, port: int, unique_id: str):
"""Initialize."""
self._attr_name = name
self._attr_unique_id = unique_id
self._host = host
self._port = port
# Internal states
self._state = STATE_UNAVAILABLE
self._volume_level = 0.0
self._is_muted = False
self._board: Optional[BenQSmartBoard] = None
# Setup device info so it appears as a device in HA
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, unique_id)},
name=name,
manufacturer="BenQ",
model="Smart Board",
configuration_url=f"http://{host}", # or other relevant link
)
@property
def state(self):
"""Return the current power state: ON, OFF, or if unreachable -> UNAVAILABLE."""
return self._state
@property
def volume_level(self):
"""Volume level (0..1)."""
return self._volume_level
@property
def is_volume_muted(self):
"""Return True if muted."""
return self._is_muted
def _ensure_board(self):
"""Ensure we have a connected BenQSmartBoard instance."""
if self._board is None:
self._board = BenQSmartBoard(self._host, self._port, timeout=3.0)
return self._board
def _refresh_from_device(self):
"""
Attempt to query the current state from the device (power, volume, mute).
Called in update().
"""
board = self._ensure_board()
try:
# Connect if not connected
board.connect()
# Get power
pwr = board.get_power() # string e.g. "001" -> ON
if pwr == PowerState.ON.value:
self._state = STATE_ON
elif pwr == PowerState.STANDBY.value:
# Some prefer modeling STANDBY as OFF or custom
self._state = STATE_OFF
elif pwr == PowerState.OFF.value:
self._state = STATE_OFF
else:
# If unknown power code, consider it OFF or UNAVAILABLE
self._state = STATE_OFF
# Volume: e.g. "030"
vol_str = board.get_volume()
volume_int = int(vol_str)
self._volume_level = volume_int / 100.0
# Mute: "000" => Unmuted, "001" => Muted
mute_str = board.get_mute()
self._is_muted = (mute_str == "001")
except BenQSmartBoardError as err:
_LOGGER.warning("Error talking to BenQ Smart Board: %s", err)
self._state = STATE_UNAVAILABLE
except ConnectionError as err:
_LOGGER.warning("Connection to BenQ Smart Board lost: %s", err)
self._state = STATE_UNAVAILABLE
async def async_update(self):
"""Update the entity state by polling the device."""
await self.hass.async_add_executor_job(self._refresh_from_device)
async def async_turn_on(self):
"""Turn on the Smart Board (if off or in standby)."""
board = self._ensure_board()
try:
board.connect()
success = board.set_power(PowerState.ON)
if success:
self._state = STATE_ON
else:
_LOGGER.warning("Smart Board refused 'power on' command.")
except BenQSmartBoardError as e:
_LOGGER.error("Failed to turn on: %s", e)
self._state = STATE_UNAVAILABLE
self.async_write_ha_state()
async def async_turn_off(self):
"""Turn off the Smart Board."""
board = self._ensure_board()
try:
board.connect()
success = board.set_power(PowerState.OFF)
if success:
self._state = STATE_OFF
else:
_LOGGER.warning("Smart Board refused 'power off' command.")
except BenQSmartBoardError as e:
_LOGGER.error("Failed to turn off: %s", e)
self._state = STATE_UNAVAILABLE
self.async_write_ha_state()
async def async_set_volume_level(self, volume: float):
"""Set volume level (0..1)."""
board = self._ensure_board()
vol_int = int(volume * 100)
try:
board.connect()
success = board.set_volume(vol_int)
if success:
self._volume_level = volume
else:
_LOGGER.warning("Smart Board refused 'set volume' command.")
except BenQSmartBoardError as e:
_LOGGER.error("Failed to set volume: %s", e)
self._state = STATE_UNAVAILABLE
self.async_write_ha_state()
async def async_mute_volume(self, mute: bool):
"""Mute or unmute the volume."""
board = self._ensure_board()
try:
board.connect()
success = board.set_mute(mute)
if success:
self._is_muted = mute
else:
_LOGGER.warning("Smart Board refused 'mute' command.")
except BenQSmartBoardError as e:
_LOGGER.error("Failed to mute: %s", e)
self._state = STATE_UNAVAILABLE
self.async_write_ha_state()
async def async_will_remove_from_hass(self):
"""Disconnect from device on removal."""
if self._board:
await self.hass.async_add_executor_job(self._board.disconnect)