Add background switch/select support with BKG API integration (v0.2.27)

Author: Ted Roberts

custom_components/novastar_h/api.py

@@ -61,10 +61,13 @@ class NovastarState:
     ftb_active: bool = False  # Fade to black (blackout) active
     freeze_active: bool = False  # Screen freeze active
     current_preset_id: int = -1  # -1 means no preset active
+    background_enabled: bool = False
+    background_id: int = 0
     screens: list[NovastarScreen] = field(default_factory=list)
     presets: list[NovastarPreset] = field(default_factory=list)
     inputs: list[dict[str, Any]] = field(default_factory=list)
     layers: list[dict[str, Any]] = field(default_factory=list)
+    backgrounds: list[dict[str, Any]] = field(default_factory=list)
 
 
 class NovastarClient:
@@ -109,6 +112,9 @@ def __init__(
         self._last_preset_id: int | None = None
         self._force_refresh_input_details = False
         self._force_refresh_layer_details = False
+        self._background_list_cache: list[dict[str, Any]] = []
+        self._background_refresh_counter = 0
+        self._force_refresh_backgrounds = False
 
     @property
     def host(self) -> str:
@@ -196,7 +202,7 @@ def _build_request(self, body: dict[str, Any]) -> dict[str, Any]:
 
     async def _async_request(
         self, endpoint: str, body: dict[str, Any]
-    ) -> dict[str, Any] | None:
+    ) -> Any | None:
         """Send POST request to Novastar API.
 
         Args:
@@ -235,7 +241,9 @@ async def _async_request(
                     body_data = data.get("body") or data.get("data") or {}
                     if self._encryption and isinstance(body_data, str):
                         return self._decrypt_body(body_data)
-                    return body_data if isinstance(body_data, dict) else {}
+                    if isinstance(body_data, (dict, list)):
+                        return body_data
+                    return {}
 
         except aiohttp.ClientError as ex:
             _LOGGER.debug("Connection error to %s: %s", url, ex)
@@ -431,9 +439,44 @@ async def async_get_state(
         state.signal_status = temp_data.get("signal_status")
         state.inputs = await self.async_get_inputs_with_details(device_id)
         state.layers = await self.async_get_layers_with_details(device_id, screen_id)
+        state.backgrounds = await self.async_get_background_list(device_id)
 
         return state
 
+    async def async_get_background_list(
+        self, device_id: int = 0
+    ) -> list[dict[str, Any]]:
+        """Get available backgrounds from bkg/readAllList with lightweight caching."""
+        self._background_refresh_counter += 1
+        periodic_refresh = self._background_refresh_counter % 12 == 0
+        should_refresh = periodic_refresh or self._force_refresh_backgrounds
+
+        if not self._background_list_cache or should_refresh:
+            self._force_refresh_backgrounds = False
+            data = await self._async_request("bkg/readAllList", {"deviceId": device_id})
+            if isinstance(data, list):
+                parsed: list[dict[str, Any]] = []
+                for item in data:
+                    if not isinstance(item, dict):
+                        continue
+                    bkg_id = item.get("bkgId")
+                    if not isinstance(bkg_id, int):
+                        continue
+                    general = item.get("general")
+                    name = item.get("name")
+                    if isinstance(general, dict) and isinstance(general.get("name"), str):
+                        name = general.get("name")
+                    parsed.append(
+                        {
+                            "bkgId": bkg_id,
+                            "name": name if isinstance(name, str) else f"BKG {bkg_id}",
+                        }
+                    )
+                parsed.sort(key=lambda item: item.get("bkgId", 0))
+                self._background_list_cache = parsed
+
+        return list(self._background_list_cache)
+
     async def async_get_input_list(self, device_id: int = 0) -> list[dict[str, Any]]:
         """Read all available inputs from input/readList."""
         data = await self._async_request("input/readList", {"deviceId": device_id})
@@ -654,7 +697,7 @@ async def async_get_device_status_info(
 
     async def async_send_raw_command(
         self, endpoint: str, body: dict[str, Any]
-    ) -> dict[str, Any] | None:
+    ) -> Any | None:
         """Send a raw API command.
 
         Args:
