本地运行

micropython 支持多平台，可以在本地运行测试APP基础代码逻辑。

使用系统

cat /etc/issue
Ubuntu 22.04 LTS \n \l

uname -a
Linux VM-0-9-ubuntu 5.15.0-106-generic

下载源码

git clone --branch v1.24.1 --depth 1 https://github.com/micropython/micropython.git
git submodule update --init --recursive

安装依赖

sudo apt update
sudo apt install -y build-essential git pkg-config \
    libffi-dev libssl-dev libbz2-dev liblzma-dev \
    libreadline-dev libsqlite3-dev libgdbm-dev \
    libncurses5-dev zlib1g-dev libmpdec-dev \
    libmbedtls-dev libdb5.3-dev uuid-dev

编译构建

cd micropython/ports/unix
make clean
make -j$(nproc) MICROPY_PY_USER_CMODULE=0


# LINK build-standard/micropython
#   text       data     bss     dec     hex filename
# 764607      69328    7088  841023   cd53f build-standard/micropython

查看版本

./build-standard/micropython --version

# MicroPython v1.24.1 on 2025-09-09; linux [GCC 11.2.0] version

准备脚本

准备hello.py，脚本内容如下

echo 'print("hello")' > hello.py

运行脚本

./build-standard/micropython hello.py

hello

Cassiablue API Mock

需要调试Cassiablue API相关的代码，可以参考下面使用网关RESTful API Mock Cassiablue API实现，用于快速测试基本逻辑功能。

备注

  • 仅仅用于本地调试代码基本逻辑，勿用于压力测和稳定性测试

  • 脚本实际运行结果，请以M2000网关内运行为准

  1"""
  2Module name: cassiablue.py
  3Purpose: Local MicroPython debugging mock. Exposes the same high-level API as the real gateway by forwarding calls to its REST + Server-Sent-Events interface, so you can develop and test your MicroPython logic on a PC without flashing firmware.
  4MicroPython compatibility: v1.24.1
  5Important: Do **not** copy this file into the gateway itself—the gateway already contains a native C implementation. This module is **only** for convenient desktop debugging and has no performance guarantees.
  6Usage: Change the variable `GATEWAY` to the actual IP address of your gateway.
  7"""
  8
  9GATEWAY = ""
 10
 11import json
 12import asyncio
 13
 14from cassia_log import get_logger
 15import aiohttp
 16
 17try:
 18    from typing import Optional
 19except ImportError:
 20    pass
 21
 22
 23log = get_logger("__mock_cassiablue__")
 24
 25
 26class AsyncQueue:
 27    def __init__(self):
 28        self._buffer = []
 29        self._flag = asyncio.ThreadSafeFlag()
 30
 31    def put_nowait(self, item):
 32        self._buffer.append(item)
 33        self._flag.set()
 34
 35    async def get(self):
 36        while not self._buffer:
 37            await self._flag.wait()
 38        return self._buffer.pop(0)
 39
 40
 41async def send_cmd(url, method="GET", query=None, body=None):
 42    global GATEWAY
 43    url = f"http://{GATEWAY}{url}"
 44
 45    print("send cmd start:", method, url, query, body)
 46
 47    async with aiohttp.ClientSession() as session:
 48        if body is not None:
 49            async with session.request(
 50                method=method, url=url, json=json.loads(body)
 51            ) as resp:
 52                if resp.status == 200:
 53                    text = await resp.text()
 54                    return True, text
 55                else:
 56                    text = await resp.text()
 57                    return False, text
 58        else:
 59            async with session.request(method=method, url=url) as resp:
 60                if resp.status == 200:
 61                    text = await resp.text()
 62                    return True, text
 63                else:
 64                    text = await resp.text()
 65                    return False, text
 66
 67
 68# address is a string like "00:11:22:33:44:55"
 69# params is a json string like '{"param1": "value1", "param2": "value2"}'
 70async def connect(addr, params=None):
 71    if not addr:
 72        raise ValueError("Address cannot be empty")
 73    url = "/gap/nodes/{}/connection".format(addr)
 74    return await send_cmd(url, "POST", None, body=params)
 75
 76
 77# address is a string like "00:11:22:33:44:55"
 78async def disconnect(addr):
 79    if not addr:
 80        raise ValueError("Address cannot be empty")
 81    url = "/gap/nodes/{}/connection".format(addr)
 82    return await send_cmd(url, "DELETE")
 83
 84
 85async def get_connected_devices():
 86    return await send_cmd("/gap/nodes", "GET")
 87
 88
 89async def gatt_discover(addr):
 90    if not addr:
 91        raise ValueError("Address cannot be empty")
 92    url = "/gatt/nodes/{}/services/characteristics/descriptors".format(addr)
 93    return await send_cmd(url, "GET")
 94
 95
 96async def gatt_read(addr, handle):
 97    if not addr or not handle:
 98        raise ValueError("Address and handle cannot be empty")
 99    url = "/gatt/nodes/{}/handle/{}/value".format(addr, handle)
