python-xiaomi-miwifi

Talks to one MiWiFi router (the gateway).

Source code in src/xiaomi_miwifi/client.py

class MiWiFiClient:
    """Talks to one MiWiFi router (the gateway)."""

    def __init__(
        self,
        host: str,
        password: str,
        *,
        username: str = "admin",
        port: int = DEFAULT_PORT,
        session: aiohttp.ClientSession | None = None,
    ) -> None:
        self._host = host
        self._password = password
        self._username = username
        self._port = port
        self._session = session
        self._owns_session = session is None
        self._token: str | None = None

    # -- properties -----------------------------------------------------
    @property
    def host(self) -> str:
        return self._host

    @property
    def port(self) -> int:
        return self._port

    @property
    def token(self) -> str | None:
        return self._token

    @property
    def _base(self) -> str:
        suffix = "" if self._port == 80 else f":{self._port}"
        return f"http://{self._host}{suffix}/cgi-bin/luci"

    # -- session --------------------------------------------------------
    async def _ensure_session(self) -> aiohttp.ClientSession:
        if self._session is None:
            # No session was ever supplied: create and own one.
            self._session = aiohttp.ClientSession()
            self._owns_session = True
        elif self._session.closed:
            if not self._owns_session:
                raise MiWiFiConnectionError("provided session is closed")
            # We own a session that has been closed: recreate it.
            self._session = aiohttp.ClientSession()
        return self._session

    async def async_close(self) -> None:
        if self._owns_session and self._session and not self._session.closed:
            await self._session.close()

    # -- auth -----------------------------------------------------------
    def _make_nonce(self, device_id: str) -> str:
        return f"0_{device_id}_{int(time.time())}_{random.randint(0, 9999999)}"

    async def async_login(self) -> str:
        session = await self._ensure_session()
        timeout = aiohttp.ClientTimeout(total=HTTP_TIMEOUT)
        try:
            async with session.get(f"{self._base}/web", timeout=timeout) as resp:
                page = await resp.text()
        except aiohttp.ClientError as err:
            raise MiWiFiConnectionError(f"cannot reach {self._host}: {err}") from err

        key_match = _KEY_RE.search(page)
        if not key_match:
            raise MiWiFiAuthError("login key not found on web page")
        key = key_match.group(1)
        device_match = _DEVICEID_RE.search(page)
        device_id = device_match.group(1) if device_match else "0"

        nonce = self._make_nonce(device_id)
        inner = hashlib.sha1((self._password + key).encode()).hexdigest()
        pwd_hash = hashlib.sha1((nonce + inner).encode()).hexdigest()

        form = {
            "username": self._username,
            "logtype": "2",
            "password": pwd_hash,
            "nonce": nonce,
        }
        try:
            async with session.post(
                f"{self._base}/{PATH_LOGIN}", data=form, timeout=timeout
            ) as resp:
                text = await resp.text()
        except aiohttp.ClientError as err:
            raise MiWiFiConnectionError(f"login transport error: {err}") from err

        try:
            payload = json.loads(text)
        except json.JSONDecodeError as err:
            raise MiWiFiConnectionError(f"bad JSON from login: {err}") from err

        token = payload.get("token") if isinstance(payload, dict) else None
        if not token:
            raise MiWiFiAuthError(f"login failed: {payload}")
        self._token = token
        return token

    # -- raw authenticated requests -------------------------------------
    async def _request(
        self,
        method: str,
        path: str,
        *,
        data: dict | None = None,
        timeout_s: int = HTTP_TIMEOUT,
        _retry: bool = True,
    ) -> dict[str, Any]:
        if self._token is None:
            await self.async_login()
        session = await self._ensure_session()
        timeout = aiohttp.ClientTimeout(total=timeout_s)
        url = f"{self._base}/;stok={self._token}/{path}"
        try:
            async with session.request(
                method, url, data=data, timeout=timeout
            ) as resp:
                text = await resp.text()
        except aiohttp.ClientError as err:
            raise MiWiFiConnectionError(f"request to {path} failed: {err}") from err

        try:
            payload = json.loads(text)
        except json.JSONDecodeError as err:
            raise MiWiFiConnectionError(f"bad JSON from {path}: {err}") from err

        if isinstance(payload, dict) and payload.get("code") == 401 and _retry:
            self._token = None
            await self.async_login()
            return await self._request(
                method, path, data=data, timeout_s=timeout_s, _retry=False
            )
        return payload

    async def _get(self, path: str, *, timeout_s: int = HTTP_TIMEOUT) -> dict[str, Any]:
        return await self._request("GET", path, timeout_s=timeout_s)

    async def _safe_get(self, path: str) -> dict:
        """GET returning {} on connection error (for optional endpoints)."""
        try:
            return await self._get(path)
        except MiWiFiConnectionError:
            return {}

    async def _post(self, path: str, data: dict) -> dict[str, Any]:
        return await self._request("POST", path, data=data)

    # -- read endpoints -------------------------------------------------
    async def async_get_newstatus(self) -> dict:
        return await self._get(PATH_NEWSTATUS)

    async def async_get_wan_info(self) -> dict:
        return await self._get(PATH_WAN_INFO)

    async def async_get_raw_status(self) -> dict:
        return await self._get(PATH_STATUS)

    async def async_get_topology(self) -> dict:
        return await self._get(PATH_TOPO_GRAPH)

    async def async_get_rom_update(self) -> dict:
        return await self._get(PATH_CHECK_ROM)

    async def async_get_wan_statistics(self) -> dict:
        return await self._get(PATH_WAN_STATISTICS)

    async def async_get_wifi_detail(self) -> dict:
        return await self._get(PATH_WIFI_DETAIL)

    async def async_get_macfilter(self) -> dict:
        """Read the MAC filter list. Slow endpoint — uses a longer timeout."""
        return await self._get(PATH_MACFILTER_INFO, timeout_s=HTTP_TIMEOUT_SLOW)

    async def async_get_led(self) -> dict:
        return await self._get(PATH_LED)

    async def async_get_router_info(self) -> dict:
        return await self._get(PATH_ROUTER_INFO)

    async def async_get_wifi_connect_devices(self) -> dict:
        return await self._get(PATH_WIFI_CONNECT_DEVICES)

    async def async_get_lan_info(self) -> dict:
        return await self._get(PATH_LAN_INFO)

    async def async_get_available_channels(self, wifi_index: int) -> list[str]:
        """Return the list of channel numbers (as strings; "0" = auto)
        available for a band. wifi_index: 1 = 2.4G, 2 = 5G.
        """
        data = await self._get(f"{PATH_AVAILABLE_CHANNELS}?wifiIndex={wifi_index}")
        items = data.get("list", []) if isinstance(data, dict) else []
        return [str(i.get("c")) for i in items if isinstance(i, dict) and "c" in i]

    async def async_get_init_info(self) -> dict:
        return await self._get(PATH_INIT_INFO)

    async def async_get_qos_info(self) -> dict:
        return await self._get(PATH_QOS_INFO)

    async def async_get_bandwidth_history(self) -> dict:
        return await self._get(f"{PATH_BANDWIDTH_TEST}?history=1")

    async def async_get_pppoe_status(self) -> dict:
        return await self._get(PATH_PPPOE_STATUS)

    async def async_get_ddns(self) -> dict:
        return await self._get(PATH_DDNS)

    async def async_get_dmz(self) -> dict:
        return await self._get(PATH_DMZ)

    async def async_get_portforward(self, ftype: int = 1) -> dict:
        return await self._get(f"{PATH_PORTFORWARD}?ftype={ftype}")

    async def async_get_lan_dhcp(self) -> dict:
        return await self._get(PATH_LAN_DHCP)

    async def async_get_sys_time(self) -> dict:
        return await self._get(PATH_SYS_TIME)

    async def async_get_ipv6_status(self) -> dict:
        return await self._get(PATH_IPV6_STATUS)

    async def async_luci_request(self, path: str) -> dict:
        """Generic READ-ONLY GET passthrough to any LuCI API path.

        ``path`` is the API path WITHOUT the leading ``;stok=`` prefix,
        e.g. ``"api/misystem/router_info"``. Strips a leading slash.

        Mutating paths (reboot, set_*, upgrade, bind/unbind, etc.) are
        rejected so this cannot change router state.
        """
        clean = path.lstrip("/")
        lowered = clean.lower()
        if any(b in lowered for b in _LUCI_BLOCKED_SUBSTRINGS):
            raise MiWiFiError(f"refusing mutating path via read-only request: {path}")
        return await self._get(clean)

    async def async_get_status(self) -> MiWiFiStatus:
        """Aggregate all read endpoints into a MiWiFiStatus.

        The three secondary endpoints (wan_statistics, wifi_detail, led) are
        fetched best-effort: a MiWiFiConnectionError on any one of them yields
        an empty dict rather than failing the whole status.
        """
        newstatus = await self.async_get_newstatus()
        wan = await self.async_get_wan_info()
        status = await self.async_get_raw_status()
        topo = await self.async_get_topology()
        rom = await self.async_get_rom_update()

        wan_stats = await self._safe_get(PATH_WAN_STATISTICS)
        wifi_detail = await self._safe_get(PATH_WIFI_DETAIL)
        led = await self._safe_get(PATH_LED)
        router_info = await self._safe_get(PATH_ROUTER_INFO)
        lan_info = await self._safe_get(PATH_LAN_INFO)
        init_info = await self._safe_get(PATH_INIT_INFO)
        qos_info = await self._safe_get(PATH_QOS_INFO)
        bandwidth_history = await self._safe_get(f"{PATH_BANDWIDTH_TEST}?history=1")
        pppoe = await self._safe_get(PATH_PPPOE_STATUS)
        ddns = await self._safe_get(PATH_DDNS)
        dmz = await self._safe_get(PATH_DMZ)
        portforward = await self._safe_get(f"{PATH_PORTFORWARD}?ftype=1")
        lan_dhcp = await self._safe_get(PATH_LAN_DHCP)
        sys_time = await self._safe_get(PATH_SYS_TIME)
        ipv6 = await self._safe_get(PATH_IPV6_STATUS)

        return parse_status(
            newstatus=newstatus,
            wan=wan,
            status=status,
            topo=topo,
            rom=rom,
            wan_stats=wan_stats,
            wifi_detail=wifi_detail,
            led=led,
            router_info=router_info,
            lan_info=lan_info,
            init_info=init_info,
            qos_info=qos_info,
            bandwidth_history=bandwidth_history,
            pppoe=pppoe,
            ddns=ddns,
            dmz=dmz,
            portforward=portforward,
            lan_dhcp=lan_dhcp,
            sys_time=sys_time,
            ipv6=ipv6,
        )

    async def async_get_blocked_devices(self) -> list[dict]:
        data = await self.async_get_macfilter()
        return data.get("flist", []) if isinstance(data, dict) else []

    async def async_get_clients(self) -> list[ClientDevice]:
        data = await self._get(PATH_DEVICELIST)
        entries = data.get("list", []) if isinstance(data, dict) else []
        clients = [ClientDevice.from_entry(e) for e in entries]
        status_devs: list = []
        connect_devs: list = []
        try:
            raw = await self._get(PATH_STATUS)
            status_devs = raw.get("dev", []) if isinstance(raw, dict) else []
        except MiWiFiConnectionError:
            pass
        try:
            cd = await self._get(PATH_WIFI_CONNECT_DEVICES)
            connect_devs = cd.get("list", []) if isinstance(cd, dict) else []
        except MiWiFiConnectionError:
            pass
        return merge_client_telemetry(clients, status_devs, connect_devs)

    async def async_get_dhcp_reservations(self) -> list[dict]:
        data = await self._get(PATH_MACBIND_INFO)
        return data.get("list", []) if isinstance(data, dict) else []

    # -- write controls -------------------------------------------------
    @staticmethod
    def _ok(payload: dict) -> bool:
        return isinstance(payload, dict) and payload.get("code") == 0

    async def async_reboot(self) -> bool:
        """Reboot the router. Disruptive — never call from tests."""
        return self._ok(await self._get(PATH_REBOOT))

    async def async_set_wifi_enabled(self, ifname: str, enabled: bool) -> bool:
        """Enable/disable a radio (wl0=5G, wl1=2.4G). Disruptive."""
        path = PATH_WIFI_UP if enabled else PATH_WIFI_DOWN
        return self._ok(await self._post(path, {"ifname": ifname}))

    async def async_add_dhcp_reservation(
        self, mac: str, ip: str, name: str
    ) -> bool:
        """Add/replace a DHCP reservation.

        The router uses replace semantics: the full list must be POSTed or
        existing reservations are wiped. We therefore read the current list,
        merge the new entry (replacing any with the same MAC), and POST all.
        """
        current = await self.async_get_dhcp_reservations()
        merged = [r for r in current if r.get("mac", "").upper() != mac.upper()]
        merged.append({"ip": ip, "mac": mac, "name": name})
        body = {"data": json.dumps(
            [
                {"ip": r["ip"], "mac": r["mac"], "name": r.get("name", "")}
                for r in merged
            ]
        )}
        return self._ok(await self._post(PATH_MAC_BIND, body))

    async def async_remove_dhcp_reservation(self, mac: str) -> bool:
        """Remove a single DHCP reservation by MAC."""
        return self._ok(await self._post(PATH_MAC_UNBIND, {"mac": mac}))

    async def async_block_device(self, mac: str) -> bool:
        """Block a device's internet access (set_mac_filter wan=0)."""
        return self._ok(
            await self._post(PATH_SET_MAC_FILTER, {"mac": mac, "wan": "0"})
        )

    async def async_unblock_device(self, mac: str) -> bool:
        """Restore a device's internet access (set_mac_filter wan=1)."""
        return self._ok(
            await self._post(PATH_SET_MAC_FILTER, {"mac": mac, "wan": "1"})
        )

    _IFNAME_FOR_INDEX = {WIFI_INDEX_24G: "wl1", WIFI_INDEX_5G: "wl0"}

    async def _radio_config(self, wifi_index: int) -> dict:
        """Return the current wifi_detail_all entry for a band, or raise."""
        detail = await self._get(PATH_WIFI_DETAIL)
        ifname = self._IFNAME_FOR_INDEX.get(wifi_index)
        info = detail.get("info", []) if isinstance(detail, dict) else []
        for radio in info:
            if isinstance(radio, dict) and radio.get("ifname") == ifname:
                return radio
        raise MiWiFiError(f"radio wifiIndex={wifi_index} not found")

    async def _set_wifi(self, wifi_index: int, overrides: dict) -> bool:
        """Read-modify-write set_wifi: resend the full current radio config
        with the given field(s) overridden. Disruptive (restarts the radio).
        """
        radio = await self._radio_config(wifi_index)
        chan_info = radio.get("channelInfo") or {}
        # Prefer the resolved operating channel (channelInfo) over the stored
        # "channel" field, which is "0" (auto) when the band is on auto-select.
        channel = chan_info.get("channel") or radio.get("channel") or "0"
        bandwidth = chan_info.get("bandwidth") or radio.get("bandwidth") or "0"
        params = {
            "wifiIndex": str(wifi_index),
            "on": "1",
            "ssid": radio.get("ssid", ""),
            "pwd": radio.get("password", ""),
            "encryption": radio.get("encryption", ""),
            "channel": str(channel),
            "bandwidth": str(bandwidth),
            "txpwr": radio.get("txpwr", "max"),
            "hidden": str(radio.get("hidden", "0")),
            "ax": str(radio.get("ax", "0")),
            "txbf": str(radio.get("txbf", "3")),
        }
        params.update(overrides)
        query = urllib.parse.urlencode(params)
        return self._ok(await self._get(f"{PATH_SET_WIFI}?{query}"))

    async def async_set_wifi_channel(self, wifi_index: int, channel: str) -> bool:
        """Set a band's channel (1=2.4G, 2=5G). "0" = auto. Disruptive."""
        return await self._set_wifi(wifi_index, {"channel": str(channel)})

    async def async_set_wifi_txpwr(self, wifi_index: int, txpwr: str) -> bool:
        """Set a band's transmit power (max/mid/min). Disruptive."""
        return await self._set_wifi(wifi_index, {"txpwr": txpwr})

    async def async_set_qos(self, enabled: bool) -> bool:
        """Enable/disable QoS."""
        on = 1 if enabled else 0
        return self._ok(await self._get(f"{PATH_QOS_SWITCH}?on={on}"))

    async def async_run_speed_test(self) -> dict:
        """Trigger a WAN speed test and return its result. Saturates the WAN
        link briefly — disruptive; never called live in tests."""
        return await self._get(PATH_BANDWIDTH_TEST)

    async def async_add_port_forward(
        self, ip: str, name: str, proto: int, sport: int, dport: int
    ) -> bool:
        """Add a single-port forward (proto: 1=TCP, 2=UDP, 3=TCP+UDP) and apply."""
        params = urllib.parse.urlencode(
            {"ip": ip, "name": name, "proto": str(proto),
             "sport": str(sport), "dport": str(dport)}
        )
        if not self._ok(await self._get(f"{PATH_ADD_REDIRECT}?{params}")):
            return False
        return self._ok(await self._get(PATH_REDIRECT_APPLY))

    async def async_delete_port_forward(self, sport: int) -> bool:
        """Delete a port forward by its external (source) port, then apply."""
        if not self._ok(await self._get(f"{PATH_DELETE_REDIRECT}?port={sport}")):
            return False
        return self._ok(await self._get(PATH_REDIRECT_APPLY))

    async def async_set_dmz(self, ip: str, mac: str = "") -> bool:
        """Enable DMZ forwarding all WAN traffic to ``ip``."""
        params = urllib.parse.urlencode({"ip": ip, "mac": mac, "mode": "1"})
        return self._ok(await self._get(f"{PATH_SET_DMZ}?{params}"))

    async def async_clear_dmz(self) -> bool:
        """Disable DMZ."""
        return self._ok(await self._get(f"{PATH_DMZ_OFF}?mode=1"))

    async def async_set_ddns(self, enabled: bool) -> bool:
        """Enable/disable DDNS."""
        return self._ok(
            await self._get(f"{PATH_DDNS_SWITCH}?on={1 if enabled else 0}")
        )

    async def async_set_dhcp(self, start: int, end: int, leasetime: str) -> bool:
        """Set DHCP pool start/end (last octet) and lease time (e.g. '720m')."""
        params = urllib.parse.urlencode(
            {"start": str(start), "end": str(end),
             "leasetime": leasetime, "ignore": "0"}
        )
        return self._ok(await self._get(f"{PATH_SET_LAN_DHCP}?{params}"))

    async def __aenter__(self) -> MiWiFiClient:
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.async_close()

