GEEKROS 并非传统意义上单一内核的“操作系统”，而是一个面向机器人与嵌入式场景、覆盖“上位机—ROS2—MCU”多层协同的开发平台与运行环境。它由 GeekROS 团队研发，目标是通过低代码、Python SDK、ROS2 集成和 MCU 控制能力，降低机器人、智能硬件和具身智能应用的开发门槛。

从架构看，GEEKROS 至少包含三层：

  • MCU 层：基于 STM32 等微控制器，负责电机、舵机、传感器、IO、PWM、UART、CAN 等硬件直接控制；
  • 上位机层：通过 USB 串口或网络与 MCU 通信，提供 Python SDK、图形化编程和 ROS2 桥接能力；
  • ROS2 层：运行于 Ubuntu 等上位机系统，承载导航、感知、控制、消息分发等机器人核心能力。

同时，GEEKROS 平台也在向具身智能语音交互方向演进，融合大模型与多模态能力，为机器人提供“智慧大脑”，强调毫秒级实时交互、自然语言理解与多设备适配。

因此，本报告中的“GEEKROS”应理解为：面向智能硬件的轻量运行时 + 机器人开发框架 + Python/MCU 协同生态。它适合用于教育机器人、服务机器人、机器狗、嵌入式控制板、语音交互终端以及需要 ROS2 与 MCU 联调的场景。

二、系统架构分析

2.1 MCU 层：硬件执行层

MCU 层通常运行在 STM32 等嵌入式芯片上，负责低延迟硬件控制。典型任务包括：

  • 读取编码器、IMU、超声波、红外等传感器数据；
  • 控制电机、舵机、LED、蜂鸣器、继电器等执行器；
  • 通过串口向上位机发送传感器帧；
  • 接收上位机下发的速度、角度、状态指令。

这一层通常使用 C/C++ 或 FreeRTOS 等实时环境开发，强调确定性、低功耗和硬件适配能力。

2.2 上位机层：Python 与 ROS2 协同层

上位机层是 GEEKROS 与 Python 关系最紧密的部分。它一方面通过串口协议与 MCU 通信，另一方面通过 rosbridge 或原生 ROS2 接口与机器人系统交互。

Python 在该层主要承担三类工作：

  1. 串口通信封装：将 MCU 的二进制/文本协议封装成结构化对象；
  2. 机器人业务逻辑：实现路径规划、状态机、任务调度、传感器融合；
  3. ROS2 桥接：通过 WebSocket 或 rosbridge 将 Python 应用接入 ROS2 话题、服务与动作。

2.3 云端与交互层：具身智能扩展

在更高层，GEEKROS 云平台引入了大模型、语音交互和多模态理解能力，可用于实现自然语言控制机器人、实时语音打断、任务切换和场景理解。

这一层并不一定运行在嵌入式设备上，而是通过 API 或云端服务与上位机协同。Python 可以作为云端交互客户端，将用户指令转化为机器人任务，再将机器人状态反馈给用户。

三、Python 在 GEEKROS 中的角色

Python 在 GEEKROS 生态中适合作为上位机控制语言、原型验证语言和系统集成语言。它不一定适合直接运行在资源受限的 MCU 上，但非常适合承担以下任务：

  • 封装串口协议，实现 MCU 控制接口；
  • 连接 ROS2 rosbridge，发布/订阅机器人状态；
  • 构建命令行调试工具，快速验证硬件接口；
  • 接入云端语音或大模型服务，实现自然语言控制；
  • 批量处理传感器日志、图像数据或实验记录；
  • 搭建轻量仿真或状态机，用于教学与算法验证。

对于嵌入式与机器人开发，Python 的价值主要在于缩短从硬件协议到业务逻辑的距离。相比 C/C++，它更适合快速搭建控制框架、调试通信协议和验证上层算法。

四、示例代码：基于串口协议的 MCU 控制封装

以下代码演示一个典型的 GEEKROS 上位机 Python 控制层。它不依赖真实硬件，而是通过模拟串口展示协议封装、命令发送、状态解析和异常处理思路。

import time
import struct
import threading
from dataclasses import dataclass, field
from typing import Optional, Callable


@dataclass
class RobotState:
    """机器人状态对象，用于上位机内部共享。"""
    timestamp: float = 0.0
    left_encoder: int = 0
    right_encoder: int = 0
    battery_voltage: float = 0.0
    imu_yaw: float = 0.0
    status_code: int = 0