@@ -666,6 +709,26 @@ async def async_send_raw_command(
         """
         return await self._async_request(endpoint, body)
 
+    async def async_set_background(
+        self,
+        background_id: int,
+        enabled: bool,
+        screen_id: int = 0,
+        device_id: int = 0,
+    ) -> bool:
+        """Set screen background using screen/writeBKG."""
+        payload = {
+            "screenId": int(screen_id),
+            "deviceId": int(device_id),
+            "enable": 0 if enabled else 1,
+            "bkgId": max(0, int(background_id)),
+        }
+        data = await self._async_request("screen/writeBKG", payload)
+        if data is not None:
+            self._force_refresh_backgrounds = True
+            return True
+        return False
+
     async def async_set_layer_source(
         self,
         layer_id: int,

custom_components/novastar_h/const.py

@@ -20,7 +20,7 @@
 DEFAULT_SCREEN_ID = 0
 DEFAULT_ENCRYPTION = False
 DEFAULT_ALLOW_RAW_COMMANDS = False
-DEFAULT_LAYER_SELECT_PREPOPULATE_COUNT = 4
+DEFAULT_LAYER_SELECT_PREPOPULATE_COUNT = 3
 
 PLATFORMS: list[Platform] = [
     Platform.SWITCH,

custom_components/novastar_h/coordinator.py

@@ -30,6 +30,8 @@ def __init__(
         self._screen_id = screen_id
         self._ftb_active = False  # Track FTB state locally (API doesn't expose read)
         self._freeze_active = False  # Track freeze state locally
+        self._background_enabled = False
+        self._background_id = 0
         super().__init__(
             hass,
             _LOGGER,
@@ -112,10 +114,47 @@ async def async_set_layer_source(
             await self.async_request_refresh()
         return result
 
+    async def async_set_background_enabled(self, enabled: bool) -> bool:
+        """Enable or disable current selected background."""
+        result = await self._client.async_set_background(
+            background_id=self._background_id,
+            enabled=enabled,
+            screen_id=self._screen_id,
+            device_id=self._device_id,
+        )
+        if result:
+            self._background_enabled = enabled
+            if self.data:
+                self.data.background_enabled = enabled
+                self.data.background_id = self._background_id
+                self.async_set_updated_data(self.data)
+            await self.async_request_refresh()
+        return result
+
+    async def async_set_background(self, background_id: int, enabled: bool = True) -> bool:
+        """Set active background id and optional enabled state."""
+        result = await self._client.async_set_background(
+            background_id=background_id,
+            enabled=enabled,
+            screen_id=self._screen_id,
+            device_id=self._device_id,
+        )
+        if result:
+            self._background_id = max(0, int(background_id))
+            self._background_enabled = enabled
+            if self.data:
+                self.data.background_id = self._background_id
+                self.data.background_enabled = self._background_enabled
+                self.async_set_updated_data(self.data)
+            await self.async_request_refresh()
+        return result
+
     async def _async_update_data(self) -> NovastarState:
         """Fetch data from the device."""
         state = await self._client.async_get_state(self._screen_id, self._device_id)
         # Preserve locally tracked states
         state.ftb_active = self._ftb_active
         state.freeze_active = self._freeze_active
+        state.background_enabled = self._background_enabled
+        state.background_id = self._background_id
         return state

custom_components/novastar_h/manifest.json

@@ -21,5 +21,5 @@
     }
   ],
   "zeroconf": ["_novastar._tcp.local."],
-  "version": "0.2.26"
+  "version": "0.2.27"
 }

custom_components/novastar_h/select.py

@@ -63,6 +63,22 @@ def _layer_input_id(layer: dict[str, Any]) -> int | None:
     return _coerce_int(input_id)
 
 
+def _background_label(background: dict[str, Any]) -> str:
+    """Build stable display label for one background item."""
+    bkg_id = _coerce_int(background.get("bkgId"))
+    name = background.get("name")
+    if isinstance(name, str) and name.strip():
+        clean_name = name.strip()
+    elif bkg_id is not None:
+        clean_name = f"BKG {bkg_id}"
+    else:
+        clean_name = "Background"
+
+    if bkg_id is not None:
+        return f"{clean_name} ({bkg_id})"
+    return clean_name
+
+
 async def async_setup_entry(
     hass: HomeAssistant,
     entry: ConfigEntry,
@@ -81,6 +97,7 @@ async def async_setup_entry(
     layer_count = _coerce_int(layer_count) or DEFAULT_LAYER_SELECT_PREPOPULATE_COUNT
 
     entities: list[SelectEntity] = [NovastarPresetSelect(entry, coordinator, device_info)]
+    entities.append(NovastarBackgroundSelect(entry, coordinator, device_info))
     entities.extend(
         [
             NovastarLayerSourceSelect(entry, coordinator, device_info, layer_id)
@@ -317,3 +334,101 @@ async def async_select_option(self, option: str) -> None:
             slot_id=slot_id or 0,
             crop_id=255,
         )
+
+
+class NovastarBackgroundSelect(CoordinatorEntity[NovastarCoordinator], SelectEntity):
+    """Select entity for active background."""
+
+    _attr_has_entity_name = True
+    _attr_name = "Background"
+    _attr_translation_key = "background"
+
+    def __init__(
+        self,
+        entry: ConfigEntry,
+        coordinator: NovastarCoordinator,
+        device_info: NovastarDeviceInfo,
+    ) -> None:
+        """Initialize background select."""
+        super().__init__(coordinator)
+        self._entry = entry
+        self._device_info = device_info
+        self._attr_unique_id = f"{entry.entry_id}_background"
+
+    @property
+    def device_info(self):
+        """Return device info."""
+        model = "H Series"
+        if self._device_info.model_id:
+            model = f"H Series (Model {self._device_info.model_id})"
+        return {
+            "identifiers": {(DOMAIN, self._entry.entry_id)},
+            "manufacturer": "Novastar",
+            "model": model,
+            "name": self._entry.data.get(CONF_NAME, DEFAULT_NAME),
+            "sw_version": self._device_info.firmware,
+            "serial_number": self._device_info.serial,
+        }
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+        return self.coordinator.last_update_success
+
+    def _background_map(self) -> dict[str, int]:
+        """Return label->background_id map."""
+        if not self.coordinator.data:
+            return {}
+
+        mapped: dict[str, int] = {}
+        for background in self.coordinator.data.backgrounds:
+            bkg_id = _coerce_int(background.get("bkgId"))
+            if bkg_id is None:
+                continue
+            mapped[_background_label(background)] = bkg_id
+        return dict(sorted(mapped.items(), key=lambda item: item[0]))
+
+    @property
+    def options(self) -> list[str]:
+        """Return available background options."""
+        options = list(self._background_map().keys())
+        current = self.current_option
+        if current and current not in options:
+            options.append(current)
+        if not options:
+            options.append("Background 0")
+        return options
+
+    @property
+    def current_option(self) -> str | None:
+        """Return currently selected background option."""
+        if not self.coordinator.data:
+            return None
+
+        current_id = _coerce_int(self.coordinator.data.background_id)
+        if current_id is None:
+            return None
+
+        for label, bkg_id in self._background_map().items():
+            if bkg_id == current_id:
+                return label
+
+        return f"Background {current_id}"
+
+    async def async_select_option(self, option: str) -> None:
+        """Set active background and enable it."""
+        bkg_id = self._background_map().get(option)
+        if bkg_id is None:
+            text = option.strip()
+            try:
+                if text.startswith("Background "):
+                    bkg_id = int(text.replace("Background ", "").strip())
+                elif text.startswith("BKG "):
+                    bkg_id = int(text.replace("BKG ", "").strip())
+            except ValueError:
+                return
+
+        if bkg_id is None:
+            return
+
+        await self.coordinator.async_set_background(background_id=bkg_id, enabled=True)

custom_components/novastar_h/strings.json

@@ -107,12 +107,18 @@
       },
       "freeze": {
         "name": "Freeze Screen"
+      },
+      "background": {
+        "name": "Background"
       }
     },
     "select": {
       "preset": {
         "name": "Preset"
       },
+      "background": {
+        "name": "Background"
+      },
       "layer_source": {
         "name": "Layer Source"
       }

custom_components/novastar_h/switch.py

@@ -26,6 +26,7 @@ async def async_setup_entry(
     entities = [
         NovastarFTBSwitch(entry, coordinator, device_info),
         NovastarFreezeSwitch(entry, coordinator, device_info),
+        NovastarBackgroundSwitch(entry, coordinator, device_info),
     ]
     async_add_entities(entities)
 
@@ -138,3 +139,35 @@ async def async_turn_on(self, **kwargs: Any) -> None:
     async def async_turn_off(self, **kwargs: Any) -> None:
         """Unfreeze screen."""
         await self.coordinator.async_set_freeze(freeze=False)
+
+
+class NovastarBackgroundSwitch(NovastarSwitchBase):
+    """Switch for background display control."""
+
+    _attr_name = "Background"
+    _attr_translation_key = "background"
+
+    def __init__(
+        self,
+        entry: ConfigEntry,
+        coordinator: NovastarCoordinator,
+        device_info: NovastarDeviceInfo,
+    ) -> None:
+        """Initialize background switch."""
+        super().__init__(entry, coordinator, device_info)
+        self._attr_unique_id = f"{entry.entry_id}_background"
+
+    @property
+    def is_on(self) -> bool:
+        """Return True if background is enabled."""
+        if self.coordinator.data:
+            return self.coordinator.data.background_enabled
+        return False
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Enable background."""
+        await self.coordinator.async_set_background_enabled(enabled=True)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Disable background."""
+        await self.coordinator.async_set_background_enabled(enabled=False)