async_add_dhcp_reservation(mac, ip, name) async

Add/replace a DHCP reservation.

The router uses replace semantics: the full list must be POSTed or existing reservations are wiped. We therefore read the current list, merge the new entry (replacing any with the same MAC), and POST all.

Source code in src/xiaomi_miwifi/client.py

async def async_add_dhcp_reservation(
    self, mac: str, ip: str, name: str
) -> bool:
    """Add/replace a DHCP reservation.

    The router uses replace semantics: the full list must be POSTed or
    existing reservations are wiped. We therefore read the current list,
    merge the new entry (replacing any with the same MAC), and POST all.
    """
    current = await self.async_get_dhcp_reservations()
    merged = [r for r in current if r.get("mac", "").upper() != mac.upper()]
    merged.append({"ip": ip, "mac": mac, "name": name})
    body = {"data": json.dumps(
        [
            {"ip": r["ip"], "mac": r["mac"], "name": r.get("name", "")}
            for r in merged
        ]
    )}
    return self._ok(await self._post(PATH_MAC_BIND, body))

async_add_port_forward(ip, name, proto, sport, dport) async

Add a single-port forward (proto: 1=TCP, 2=UDP, 3=TCP+UDP) and apply.

Source code in src/xiaomi_miwifi/client.py

async def async_add_port_forward(
    self, ip: str, name: str, proto: int, sport: int, dport: int
) -> bool:
    """Add a single-port forward (proto: 1=TCP, 2=UDP, 3=TCP+UDP) and apply."""
    params = urllib.parse.urlencode(
        {"ip": ip, "name": name, "proto": str(proto),
         "sport": str(sport), "dport": str(dport)}
    )
    if not self._ok(await self._get(f"{PATH_ADD_REDIRECT}?{params}")):
        return False
    return self._ok(await self._get(PATH_REDIRECT_APPLY))

