本地运行
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
Cassiamqtt API Mock
需要调试 Cassiamqtt API 相关的代码，可以参考下面使用 mqtt_as 实现，用于快速测试基本逻辑功能。
备注
仅用于本地调试代码基本逻辑，勿用于压力测试和稳定性测试
脚本实际运行结果，请以M2000网关内运行为准
1"""
2Module name: cassiamqtt.py
3Purpose: Local MicroPython debugging mock. Exposes the same high-level API as the real gateway by forwarding calls to mqtt_as 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.
6"""
7
8import asyncio
9
10from mqtt_as import MQTTClient
11from cassia_log import get_logger
12
13
14def _parse_uri(uri):
15 try:
16 host_port = uri.split("://", 1)[1]
17 host, port = host_port.rsplit(":", 1)
18 return {
19 "host": host,
20 "port": port,
21 }
22 except (IndexError, ValueError):
23 return None
24
25
26class CassiaMQTTClient:
27 def __init__(self, uri, username=None, password=None, client_id=None):
28 self.log = get_logger("__mock_cassiamqtt__")
29
30 addr = _parse_uri(uri)
31 if addr is None:
32 self.log.warn(f"parse uri failed: {uri}")
33 return
34
35 config = {
36 "client_id": client_id,
37 "server": addr.get("host", None),
38 "port": addr.get("port", None),
39 "user": username,
40 "password": password,
41 "keepalive": 60,
42 "ping_interval": 0,
43 "ssl": False,
44 "ssl_params": {},
45 "response_time": 10,
46 "clean_init": True,
47 "clean": True,
48 "max_repubs": 4,
49 "will": None,
50 "subs_cb": lambda *_: None,
51 "ssid": None,
52 "wifi_pw": None,
53 "queue_len": 64,
54 "gateway": False,
55 "mqttv5": False,
56 "mqttv5_con_props": None,
57 }
58
59 MQTTClient.DEBUG = True
60 self.client = MQTTClient(config=config)
61
62 async def __aenter__(self):
63 while True:
64 try:
65 self.log.info("connect start...")
66 await self.client.connect()
67 self.log.info("connect ok")
68 break
69 except Exception as e:
70 self.log.error(f"connect failed: {e}, wait next retry...")
71 await asyncio.sleep(3)
72 return self
73
74 async def __aexit__(self, exc_type, exc_val, exc_tb):
75 self.log.info("disconnect start")
76 await self.client.disconnect()
77 self.client = None
78 self.log.info("disconnect ok")
79
80 def __aiter__(self):
81 return self
82
83 async def __anext__(self):
84 if self.client is None:
85 raise StopAsyncIteration
86
87 (topic, msg, retained) = await self.client.queue.__anext__()
88
89 return {
90 "topic": topic.decode(),
91 "payload": msg.decode(),
92 "qos": 0,
93 }
94
95 async def publish(self, topic, payload: str, qos=0, retain=False):
96 try:
97 self.log.info(f"pub start: {topic} {qos} {retain} {payload[:32]}...")
98 await self.client.publish(topic, payload, qos=qos, retain=retain)
99 self.log.info(f"pub ok")
100 return True, ""
101 except Exception as e:
102 self.log.warn(f"pub failed: {e}")
103 return False, e
104
105 async def subscribe(self, topic, qos=0):
106 try:
107 self.log.info(f"sub start: {topic} {qos}")
108 await self.client.subscribe(topic, qos=qos)
109 self.log.info(f"sub ok")
110 return True, ""
111 except Exception as e:
112 self.log.warn(f"sub failed: {e}")
113 return False, e
114
115 async def unsubscribe(self, topic):
116 try:
117 self.log.info(f"unsub start: {topic}")
118 await self.client.unsubscribe(topic)
119 self.log.info(f"unsub ok")
120 return True, ""
121 except Exception as e:
122 self.log.warn(f"unsub failed: {e}")
123 return False, e