diff --git a/.gitignore b/.gitignore index 2d635fd..b549013 100644 --- a/.gitignore +++ b/.gitignore @@ -168,3 +168,4 @@ custom_components/opendisplay/*.jpg custom_components/opendisplay/lastapinteraction.txt tests/drawcustom/test_images/rename_me.png +.worktrees/ diff --git a/custom_components/opendisplay/__init__.py b/custom_components/opendisplay/__init__.py index db94cbd..9749ba5 100644 --- a/custom_components/opendisplay/__init__.py +++ b/custom_components/opendisplay/__init__.py @@ -11,7 +11,7 @@ from homeassistant.helpers import entity_registry as er, device_registry as dr, storage from homeassistant.const import __version__ as HA_VERSION from homeassistant.helpers.typing import ConfigType -from .ble import BLEDeviceMetadata +from .metadata import BLEDeviceMetadata from .const import DOMAIN from .coordinator import Hub from .runtime_data import OpenDisplayConfigEntry, OpenDisplayBLERuntimeData @@ -114,7 +114,7 @@ async def async_remove_invalid_ble_entities( mac_address = entry.data.get("mac_address", "") # Check power mode - remove battery sensors if not battery/solar powered - from .ble import BLEDeviceMetadata + from .metadata import BLEDeviceMetadata metadata = BLEDeviceMetadata(device_metadata) if metadata.power_mode not in (1, 3): # Not battery (1) or solar (3) for entity in er.async_entries_for_config_entry(entity_registry, entry.entry_id): @@ -308,7 +308,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry) mac_address = entry.data.get("mac_address") name = entry.data.get("name") device_metadata = entry.data.get("device_metadata", {}) - protocol_type = entry.data.get("protocol_type", "atc") # Default to ATC for backward compatibility + protocol_type = "open_display" # Get protocol handler for this device protocol = get_protocol_by_name(protocol_type) @@ -378,24 +378,7 @@ def _ble_device_found( _LOGGER.debug("Failed to parse advertising data for %s: %s", mac_address, err, exc_info=True) return - # Dynamically update device attributes (skip OpenDisplay fw to avoid incorrect value) - if advertising_data.fw_version and protocol_type != "open_display": - device_registry = dr.async_get(hass) - device_entry = device_registry.async_get_device( - identifiers={(DOMAIN, f"ble_{mac_address}")} - ) - new_fw_string = str(advertising_data.fw_version) - if device_entry and device_entry.sw_version != new_fw_string: - _LOGGER.debug( - "Device %s firmware updated from %s to %s", - mac_address, - device_entry.sw_version, - new_fw_string, - ) - device_registry.async_update_device( - device_entry.id, - sw_version=new_fw_string - ) + # Build sensor data sensor_data = { @@ -410,11 +393,6 @@ def _ble_device_found( for sensor in ble_data.sensors.values(): sensor.update_from_advertising_data(sensor_data) - # Remove deprecated clock mode button entities - removed_clock_buttons = await async_remove_clock_mode_buttons(hass, entry) - if removed_clock_buttons: - _LOGGER.info("Removed deprecated clock mode buttons: %s", removed_clock_buttons) - # Remove invalid entities based on the current device config removed_invalid = await async_remove_invalid_ble_entities(hass, entry, device_metadata) if removed_invalid: diff --git a/custom_components/opendisplay/ble/image_upload.py b/custom_components/opendisplay/ble/image_upload.py index b239ae4..1007bb3 100644 --- a/custom_components/opendisplay/ble/image_upload.py +++ b/custom_components/opendisplay/ble/image_upload.py @@ -11,7 +11,7 @@ from .exceptions import BLEError from .image_processing import process_image_for_device -from .metadata import BLEDeviceMetadata +from ..metadata import BLEDeviceMetadata _LOGGER = logging.getLogger(__name__) diff --git a/custom_components/opendisplay/config_flow.py b/custom_components/opendisplay/config_flow.py index 9c73192..d96967a 100644 --- a/custom_components/opendisplay/config_flow.py +++ b/custom_components/opendisplay/config_flow.py @@ -43,8 +43,6 @@ def _format_ble_protocol_label(protocol_type: str) -> str: """Return a user-facing label for a BLE protocol.""" if protocol_type == "open_display": return "OpenDisplay (OD)" - if protocol_type == "atc": - return "OEPL / ATC" return protocol_type @@ -75,15 +73,6 @@ def _bluetooth_description_placeholders( """Build placeholders for the Bluetooth confirmation dialog.""" device = self._discovered_device advertised_details = "" - if device["protocol_type"] == "atc": - battery = f"{device['battery_mv']/1000:.2f}V" if device["battery_mv"] > 0 else "Unknown" - fw_version = str(device["fw_version"]) if device["fw_version"] > 0 else "Unknown" - config_version = str(device["version"]) if device["version"] > 0 else "Unknown" - advertised_details = ( - f"\n- Battery: {battery}" - f"\n- Firmware: {fw_version}" - f"\n- Config Version: {config_version}" - ) placeholders = { "name": device["name"], @@ -223,9 +212,9 @@ async def async_step_bluetooth( manufacturer_id = None manufacturer_data = b'' - # Check for known manufacturer IDs (ATC: 4919, OpenDisplay: 9286) + # Check for known manufacturer IDs (OpenDisplay: 9286) for mfg_id, mfg_data in discovery_info.manufacturer_data.items(): - if mfg_id in (4919, 9286): + if mfg_id == 9286: manufacturer_id = mfg_id manufacturer_data = mfg_data break @@ -285,9 +274,7 @@ async def async_step_bluetooth_confirm( try: # Get protocol handler for this device - protocol = get_protocol_by_manufacturer_id( - 9286 if self._discovered_device["protocol_type"] == "open_display" else 4919 - ) + protocol = get_protocol_by_manufacturer_id(9286) # Interrogate device using protocol-specific method fw_info: dict[str, Any] | None = None @@ -322,93 +309,39 @@ async def async_step_bluetooth_confirm( # Generate model name based on protocol type hw_type = self._discovered_device["hw_type"] - if self._discovered_device["protocol_type"] == "open_display": - # OpenDisplay devices: Store complete config, generate model name from DisplayConfig - from .ble.tlv_parser import config_to_dict, generate_model_name - - if hasattr(protocol, '_last_config') and protocol._last_config: - # Store complete OpenDisplay config for future use - device_metadata = { - "open_display_config": config_to_dict(protocol._last_config), - } - if fw_info: - device_metadata["fw_version"] = fw_info.get("version") - device_metadata["fw_version_raw"] = fw_info.get("raw") - if fw_info.get("sha"): - device_metadata["fw_sha"] = fw_info["sha"] - - # Generate model name from display config - if protocol._last_config.displays: - model_name = generate_model_name(protocol._last_config.displays[0]) - device_metadata["model_name"] = model_name - _LOGGER.debug("Generated model name from config: %s", model_name) - else: - _LOGGER.warning("OpenDisplay config has no display config") + # OpenDisplay devices: Store complete config, generate model name from DisplayConfig + from .ble.tlv_parser import config_to_dict, generate_model_name + + if hasattr(protocol, '_last_config') and protocol._last_config: + # Store complete OpenDisplay config for future use + device_metadata = { + "open_display_config": config_to_dict(protocol._last_config), + } + if fw_info: + device_metadata["fw_version"] = fw_info.get("version") + device_metadata["fw_version_raw"] = fw_info.get("raw") + if fw_info.get("sha"): + device_metadata["fw_sha"] = fw_info["sha"] + + # Generate model name from display config + if protocol._last_config.displays: + model_name = generate_model_name(protocol._last_config.displays[0]) + device_metadata["model_name"] = model_name + _LOGGER.debug("Generated model name from config: %s", model_name) else: - # Fallback if config unavailable (shouldn't happen for OpenDisplay) - model_name = get_hw_string(hw_type) if hw_type else "Unknown" - _LOGGER.warning("OpenDisplay config unavailable, using tagtypes fallback: %s", model_name) - # Store individual fields as fallback - device_metadata = { - "hw_type": hw_type, - "fw_version": self._discovered_device["fw_version"], - "width": capabilities.width, - "height": capabilities.height, - "rotatebuffer": capabilities.rotatebuffer, - "color_scheme": capabilities.color_scheme, - "model_name": model_name, - } + _LOGGER.warning("OpenDisplay config has no display config") else: - # ATC devices: Use tagtypes.json lookup and store individual fields - # Try to get tag types manager, but don't fail if unavailable - tag_types_manager = None - try: - tag_types_manager = await get_tag_types_manager(self.hass) - _LOGGER.debug("Tag types manager loaded successfully") - except Exception as tag_err: - _LOGGER.warning( - "Could not load tag types during config flow, will use fallback values: %s", - tag_err - ) - + # Fallback if config unavailable (shouldn't happen for OpenDisplay) model_name = get_hw_string(hw_type) if hw_type else "Unknown" - _LOGGER.debug("Resolved hw_type %s to model: %s", hw_type, model_name) - - # Refine color_scheme using TagTypes db if available - if tag_types_manager and tag_types_manager.is_in_hw_map(hw_type): - tag_type = await tag_types_manager.get_tag_info(hw_type) - color_table = tag_type.color_table - - if 'yellow' in color_table and 'red' in color_table: - color_scheme = 3 # BWRY - elif 'yellow' in color_table: - color_scheme = 2 # BWY - elif 'red' in color_table: - color_scheme = 1 # BWR - else: - color_scheme = 0 # BW - else: - # Fallback to protocol detection - color_scheme = capabilities.color_scheme - if not tag_types_manager: - _LOGGER.info( - "Tag types not available, using protocol-detected color_scheme: %d", - color_scheme - ) - else: - _LOGGER.warning( - "hw_type %s not in TagTypes, using protocol color_scheme: %d", - hw_type, color_scheme - ) - - # Build device metadata from capabilities + _LOGGER.warning("OpenDisplay config unavailable, using tagtypes fallback: %s", model_name) + # Store individual fields as fallback device_metadata = { "hw_type": hw_type, "fw_version": self._discovered_device["fw_version"], "width": capabilities.width, "height": capabilities.height, "rotatebuffer": capabilities.rotatebuffer, - "color_scheme": color_scheme, + "color_scheme": capabilities.color_scheme, "model_name": model_name, } diff --git a/custom_components/opendisplay/entity.py b/custom_components/opendisplay/entity.py index 828b1ea..0ef69ba 100644 --- a/custom_components/opendisplay/entity.py +++ b/custom_components/opendisplay/entity.py @@ -6,7 +6,7 @@ from homeassistant.core import callback from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity import DeviceInfo, Entity -from .ble import BLEDeviceMetadata +from .metadata import BLEDeviceMetadata from .const import DOMAIN, OPENDISPLAY_CONFIG_URL, ATC_CONFIG_URL from .tag_types import get_hw_string, get_hw_dimensions diff --git a/custom_components/opendisplay/imagegen/core.py b/custom_components/opendisplay/imagegen/core.py index 42b90af..e1b597d 100644 --- a/custom_components/opendisplay/imagegen/core.py +++ b/custom_components/opendisplay/imagegen/core.py @@ -232,7 +232,7 @@ async def get_ble_tag_info(self, hass: HomeAssistant, entity_id: str) -> tuple[i translation_placeholders={"entity_id": entity_id} ) # Wrap metadata for clean access - from ..ble import BLEDeviceMetadata + from ..metadata import BLEDeviceMetadata metadata = BLEDeviceMetadata(device_metadata) # Extract device capabilities diff --git a/custom_components/opendisplay/manifest.json b/custom_components/opendisplay/manifest.json index 7558304..7ad2973 100644 --- a/custom_components/opendisplay/manifest.json +++ b/custom_components/opendisplay/manifest.json @@ -2,9 +2,6 @@ "domain": "opendisplay", "name": "OpenDisplay", "bluetooth": [ - { - "manufacturer_id": 4919 - }, { "manufacturer_id": 9286 } @@ -34,6 +31,7 @@ } }, "requirements": [ + "py-opendisplay==7.3.0", "qrcode[pil]==7.4.2", "requests_toolbelt==1.0.0", "websocket-client==1.7.0", diff --git a/custom_components/opendisplay/metadata.py b/custom_components/opendisplay/metadata.py new file mode 100644 index 0000000..637362f --- /dev/null +++ b/custom_components/opendisplay/metadata.py @@ -0,0 +1,445 @@ +"""New BLE Device Metadata Abstraction using py-opendisplay. + +Provides a clean interface for accessing device metadata that transparently +handles differences between ATC (flat structure) and OpenDisplay (nested config) formats, +utilizing py-opendisplay library configuration models. +""" +from __future__ import annotations + +from typing import Any + +from opendisplay import ColorScheme as ODColorScheme +from opendisplay.models.config import ( + GlobalConfig, + SystemConfig, + ManufacturerData, + PowerOption, + DisplayConfig, + LedConfig, + SensorData, + DataBus, + BinaryInputs, + WifiConfig, + SecurityConfig, + TouchController, + PassiveBuzzer, +) + +def _dict_to_global_config(data: dict[str, Any]) -> GlobalConfig: + """Reconstruct a GlobalConfig dataclass from its nested dict representation.""" + def _to_bytes(val: Any) -> bytes: + if isinstance(val, str): + return bytes.fromhex(val) + return val + + # Extract single instances + sys_data = data.get("system") or {} + system = SystemConfig( + ic_type=sys_data.get("ic_type", 0), + communication_modes=sys_data.get("communication_modes", 0), + device_flags=sys_data.get("device_flags", 0), + pwr_pin=sys_data.get("pwr_pin", 0xFF), + reserved=_to_bytes(sys_data.get("reserved", b"\x00" * 15)), + pwr_pin_2=sys_data.get("pwr_pin_2", 0xFF), + pwr_pin_3=sys_data.get("pwr_pin_3", 0xFF), + ) + + mfr_data = data.get("manufacturer") or {} + manufacturer = ManufacturerData( + manufacturer_id=mfr_data.get("manufacturer_id", 0x2446), + board_type=mfr_data.get("board_type", 0), + board_revision=mfr_data.get("board_revision", 0), + reserved=_to_bytes(mfr_data.get("reserved", b"\x00" * 18)), + ) + + pwr_data = data.get("power") or {} + power = PowerOption( + power_mode=pwr_data.get("power_mode", 1), + battery_capacity_mah=_to_bytes(pwr_data.get("battery_capacity_mah", b"\x00" * 3)), + sleep_timeout_ms=pwr_data.get("sleep_timeout_ms", 0), + tx_power=pwr_data.get("tx_power", 0), + sleep_flags=pwr_data.get("sleep_flags", 0), + battery_sense_pin=pwr_data.get("battery_sense_pin", 0xFF), + battery_sense_enable_pin=pwr_data.get("battery_sense_enable_pin", 0xFF), + battery_sense_flags=pwr_data.get("battery_sense_flags", 0), + capacity_estimator=pwr_data.get("capacity_estimator", 0), + voltage_scaling_factor=pwr_data.get("voltage_scaling_factor", 0), + deep_sleep_current_ua=pwr_data.get("deep_sleep_current_ua", 0), + deep_sleep_time_seconds=pwr_data.get("deep_sleep_time_seconds", 0), + reserved=_to_bytes(pwr_data.get("reserved", b"\x00" * 10)), + ) + + # Extract displays + displays = [] + for d in data.get("displays", []): + displays.append( + DisplayConfig( + instance_number=d.get("instance_number", 0), + display_technology=d.get("display_technology", 0), + panel_ic_type=d.get("panel_ic_type", 0), + pixel_width=d.get("pixel_width", 0), + pixel_height=d.get("pixel_height", 0), + active_width_mm=d.get("active_width_mm", 0), + active_height_mm=d.get("active_height_mm", 0), + tag_type=d.get("open_display_tagtype") or d.get("tag_type") or 0, + rotation=d.get("rotation", 0), + reset_pin=d.get("reset_pin", 0xFF), + busy_pin=d.get("busy_pin", 0xFF), + dc_pin=d.get("dc_pin", 0xFF), + cs_pin=d.get("cs_pin", 0xFF), + data_pin=d.get("data_pin", 0xFF), + partial_update_support=d.get("partial_update_support", 0), + color_scheme=d.get("color_scheme", 0), + transmission_modes=d.get("transmission_modes", 0), + clk_pin=d.get("clk_pin", 0xFF), + reserved_pins=_to_bytes(d.get("reserved_pins", b"\x00" * 7)), + full_update_mC=d.get("full_update_mC", 0), + reserved=_to_bytes(d.get("reserved", b"\x00" * 13)), + ) + ) + + # Extract leds + leds = [] + for l in data.get("leds", []): + leds.append( + LedConfig( + instance_number=l.get("instance_number", 0), + led_type=l.get("led_type", 0), + led_1_r=l.get("led_1_r", 0), + led_2_g=l.get("led_2_g", 0), + led_3_b=l.get("led_3_b", 0), + led_4=l.get("led_4", 0), + led_flags=l.get("led_flags", 0), + reserved=_to_bytes(l.get("reserved", b"\x00" * 15)), + ) + ) + + # Extract sensors + sensors = [] + for s in data.get("sensors", []): + sensors.append( + SensorData( + instance_number=s.get("instance_number", 0), + sensor_type=s.get("sensor_type", 0), + bus_id=s.get("bus_id", 0), + i2c_addr_7bit=s.get("i2c_addr_7bit", 0), + msd_data_start_byte=s.get("msd_data_start_byte", 0), + reserved=_to_bytes(s.get("reserved", b"\x00" * 24)), + ) + ) + + # Extract data buses + data_buses = [] + buses_list = data.get("data_buses") or data.get("buses") or [] + for b in buses_list: + data_buses.append( + DataBus( + instance_number=b.get("instance_number", 0), + bus_type=b.get("bus_type", 0), + pin_1=b.get("pin_1", 0xFF), + pin_2=b.get("pin_2", 0xFF), + pin_3=b.get("pin_3", 0xFF), + pin_4=b.get("pin_4", 0xFF), + pin_5=b.get("pin_5", 0xFF), + pin_6=b.get("pin_6", 0xFF), + pin_7=b.get("pin_7", 0xFF), + bus_speed_hz=b.get("bus_speed_hz", 0), + bus_flags=b.get("bus_flags", 0), + pullups=b.get("pullups", 0), + pulldowns=b.get("pulldowns", 0), + reserved=_to_bytes(b.get("reserved", b"\x00" * 14)), + ) + ) + + # Extract binary inputs + binary_inputs = [] + inputs_list = data.get("binary_inputs") or data.get("inputs") or [] + for bi in inputs_list: + binary_inputs.append( + BinaryInputs( + instance_number=bi.get("instance_number", 0), + input_type=bi.get("input_type", 0), + display_as=bi.get("display_as", 0), + reserved_pins=_to_bytes(bi.get("reserved_pins", b"\x00" * 8)), + input_flags=bi.get("input_flags", 0), + invert=bi.get("invert", 0), + pullups=bi.get("pullups", 0), + pulldowns=bi.get("pulldowns", 0), + button_data_byte_index=bi.get("button_data_byte_index", 0), + reserved=_to_bytes(bi.get("reserved", b"\x00" * 14)), + ) + ) + + # Extract optional configs + wifi_config = None + if "wifi_config" in data and data["wifi_config"] is not None: + w = data["wifi_config"] + wifi_config = WifiConfig( + ssid=_to_bytes(w.get("ssid", b"\x00" * 32)), + password=_to_bytes(w.get("password", b"\x00" * 32)), + encryption_type=w.get("encryption_type", 0), + server_url=_to_bytes(w.get("server_url", b"\x00" * 64)), + server_port=w.get("server_port", 2446), + reserved=_to_bytes(w.get("reserved", b"\x00" * 29)), + ) + + security_config = None + if "security_config" in data and data["security_config"] is not None: + sec = data["security_config"] + security_config = SecurityConfig( + encryption_enabled=sec.get("encryption_enabled", 0), + encryption_key=_to_bytes(sec.get("encryption_key", b"\x00" * 16)), + session_timeout_seconds=sec.get("session_timeout_seconds", 0), + flags=sec.get("flags", 0), + reset_pin=sec.get("reset_pin", 0xFF), + reserved=_to_bytes(sec.get("reserved", b"\x00" * 43)), + ) + + touch_controllers = [] + for tc in data.get("touch_controllers", []): + touch_controllers.append( + TouchController( + instance_number=tc.get("instance_number", 0), + touch_ic_type=tc.get("touch_ic_type", 0), + bus_id=tc.get("bus_id", 0xFF), + i2c_addr_7bit=tc.get("i2c_addr_7bit", 0), + int_pin=tc.get("int_pin", 0xFF), + rst_pin=tc.get("rst_pin", 0xFF), + display_instance=tc.get("display_instance", 0), + flags=tc.get("flags", 0), + poll_interval_ms=tc.get("poll_interval_ms", 0), + touch_data_start_byte=tc.get("touch_data_start_byte", 0), + reserved=_to_bytes(tc.get("reserved", b"\x00" * 21)), + ) + ) + + buzzers = [] + for bz in data.get("buzzers", []): + buzzers.append( + PassiveBuzzer( + instance_number=bz.get("instance_number", 0), + drive_pin=bz.get("drive_pin", 0), + enable_pin=bz.get("enable_pin", 0xFF), + flags=bz.get("flags", 0), + duty_percent=bz.get("duty_percent", 0), + reserved=_to_bytes(bz.get("reserved", b"\x00" * 27)), + ) + ) + + return GlobalConfig( + system=system, + manufacturer=manufacturer, + power=power, + displays=displays, + leds=leds, + sensors=sensors, + data_buses=data_buses, + binary_inputs=binary_inputs, + wifi_config=wifi_config, + security_config=security_config, + touch_controllers=touch_controllers, + buzzers=buzzers, + version=data.get("version", 0), + minor_version=data.get("minor_version", 0), + loaded=data.get("loaded", False), + ) + + +class BLEDeviceMetadata: + """Abstraction for BLE device metadata. + + Wraps raw metadata dictionary and provides clean property-based access + to device capabilities, using py-opendisplay configuration models. + + Args: + raw_metadata: Dictionary containing device metadata + """ + + def __init__(self, raw_metadata: dict[str, Any]) -> None: + """Initialize BLE device metadata wrapper. + + Args: + raw_metadata: Device metadata dictionary from config entry + """ + if "open_display_config" not in raw_metadata and "oepl_config" in raw_metadata: + self._metadata = {**raw_metadata, "open_display_config": raw_metadata["oepl_config"]} + else: + self._metadata = raw_metadata + self._is_open_display = "open_display_config" in self._metadata + + self._config: GlobalConfig | None = None + if self._is_open_display: + try: + self._config = _dict_to_global_config(self._metadata["open_display_config"]) + except Exception: + pass + + @property + def width(self) -> int: + """Get display width in pixels. + + Returns: + Display width, or 0 if not available + """ + if self._is_open_display and self._config and self._config.displays: + return self._config.displays[0].pixel_width + return self._metadata.get("width", 0) + + @property + def height(self) -> int: + """Get display height in pixels. + + Returns: + Display height, or 0 if not available + """ + if self._is_open_display and self._config and self._config.displays: + return self._config.displays[0].pixel_height + return self._metadata.get("height", 0) + + @property + def model_name(self) -> str: + """Get device model name. + + Returns: + Model name string, or "Unknown" if not available + """ + return self._metadata.get("model_name", "Unknown") + + @property + def fw_version(self) -> int | str: + """Get firmware version. + + Returns: + Firmware version number or string, or 0/"" if not available + """ + if self._is_open_display: + # Prefer explicit string/parsed version saved from interrogation + if "fw_version" in self._metadata: + return self._metadata.get("fw_version", "") + major = self._metadata.get("fw_version_major") + minor = self._metadata.get("fw_version_minor") + if major is not None and minor is not None: + return f"{major}.{minor}" + return self._metadata.get("fw_version", 0) + + def formatted_fw_version(self) -> str | None: + """Return firmware version formatted for display.""" + fw = self.fw_version + if fw in (None, ""): + return None + if isinstance(fw, int): + return f"0x{fw:04x}" + return str(fw) + + @property + def rotatebuffer(self) -> int: + """Get rotation setting. + + For OpenDisplay devices, returns the rotation value from display config. + For ATC devices, returns the rotatebuffer flag. + + Returns: + Rotation value (0, 1, 2, or 3) or rotatebuffer flag (0 or 1) + """ + if self._is_open_display and self._config and self._config.displays: + return self._config.displays[0].rotation + return self._metadata.get("rotatebuffer", 0) + + @property + def hw_type(self) -> int: + """Get hardware type identifier. + + Returns: + Hardware type code, or 0 if not available + """ + if self._is_open_display and self._config and self._config.displays: + return self._config.displays[0].tag_type + return self._metadata.get("hw_type", 0) + + @property + def power_mode(self) -> int: + """Get power mode setting. + + Returns: + Power mode: 1=battery, 2=USB, 3=solar + ATC devices always return 1 (battery) + """ + if self._is_open_display and self._config and self._config.power: + return self._config.power.power_mode + return 1 # ATC devices always have batteries + + @property + def is_open_display(self) -> bool: + """Check if this is an OpenDisplay device. + + Returns: + True if OpenDisplay device, False if ATC device + """ + return self._is_open_display + + @property + def color_scheme(self) -> ODColorScheme: + """Get ColorScheme enum for this device.""" + if self._is_open_display and self._config and self._config.displays: + raw_scheme = self._config.displays[0].color_scheme + else: + raw_scheme = self._metadata.get("color_scheme", 0) + return ODColorScheme.from_value(raw_scheme) + + @property + def accent_color(self) -> str: + """Get accent color name. + + Returns: + Accent color name from color scheme palette + """ + return self.color_scheme.accent_color + + @property + def is_multi_color(self) -> bool: + """Check if device supports multiple colors. + + Returns: + True if color scheme has more than 2 colors, False otherwise + """ + return len(self.color_scheme.palette.colors) > 2 + + @property + def transmission_modes(self) -> int: + """Get supported transmission modes (bitfield). + + Bit flags: + - Bit 0 (0x01): raw transfer (block-based uncompressed) + - Bit 1 (0x02): zip compressed transfer (block-based compressed) + - Bit 3 (0x08): direct_write mode + + Returns: + Transmission modes bitfield, or 0 if not available + ATC devices return 0 (assume block-based only for backward compatibility) + """ + if self._is_open_display and self._config and self._config.displays: + return self._config.displays[0].transmission_modes + return 0 # ATC devices don't support direct_write + + @property + def supports_zip_compression(self) -> bool: + """Return true if the device advertises zip-compressed transfer support.""" + return (self.transmission_modes & 0x02) != 0 + + def get_best_upload_method(self) -> str: + """Determine the best upload method based on device capabilities. + + Priority order: + 1. direct_write: If direct_write (0x08) is supported + 2. block: Fallback to block-based upload (always supported) + + Returns: + Upload method string: "direct_write" or "block" + """ + modes = self.transmission_modes + has_direct_write = (modes & 0x08) != 0 + + if has_direct_write: + return "direct_write" + else: + return "block" diff --git a/custom_components/opendisplay/sensor.py b/custom_components/opendisplay/sensor.py index af87d4e..73c5ec2 100644 --- a/custom_components/opendisplay/sensor.py +++ b/custom_components/opendisplay/sensor.py @@ -625,22 +625,18 @@ async def async_setup_entry(hass: HomeAssistant, entry: OpenDisplayConfigEntry, mac_address = entry_data.mac_address name = entry_data.name device_metadata = entry_data.device_metadata - protocol_type = entry_data.protocol_type # Default to ATC for backward compatibility + protocol_type = "open_display" # Create sensors for each description - from .ble import BLEDeviceMetadata + from .metadata import BLEDeviceMetadata metadata = BLEDeviceMetadata(device_metadata) sensors = [] for description in BLE_SENSOR_TYPES: # Handle battery sensors based on device protocol if description.key in ("battery_percentage", "battery_voltage"): - if protocol_type == "atc": - # ATC devices always have batteries - pass # Continue to create sensor - elif protocol_type == "open_display": - # OpenDisplay devices: only create battery sensors for battery/solar power - if metadata.power_mode not in (1, 3): # Not battery (1) or solar (3) - continue # Skip battery sensors + # OpenDisplay devices: only create battery sensors for battery/solar power + if metadata.power_mode not in (1, 3): # Not battery (1) or solar (3) + continue # Skip battery sensors sensor = OpenDisplayBLESensor( hass=hass, diff --git a/custom_components/opendisplay/services.py b/custom_components/opendisplay/services.py index db5b444..1ff195e 100644 --- a/custom_components/opendisplay/services.py +++ b/custom_components/opendisplay/services.py @@ -10,7 +10,8 @@ from homeassistant.helpers import device_registry as dr from homeassistant.helpers.dispatcher import async_dispatcher_send from .coordinator import Hub -from .ble import BLEConnectionError, BLETimeoutError, BLEProtocolError, BLEDeviceMetadata +from .ble import BLEConnectionError, BLETimeoutError, BLEProtocolError +from .metadata import BLEDeviceMetadata from .const import DOMAIN, SIGNAL_TAG_IMAGE_UPDATE from .imagegen import ImageGen from .tag_types import get_tag_types_manager diff --git a/custom_components/opendisplay/update.py b/custom_components/opendisplay/update.py index 104493f..9b74821 100644 --- a/custom_components/opendisplay/update.py +++ b/custom_components/opendisplay/update.py @@ -15,7 +15,7 @@ from homeassistant.helpers.entity import EntityCategory from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .ble import BLEDeviceMetadata +from .metadata import BLEDeviceMetadata from .const import DOMAIN from .entity import OpenDisplayBLEEntity from .runtime_data import OpenDisplayBLERuntimeData diff --git a/custom_components/opendisplay/upload.py b/custom_components/opendisplay/upload.py index 28a39cb..72692a4 100644 --- a/custom_components/opendisplay/upload.py +++ b/custom_components/opendisplay/upload.py @@ -17,8 +17,9 @@ from homeassistant.helpers.dispatcher import async_dispatcher_send from .runtime_data import OpenDisplayBLERuntimeData from .const import DOMAIN, SIGNAL_TAG_IMAGE_UPDATE -from .ble import BLEConnection, BLEImageUploader, BLEDeviceMetadata, get_protocol_by_name, BLEConnectionError, \ +from .ble import BLEConnection, BLEImageUploader, get_protocol_by_name, BLEConnectionError, \ BLETimeoutError, BLEProtocolError +from .metadata import BLEDeviceMetadata _LOGGER: Final = logging.getLogger(__name__) @@ -371,7 +372,7 @@ async def upload_to_ble_block( try: # Get device metadata from Home Assistant data device_metadata = None - protocol_type = "atc" # Default to ATC for backward compatibility + protocol_type = "open_display" # Find the config entry for this BLE device for entry in hass.config_entries.async_entries(DOMAIN): @@ -379,7 +380,7 @@ async def upload_to_ble_block( if runtime_data is not None and isinstance(runtime_data, OpenDisplayBLERuntimeData): if runtime_data.mac_address.upper() == mac: device_metadata = runtime_data.device_metadata - protocol_type = runtime_data.protocol_type + protocol_type = "open_display" break @@ -416,10 +417,7 @@ async def upload_to_ble_block( ) if processed_image is not None: - # Undo rotation for display (ATC rotation is for device memory, not viewing) display_image = processed_image - if protocol_type == "atc" and metadata.rotatebuffer == 1: - display_image = processed_image.transpose(Image.Transpose.ROTATE_270) jpeg_bytes = await hass.async_add_executor_job( image_to_jpeg_bytes, display_image, 95 diff --git a/requirements.txt b/requirements.txt index 49f3bf7..4ac4e9f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,4 +7,5 @@ requests_toolbelt==1.0.0 websockets websocket-client==1.7.0 python-resize-image==1.1.20 +py-opendisplay==7.3.0 bleak>=1.0.1 diff --git a/tests/test_ble_metadata.py b/tests/test_ble_metadata.py new file mode 100644 index 0000000..b45b49e --- /dev/null +++ b/tests/test_ble_metadata.py @@ -0,0 +1,146 @@ +"""Tests for the new BLEDeviceMetadata implementation.""" +import os +import sys + +# Add metadata paths directly to sys.path to bypass importing custom_components/__init__.py +# which would trigger Home Assistant imports that aren't installed in the test venv. +CURR_DIR = os.path.dirname(os.path.abspath(__file__)) +OPENDISPLAY_DIR = os.path.abspath(os.path.join(CURR_DIR, "../custom_components/opendisplay")) + +sys.path.insert(0, OPENDISPLAY_DIR) +import metadata as new_metadata_mod +NewBLEDeviceMetadata = new_metadata_mod.BLEDeviceMetadata + +sys.path.insert(0, os.path.join(OPENDISPLAY_DIR, "ble")) +import metadata as old_metadata_mod +OldBLEDeviceMetadata = old_metadata_mod.BLEDeviceMetadata + + +def test_atc_metadata_compatibility(): + """Verify that both old and new BLEDeviceMetadata produce identical results for ATC (flat) devices.""" + raw_metadata = { + "width": 296, + "height": 128, + "model_name": "ATC_2.9", + "fw_version": 25, + "rotatebuffer": 1, + "hw_type": 15, + "color_scheme": 1, # BWR + } + + old_metadata = OldBLEDeviceMetadata(raw_metadata) + new_metadata = NewBLEDeviceMetadata(raw_metadata) + + # Core properties + assert new_metadata.width == old_metadata.width == 296 + assert new_metadata.height == old_metadata.height == 128 + assert new_metadata.model_name == old_metadata.model_name == "ATC_2.9" + assert new_metadata.fw_version == old_metadata.fw_version == 25 + assert new_metadata.formatted_fw_version() == old_metadata.formatted_fw_version() == "0x0019" + assert new_metadata.rotatebuffer == old_metadata.rotatebuffer == 1 + assert new_metadata.hw_type == old_metadata.hw_type == 15 + assert new_metadata.power_mode == old_metadata.power_mode == 1 + assert new_metadata.is_open_display == old_metadata.is_open_display is False + + # Color scheme properties + assert new_metadata.color_scheme.value == old_metadata.color_scheme.value == 1 + assert new_metadata.accent_color == old_metadata.accent_color == "red" + assert new_metadata.is_multi_color == old_metadata.is_multi_color is True + + # Transmission / Upload properties + assert new_metadata.transmission_modes == old_metadata.transmission_modes == 0 + assert new_metadata.supports_zip_compression == old_metadata.supports_zip_compression is False + assert new_metadata.get_best_upload_method() == old_metadata.get_best_upload_method() == "block" + + +def test_opendisplay_metadata_compatibility(): + """Verify that both old and new BLEDeviceMetadata produce identical results for OpenDisplay devices.""" + raw_metadata = { + "fw_version": "2.0.2", + "model_name": "OD_7.5_BWR", + "open_display_config": { + "version": 1, + "minor_version": 0, + "loaded": True, + "system": { + "ic_type": 1, + "communication_modes": 1, + "device_flags": 0, + "pwr_pin": 255, + "reserved": "000000000000000000000000000000", + "pwr_pin_2": 255, + "pwr_pin_3": 255, + }, + "manufacturer": { + "manufacturer_id": 9286, + "board_type": 1, + "board_revision": 0, + "reserved": "000000000000000000000000000000000000", + }, + "power": { + "power_mode": 2, # USB power + "battery_capacity_mah": "000000", + "sleep_timeout_ms": 0, + "tx_power": 0, + "sleep_flags": 0, + "battery_sense_pin": 255, + "battery_sense_enable_pin": 255, + "battery_sense_flags": 0, + "capacity_estimator": 0, + "voltage_scaling_factor": 0, + "deep_sleep_current_ua": 0, + "deep_sleep_time_seconds": 0, + "reserved": "00000000000000000000", + }, + "displays": [ + { + "instance_number": 0, + "display_technology": 0, + "panel_ic_type": 0, + "pixel_width": 800, + "pixel_height": 480, + "active_width_mm": 0, + "active_height_mm": 0, + "open_display_tagtype": 12, + "tag_type": 12, + "rotation": 90, + "reset_pin": 255, + "busy_pin": 255, + "dc_pin": 255, + "cs_pin": 255, + "data_pin": 255, + "partial_update_support": 1, + "color_scheme": 3, # BWRY + "transmission_modes": 10, # 0x0A (supports ZIP (0x02) and direct_write (0x08)) + "clk_pin": 255, + "reserved_pins": "00000000000000", + "full_update_mC": 0, + "reserved": "0000000000000000000000000000", + } + ], + } + } + + old_metadata = OldBLEDeviceMetadata(raw_metadata) + new_metadata = NewBLEDeviceMetadata(raw_metadata) + + # Core properties + assert new_metadata.width == old_metadata.width == 800 + assert new_metadata.height == old_metadata.height == 480 + assert new_metadata.model_name == old_metadata.model_name == "OD_7.5_BWR" + assert new_metadata.fw_version == old_metadata.fw_version == "2.0.2" + assert new_metadata.formatted_fw_version() == old_metadata.formatted_fw_version() == "2.0.2" + assert new_metadata.rotatebuffer == old_metadata.rotatebuffer == 90 + assert new_metadata.hw_type == old_metadata.hw_type == 12 + assert new_metadata.power_mode == old_metadata.power_mode == 2 + assert new_metadata.is_open_display == old_metadata.is_open_display is True + + # Color scheme properties + assert new_metadata.color_scheme.value == old_metadata.color_scheme.value == 3 + assert new_metadata.accent_color == old_metadata.accent_color == "red" + assert new_metadata.is_multi_color == old_metadata.is_multi_color is True + + # Transmission / Upload properties + assert new_metadata.transmission_modes == old_metadata.transmission_modes == 10 + assert new_metadata.supports_zip_compression == old_metadata.supports_zip_compression is True + assert new_metadata.get_best_upload_method() == old_metadata.get_best_upload_method() == "direct_write"