async_block_device(mac) async

Block a device's internet access (set_mac_filter wan=0).

Source code in src/xiaomi_miwifi/client.py

async def async_block_device(self, mac: str) -> bool:
    """Block a device's internet access (set_mac_filter wan=0)."""
    return self._ok(
        await self._post(PATH_SET_MAC_FILTER, {"mac": mac, "wan": "0"})
    )

async_clear_dmz() async

Disable DMZ.

Source code in src/xiaomi_miwifi/client.py

async def async_clear_dmz(self) -> bool:
    """Disable DMZ."""
    return self._ok(await self._get(f"{PATH_DMZ_OFF}?mode=1"))

async_delete_port_forward(sport) async

Delete a port forward by its external (source) port, then apply.

Source code in src/xiaomi_miwifi/client.py

async def async_delete_port_forward(self, sport: int) -> bool:
    """Delete a port forward by its external (source) port, then apply."""
    if not self._ok(await self._get(f"{PATH_DELETE_REDIRECT}?port={sport}")):
        return False
    return self._ok(await self._get(PATH_REDIRECT_APPLY))

async_get_available_channels(wifi_index) async

Return the list of channel numbers (as strings; "0" = auto) available for a band. wifi_index: 1 = 2.4G, 2 = 5G.

Source code in src/xiaomi_miwifi/client.py

