Run Locally
micropython supports multiple platforms and can be used to run and test the basic APP logic locally.
System Information
cat /etc/issue
Ubuntu 22.04 LTS \n \l
uname -a
Linux VM-0-9-ubuntu 5.15.0-106-generic
Download Source Code
git clone --branch v1.24.1 --depth 1 https://github.com/micropython/micropython.git
git submodule update --init --recursive
Install Dependencies
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
Compile and Build
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
Check Version
./build-standard/micropython --version
# MicroPython v1.24.1 on 2025-09-09; linux [GCC 11.2.0] version
Prepare Script
Prepare hello.py with the following script content:
echo 'print("hello")' > hello.py
Run the Script
./build-standard/micropython hello.py
hello
Cassiablue API Mock
To debug code related to the Cassiablue API, you can refer to the following implementation using the gateway RESTful API to mock the Cassiablue API, which allows for quick testing of basic logic and functionality.
Note
仅用于本地调试代码基本逻辑，勿用于压力测试和稳定性测试
The actual script execution results should be based on operation within the M2000 gateway.
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
To debug code related to the Cassiamqtt API , you can refer to the following implementation using the mqtt_as API to mock the Cassiamqtt API , which allows for quick testing of basic logic and functionality.
Note
仅用于本地调试代码基本逻辑，勿用于压力测试和稳定性测试
The actual script execution results should be based on operation within the M2000 gateway.
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