class MockSerialPort:
    """模拟串口，用于离线验证协议逻辑。"""

    def __init__(self) -> None:
        self._buffer: bytearray = bytearray()
        self._running = True

    def write(self, data: bytes) -> None:
        self._buffer.extend(data)

    def read(self, size: int) -> bytes:
        if size <= 0:
            return b""
        result = bytes(self._buffer[:size])
        self._buffer = self._buffer[size:]
        return result

    def in_waiting(self) -> int:
        return len(self._buffer)

    def close(self) -> None:
        self._running = False


class GeekRosMCUClient:
    """
    GEEKROS 上位机 MCU 客户端示例。
    真实项目中，MockSerialPort 可替换为 serial.Serial。
    """

    HEADER = 0xAA
    CMD_PING = 0x01
    CMD_SET_SPEED = 0x02
    CMD_GET_STATE = 0x03

    def __init__(self, port, timeout: float = 0.5) -> None:
        self._port = port
        self._timeout = timeout
        self._lock = threading.Lock()
        self._state = RobotState()
        self._on_state: Optional[Callable = None

    def set_state_callback(self, handler: Callable[[RobotState], None]) -> None:
        self._on_state = handler

    def _build_frame(self, cmd: int, payload: bytes) -> bytes:
        length = len(payload)
        header = struct.pack("B", self.HEADER)
        meta = struct.pack("BB", cmd, length)
        checksum = (cmd + length + sum(payload)) & 0xFF
        return header + meta + payload + bytes([checksum])

    def _read_frame(self) -> Optional[bytes]:
        start = time.time()
        while time.time() - start < self._timeout:
            if self._port.in_waiting() >= 3:
                header = self._port.read(1)
                if header[0] != self.HEADER:
                    continue
                cmd, length = struct.unpack("BB", self._port.read(2))
                if self._port.in_waiting() < length + 1:
                    time.sleep(0.01)
                    continue
                payload = self._port.read(length)
                checksum_byte = self._port.read(1)
                expected = (cmd + length + sum(payload)) & 0xFF
                if checksum_byte[0] != expected:
                    raise ValueError("checksum error")
                return struct.pack("BB", cmd, length) + payload
        return None

    def ping(self) -> bool:
        with self._lock:
            frame = self._build_frame(self.CMD_PING, b"")
            self._port.write(frame)
            resp = self._read_frame()
            return resp is not None and resp[0] == self.CMD_PING

    def set_speed(self, left: float, right: float) -> None:
        payload = struct.pack("<ff", left, right)
        with self._lock:
            frame = self._build_frame(self.CMD_SET_SPEED, payload)
            self._port.write(frame)

    def get_state(self) -> RobotState:
        with self._lock:
            frame = self._build_frame(self.CMD_GET_STATE, b"")
            self._port.write(frame)
            resp = self._read_frame()
            if resp is None:
                raise TimeoutError("MCU state timeout")
            # 假设响应体紧跟在 cmd+len 之后
            payload = resp[2:]
            if len(payload) < 18:
                raise ValueError("invalid state payload")
            ts, le, re, volt, yaw, code = struct.unpack("<dii ff i", payload[:18])
            self._state = RobotState(
                timestamp=ts,
                left_encoder=le,
                right_encoder=re,
                battery_voltage=volt,
                imu_yaw=yaw,
                status_code=code,
            )
            if self._on_state:
                self._on_state(self._state)
            return self._state


def simulate_mcu_reply(port: MockSerialPort) -> None:
    """模拟 MCU 回复状态帧。"""
    time.sleep(0.05)
    cmd = 0x03
    ts = time.time()
    payload = struct.pack("<dii ff i", ts, 120, 118, 11.8, 1.23, 0)
    length = len(payload)
    checksum = (cmd + length + sum(payload)) & 0xFF
    frame = struct.pack("B", 0xAA) + struct.pack("BB", cmd, length) + payload + bytes([checksum])
    port.write(frame)


if __name__ == "__main__":
    port = MockSerialPort()
    client = GeekRosMCUClient(port)

    def on_state(state: RobotState) -> None:
        print(
            f"[State] t={state.timestamp:.3f} "
            f"L={state.left_encoder} R={state.right_encoder} "
            f"V={state.battery_voltage:.2f} yaw={state.imu_yaw:.2f} "
            f"code={state.status_code}"
        )

    client.set_state_callback(on_state)

    print("ping:", client.ping())
    client.set_speed(0.3, 0.3)

    threading.Thread(target=simulate_mcu_reply, args=(port,), daemon=True).start()
    state = client.get_state()
    print("battery:", state.battery_voltage)

代码解析

  1. 协议帧设计
    示例采用 HEADER + CMD + LENGTH + PAYLOAD + CHECKSUM 的简单帧结构。嵌入式串口通信中，帧头用于同步，长度字段用于分包，校验和用于防止数据错误。

  2. 命令封装
    set_speed 将左右轮速度打包为二进制 payload；get_state 发送查询命令并等待 MCU 返回。真实项目中，CMD 编号、字段顺序和字节序需要与 MCU 固件严格一致。

  3. 线程安全
    _lock 保证同一时刻只有一个线程发送并读取响应，避免串口读写竞争。对于机器人上位机，串口通常属于共享资源，必须串行访问。

  4. 状态回调
    set_state_callback 允许上层业务订阅机器人状态，适合驱动界面刷新、日志记录或 ROS2 话题发布。

  5. 可替换串口实现
    MockSerialPort 仅用于演示。实际接入 GEEKROS MCU 时，可替换为 serial.Serial，并增加重连、超时、断帧恢复等机制。

五、示例代码：ROS2 rosbridge 状态桥接

如果上位机运行在支持 ROS2 的环境中，Python 可以通过 rosbridge WebSocket 将 MCU 状态发布到 ROS2 话题。

import json
import time
import threading

try:
    import websocket  # pip install websocket-client
except ImportError:
    websocket = None


class RosBridgePublisher:
    """通过 rosbridge 将机器人状态发布到 ROS2 话题。"""

    def __init__(self, ws_url: str = "ws://localhost:9090", topic: str = "/robot/state") -> None:
        self.ws_url = ws_url
        self.topic = topic
        self._ws = None
        self._connected = threading.Event()

    def connect(self) -> None:
        if websocket is None:
            raise RuntimeError("websocket-client is not installed")
        self._ws = websocket.WebSocketApp(
            self.ws_url,
            on_open=lambda _: self._connected.set(),
        )
        threading.Thread(target=self._ws.run_forever, daemon=True).start()
        if not self._connected.wait(timeout=5):
            raise TimeoutError("rosbridge connection timeout")

    def publish(self, msg: dict) -> None:
        if self._ws is None:
            raise RuntimeError("not connected")
        frame = {
            "op": "publish",
            "topic": self.topic,
            "msg": msg,
        }
        self._ws.send(json.dumps(frame))

    def close(self) -> None:
        if self._ws:
            self._ws.close()


if __name__ == "__main__":
    pub = RosBridgePublisher()
    pub.connect()
    for i in range(5):
        pub.publish({
            "header": {"stamp": {"secs": int(time.time()), "nsecs": 0}},
            "left_encoder": 100 + i,
            "right_encoder": 98 + i,
            "battery_voltage": 11.8,
        })
        time.sleep(0.2)
    pub.close()

代码解析

  • rosbridge 通过 WebSocket 提供 JSON 格式的发布/订阅能力，适合 Python 快速接入 ROS2；
  • op: publish 表示发布消息，topic 指定话题名，msg 为消息内容；
  • 实际项目中应定义明确的 ROS2 消息类型，例如 sensor_msgs 或自定义 msg；
  • 该示例适合原型验证，生产环境建议使用原生 ROS2 Python 客户端以获得更低延迟和更强类型约束。

六、示例代码：自然语言任务路由

GEEKROS 平台强调实时交互与具身智能能力，上层可通过云端服务将自然语言指令转换为机器人动作。

from dataclasses import dataclass
from typing import Optional


@dataclass
class RobotCommand:
    action: str
    params: dict


class IntentRouter:
    """
    简易意图路由器。
    真实项目中可替换为 GEEKROS 云平台或大模型 API。
    """

    RULES = [
        (["前进", "向前", "go"], "move_forward"),
        (["后退", "返回", "back"], "move_backward"),
        (["左转", "left"], "turn_left"),
        (["右转", "right"], "turn_right"),
        (["停止", "停", "stop"], "stop"),
    ]

    def parse(self, text: str) -> Optional[RobotCommand]:
        text = text.strip().lower()
        for keywords, action in self.RULES:
            if any(k in text for k in keywords):
                return RobotCommand(action=action, params={"source": "voice"})
        return None


class RobotController:
    def __init__(self, mcu_client) -> None:
        self._mcu = mcu_client
        self._router = IntentRouter()

    def execute_voice(self, text: str) -> None:
        cmd = self._router.parse(text)
        if cmd is None:
            print("未识别指令:", text)
            return

        print("执行动作:", cmd.action, cmd.params)
        if cmd.action == "move_forward":
            self._mcu.set_speed(0.3, 0.3)
        elif cmd.action == "move_backward":
            self._mcu.set_speed(-0.3, -0.3)
        elif cmd.action == "turn_left":
            self._mcu.set_speed(-0.2, 0.2)
        elif cmd.action == "turn_right":
            self._mcu.set_speed(0.2, -0.2)
        elif cmd.action == "stop":
            self._mcu.set_speed(0.0, 0.0)

代码解析

  • IntentRouter 负责将自然语言映射为结构化动作；
  • RobotController 将动作进一步转换为 MCU 速度指令；
  • 该结构可以平滑替换为云端大模型接口，实现更复杂的任务规划；
  • 对于实时交互场景，需要重点控制端到端延迟，避免指令响应过慢影响体验。

七、技术亮点

  1. MCU—上位机—ROS2 分层清晰
    GEEKROS 把硬件控制、业务逻辑和机器人中间件分层，便于不同角色协同开发。嵌入式工程师负责 MCU，算法工程师负责 ROS2，应用工程师负责 Python 上层逻辑。

  2. Python 可作为快速接入层
    通过串口封装、rosbridge 桥接和云端 API 接入，Python 能快速连接硬件、机器人系统与交互服务，适合教学、原型验证和中小型智能硬件项目。

  3. 低代码与 SDK 并行
    GEEKROS 同时提供图形化/低代码方式和 Python SDK，降低了入门门槛，也保留了代码级可扩展性。

  4. 面向实时交互演进
    平台从机器人开发工具扩展到具身智能语音交互平台，强调毫秒级响应、自然语言理解和多模态能力，符合智能硬件从“可控制”走向“可对话、可协作”的趋势。

  5. 适合二次开发
    开发者可以基于串口协议扩展传感器驱动，基于 ROS2 扩展导航与感知模块，基于云平台扩展语音交互与任务编排。

八、局限与改进方向

  • 公开资料中对 GEEKROS 的完整协议、SDK 接口和开源仓库细节披露有限，实际开发应以官方文档和源码为准；
  • Python 不适合直接承担高实时 MCU 控制，应主要用于上位机、桥接层和业务编排；
  • rosbridge 适合快速验证，但高频率控制场景建议使用原生 ROS2 通信；
  • 自然语言控制需要增加权限校验、安全围栏和失败回退机制，避免误触发危险动作；
  • 工程化项目中应补充日志系统、配置管理、单元测试、异常恢复和设备健康监控。

九、结论

GEEKROS 的价值在于把机器人开发中常见的“硬件控制—ROS2 通信—上层交互”三层能力整合到一个更友好的开发框架中。Python 在其中主要承担上位机控制、协议桥接、快速原型和交互编排的角色。对于开发者而言，理解 GEEKROS 的关键不是把它简单视为一个传统操作系统，而是把它看作面向智能硬件的协同运行时与开发平台：MCU 负责执行，ROS2 负责机器人能力，Python 负责连接与扩展，云端交互负责提升智能水平。通过上述代码示例可以看出，基于串口帧封装、rosbridge 发布和意图路由，可以快速构建一个可演进的原型系统，并进一步接入更完整的 GEEKROS 生态能力。

在这里插入图片描述

# 机器人 # 单片机 # 嵌入式硬件
Logo
openEuler 社区

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目，面向数字基础设施四大核心场景（服务器、云计算、边缘计算、嵌入式），全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐

cover

华为昇腾NPU开发实战：YOLOv11从GPU到Atlas 300I Duo全流程部署

avatar openEuler 社区
cover

电子信息类专业考研专业课考哪些课程

avatar openEuler 社区
cover

用C语言手写微型OS，从引导到进程调度，新手也能跑通

avatar openEuler 社区