async def async_get_available_channels(self, wifi_index: int) -> list[str]:
    """Return the list of channel numbers (as strings; "0" = auto)
    available for a band. wifi_index: 1 = 2.4G, 2 = 5G.
    """
    data = await self._get(f"{PATH_AVAILABLE_CHANNELS}?wifiIndex={wifi_index}")
    items = data.get("list", []) if isinstance(data, dict) else []
    return [str(i.get("c")) for i in items if isinstance(i, dict) and "c" in i]

async_get_macfilter() async

Read the MAC filter list. Slow endpoint — uses a longer timeout.

Source code in src/xiaomi_miwifi/client.py

async def async_get_macfilter(self) -> dict:
    """Read the MAC filter list. Slow endpoint — uses a longer timeout."""
    return await self._get(PATH_MACFILTER_INFO, timeout_s=HTTP_TIMEOUT_SLOW)

async_get_status() async

Aggregate all read endpoints into a MiWiFiStatus.

The three secondary endpoints (wan_statistics, wifi_detail, led) are fetched best-effort: a MiWiFiConnectionError on any one of them yields an empty dict rather than failing the whole status.

Source code in src/xiaomi_miwifi/client.py

async def async_get_status(self) -> MiWiFiStatus:
    """Aggregate all read endpoints into a MiWiFiStatus.

    The three secondary endpoints (wan_statistics, wifi_detail, led) are
    fetched best-effort: a MiWiFiConnectionError on any one of them yields
    an empty dict rather than failing the whole status.
    """
    newstatus = await self.async_get_newstatus()
    wan = await self.async_get_wan_info()
    status = await self.async_get_raw_status()
    topo = await self.async_get_topology()
    rom = await self.async_get_rom_update()

    wan_stats = await self._safe_get(PATH_WAN_STATISTICS)
    wifi_detail = await self._safe_get(PATH_WIFI_DETAIL)
    led = await self._safe_get(PATH_LED)
    router_info = await self._safe_get(PATH_ROUTER_INFO)
    lan_info = await self._safe_get(PATH_LAN_INFO)
    init_info = await self._safe_get(PATH_INIT_INFO)
    qos_info = await self._safe_get(PATH_QOS_INFO)
    bandwidth_history = await self._safe_get(f"{PATH_BANDWIDTH_TEST}?history=1")
    pppoe = await self._safe_get(PATH_PPPOE_STATUS)
    ddns = await self._safe_get(PATH_DDNS)
    dmz = await self._safe_get(PATH_DMZ)
    portforward = await self._safe_get(f"{PATH_PORTFORWARD}?ftype=1")
    lan_dhcp = await self._safe_get(PATH_LAN_DHCP)
    sys_time = await self._safe_get(PATH_SYS_TIME)
    ipv6 = await self._safe_get(PATH_IPV6_STATUS)

    return parse_status(
        newstatus=newstatus,
        wan=wan,
        status=status,
        topo=topo,
        rom=rom,
        wan_stats=wan_stats,
        wifi_detail=wifi_detail,
        led=led,
        router_info=router_info,
        lan_info=lan_info,
        init_info=init_info,
        qos_info=qos_info,
        bandwidth_history=bandwidth_history,
        pppoe=pppoe,
        ddns=ddns,
        dmz=dmz,
        portforward=portforward,
        lan_dhcp=lan_dhcp,
        sys_time=sys_time,
        ipv6=ipv6,
    )

