#!/usr/bin/env python3 """ Маркет-Пилот · Bluetooth Wi-Fi онбординг безэкранной коробки (стандарт Improv Wi-Fi over BLE). Клиент с телефона (Android Chrome/Edge — прямо в браузере через Web Bluetooth; iPhone — приложение Bluefy) открывает страницу онбординга, находит коробку по Bluetooth, вводит имя Wi-Fi и пароль -> коробка реально подключается (nmcli/NetworkManager) и сообщает статус. Реализация: BLE GATT-сервер на BlueZ через библиотеку `bless`. Протокол Improv: service 00467768-6228-2272-4663-277478268000 state ...268001 (read/notify) Current State error ...268002 (read/notify) Error State rpc ...268003 (write) RPC Command (0x01 = Send Wi-Fi: ssid+password) result ...268004 (read/notify) RPC Result (после provisioned) caps ...268005 (read) Capabilities ВАЖНО (в отличие от reference pyImprov, где connect — заглушка): здесь реальный nmcli-коннект. """ import asyncio import logging import subprocess import sys from bless import ( # type: ignore BlessServer, GATTCharacteristicProperties, GATTAttributePermissions, ) logging.basicConfig(level=logging.INFO, format="%(asctime)s mp-improv %(levelname)s %(message)s") log = logging.getLogger("mp-improv") SVC = "00467768-6228-2272-4663-277478268000" CH_STATE = "00467768-6228-2272-4663-277478268001" CH_ERROR = "00467768-6228-2272-4663-277478268002" CH_RPC = "00467768-6228-2272-4663-277478268003" CH_RESULT = "00467768-6228-2272-4663-277478268004" CH_CAPS = "00467768-6228-2272-4663-277478268005" # Improv states ST_AUTHORIZED = 0x02 ST_PROVISIONING = 0x03 ST_PROVISIONED = 0x04 # Improv errors ERR_NONE = 0x00 ERR_UNABLE_TO_CONNECT = 0x03 # RPC commands CMD_SEND_WIFI = 0x01 DEVICE_NAME = "MARKET-PILOT" def _checksum(payload: bytes) -> int: return sum(payload) & 0xFF def _encode_result(strings): """Improv RPC result: [cmd][len][ (str-len)(str) ... ][checksum] — список строк (напр. URL).""" body = bytes([CMD_SEND_WIFI]) data = b"" for s in strings: b = s.encode() data += bytes([len(b)]) + b body += bytes([len(data)]) + data return body + bytes([_checksum(body)]) def nmcli_connect(ssid: str, password: str) -> bool: """Реальное подключение к Wi-Fi через NetworkManager. True при успехе.""" log.info("nmcli connect ssid=%r", ssid) try: subprocess.run(["nmcli", "radio", "wifi", "on"], check=False, timeout=15) subprocess.run(["nmcli", "device", "wifi", "rescan"], check=False, timeout=25) r = subprocess.run( ["nmcli", "-w", "40", "device", "wifi", "connect", ssid, "password", password], capture_output=True, text=True, timeout=60, ) ok = r.returncode == 0 log.info("nmcli rc=%s out=%s err=%s", r.returncode, r.stdout.strip(), r.stderr.strip()) return ok except Exception as e: # noqa log.error("nmcli failed: %s", e) return False class Improv: def __init__(self): self.server: BlessServer | None = None self.state = ST_AUTHORIZED self.error = ERR_NONE self._rpc_buf = bytearray() async def set_char(self, uuid, value: bytes): self.server.get_characteristic(uuid).value = bytearray(value) # bless 0.3.0: update_value — синхронный (возвращает bool), НЕ await self.server.update_value(SVC, uuid) async def push_state(self): await self.set_char(CH_STATE, bytes([self.state])) async def push_error(self): await self.set_char(CH_ERROR, bytes([self.error])) def read_request(self, characteristic, **_): return characteristic.value def write_request(self, characteristic, value, **_): # RPC Command write — парсим [cmd][len][data...][checksum] if characteristic.uuid.lower() != CH_RPC.lower(): return data = bytes(value) log.info("rpc write %s", data.hex()) asyncio.create_task(self._handle_rpc(data)) async def _handle_rpc(self, data: bytes): if len(data) < 3 or data[0] != CMD_SEND_WIFI: return ln = data[1] body = data[2:2 + ln] # body = [ssid-len][ssid][pwd-len][pwd] try: sl = body[0] ssid = body[1:1 + sl].decode() pl = body[1 + sl] pwd = body[2 + sl:2 + sl + pl].decode() except Exception as e: # noqa log.error("bad wifi payload: %s", e) self.error = ERR_UNABLE_TO_CONNECT await self.push_error() return self.state = ST_PROVISIONING self.error = ERR_NONE await self.push_state() await self.push_error() ok = await asyncio.get_event_loop().run_in_executor(None, nmcli_connect, ssid, pwd) if ok: self.state = ST_PROVISIONED await self.push_state() # result: можно отдать URL статуса; пусто = просто успех await self.set_char(CH_RESULT, _encode_result([])) log.info("PROVISIONED ok") else: self.state = ST_AUTHORIZED self.error = ERR_UNABLE_TO_CONNECT await self.push_state() await self.push_error() log.info("provision failed") async def run(self): self.server = BlessServer(name=DEVICE_NAME) self.server.read_request_func = self.read_request self.server.write_request_func = self.write_request await self.server.add_new_service(SVC) notify = GATTCharacteristicProperties.read | GATTCharacteristicProperties.notify write = GATTCharacteristicProperties.write read = GATTCharacteristicProperties.read perm_r = GATTAttributePermissions.readable perm_w = GATTAttributePermissions.writeable await self.server.add_new_characteristic(SVC, CH_CAPS, read, bytearray([0x00]), perm_r) await self.server.add_new_characteristic(SVC, CH_STATE, notify, bytearray([self.state]), perm_r) await self.server.add_new_characteristic(SVC, CH_ERROR, notify, bytearray([self.error]), perm_r) await self.server.add_new_characteristic(SVC, CH_RPC, write, bytearray(), perm_w) await self.server.add_new_characteristic(SVC, CH_RESULT, notify, bytearray(), perm_r) await self.server.start() log.info("Improv BLE advertising as %s (service %s)", DEVICE_NAME, SVC) while True: await asyncio.sleep(3600) def have_internet() -> bool: r = subprocess.run(["bash", "-c", "ip route get 1.1.1.1 >/dev/null 2>&1 && timeout 3 curl -fsS https://1.1.1.1 >/dev/null 2>&1"], capture_output=True) return r.returncode == 0 if __name__ == "__main__": # Если уже есть интернет (Ethernet/Wi-Fi) — онбординг не нужен, но advertising всё равно # полезен для смены сети; держим всегда. (Можно гейтить через --only-if-offline.) if "--only-if-offline" in sys.argv and have_internet(): log.info("internet present — improv onboarding not needed, exiting 0") sys.exit(0) asyncio.run(Improv().run())