Add MOS gate control via select entity

Adds a Select entity with four options (Normal / Charge Disabled /
Discharge Disabled / Both Disabled) that sends the 0xE1 write command
to the BMS and immediately refreshes sensor state.  Current option is
derived from the mos_charge_enabled / mos_discharge_enabled bits
already parsed on every poll.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

custom_components/xiaoxiang_bms/__init__.py

@@ -7,7 +7,7 @@ from homeassistant.core import HomeAssistant
 from .const import CONF_ADDRESS, CONF_POLL_INTERVAL, DEFAULT_POLL_INTERVAL, DOMAIN
 from .coordinator import BmsCoordinator
 
-PLATFORMS = ["sensor", "binary_sensor", "number"]
+PLATFORMS = ["sensor", "binary_sensor", "number", "select"]
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:

custom_components/xiaoxiang_bms/bluetooth_handler.py

@@ -161,6 +161,53 @@ class BmsBluetoothHandler:
 
         return None
 
+    # ------------------------------------------------------------------
+    # MOS write command
+    # ------------------------------------------------------------------
+
+    async def write_mos(self, ble_device: BLEDevice, value: int) -> bool:
+        """Send a MOS control write command and return True on ACK.
+
+        Follows the same connect → send → disconnect pattern as poll() so
+        it doesn't interfere with the normal poll cycle.
+        """
+        command = self._build_mos_command(value)
+        _LOGGER.debug("Writing MOS value 0x%02X to BMS at %s", value, self._address)
+        client = BleakClient(ble_device)
+        try:
+            await client.connect()
+            await client.start_notify(RX_CHAR_UUID, self._on_notify)
+            await asyncio.sleep(0.5)
+            response = await self._request(client, command, timeout=3.0, retries=2)
+            # Response: DD E1 00 00 CHK_H CHK_L 77  (status byte 0x00 = OK)
+            return response is not None and response[2] == 0x00
+        finally:
+            try:
+                await client.disconnect()
+            except Exception:
+                pass
+            self._buffer.clear()
+
+    @staticmethod
+    def _build_mos_command(value: int) -> bytes:
+        """Build a MOS control write frame with correct checksum.
+
+        Frame:  DD  5A  E1  02  00  XX  CHK_H  CHK_L  77
+        Checked bytes (per spec): command_code + length + data bytes
+                                = 0xE1 + 0x02 + 0x00 + XX
+        Checksum = two's complement of sum, high byte first.
+
+        Verified against spec example:
+          XX=0x02 → sum=0xE5 → ~0xE5+1=0xFF1B → CHK FF 1B ✓
+        """
+        checked = [0xE1, 0x02, 0x00, value & 0xFF]
+        checksum = (~sum(checked) + 1) & 0xFFFF
+        return bytes([
+            0xDD, 0x5A, 0xE1, 0x02, 0x00, value & 0xFF,
+            (checksum >> 8) & 0xFF, checksum & 0xFF,
+            0x77,
+        ])
+
     # ------------------------------------------------------------------
     # Frame parsers
     # ------------------------------------------------------------------

custom_components/xiaoxiang_bms/const.py

@@ -23,6 +23,12 @@ CMD_VERSION = bytes([0xDD, 0xA5, 0x05, 0x00, 0xFF, 0xFB, 0x77])  # hardware vers
 FRAME_START = 0xDD
 FRAME_END   = 0x77
 
+# MOS control values (XX byte in write command DD 5A E1 02 00 XX CHK_H CHK_L 77)
+MOS_NORMAL        = 0x00  # release software lock — both gates open
+MOS_CHARGE_OFF    = 0x01  # disable charge MOS, keep discharge MOS on
+MOS_DISCHARGE_OFF = 0x02  # disable discharge MOS, keep charge MOS on
+MOS_BOTH_OFF      = 0x03  # disable both charge and discharge MOS
+
 # Response command IDs (byte 1 of frame)
 CMD_ID_GENERAL = 0x03
 CMD_ID_CELL    = 0x04

custom_components/xiaoxiang_bms/coordinator.py

@@ -7,6 +7,7 @@ from datetime import timedelta
 
 from homeassistant.components.bluetooth import async_ble_device_from_address
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
@@ -71,6 +72,19 @@ class BmsCoordinator(DataUpdateCoordinator[dict]):
     async def async_teardown(self) -> None:
         """No-op — each poll disconnects itself."""
 
+    async def async_write_mos(self, value: int) -> None:
+        """Send a MOS control command to the BMS, then refresh sensor state."""
+        device = async_ble_device_from_address(self.hass, self.address, connectable=True)
+        if device is None:
+            raise HomeAssistantError(
+                f"BMS ({self.address}) not reachable — cannot send MOS command"
+            )
+        success = await self._handler.write_mos(device, value)
+        if not success:
+            raise HomeAssistantError("BMS did not acknowledge the MOS command")
+        # Refresh immediately so sensors reflect the new MOS state
+        await self.async_request_refresh()
+
     # ------------------------------------------------------------------
     # Poll
     # ------------------------------------------------------------------

custom_components/xiaoxiang_bms/select.py

@@ -0,0 +1,73 @@
+"""Select entity for Xiaoxiang Smart BMS MOS gate control."""
+from __future__ import annotations
+
+from homeassistant.components.select import SelectEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import (
+    DOMAIN,
+    MOS_BOTH_OFF,
+    MOS_CHARGE_OFF,
+    MOS_DISCHARGE_OFF,
+    MOS_NORMAL,
+)
+from .coordinator import BmsCoordinator
+
+# Maps the human-readable option to the XX byte value in the write command
+_OPTION_TO_VALUE: dict[str, int] = {
+    "Normal":           MOS_NORMAL,
+    "Charge Disabled":  MOS_CHARGE_OFF,
+    "Discharge Disabled": MOS_DISCHARGE_OFF,
+    "Both Disabled":    MOS_BOTH_OFF,
+}
+_OPTIONS = list(_OPTION_TO_VALUE)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    coordinator: BmsCoordinator = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities([BmsMosSelect(coordinator)])
+
+
+class BmsMosSelect(CoordinatorEntity[BmsCoordinator], SelectEntity):
+    """Dropdown to control the BMS charge/discharge MOSFET gates.
+
+    Current state is derived from the MOS status bits returned by the BMS
+    on every poll.  Selecting an option sends the write command (0xE1)
+    immediately and triggers a coordinator refresh so sensors update.
+    """
+
+    _attr_has_entity_name = True
+    _attr_name = "MOS Control"
+    _attr_options = _OPTIONS
+    _attr_icon = "mdi:electric-switch"
+
+    def __init__(self, coordinator: BmsCoordinator) -> None:
+        super().__init__(coordinator)
+        self._attr_unique_id = f"{coordinator.address}_mos_control"
+        self._attr_device_info = coordinator.device_info
+
+    @property
+    def current_option(self) -> str | None:
+        data = self.coordinator.data
+        if not data:
+            return None
+        charge    = data.get("mos_charge_enabled", True)
+        discharge = data.get("mos_discharge_enabled", True)
+        if charge and discharge:
+            return "Normal"
+        if not charge and discharge:
+            return "Charge Disabled"
+        if charge and not discharge:
+            return "Discharge Disabled"
+        return "Both Disabled"
+
+    async def async_select_option(self, option: str) -> None:
+        value = _OPTION_TO_VALUE[option]
+        await self.coordinator.async_write_mos(value)