async_luci_request(path) async

Generic READ-ONLY GET passthrough to any LuCI API path.

path is the API path WITHOUT the leading ;stok= prefix, e.g. "api/misystem/router_info". Strips a leading slash.

Mutating paths (reboot, set_*, upgrade, bind/unbind, etc.) are rejected so this cannot change router state.

Source code in src/xiaomi_miwifi/client.py

async def async_luci_request(self, path: str) -> dict:
    """Generic READ-ONLY GET passthrough to any LuCI API path.

    ``path`` is the API path WITHOUT the leading ``;stok=`` prefix,
    e.g. ``"api/misystem/router_info"``. Strips a leading slash.

    Mutating paths (reboot, set_*, upgrade, bind/unbind, etc.) are
    rejected so this cannot change router state.
    """
    clean = path.lstrip("/")
    lowered = clean.lower()
    if any(b in lowered for b in _LUCI_BLOCKED_SUBSTRINGS):
        raise MiWiFiError(f"refusing mutating path via read-only request: {path}")
    return await self._get(clean)

async_reboot() async

Reboot the router. Disruptive — never call from tests.

Source code in src/xiaomi_miwifi/client.py

async def async_reboot(self) -> bool:
    """Reboot the router. Disruptive — never call from tests."""
    return self._ok(await self._get(PATH_REBOOT))