100    return await send_cmd(url, "GET")
101
102
103async def gatt_write(addr, handle, value):
104    if not addr or not handle or value is None:
105        raise ValueError("Address, handle, and value cannot be empty")
106    url = "/gatt/nodes/{}/handle/{}/value/{}".format(addr, handle, value)
107    return await send_cmd(url, "GET")
108
109
110class SSEClient:
111    def __init__(
112        self,
113        host: str,
114        path: str,
115        reconnect_delay: int = 3,
116    ):
117        self.log = get_logger(self.__class__.__name__)
118        self.host = host
119        self.path = path
120        self.reconnect_delay = reconnect_delay
121        self.running = True
122        self.queue = AsyncQueue()
123        self.reader = None
124        self.writer = None
125
126    async def connect(self):
127        self.log.info(f"connect to sse start: {self.host}{self.path}")
128        self.reader, self.writer = await asyncio.open_connection(self.host, 80)
129        req = (
130            f"GET {self.path} HTTP/1.1\r\n"
131            f"Accept: text/event-stream\r\n"
132            f"Host: {self.host}\r\n"
133            f"Connection: keep-alive\r\n"
134            "\r\n"
135        )
136        self.writer.write(req.encode())
137        await self.writer.drain()
138        self.log.info(f"connect to sse ok: {self.host}{self.path}")
139
140    async def co_read(self):
141        while self.running:
142            try:
143                line = await self.reader.readline()
144
145                if not line:
146                    raise OSError("connection closed")
147
148                line = line.decode().strip()
149
150                if not line:
151                    continue
152
153                log.debug("raw line:", line)
154                if not (line[0] in ("{", "[") or line.startswith("data: {")):
155                    continue
156
157                log.debug("raw line:", line)
158                line = line.replace("data: ", "")
159                line = line.replace("\n", "")
160                line = line.replace("\r", "")
161                line = line.replace("\r\n", "")
162
163                try:
164                    data = json.loads(line)
165
166                    if "bdaddrs" in data:
167                        data["bdaddr"] = data["bdaddrs"][0]["bdaddr"]
168                        data["bdaddrType"] = data["bdaddrs"][0]["bdaddrType"]
169                        del data["bdaddrs"]
170
171                    self.queue.put_nowait(data)
172                except Exception as e:
173                    self.log.info("parse data error:", e, line)
174
175            except Exception as e:
176                self.log.error("sse disconnected:", e)
177                await asyncio.sleep(self.reconnect_delay)
178                self.log.info("reconnecting...")
179
180    async def stop(self):
181        self.running = False
182
183
184#######################
185# scan sse
186#######################
187
188scan_sse_client: Optional[SSEClient] = None
189
190
191class BLEScanResult:
192    def __aiter__(self):
193        return self
194
195    async def __anext__(self):
196        global scan_sse_client
197        item = await scan_sse_client.queue.get()
198        return item
199
200
201def scan_result():
202    return BLEScanResult()
203
204
205async def start_scan(query: str = None):
206    global scan_sse_client
207
208    qs = ""
209    if query is None:
210        qs = "event=1"
211    else:
212        if "event=1" not in query:
213            qs = qs + "event=1&"
214        qs = qs + query
215
216    log.info("start scan:", qs)
217
218    scan_sse_client = SSEClient(
219        host=GATEWAY,
220        path=f"/gap/nodes?{qs}",
221    )
222
223    await scan_sse_client.connect()
224
225    asyncio.create_task(scan_sse_client.co_read())
226
227    return True, "OK"
228
229
230#######################
231# notify sse
232#######################
233
234notify_sse_client: Optional[SSEClient] = None
235
236
237class BLENotifyResult:
238    def __aiter__(self):
239        return self
240
241    async def __anext__(self):
242        global notify_sse_client
243        item = await notify_sse_client.queue.get()
244        return item
245
246
247def notify_result():
248    return BLENotifyResult()
249
250
251async def start_recv_notify(query: str = None):
252    global notify_sse_client
253
254    qs = ""
255    if query is None:
256        qs = "event=1"
257    else:
258        if "event=1" not in query:
259            qs = qs + "event=1&"
260        qs = qs + query
261
262    log.info("start notify:", qs)
263
264    notify_sse_client = SSEClient(
265        host=GATEWAY,
266        path=f"/gatt/nodes?{qs}",
267    )
268
269    await notify_sse_client.connect()
270
271    asyncio.create_task(notify_sse_client.co_read())
272
273    return True, "OK"
274
275
276#######################
277# state sse
278#######################
279
280state_sse_client: Optional[SSEClient] = None
281
282
283class BLEConnectionResult:
284    def __aiter__(self):
285        return self
286
287    async def __anext__(self):
288        global state_sse_client
289        item = await state_sse_client.queue.get()
290        return item
291
292
293def connection_result():
294    return BLEConnectionResult()
295
296
297async def start_recv_connection_state():
298    global state_sse_client
299
300    log.info("start state")
301
302    state_sse_client = SSEClient(
303        host=GATEWAY,
304        path=f"/management/nodes/connection-state",
305    )
306
307    await state_sse_client.connect()
308    asyncio.create_task(state_sse_client.co_read())
309
310    return True, "OK"
311
312
313def set_gateway(ip: str):
314    global GATEWAY
315    GATEWAY = ip