第1章 Python基础语法:场景驱动与”够用”哲学#
方法论核心:每个语法点都绑定工程场景,强调类型提示对AI协作的巨大价值
开篇:为什么嵌入式工程师需要重新学Python基础语法?#
如果你是嵌入式工程师,你可能已经学过C语言,甚至对指针、内存管理、位操作了如指掌。但当你第一次打开Python教程,看到那些”Hello World”和”猜数字游戏”时,内心可能会产生一种微妙的抗拒——“这些跟我写驱动、调协议有什么关系?”
这种抗拒是有道理的。
传统Python教程往往从”通用编程”的角度出发,教你写网站、做数据分析、搞机器学习。但嵌入式工程师的日常完全不同:你要解析串口发来的二进制协议、要读写寄存器、要把十六进制的传感器数据转换成人类可读的温度值、要把配置参数保存到JSON文件里方便产线工人修改。
这就是本章存在的意义。
我们不会按照传统教程的方式罗列Python语法——那种方式会让你觉得”学了也没用”。相反,我们采用场景驱动的方法:每一个语法点,都会绑定一个真实的嵌入式工程场景。你学的不是”Python语法”,而是”用Python解决嵌入式问题的工具箱”。
同时,本章会反复强调一个现代理念:类型提示(Type Hints)。在AI时代,你写的代码不仅要给编译器看、给自己看,还要给AI看。类型提示就像一份”数据流说明书”,它告诉AI(以及未来的你):这个变量是什么类型、这个函数接收什么参数、返回什么结果。这不是可选的装饰,而是AI时代工程师的核心素养。
让我们开始。
1.1 变量、数据类型与类型提示 (Type Hints)#
1.1.1 变量:给数据贴标签#
在C语言中,声明变量需要指定类型:
int temperature = 25;
float voltage = 3.3;
char buffer[64];cPython简化了这个过程——你只需要给变量起个名字,然后赋值:
temperature = 25
voltage = 3.3
buffer = b'\x01\x02\x03' # 字节流pythonPython会自动推断变量的类型。这叫做动态类型,它是Python的核心特性之一。
嵌入式场景:假设你从串口读取了一个传感器的原始值:
raw_value = ser.read(2) # 读取2个字节pythonraw_value的类型是bytes(字节序列),Python自动帮你记住了这一点。你不需要像C那样手动管理unsigned char buffer[2]。
1.1.2 核心数据类型:嵌入式工程师最常用的#
| 类型 | 含义 | 嵌入式场景示例 |
|---|---|---|
int | 整数 | 寄存器地址、传感器原始值、波特率 |
float | 浮点数 | 温度值、电压值、PID参数 |
bool | 布尔值 | 设备状态(在线/离线)、开关量 |
str | 字符串 | 设备名称、日志信息、配置文件内容 |
bytes | 字节序列 | 串口数据、网络包、二进制协议帧 |
bytearray | 可变字节序列 | 需要原地修改的协议缓冲区 |
None | 空值 | 表示”没有数据”或”未初始化” |
关键区别:str是人类可读的文本,bytes是机器可读的二进制数据。在嵌入式开发中,串口、网络、文件读写的底层数据都是bytes,你需要学会在两者之间转换:
# 字符串 → 字节(发送数据前)
text = "Hello"
data = text.encode('utf-8') # b'Hello'
# 字节 → 字符串(解析响应后)
response = b'OK\x00\x01'
status = response[:2].decode('ascii') # 'OK'python1.1.3 类型提示:给AI看的代码契约#
Python 3.5引入了类型提示(Type Hints),它允许你”标注”变量和函数的类型,但不会强制检查。这不是语法错误,而是一种文档——它告诉阅读代码的人(和AI):“这个变量应该是这个类型”。
为什么类型提示对嵌入式工程师特别重要?
- 字节流的边界:串口数据是
bytes,但你需要知道它是2字节还是4字节、是大端还是小端。类型提示帮你明确这个边界。 - 寄存器值的范围:一个寄存器可能是8位(0-255)或16位(0-65535)。类型提示帮你记录这个约束。
- AI协作的接口:当你把代码交给AI优化或调试时,类型提示就像一份”需求文档”,让AI准确理解你的意图。
基本语法:
# 变量类型提示
temperature: float = 25.0
raw_data: bytes = b'\x01\x02\x03'
device_name: str = "STM32_Board_01"
# 函数类型提示
def parse_sensor_data(raw: bytes) -> float:
"""将2字节原始数据转换为温度值"""
# ...处理逻辑
return temperature
def send_command(port: str, command: bytes) -> bool:
"""发送串口命令,返回是否成功"""
# ...发送逻辑
return successpython注意:Python解释器不会因为类型错误而报错。类型提示的真正价值是:
- 让IDE(如VS Code、PyCharm)提供更智能的代码补全和错误提示
- 让静态检查工具(如mypy)在运行前发现类型错误
- 让AI工具(如Copilot、ChatGPT)更准确地理解你的代码意图
- 让未来的你(或同事)在三个月后还能看懂这段代码
嵌入式实践建议:
# 明确字节流的长度
def read_register(address: int) -> bytes:
"""读取寄存器,返回4字节数据"""
...
# 明确返回值的结构
def get_device_status() -> dict[str, int]:
"""返回设备状态字典,包含 temperature, voltage, status_code"""
...
# 使用 TypeAlias 定义复杂类型(Python 3.10+)
from typing import TypeAlias
RegisterMap: TypeAlias = dict[str, int] # 寄存器名称 → 值的映射python1.2 字符串、列表、字典、元组——工程数据的常见载体#
1.2.1 字符串 (str):日志与协议文本#
字符串是文本数据的载体。在嵌入式场景中,它常用于:
- 日志输出:
f"温度={temperature}℃,电压={voltage}V" - 文本协议解析:AT命令、文本格式的传感器数据
- 配置文件:INI、JSON中的文本内容
核心操作:
# f-string 格式化(Python 3.6+,推荐)
temperature = 25.3
log_msg = f"[{datetime.now()}] 温度={temperature:.1f}℃"
# 字符串分割(解析文本协议)
response = "+CIPSTATUS:0,CONNECTED,TCP,192.168.1.100,8080"
parts = response.split(',')
status = parts[1] # "CONNECTED"
ip = parts[3] # "192.168.1.100"
# 字符串查找
if "ERROR" in response:
print("命令执行失败")python嵌入式技巧:文本协议虽然可读性好,但解析容易出错。对于高性能场景,建议使用二进制协议(见1.8节)。
1.2.2 列表 (list):有序数据的瑞士军刀#
列表是Python中最常用的数据结构,它是一个有序、可变的序列。
嵌入式场景:
# 存储历史数据
temperature_history: list[float] = []
def record_temperature(value: float) -> None:
temperature_history.append(value)
if len(temperature_history) > 100:
temperature_history.pop(0) # 保持最近100条
# 批量处理寄存器
registers: list[int] = [0x01, 0x02, 0x03, 0x04]
for reg in registers:
write_register(reg, 0x00)
# 列表推导式(一行代码完成批量操作)
raw_values = [0x0100, 0x0200, 0x0300]
temperatures = [v / 256.0 for v in raw_values] # [1.0, 2.0, 3.0]python列表推导式是Python的”语法糖”,它让批量数据处理变得极其简洁。嵌入式工程师经常需要处理一批传感器数据,列表推导式是你的好帮手。
1.2.3 字典 (dict):键值对的快速查找#
字典是键值对的集合,用于快速查找。在嵌入式场景中,它常用于:
# 寄存器映射表
REGISTER_MAP: dict[str, int] = {
"temperature": 0x0100,
"voltage": 0x0102,
"status": 0x0104,
"control": 0x0200,
}
# 通过名称查找地址
temp_addr = REGISTER_MAP["temperature"] # 0x0100
# 设备配置
device_config: dict[str, any] = {
"port": "COM3",
"baudrate": 115200,
"timeout": 1.0,
"retry_count": 3,
}
# 构建协议帧
def build_command(cmd_type: str, params: dict[str, int]) -> bytes:
"""根据命令类型和参数构建二进制帧"""
addr = REGISTER_MAP.get(cmd_type)
if addr is None:
raise ValueError(f"未知命令类型: {cmd_type}")
# ...构建帧逻辑python字典的”防御性”用法:使用.get()方法避免KeyError:
# 不推荐:可能报错
temp_addr = REGISTER_MAP["temperature_typo"] # KeyError!
# 推荐:安全访问
temp_addr = REGISTER_MAP.get("temperature_typo") # 返回 None
temp_addr = REGISTER_MAP.get("temperature_typo", 0x0000) # 返回默认值python1.2.4 元组 (tuple):不可变的数据打包#
元组与列表类似,但不可修改。它常用于:
# 返回多个值
def read_sensor() -> tuple[float, float]:
"""读取传感器,返回 (温度, 电压)"""
temperature = 25.3
voltage = 3.31
return (temperature, voltage)
# 解包赋值
temp, volt = read_sensor()
# 作为字典的键(列表不行)
sensor_cache: dict[tuple[int, int], float] = {}
sensor_cache[(1, 0x0100)] = 25.3 # (设备ID, 寄存器地址) → 值
# 函数返回值的类型提示
def get_device_info() -> tuple[str, int, bool]:
"""返回 (设备名, 设备ID, 是否在线)"""
return ("STM32_Board_01", 1001, True)python选择建议:
- 数据需要修改 → 用
list - 数据不需要修改 → 用
tuple(更安全,语义更清晰) - 需要快速查找 → 用
dict - 需要去重 → 用
set(本章不展开)
1.3 条件判断与循环控制#
1.3.1 条件判断:让程序学会”做选择”#
基本语法:
temperature = 38.5
if temperature > 40:
print("警告:温度过高!")
send_alarm()
elif temperature > 30:
print("温度偏高,注意散热")
else:
print("温度正常")pythonPython的独特之处——没有switch(3.10之前):
在C语言中,你可能习惯用switch处理多种状态。Python 3.10之前没有switch,但可以用字典模拟:
# 用字典模拟 switch-case
STATUS_HANDLERS = {
"IDLE": handle_idle,
"RUNNING": handle_running,
"ERROR": handle_error,
}
status = get_device_status()
handler = STATUS_HANDLERS.get(status)
if handler:
handler()
else:
print(f"未知状态: {status}")pythonPython 3.10+引入了match-case语法,但字典方式在嵌入式场景中仍然更灵活(可以动态注册处理器)。
嵌入式技巧:位操作判断
# 检查状态寄存器的某一位
status_reg = read_register(0x0100)
# 检查第3位(bit 3)是否为1
if status_reg & (1 << 3):
print("设备故障")
# 检查多个位
ERROR_MASK = 0b11110000 # 高4位是错误码
error_code = (status_reg & ERROR_MASK) >> 4python1.3.2 循环控制:让程序学会”重复劳动”#
for循环:遍历序列
# 遍历寄存器列表
registers = [0x0100, 0x0102, 0x0104]
for addr in registers:
value = read_register(addr)
print(f"寄存器 0x{addr:04X} = 0x{value:04X}")
# 遍历字典
config = {"port": "COM3", "baudrate": 115200, "timeout": 1}
for key, value in config.items():
print(f"{key}: {value}")
# 使用 enumerate 获取索引
sensors = ["温度", "湿度", "气压"]
for i, name in enumerate(sensors):
value = read_sensor(i)
print(f"{name} = {value}")pythonwhile循环:条件循环
# 等待设备就绪
max_retries = 10
retry_count = 0
while retry_count < max_retries:
if device_is_ready():
break
retry_count += 1
time.sleep(0.5)
else:
# while循环正常结束(没有break)时执行
raise TimeoutError("设备未就绪")
# 持续监听串口
while True:
data = ser.read(100)
if data:
process_data(data)python循环控制语句:
# break:跳出循环
for i in range(100):
if check_emergency_stop():
break # 紧急停止
# continue:跳过本次迭代
for data in data_stream:
if is_noise(data):
continue # 跳过噪声数据
process(data)
# pass:占位符(保持代码结构完整)
for cmd in command_queue:
if cmd.type == "TODO":
pass # TODO: 待实现
else:
execute(cmd)python1.4 函数定义与参数传递#
1.4.1 函数的基本定义#
函数是代码复用的基本单元。在嵌入式场景中,每个硬件操作都应该封装为函数:
def read_register(address: int, length: int = 2) -> int:
"""
读取寄存器的值
Args:
address: 寄存器地址
length: 读取长度(字节数),默认为2
Returns:
寄存器的值(整数)
"""
# 发送读取命令
cmd = build_read_command(address, length)
ser.write(cmd)
# 接收响应
response = ser.read(length + 2) # 数据 + CRC
if len(response) < length + 2:
raise IOError("读取超时")
# 解析数据
value = int.from_bytes(response[:length], byteorder='big')
return valuepython类型提示的完整形式:
from typing import Optional
def connect_device(port: str, baudrate: int = 115200) -> Optional[serial.Serial]:
"""
连接设备
Args:
port: 串口号,如 'COM3' 或 '/dev/ttyUSB0'
baudrate: 波特率,默认115200
Returns:
成功返回 Serial 对象,失败返回 None
"""
try:
ser = serial.Serial(port, baudrate, timeout=1)
return ser
except serial.SerialException:
return Nonepython1.4.2 参数传递:位置参数、关键字参数、默认值#
# 位置参数(必须按顺序传递)
def build_frame(cmd: int, addr: int, data: bytes) -> bytes:
...
frame = build_frame(0x03, 0x0100, b'\x00\x01')
# 关键字参数(可以不按顺序,更清晰)
frame = build_frame(data=b'\x00\x01', cmd=0x03, addr=0x0100)
# 默认参数(简化常见配置)
def send_command(cmd: bytes, timeout: float = 1.0, retry: int = 3) -> bool:
...
# 常用配置可以省略
send_command(b'\x01\x02') # 使用默认超时和重试次数
# 特殊配置需要显式指定
send_command(b'\x01\x02', timeout=5.0, retry=10)python警告:可变对象作为默认参数:
# 错误示范!
def add_value(value: int, history: list[int] = []) -> list[int]:
history.append(value)
return history
# 每次调用都会累积历史数据
add_value(1) # [1]
add_value(2) # [1, 2] ← 不是 [2]!
# 正确做法:使用 None 作为默认值
def add_value(value: int, history: list[int] | None = None) -> list[int]:
if history is None:
history = []
history.append(value)
return historypython1.4.3 可变参数:*args 和 **kwargs#
当你不确定函数会接收多少参数时,可以使用可变参数:
# *args:接收任意数量的位置参数,打包成元组
def log_values(*values: float) -> None:
"""打印任意数量的值"""
for v in values:
print(f"{v:.2f}")
log_values(1.0, 2.5, 3.7) # 1.00, 2.50, 3.70
# **kwargs:接收任意数量的关键字参数,打包成字典
def configure_device(**settings) -> None:
"""配置设备参数"""
for key, value in settings.items():
write_config(key, value)
configure_device(baudrate=115200, timeout=1.0, retry=3)
# 混合使用
def send_advanced_command(cmd: bytes, *args, **kwargs) -> bool:
...python1.5 文件读写与配置文件解析 (JSON/INI) —— 告别硬编码#
1.5.1 基本文件操作#
在嵌入式开发中,文件操作常用于:
- 保存测试数据到CSV/日志文件
- 读取配置文件
- 导出固件升级包
# 写入文件
with open("test_log.txt", "w", encoding="utf-8") as f:
f.write(f"测试时间: {datetime.now()}\n")
f.write(f"温度: {temperature:.1f}℃\n")
# 读取文件
with open("test_log.txt", "r", encoding="utf-8") as f:
content = f.read() # 读取全部内容
# 逐行读取(适合大文件)
with open("sensor_data.csv", "r") as f:
for line in f:
values = line.strip().split(',')
process_data(values)pythonwith语句:自动关闭文件,即使发生异常也能正确释放资源。这是Python的”上下文管理器”,比手动open/close更安全。
1.5.2 JSON:结构化配置的首选#
JSON是嵌入式项目中最常用的配置格式,因为它:
- 人类可读
- 支持嵌套结构
- Python原生支持(无需安装第三方库)
示例配置文件 (device_config.json):
{
"device": {
"name": "STM32_Board_01",
"port": "COM3",
"baudrate": 115200,
"timeout": 1.0
},
"sensors": {
"temperature": {
"enabled": true,
"calibration_offset": -0.5
},
"humidity": {
"enabled": true,
"calibration_offset": 0.0
}
},
"logging": {
"level": "INFO",
"file": "logs/sensor.log"
}
}jsonPython读写JSON:
import json
from pathlib import Path
# 读取JSON配置
def load_config(config_path: str = "device_config.json") -> dict:
"""加载配置文件"""
with open(config_path, "r", encoding="utf-8") as f:
return json.load(f)
# 使用配置
config = load_config()
port = config["device"]["port"] # "COM3"
baudrate = config["device"]["baudrate"] # 115200
temp_offset = config["sensors"]["temperature"]["calibration_offset"] # -0.5
# 保存配置
def save_config(config: dict, config_path: str = "device_config.json") -> None:
"""保存配置文件"""
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=4, ensure_ascii=False)
# 修改配置
config["device"]["baudrate"] = 9600
save_config(config)python1.5.3 INI:简单配置的替代方案#
INI格式比JSON更简单,适合扁平化的配置:
示例 (config.ini):
[device]
port = COM3
baudrate = 115200
timeout = 1.0
[calibration]
temperature_offset = -0.5
humidity_offset = 0.0
[logging]
level = INFO
file = logs/sensor.loginiPython读写INI:
import configparser
# 读取INI配置
config = configparser.ConfigParser()
config.read("config.ini", encoding="utf-8")
port = config["device"]["port"]
temp_offset = config.getfloat("calibration", "temperature_offset")
# 修改并保存
config["device"]["baudrate"] = "9600"
with open("config.ini", "w") as f:
config.write(f)python选型建议:
- 配置有嵌套结构 → JSON
- 配置是扁平的键值对 → INI
- 配置需要支持注释 → INI(JSON不支持注释)
1.6 异常处理与工程级日志 (Logging) —— 告别 print() 调试#
1.6.1 异常处理:让程序更健壮#
嵌入式设备的运行环境充满了不确定性:串口可能断开、设备可能无响应、数据可能损坏。异常处理让你的程序在遇到这些情况时不会崩溃,而是优雅地处理错误。
基本语法:
try:
# 可能出错的代码
ser = serial.Serial("COM3", 115200, timeout=1)
data = ser.read(10)
result = parse_data(data)
except serial.SerialException as e:
# 处理串口错误
print(f"串口错误: {e}")
reconnect()
except ValueError as e:
# 处理数据解析错误
print(f"数据格式错误: {e}")
except Exception as e:
# 处理其他所有错误
print(f"未知错误: {e}")
else:
# 没有异常时执行
print(f"读取成功: {result}")
finally:
# 无论是否有异常都会执行(清理资源)
if ser and ser.is_open:
ser.close()python嵌入式场景:带重试的操作
import time
from typing import Optional
def read_sensor_with_retry(
port: str,
address: int,
max_retries: int = 3,
retry_delay: float = 0.5
) -> Optional[int]:
"""
带重试的传感器读取
Args:
port: 串口号
address: 寄存器地址
max_retries: 最大重试次数
retry_delay: 重试间隔(秒)
Returns:
成功返回读取的值,失败返回 None
"""
ser = None
try:
ser = serial.Serial(port, 115200, timeout=1)
for attempt in range(max_retries):
try:
cmd = build_read_command(address)
ser.write(cmd)
response = ser.read(4)
if len(response) < 4:
raise IOError("响应不完整")
value = int.from_bytes(response[:2], byteorder='big')
return value
except (IOError, serial.SerialException) as e:
if attempt < max_retries - 1:
print(f"读取失败(第{attempt+1}次),{retry_delay}秒后重试...")
time.sleep(retry_delay)
else:
print(f"读取失败,已重试{max_retries}次")
return None
except serial.SerialException as e:
print(f"串口连接失败: {e}")
return None
finally:
if ser and ser.is_open:
ser.close()python自定义异常:
class DeviceError(Exception):
"""设备相关错误的基类"""
pass
class ConnectionError(DeviceError):
"""连接错误"""
pass
class CommunicationError(DeviceError):
"""通信错误"""
pass
class TimeoutError(DeviceError):
"""超时错误"""
pass
# 使用自定义异常
def send_command(cmd: bytes) -> bytes:
if not is_connected():
raise ConnectionError("设备未连接")
try:
ser.write(cmd)
response = ser.read(100)
except serial.SerialTimeoutException:
raise TimeoutError("命令响应超时")
if len(response) == 0:
raise CommunicationError("无响应")
return responsepython1.6.2 工程级日志:告别 print() 调试#
初学者常用print()调试,但在工程项目中,这有几个严重问题:
- 无法控制输出级别(调试信息和错误信息混在一起)
- 无法输出到文件(程序崩溃后日志丢失)
- 无法添加时间戳、模块名等上下文信息
- 无法动态开关(发布版本还要手动删除print)
Logging模块是Python的标准日志库,它解决了所有这些问题。
基本配置:
import logging
# 配置日志格式
logging.basicConfig(
level=logging.DEBUG, # 最低级别
format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
handlers=[
logging.StreamHandler(), # 输出到控制台
logging.FileHandler('app.log', encoding='utf-8') # 输出到文件
]
)
# 创建日志记录器
logger = logging.getLogger(__name__)python日志级别:
| 级别 | 数值 | 用途 |
|---|---|---|
DEBUG | 10 | 调试信息,只在开发时使用 |
INFO | 20 | 程序正常运行的信息 |
WARNING | 30 | 警告,程序还能运行但有问题 |
ERROR | 40 | 错误,某个功能无法执行 |
CRITICAL | 50 | 严重错误,程序可能无法继续 |
使用示例:
def read_sensor(address: int) -> Optional[float]:
logger.debug(f"读取传感器,地址=0x{address:04X}")
try:
raw = read_register(address)
logger.info(f"传感器读取成功,原始值=0x{raw:04X}")
temperature = raw / 256.0
logger.debug(f"转换后温度={temperature:.2f}℃")
if temperature > 80:
logger.warning(f"温度偏高: {temperature:.2f}℃")
return temperature
except IOError as e:
logger.error(f"读取失败: {e}")
return None
except Exception as e:
logger.critical(f"未预期的错误: {e}", exc_info=True)
raisepython在嵌入式项目中的日志策略:
import logging
from logging.handlers import RotatingFileHandler
def setup_logger(name: str, log_file: str, level: int = logging.INFO) -> logging.Logger:
"""
创建项目日志记录器
Args:
name: 记录器名称
log_file: 日志文件路径
level: 日志级别
Returns:
配置好的日志记录器
"""
logger = logging.getLogger(name)
logger.setLevel(level)
# 日志格式
formatter = logging.Formatter(
'%(asctime)s [%(levelname)s] %(name)s (%(filename)s:%(lineno)d): %(message)s'
)
# 控制台输出(只显示 INFO 及以上)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)
# 文件输出(记录 DEBUG 及以上,自动轮转)
file_handler = RotatingFileHandler(
log_file,
maxBytes=10*1024*1024, # 10MB
backupCount=5,
encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
logger.addHandler(console_handler)
logger.addHandler(file_handler)
return logger
# 使用
logger = setup_logger('sensor_monitor', 'logs/sensor.log')pythonprint() vs logging 对比:
# print():无法控制,无上下文
print("温度过高")
# logging:可控、可配置、有上下文
logger.warning(f"温度过高: {temperature:.1f}℃ (阈值={threshold}℃)")python1.7 类与对象(基础概念与封装思维)#
1.7.1 为什么要用类?#
在嵌入式开发中,你经常需要操作”设备”——串口设备、传感器、执行器等。每个设备都有自己的状态(是否连接、当前配置)和行为(读取、写入、重启)。
类就是把这些状态和行为打包在一起的工具。
不用类的代码:
# 每次操作都要传递一大堆参数
def read_sensor(port, baudrate, timeout, address):
...
def write_sensor(port, baudrate, timeout, address, value):
...
# 很容易传错参数
read_sensor("COM3", 115200, 1.0, 0x0100)
write_sensor("COM3", 115200, 1.0, 0x0100, 25.5) # 一长串参数python用类的代码:
sensor = Sensor("COM3", 115200)
sensor.connect()
sensor.read(0x0100)
sensor.write(0x0100, 25.5)
sensor.disconnect()python类把相关的数据和操作组织在一起,代码更清晰、更安全。
1.7.2 定义一个基本的类#
class Sensor:
"""
传感器设备类
"""
def __init__(self, port: str, baudrate: int = 115200):
"""
初始化传感器
Args:
port: 串口号
baudrate: 波特率
"""
self.port = port
self.baudrate = baudrate
self.serial = None # 串口对象,稍后初始化
self.is_connected = False
def connect(self) -> bool:
"""连接传感器"""
try:
self.serial = serial.Serial(
port=self.port,
baudrate=self.baudrate,
timeout=1.0
)
self.is_connected = True
return True
except serial.SerialException as e:
logger.error(f"连接失败: {e}")
return False
def disconnect(self) -> None:
"""断开连接"""
if self.serial and self.serial.is_open:
self.serial.close()
self.is_connected = False
def read(self, address: int) -> Optional[int]:
"""读取寄存器"""
if not self.is_connected:
raise ConnectionError("设备未连接")
# 发送读取命令
cmd = self._build_read_command(address)
self.serial.write(cmd)
# 接收响应
response = self.serial.read(4)
if len(response) < 4:
return None
return int.from_bytes(response[:2], byteorder='big')
def write(self, address: int, value: int) -> bool:
"""写入寄存器"""
if not self.is_connected:
raise ConnectionError("设备未连接")
# 发送写入命令
cmd = self._build_write_command(address, value)
self.serial.write(cmd)
# 等待应答
response = self.serial.read(2)
return response == b'\x06' # ACK
def _build_read_command(self, address: int) -> bytes:
"""构建读取命令(内部方法)"""
# 这里只是示例,实际协议可能更复杂
return bytes([0x01, 0x03]) + address.to_bytes(2, 'big')
def _build_write_command(self, address: int, value: int) -> bytes:
"""构建写入命令(内部方法)"""
return bytes([0x01, 0x06]) + address.to_bytes(2, 'big') + value.to_bytes(2, 'big')python使用示例:
# 创建传感器对象
sensor = Sensor("COM3", 115200)
# 连接
if sensor.connect():
print("连接成功")
# 读取温度
temp_raw = sensor.read(0x0100)
if temp_raw is not None:
temperature = temp_raw / 256.0
print(f"温度: {temperature:.1f}℃")
# 写入配置
sensor.write(0x0200, 0x0001)
# 断开连接
sensor.disconnect()
else:
print("连接失败")python1.7.3 类的特殊方法:让对象更”Pythonic”#
Python类有一些”魔术方法”(以双下划线开头和结尾),它们让你的对象更符合Python的使用习惯:
class Register:
"""寄存器类"""
def __init__(self, address: int, name: str, value: int = 0):
self.address = address
self.name = name
self.value = value
def __repr__(self) -> str:
"""对象的官方表示(用于调试)"""
return f"Register(0x{self.address:04X}, '{self.name}', 0x{self.value:04X})"
def __str__(self) -> str:
"""对象的字符串表示(用于打印)"""
return f"{self.name} @ 0x{self.address:04X} = 0x{self.value:04X}"
def __eq__(self, other: object) -> bool:
"""比较两个对象是否相等"""
if not isinstance(other, Register):
return False
return self.address == other.address and self.value == other.value
# 使用
reg = Register(0x0100, "TEMPERATURE", 0x1900)
print(reg) # TEMPERATURE @ 0x0100 = 0x1900
print(repr(reg)) # Register(0x0100, 'TEMPERATURE', 0x1900)python1.7.4 上下文管理器:优雅的资源管理#
还记得with open(...)吗?你可以让自己的类也支持这种用法,确保资源(串口、文件等)被正确释放:
class Sensor:
# ...之前的代码...
def __enter__(self):
"""进入 with 语句时调用"""
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""离开 with 语句时调用"""
self.disconnect()
return False # 不抑制异常
# 使用上下文管理器
with Sensor("COM3", 115200) as sensor:
temp = sensor.read(0x0100)
print(f"温度: {temp}")
# 无论是否发生异常,都会自动调用 disconnect()python上下文管理器的价值:即使在with代码块中发生异常,__exit__也会被调用,确保串口被正确关闭。这比手动try-finally更简洁、更不容易出错。
1.8 字节操作 (bytes/bytearray) 与 struct 库——硬件通讯的基石#
这一节是嵌入式工程师最需要的Python知识。串口通信、网络通信、固件升级……所有与硬件交互的操作,本质上都是在处理字节流。
1.8.1 bytes 与 bytearray:二进制数据的载体#
# bytes:不可变的字节序列
data = b'\x01\x02\x03\x04' # 直接创建
data = bytes([1, 2, 3, 4]) # 从列表创建
data = 'Hello'.encode('utf-8') # 从字符串编码
# bytearray:可变的字节序列(可以原地修改)
buf = bytearray(b'\x01\x02\x03')
buf[0] = 0xFF # 修改第一个字节
buf.append(0x04) # 追加一个字节
# 常用操作
data = b'\x01\x02\x03\x04'
print(len(data)) # 4
print(data[0]) # 1(返回整数)
print(data[:2]) # b'\x01\x02'(切片)
print(data.hex()) # '01020304'(十六进制表示)python字节序(Endianness):多字节数据的存储顺序。嵌入式设备常用大端序(高位在前),PC常用小端序。
# 大端序(Big-Endian):高位在前
value = 0x1234
big_endian = value.to_bytes(2, byteorder='big') # b'\x12\x34'
# 小端序(Little-Endian):低位在前
little_endian = value.to_bytes(2, byteorder='little') # b'\x34\x12'
# 从字节还原为整数
restored = int.from_bytes(big_endian, byteorder='big') # 0x1234python1.8.2 struct 库:打包和解包二进制数据#
struct库是Python处理二进制数据的核心工具。它能将Python对象(整数、浮点数等)打包成字节序列,也能将字节序列解包成Python对象。
基本用法:
import struct
# 打包:将Python对象 → 字节序列
# 格式:'>' 表示大端序,'H' 表示无符号16位整数
data = struct.pack('>H', 0x1234) # b'\x12\x34'
# 解包:将字节序列 → Python对象
value = struct.unpack('>H', b'\x12\x34') # (0x1234,)
value = value[0] # 0x1234
# 打包多个值
data = struct.pack('>HBf', 0x1234, 255, 3.14)
# H: 无符号16位,B: 无符号8位,f: 32位浮点数
# 解包多个值
addr, status, temperature = struct.unpack('>HBf', data)python格式字符速查表:
| 格式 | C类型 | Python类型 | 字节数 |
|---|---|---|---|
x | padding | - | 1 |
c | char | bytes(1) | 1 |
b | signed char | int | 1 |
B | unsigned char | int | 1 |
h | short | int | 2 |
H | unsigned short | int | 2 |
i | int | int | 4 |
I | unsigned int | int | 4 |
q | long long | int | 8 |
Q | unsigned long long | int | 8 |
f | float | float | 4 |
d | double | float | 8 |
s | char[] | bytes | - |
字节序前缀:
| 前缀 | 含义 |
|---|---|
@ | 本地字节序(默认) |
< | 小端序 |
> | 大端序 |
! | 网络字节序(大端) |
1.8.3 实战:解析串口协议帧#
假设你有一个传感器,它通过串口发送如下格式的数据帧:
帧头(2字节) + 长度(1字节) + 命令(1字节) + 数据(N字节) + CRC(2字节)
帧头: 0xAA 0x55plaintext用struct解析:
import struct
from typing import Optional
class SensorFrame:
"""传感器数据帧"""
HEADER = b'\xAA\x55'
HEADER_FORMAT = '>2sBB' # 帧头(2字节) + 长度(1字节) + 命令(1字节)
HEADER_SIZE = struct.calcsize(HEADER_FORMAT) # 4字节
# 命令定义
CMD_TEMPERATURE = 0x01
CMD_HUMIDITY = 0x02
CMD_STATUS = 0x03
@staticmethod
def calculate_crc(data: bytes) -> int:
"""计算CRC-16(简化版,实际项目请使用标准算法)"""
crc = 0xFFFF
for byte in data:
crc ^= byte
for _ in range(8):
if crc & 0x0001:
crc = (crc >> 1) ^ 0xA001
else:
crc >>= 1
return crc
@classmethod
def parse(cls, data: bytes) -> Optional[dict]:
"""
解析数据帧
Args:
data: 原始字节数据
Returns:
解析结果字典,失败返回 None
"""
# 检查最小长度
if len(data) < cls.HEADER_SIZE + 2: # 头部 + CRC
return None
# 检查帧头
if data[:2] != cls.HEADER:
return None
# 解析头部
_, length, cmd = struct.unpack_from(cls.HEADER_FORMAT, data)
# 检查长度
expected_len = cls.HEADER_SIZE + length + 2 # 头部 + 数据 + CRC
if len(data) < expected_len:
return None
# 提取数据和CRC
payload = data[cls.HEADER_SIZE:cls.HEADER_SIZE + length]
crc_received = struct.unpack_from('>H', data, cls.HEADER_SIZE + length)[0]
# 验证CRC
crc_calculated = cls.calculate_crc(data[:cls.HEADER_SIZE + length])
if crc_received != crc_calculated:
return None
# 根据命令类型解析数据
result = {"command": cmd, "raw": payload}
if cmd == cls.CMD_TEMPERATURE and len(payload) == 2:
temp_raw = struct.unpack('>h', payload)[0] # 有符号16位
result["temperature"] = temp_raw / 256.0
elif cmd == cls.CMD_HUMIDITY and len(payload) == 2:
hum_raw = struct.unpack('>H', payload)[0] # 无符号16位
result["humidity"] = hum_raw / 100.0
elif cmd == cls.CMD_STATUS and len(payload) == 1:
result["status_code"] = payload[0]
return result
@classmethod
def build(cls, cmd: int, data: bytes) -> bytes:
"""
构建数据帧
Args:
cmd: 命令字节
data: 数据载荷
Returns:
完整的数据帧
"""
length = len(data)
header = struct.pack(cls.HEADER_FORMAT, cls.HEADER, length, cmd)
# 计算CRC
frame_without_crc = header + data
crc = cls.calculate_crc(frame_without_crc)
return frame_without_crc + struct.pack('>H', crc)
# 使用示例
if __name__ == "__main__":
# 构建一个温度读取命令
frame = SensorFrame.build(SensorFrame.CMD_TEMPERATURE, b'')
print(f"发送帧: {frame.hex()}") # aa550001xxxx
# 模拟接收数据并解析
received_data = bytes.fromhex("aa5502011900xxxx") # 示例数据
result = SensorFrame.parse(received_data)
if result:
print(f"解析结果: {result}")
else:
print("解析失败")python1.8.4 实战:处理寄存器数组#
嵌入式设备通常有多个寄存器,每个寄存器有不同的含义和长度:
import struct
from dataclasses import dataclass
from typing import ClassVar
@dataclass
class DeviceRegisters:
"""设备寄存器映射"""
# 寄存器地址定义
REG_DEVICE_ID: ClassVar[int] = 0x0000
REG_STATUS: ClassVar[int] = 0x0002
REG_TEMPERATURE: ClassVar[int] = 0x0010
REG_HUMIDITY: ClassVar[int] = 0x0012
REG_VOLTAGE: ClassVar[int] = 0x0014
REG_CONTROL: ClassVar[int] = 0x0100
# 数据内容
device_id: int = 0
status: int = 0
temperature: float = 0.0
humidity: float = 0.0
voltage: float = 0.0
control: int = 0
@classmethod
def from_bytes(cls, data: bytes) -> 'DeviceRegisters':
"""从字节数据解析寄存器"""
if len(data) < 20: # 假设我们需要前20个字节
raise ValueError("数据长度不足")
regs = cls()
# 解析各个寄存器
regs.device_id = struct.unpack_from('>H', data, cls.REG_DEVICE_ID)[0]
regs.status = struct.unpack_from('>H', data, cls.REG_STATUS)[0]
# 温度:有符号16位,单位0.01℃
temp_raw = struct.unpack_from('>h', data, cls.REG_TEMPERATURE)[0]
regs.temperature = temp_raw / 100.0
# 湿度:无符号16位,单位0.01%
hum_raw = struct.unpack_from('>H', data, cls.REG_HUMIDITY)[0]
regs.humidity = hum_raw / 100.0
# 电压:无符号16位,单位0.001V
volt_raw = struct.unpack_from('>H', data, cls.REG_VOLTAGE)[0]
regs.voltage = volt_raw / 1000.0
regs.control = struct.unpack_from('>H', data, cls.REG_CONTROL)[0]
return regs
def to_dict(self) -> dict:
"""转换为字典"""
return {
'device_id': f'0x{self.device_id:04X}',
'status': f'0x{self.status:04X}',
'temperature': f'{self.temperature:.2f}℃',
'humidity': f'{self.humidity:.1f}%',
'voltage': f'{self.voltage:.3f}V',
'control': f'0x{self.control:04X}',
}
def __str__(self) -> str:
"""友好的字符串表示"""
lines = ["=== Device Registers ==="]
for key, value in self.to_dict().items():
lines.append(f" {key:15s} = {value}")
return '\n'.join(lines)python1.9 模块导入与项目代码组织#
1.9.1 为什么需要模块化?#
随着项目变大,把所有代码放在一个文件里会导致:
- 文件太长,难以查找
- 多人协作时容易冲突
- 代码复用困难
模块化让你把代码拆分成多个文件,每个文件负责一个功能。
1.9.2 基本的模块导入#
# 导入整个模块
import serial
import struct
import json
# 使用时需要加模块名
ser = serial.Serial('COM3', 115200)
data = struct.pack('>H', 0x1234)
# 导入模块的特定功能
from serial import Serial
from struct import pack, unpack
# 使用时不需要加模块名
ser = Serial('COM3', 115200)
data = pack('>H', 0x1234)
# 导入并重命名(避免命名冲突)
import serial as ser_lib
from serial import Serial as SerialPortpython1.9.3 自定义模块#
创建自己的模块非常简单——把函数和类写在一个.py文件里就行了。
项目结构示例:
my_project/
├── main.py # 主程序
├── config.json # 配置文件
├── requirements.txt # 依赖列表
└── src/ # 源代码目录
├── __init__.py # 让 Python 把 src 当作包
├── sensor.py # 传感器相关功能
├── protocol.py # 协议解析
├── utils.py # 工具函数
└── config.py # 配置管理plaintextsrc/sensor.py:
"""传感器操作模块"""
import serial
import logging
from typing import Optional
logger = logging.getLogger(__name__)
class Sensor:
"""传感器类"""
def __init__(self, port: str, baudrate: int = 115200):
self.port = port
self.baudrate = baudrate
self.serial: Optional[serial.Serial] = None
def connect(self) -> bool:
try:
self.serial = serial.Serial(self.port, self.baudrate, timeout=1)
logger.info(f"连接成功: {self.port}")
return True
except serial.SerialException as e:
logger.error(f"连接失败: {e}")
return False
def read_temperature(self) -> Optional[float]:
"""读取温度"""
# ...实现细节...
passpythonmain.py:
"""主程序"""
from src.sensor import Sensor
from src.protocol import parse_frame
def main():
sensor = Sensor("COM3", 115200)
if sensor.connect():
temp = sensor.read_temperature()
print(f"温度: {temp}℃")
if __name__ == "__main__":
main()python1.9.4 __init__.py 的作用#
在Python 3中,__init__.py文件可以为空,但它告诉Python:“这个目录是一个Python包”。
src/__init__.py:
"""
传感器驱动包
"""
# 简化导入路径
from .sensor import Sensor
from .protocol import parse_frame
# 定义包的公开接口
__all__ = ['Sensor', 'parse_frame']python有了这个文件,用户可以直接:
from src import Sensor # 而不是 from src.sensor import Sensorpython1.10 四两拨千斤的实用小工具#
1.10.1 tqdm:进度条#
当你需要处理大量数据或执行耗时操作时,进度条能让用户知道程序还在运行。
安装:
pip install tqdmbash使用:
from tqdm import tqdm
import time
# 基本用法
for i in tqdm(range(100), desc="处理数据"):
time.sleep(0.05) # 模拟耗时操作
# 嵌入式场景:批量读取寄存器
registers = range(0x0000, 0x0100)
results = {}
for addr in tqdm(registers, desc="读取寄存器"):
value = read_register(addr)
results[addr] = value
# 手动更新进度
with tqdm(total=100, desc="固件升级") as pbar:
for chunk in firmware_chunks:
send_chunk(chunk)
pbar.update(1) # 更新进度python输出效果:
读取寄存器: 100%|████████████████████████████████| 256/256 [00:12<00:00, 20.3it/s]plaintext1.10.2 argparse:命令行参数解析#
很多嵌入式工具需要从命令行接收参数(串口号、波特率、配置文件路径等)。argparse是Python标准库中的命令行参数解析工具。
基本用法:
import argparse
def main():
parser = argparse.ArgumentParser(description="传感器数据采集工具")
# 添加参数
parser.add_argument(
"-p", "--port",
type=str,
default="COM3",
help="串口号 (默认: COM3)"
)
parser.add_argument(
"-b", "--baudrate",
type=int,
default=115200,
help="波特率 (默认: 115200)"
)
parser.add_argument(
"-t", "--timeout",
type=float,
default=1.0,
help="超时时间,秒 (默认: 1.0)"
)
parser.add_argument(
"-v", "--verbose",
action="store_true",
help="详细输出"
)
parser.add_argument(
"config",
type=str,
help="配置文件路径"
)
# 解析参数
args = parser.parse_args()
# 使用参数
print(f"串口: {args.port}")
print(f"波特率: {args.baudrate}")
print(f"配置文件: {args.config}")
if args.verbose:
print("详细模式已启用")
if __name__ == "__main__":
main()python使用示例:
# 使用默认参数
python main.py config.json
# 指定参数
python main.py -p COM5 -b 9600 config.json
# 查看帮助
python main.py --helpbash输出:
usage: main.py [-h] [-p PORT] [-b BAUDRATE] [-t TIMEOUT] [-v] config
传感器数据采集工具
positional arguments:
config 配置文件路径
optional arguments:
-h, --help show this help message and exit
-p PORT, --port PORT 串口号 (默认: COM3)
-b BAUDRATE, --baudrate BAUDRATE
波特率 (默认: 115200)
-t TIMEOUT, --timeout TIMEOUT
超时时间,秒 (默认: 1.0)
-v, --verbose 详细输出plaintext1.10.3 typer:更现代的命令行工具(可选)#
如果你觉得argparse的语法太繁琐,可以试试typer——它用类型提示来定义命令行参数,更简洁、更Pythonic。
安装:
pip install typerbash使用:
import typer
from typing import Optional
def main(
port: str = typer.Option("COM3", "--port", "-p", help="串口号"),
baudrate: int = typer.Option(115200, "--baudrate", "-b", help="波特率"),
timeout: float = typer.Option(1.0, "--timeout", "-t", help="超时时间"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
config: str = typer.Argument(..., help="配置文件路径"),
):
"""传感器数据采集工具"""
print(f"串口: {port}")
print(f"波特率: {baudrate}")
print(f"配置文件: {config}")
if verbose:
print("详细模式已启用")
if __name__ == "__main__":
typer.run(main)python选型建议:
- 简单脚本、标准库优先 →
argparse - 复杂命令、子命令 →
typer或click
本章小结#
恭喜你读完了第1章!让我们回顾一下你学到了什么:
- 变量与数据类型:Python的基本数据类型,以及
bytes/bytearray在嵌入式中的核心地位 - 类型提示:给AI和未来的自己看的”数据流说明书”
- 数据结构:
str、list、dict、tuple在工程数据中的应用 - 控制流:条件判断和循环的基本语法,以及位操作的技巧
- 函数:定义、参数传递、类型提示的完整用法
- 文件与配置:JSON和INI配置文件的读写
- 异常与日志:让程序更健壮,告别
print()调试 - 类与对象:封装思想,上下文管理器
- 字节操作:
struct库打包解包二进制数据 - 模块化:项目代码组织,自定义模块
- 实用工具:
tqdm进度条、argparse命令行参数
最重要的理念:
- 够用就好:不要追求精通所有语法,够用就行
- 类型提示:这是AI时代工程师的基本素养
- 场景驱动:每个知识点都绑定一个工程场景
- 工程思维:异常处理、日志记录、模块化——这些”非语法”的东西比语法本身更重要
下一章预告:
第2章将介绍现代Python工具链——包管理、虚拟环境、代码格式化、版本控制。这些是让你从”写脚本”升级到”做项目”的关键基础设施。