async_remove_dhcp_reservation(mac) async

Remove a single DHCP reservation by MAC.

Source code in src/xiaomi_miwifi/client.py

async def async_remove_dhcp_reservation(self, mac: str) -> bool:
    """Remove a single DHCP reservation by MAC."""
    return self._ok(await self._post(PATH_MAC_UNBIND, {"mac": mac}))

async_run_speed_test() async

Trigger a WAN speed test and return its result. Saturates the WAN link briefly — disruptive; never called live in tests.

Source code in src/xiaomi_miwifi/client.py

async def async_run_speed_test(self) -> dict:
    """Trigger a WAN speed test and return its result. Saturates the WAN
    link briefly — disruptive; never called live in tests."""
    return await self._get(PATH_BANDWIDTH_TEST)

async_set_ddns(enabled) async

Enable/disable DDNS.

Source code in src/xiaomi_miwifi/client.py

async def async_set_ddns(self, enabled: bool) -> bool:
    """Enable/disable DDNS."""
    return self._ok(
        await self._get(f"{PATH_DDNS_SWITCH}?on={1 if enabled else 0}")
    )

async_set_dhcp(start, end, leasetime) async

Set DHCP pool start/end (last octet) and lease time (e.g. '720m').

Source code in src/xiaomi_miwifi/client.py

async def async_set_dhcp(self, start: int, end: int, leasetime: str) -> bool:
    """Set DHCP pool start/end (last octet) and lease time (e.g. '720m')."""
    params = urllib.parse.urlencode(
        {"start": str(start), "end": str(end),
         "leasetime": leasetime, "ignore": "0"}
    )
    return self._ok(await self._get(f"{PATH_SET_LAN_DHCP}?{params}"))

async_set_dmz(ip, mac='') async

Enable DMZ forwarding all WAN traffic to ip.

Source code in src/xiaomi_miwifi/client.py

async def async_set_dmz(self, ip: str, mac: str = "") -> bool:
    """Enable DMZ forwarding all WAN traffic to ``ip``."""
    params = urllib.parse.urlencode({"ip": ip, "mac": mac, "mode": "1"})
    return self._ok(await self._get(f"{PATH_SET_DMZ}?{params}"))

async_set_qos(enabled) async

Enable/disable QoS.

Source code in src/xiaomi_miwifi/client.py

async def async_set_qos(self, enabled: bool) -> bool:
    """Enable/disable QoS."""
    on = 1 if enabled else 0
    return self._ok(await self._get(f"{PATH_QOS_SWITCH}?on={on}"))

async_set_wifi_channel(wifi_index, channel) async

Set a band's channel (1=2.4G, 2=5G). "0" = auto. Disruptive.

Source code in src/xiaomi_miwifi/client.py

async def async_set_wifi_channel(self, wifi_index: int, channel: str) -> bool:
    """Set a band's channel (1=2.4G, 2=5G). "0" = auto. Disruptive."""
    return await self._set_wifi(wifi_index, {"channel": str(channel)})

async_set_wifi_enabled(ifname, enabled) async

Enable/disable a radio (wl0=5G, wl1=2.4G). Disruptive.

Source code in src/xiaomi_miwifi/client.py

async def async_set_wifi_enabled(self, ifname: str, enabled: bool) -> bool:
    """Enable/disable a radio (wl0=5G, wl1=2.4G). Disruptive."""
    path = PATH_WIFI_UP if enabled else PATH_WIFI_DOWN
    return self._ok(await self._post(path, {"ifname": ifname}))

async_set_wifi_txpwr(wifi_index, txpwr) async

Set a band's transmit power (max/mid/min). Disruptive.

Source code in src/xiaomi_miwifi/client.py

async def async_set_wifi_txpwr(self, wifi_index: int, txpwr: str) -> bool:
    """Set a band's transmit power (max/mid/min). Disruptive."""
    return await self._set_wifi(wifi_index, {"txpwr": txpwr})

async_unblock_device(mac) async

Restore a device's internet access (set_mac_filter wan=1).

Source code in src/xiaomi_miwifi/client.py

async def async_unblock_device(self, mac: str) -> bool:
    """Restore a device's internet access (set_mac_filter wan=1)."""
    return self._ok(
        await self._post(PATH_SET_MAC_FILTER, {"mac": mac, "wan": "1"})
    )

Aggregated router status from several LuCI endpoints.

Source code in src/xiaomi_miwifi/models.py

@dataclass
class MiWiFiStatus:
    """Aggregated router status from several LuCI endpoints."""

    online: bool
    hardware: str = ""
    firmware_version: str = ""
    serial: str = ""
    mac: str = ""
    client_count: int = 0
    mesh_node_count: int = 0
    cap_count: int = 0
    clients_24g: int = 0
    clients_5g: int = 0
    ssid_24g: str = ""
    ssid_5g: str = ""
    wan_ip: str = ""
    wan_gateway: str = ""
    wan_type: str = ""
    wan_uptime: int = 0
    wan_link: bool = False
    upload_speed: int = 0
    download_speed: int = 0
    update_available: bool = False
    wan_download_total: int = 0
    wan_upload_total: int = 0
    wan_max_download: int = 0
    wan_max_upload: int = 0
    led_on: bool = False
    channel_24g: int = 0
    channel_5g: int = 0
    encryption_24g: str = ""
    encryption_5g: str = ""
    txpwr_24g: str = ""
    txpwr_5g: str = ""
    country_code: str = ""
    qos_on: bool = False
    speedtest_download: float = 0.0
    speedtest_upload: float = 0.0
    speedtest_bandwidth: float = 0.0
    wan_dns: list[str] = field(default_factory=list)
    ddns_on: bool = False
    dmz_on: bool = False
    dmz_ip: str = ""
    port_forward_count: int = 0
    dhcp_leasetime: int = 0
    dhcp_limit: int = 0
    dhcp_start: int = 0
    timezone: str = ""
    rom_changelog: str = ""
    rom_latest_version: str = ""
    mesh_nodes: list[MeshNode] = field(default_factory=list)
    mode: int = 0
    lan_ports: list[bool] = field(default_factory=list)
    ipv6_on: bool = False

    @property
    def model(self) -> str:
        return friendly_model(self.hardware)

    @property
    def mode_name(self) -> str:
        return _MODE_NAMES.get(self.mode, f"Mode {self.mode}")

    @property
    def lan_ports_active(self) -> int:
        return sum(1 for up in self.lan_ports if up)