python补习

装饰器

函数装饰器

什么是装饰器

装饰器是python的一种高级语法，本质上是函数包装器，可以在不修改函数代码的前提下为函数添加额外功能如日志记录、性能计时、权限校验，也可以修改函数的输入和输出。装饰器通过@装饰器名语法应用与函数，也是一种语法糖，简化包装的代码。

基本语法与原理

装饰器是一个接收函数作为参数，并返回新函数的函数。当用@decorator修饰函数func时，相当于执行func=decorator(func)，即原函数被替换为装饰器返回的新函数。

# 定义装饰器：打印函数调用信息
def log_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"调用函数: {func.__name__}")  # 额外功能：打印日志
        result = func(*args, **kwargs)      # 执行原函数
        print(f"函数 {func.__name__} 执行完毕")
        return result                       # 返回原函数结果
    return wrapper

# 应用装饰器
@log_decorator
def add(a, b):
    return a + b

# 调用函数
add(1, 2)

上面打印的输出为

调用函数: add
函数 add 执行完毕
3

可以看到，相当于给add函数做了一层包装。这里 log_decorator 为 add 函数添加了“调用日志”功能，原 add 函数代码未做任何修改。

带参数的装饰器

如果装饰器需要自定义参数（如 @parser.wrap() 中的空括号），需在基础装饰器外再嵌套一层参数接收函数。

# 带参数的装饰器：自定义日志前缀
def log_decorator(prefix="LOG"):
    def decorator(func):
        def wrapper(*args, **kwargs):
            print(f"[{prefix}] 调用函数: {func.__name__}")  # 使用装饰器参数
            result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

# 应用装饰器（传递参数）
@log_decorator(prefix="DEBUG")
def multiply(a, b):
    return a * b

multiply(3, 4)  # 输出: [DEBUG] 调用函数: multiply → 返回 12

装饰器在 Python 中非常常用，典型场景包括：

• 日志记录：自动记录函数调用、参数、返回值；
• 性能计时：统计函数执行时间；
• 权限校验：检查用户是否有权限调用函数；
• 输入/输出处理：自动转换参数类型、格式化返回值；
• 资源管理：自动打开/关闭文件、数据库连接等。

类装饰器

前面描述了函数装饰器，还有类的装饰器。类的装饰器是通过修改类的定义、添加/覆盖方法或属性，或返回一个新类，来增强类的功能，与函数装饰器（修饰函数）不同，类装饰器专注于修饰类本身。类装饰器通过 @装饰器名 语法应用于类，是一种“语法糖”，简化了类的动态修改逻辑。

类装饰器是一个接收类作为参数，并返回新类的函数。当用 @decorator 修饰类 MyClass 时，相当于执行 MyClass = decorator(MyClass)，即原类被“替换”为装饰器返回的新类。

简单类装饰器示例

# 定义类装饰器：为类添加一个属性和方法
def add_greeting(cls):
    cls.greeting = "Hello from decorator!"  # 新添加类属性

    def say_hello(self):  # 新定义要添加的实例方法
        return f"{self.greeting} I'm {self.name}."

    cls.say_hello = say_hello  # 将方法绑定到类
    return cls  # 返回修改后的类

# 应用装饰器
@add_greeting
class Person:
    def __init__(self, name):
        self.name = name

# 使用装饰后的类
p = Person("Alice")
print(p.greeting)       # 输出：Hello from decorator!
print(p.say_hello())    # 输出：Hello from decorator! I'm Alice.

内置装饰器@dataclass

@dataclass 是 Python 标准库 dataclasses 提供的内置类装饰器，用于快速定义数据存储类（Data Class），核心作用是：

• 自动生成 init 方法：无需手动编写 def init(self, robot, dataset, …): …，装饰器会根据类字段自动生成。
• 自动生成 repr、eq 等方法：方便打印实例（如 print(cfg)）和比较实例是否相等。
• 支持字段默认值和类型注解：如 teleop: TeleoperatorConfig | None = None 中的默认值和类型约束。

先来看看如果不适用@dataclass装饰器，定义一个普通的类，需要手动编写init，repr等方法，如下

class DatasetConfig:
    def __init__(self, repo_id: str, num_episodes: int = 50):
        self.repo_id = repo_id  # 手动绑定 self.xxx
        self.num_episodes = num_episodes
    def __repr__(self):  # 手动编写打印逻辑
        return f"DatasetConfig(repo_id={self.repo_id!r}, num_episodes={self.num_episodes!r})"

# 使用
cfg = DatasetConfig("aliberts/record-test", 2)
print(cfg)  # 输出：DatasetConfig(repo_id='aliberts/record-test', num_episodes=2)

如果使用了@dataclass，则不需要编写init等方法，但是需要添加属性的类型注解声明，如下面repo_id，num_episodes。如果有默认值的，如下的int=50,可选传递参数，如果没有默认值的，必现要传递参数如repo_id。

from dataclasses import dataclass

@dataclass
class DatasetConfig:
    repo_id: str  # 必选字段（无默认值）
    num_episodes: int = 50  # 可选字段（默认值 50）

# 使用（效果与普通类完全一致）
cfg = DatasetConfig("aliberts/record-test", 2)
print(cfg)  # 自动生成 __repr__：DatasetConfig(repo_id='aliberts/record-test', num_episodes=2)

@dataclass提供初始化后回调方法，在init执行完毕后自动调用，用于字段校验，动态修改字段值等。

from dataclasses import dataclass

@dataclass
class DatasetConfig:
    repo_id: str  # 必选字段（无默认值）
    num_episodes: int = 50  # 可选字段（默认值 50）

    def __post_init__(self):
        if self.repo_id is None:
            raise ValueError("You need to provide a repo_id as argument.")

函数返回类型注解

函数返回类型注解是python 3.0+引入的类型提示语法，用于显式声明函数预期返回值的类型。它不会改变函数的运行逻辑，只是为了提升代码的可读性、支持IDE智能提示，便于静态代码检查工具检测潜在错误。其语法格式为如下：

def 函数名(参数: 参数类型) -> 返回值类型:
    # 函数逻辑
    return 返回值

返回类型注解通过->类型语法声明，位于函数定义参数列表之后、冒号：之前。

def record(cfg: RecordConfig) -> LeRobotDataset:
    # ... 函数逻辑 ...
    return dataset  # dataset 是 LeRobotDataset 实例

这里 -> LeRobotDataset 表示：record 函数执行完毕后，预期返回一个 LeRobotDataset 类的实例。

注解都有哪些类型了，除了基础的int、float、str、bool、None（空）这几个类型外，还有容器类型、组合类型、特殊类型等。

容器类型

列表，list[值类型] ，用于标注列表、字典、元组等容器的元素类型，Python 3.9+ 支持直接用 list[int] 形式，旧版本需从 typing 模块导入（如 List[int]）。

motor_names: list[str] = ["shoulder_pan", "elbow_flex"]  # 字符串列表
positions: list[float] = [0.2, 0.5, -0.3]  # 浮点数列表

字典，dict[键类型, 值类型]如下示例

def _motors_ft(self) -> dict[str, type]:  # 键为字符串，值为类型对象（如 float）
    return {f"{motor}.pos": float for motor in self.bus.motors}

元组，tuple[类型1, 类型2, …]，如下示例：

def _cameras_ft(self) -> dict[str, tuple]:  # 值为元组（高、宽、通道数）
    return {cam: (height, width, 3) for cam in self.cameras}
# 更精确标注：tuple[int, int, int]（高、宽、3通道）

集合类型，set[元素类型]，如下示例唯一电机ID集合。

motor_ids: set[int] = {1, 2, 3, 4, 5, 6}  # 整数集合

组合类型

Union,Union[类型1, 类型2, …],允许整数或字符串的参数

from typing import Union

def get_motor(motor_id: Union[int, str]) -> Motor:  # motor_id 可为 int 或 str
    ...
# Python 3.10+ 简写：
def get_motor(motor_id: int | str) -> Motor:
    ...

Option，Optional[类型]，（等价于 Union[类型, None]），可能为None的配置参数。

from typing import Optional

def connect(port: Optional[str] = None) -> None:  # port 可为字符串或 None
    ...

特殊类型

Any，任意类型，关闭类型检查，允许任何类型（常用于动态数据，如lerobot代码中 get_observation 返回 dict[str, Any]）

from typing import Any

def get_observation(self) -> dict[str, Any]:  # 值可为任意类型（电机位置/图像等）
    ...

Callable，Callable[[参数类型1, 参数类型2], 返回值类型]，接受函数作为参数。

from typing import Callable

def calibrate(callback: Callable[[str], None]) -> None:  # callback 是 (str) -> None 的函数
    callback("Calibration done")

type,Type[类型]（标注“类型本身”而非实例，如lerobot代码中 dict[str, type]）,接受类作为参数。

from typing import Type

def create_robot(robot_class: Type[Robot]) -> Robot:  # robot_class 是 Robot 的子类
    return robot_class()

配置选择注册机制

以draccus.ChoiceRegistry为例说明，draccus.ChoiceRegistry 是 draccus 配置框架提供的子类注册与动态选择机制。它允许将基类的多个子类注册为“可选选项”，并通过配置参数（如命令行、配置文件）动态选择具体子类。在工程中，这一机制用于实现 “同一接口，多种实现” 的灵活配置（例如不同机器人型号共享 RobotConfig 接口，但有各自的硬件参数实现）。

注册与选择流程

1. 基类：继承ChoiceRegistry并声明接口。如示例基类（如 RobotConfig）继承 draccus.ChoiceRegistry，作为所有子类的“公共接口”。它定义通用字段和方法，不包含具体实现细节。

from dataclasses import dataclass
import abc
import draccus

@dataclass(kw_only=True)
class RobotConfig(draccus.ChoiceRegistry, abc.ABC):  # 继承 ChoiceRegistry
    # 通用字段（所有子类共享）
    id: str | None = None  # 机器人标识
    calibration_dir: Path | None = None  # 校准文件路径

    @property
    def type(self) -> str:
        # 获取当前子类的注册名称（核心方法）
        return self.get_choice_name(self.__class__)

2. 子类：注册为可选项。每个具体实现（如不同机器人型号）定义一个 RobotConfig 的子类，补充特有字段和逻辑。draccus 会自动将子类注册为一个可选选项，如下：

例如，so101_follower 机器人的配置子类：

# SO101FollowerConfig（so101_follower 型号）
@dataclass(kw_only=True)
class SO101FollowerConfig(RobotConfig):
    port: str  # 型号特有字段（通信端口）
    disable_torque_on_disconnect: bool = True  # 型号特有字段（扭矩控制）

# KochFollowerConfig（koch_follower 型号）
@dataclass(kw_only=True)
class KochFollowerConfig(RobotConfig):
    ip_address: str  # 型号特有字段（以太网通信地址）
    timeout_ms: int = 500  # 型号特有字段（通信超时）

SO101FollowerConfig/KochFollowerConfig继承了RobotConfig，而RobotConfig继承了draccus.ChoiceRegistry。

3. 动态选择：通过配置参数指定子类。用户通过配置参数（如命令行 –robot.type=so101_follower）指定要使用的子类。draccus 会：

• 根据参数值（so101_follower）查找注册的子类；
• 实例化该子类，并将其他配置参数（如 –robot.port=/dev/ttyACM1）映射到子类字段；
• 返回实例化后的子类对象，作为业务逻辑的输入。

# 命令行参数示例
python -m lerobot.record \
  --robot.type=so101_follower \  # 选择 SO101FollowerConfig 子类
  --robot.id=black \             # 设置通用字段 id
  --robot.port=/dev/ttyACM0 \    # 设置型号特有字段 port
  --robot.disable_torque_on_disconnect=true  # 设置型号特有字段

动作先行思维

C语言的风格是写法是条件先行，再写动作。而python支持动作先行写法，再补充条件，主要是为了简写。请看下面示例。

A if cond else B

# 简洁写法（条件表达式）
teleop = make_teleoperator_from_config(cfg.teleop) if cfg.teleop is not None else None

# 等价传统写法（if-else 块）
if cfg.teleop is not None:
    teleop = make_teleoperator_from_config(cfg.teleop)
else:
    teleop = None

使用的简洁方法是： A if cond else B。

or短路运算

# 例：若 config_path 为空，则默认使用 "./config.json"
config_path = user_provided_path or "./config.json"

for循环

C 的 for 循环强调“初始化→条件→增量”的控制流，而 Python 的 for 更关注“迭代对象→元素处理”，动作（循环体）直接跟在迭代逻辑后。

# 例：遍历数据集并处理每个帧
for frame in dataset.frames:
    process_frame(frame)  # 动作（循环体）直接跟在迭代逻辑后

Python 无需显式初始化索引、判断终止条件或手动增量（如 i++），迭代逻辑由“可迭代对象”（如列表、字典、生成器）内部处理。

列表推导式

结构是列表推导式：[表达式 for 变量 in 可迭代对象 if 条件]

Python 的列表推导式将“对元素的处理动作”放在最前面，直接表达“要生成什么样的列表”，而非 C 中“如何生成列表”的步骤式逻辑。

# 例：筛选偶数并计算平方（动作：x**2，条件：x%2==0）
even_squares = [x**2 for x in range(10) if x % 2 == 0]
# 结果：[0, 4, 16, 36, 64]

结构拆解：

• 动作:（x**2）,定义每个元素的转换方式（先明确“要做什么”）；
• 迭代逻辑:（for x in range(10)）,从哪里获取元素；
• 条件:(if x\%2 =\= 0),筛选元素的规则,后补充限制条件。

也可以直接做赋值，看下面例子

@dataclass
class Example:
    src: List[int]
    tgt: List[int]

收集列表直接赋值为data
data = [Example(s,t)] for s, t in paris if len(s) > 0]

等价于

for s, t in pairs:
    if len(s) > 0:
        src_ids = s;
        tgt_idg = s;
        data.append(Example(s,t))

字典推导式

核心逻辑是：{新键: 新值 for 键, 值 in 迭代器 if 条件}

先定义“键和值的生成动作”，再说明迭代范围和筛选规则，适用于快速构建字典。示例：将遥操作器原始动作（如 {“shoulder”: 0.2, “gripper”: 0.9}）转换为带前缀的数据集格式：

# 原始遥操作动作
teleop_action = {"shoulder": 0.2, "elbow": 0.5, "gripper": 0.9}

# 字典推导式：先定义键值动作（添加前缀），再迭代
dataset_action = {
    f"action.{key}": value  # 动作：键添加前缀，值保持不变
    for key, value in teleop_action.items()  # 迭代范围：遥操作动作字典
    if key != "gripper"  # 筛选条件：排除 gripper（假设无需记录）
}


print(dataset_action) 
# 输出：{"action.shoulder": 0.2, "action.elbow": 0.5}

再来看看几个例子：

    features = {}
    joint_fts = {key: ftype for key, ftype in hw_features.items() if ftype is float}

遍历 hw_features 的所有键值对（key, ftype），仅保留 值为 float 类型 的键值对（即电机角度等浮点型特征）。若hw_features 含 {“shoulder.pos”: float, “camera”: (480,640,3)}，则 joint_fts 为 {“shoulder.pos”: float}。函数的作用就是筛选电机特征。实际打印如下：

joint_fts : {'shoulder_pan.pos': <class 'float'>, 'shoulder_lift.pos': <class 'float'>, 'elbow_flex.pos': <class 'float'>, 'wrist_flex.pos': <class 'float'>, 'wrist_roll.pos': <class 'float'>, 'gripper.pos': <class 'float'>}

    cam_fts = {key: shape for key, shape in hw_features.items() if isinstance(shape, tuple)}

遍历 hw_features，仅保留 值为元组类型 的键值对（即相机尺寸等元组特征，如 (高, 宽, 通道数)）。示例：若 hw_features 含 {“camera”: (480,640,3)}，则 cam_fts 为 {“camera”: (480,640,3)}。函数的作用就是筛选相机特征。

cam_fts : {'handeye': (480, 640, 3), 'fixed': (480, 640, 3)}
Action Features: {'action': {'dtype': 'float32', 'shape': (6,), 'names': ['shoulder_pan.pos', 'shoulder_lift.pos', 'elbow_flex.pos', 'wrist_flex.pos', 'wrist_roll.pos', 'gripper.pos']}}

面向对象

三大特性

（1）封装

class MotorsBus:
    def __init__(self, port):
        self.port = port
        self._private_var = "内部使用"  # 私有变量

    def public_method(self):
        return self._private_var  # 通过公共方法访问私有数据

（2）继承

# 基类
class MotorsBus(abc.ABC):
    def __init__(self, port):
        self.port = port

    @abc.abstractmethod
    def write_calibration(self):
        pass

# 子类
class FeetechMotorsBus(MotorsBus):
    def write_calibration(self):
        # 具体实现
        pass

（3）多态

# 同一个接口，不同实现
bus1 = FeetechMotorsBus(port)
bus2 = DynamixelMotorsBus(port)

# 调用相同方法，但执行不同逻辑
bus1.write_calibration()  # Feetech 的实现
bus2.write_calibration()  # Dynamixel 的实现

抽象基类

定义抽象接口

import abc

class MotorsBus(abc.ABC):
    @abc.abstractmethod
    def write_calibration(self):
        pass  # 强制子类实现

子类实现

class FeetechMotorsBus(MotorsBus):
    def write_calibration(self):
        # 具体实现
        for motor, calibration in calibration_dict.items():
            self.write("Homing_Offset", motor, calibration.homing_offset)

类型注解

class MotorsBus(ABC):
    """电机总线基类"""

    def __init__(self, port: str, motors: Dict[str, 'Motor']):
        # 基础类型注解
        self.port: str = port
        self.motors: Dict[str, 'Motor'] = motors

        # 可选类型注解
        self.calibration: Optional[Dict[str, 'MotorCalibration']] = None

        # 延迟初始化的类型注解（仅声明类型，不赋值）
        self.port_handler: 'PortHandler'
        self.packet_handler: 'PacketHandler'

        # 私有变量类型注解
        self._comm_success: int
        self._no_error: int

父类只声明变量的类型，子类需要实现。

class FeetechMotorsBus(MotorsBus):
    """Feetech 电机总线实现"""

    def __init__(self, port: str, motors: Dict[str, 'Motor']):
        super().__init__(port, motors)

        # 子类中的具体初始化
        import scservo_sdk as scs
        self.port_handler = scs.PortHandler(self.port)  # 实际赋值
        self.packet_handler = scs.PacketHandler(0)

异步编程

为什么需要async

在传统python中

def download_file():
    data = requests.get("https://example.com/file")
    return data

如上示例，代码是阻塞了，程序必须等请求返回结果后才能执行下一行，但如果要同时下载上百个文件，就会非常慢，为了解决这种I/O阻塞问题，python提供了异步编程，让多个任务可以并发运行（并非正在并行，但效率更高）。

基本的语法

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

asyncio.run(say_hello())

• async def：定义一个协程函数（coroutine function），调用它不会立即执行，而是返回一个协程对象（coroutine object）。
• await：挂起当前协程，等待另外一个异步操作完成后再继续执行。

可以理解会把say_hello任务放到一个线程中运行。

那如何实现“并发”了？

import asyncio

async def download(name, delay):
    print(f"开始下载 {name}")
    await asyncio.sleep(delay)
    print(f"下载完成 {name}")

async def main():
    await asyncio.gather(
        download("文件1", 2),
        download("文件2", 1),
        download("文件3", 3),
    )

asyncio.run(main())

上面的示例就会“并发”下载3个文件，输出顺序不是严格依次的，因为 await asyncio.sleep() 会释放控制权，其他任务可同时运行。 print(f”开始下载 {name}”)这句会交叉打印，而不是依次文件1、文件2、文件3排序严格运行。可以理解会把download放到了3个线程中运行。

async with

这是异步上下文管理器，普通的with会调用：

__enter__() 和 __exit__()

对于上下文管理器语法，会给上下文对象起一个变量名，语法为：

with 表达式 as 变量:
    ...

示例

with open("file.txt") as f:
    data = f.read()

with包括的上下文中创建了对象f，在离开with上下文中，会自动销毁f。普通的with可以自动帮助打开文件->执行enter，读取内容，离开代码块是自动关闭文件，执行exit。所以这里的with是资源安全保护壳，不管有没有异常都会自动清理资源。

在异步世界里，有些资源（比如网络连接、数据库连接、WebSocket）也需要“自动清理”，但它们的清理动作本身是异步的，必须 await 才能完成。于是python提供了async with。他会自动调用。

__aenter__()  # 异步进入
__aexit__()   # 异步退出

下面举个例子：

import httpx
import asyncio

async def fetch(url):
    async with httpx.AsyncClient() as client:  # 注意这里
        response = await client.get(url)
        return response.text

async def main():
    html = await fetch("https://www.python.org")
    print(len(html))

asyncio.run(main())

上面的as client是创建异步http客户端的对象，Python 会调用它的 aenter() 方法（异步版的 enter()），这个方法返回的对象会被赋值给你在 as client 中定义的变量，离开 with 块时，会自动调用 aexit() 关闭连接、释放资源。

除了async with外，还有async for，也是类似的用法。下面举例：

import asyncio
import websockets

async def listen():
    async with websockets.connect("wss://echo.websocket.org") as ws:
        await ws.send("hello")
        async for message in ws:   # 持续监听消息流
            print("收到：", message)

asyncio.run(listen())

闭包、函数对象与回调注册机制

    def register_service(self, sd: ServiceDescriptor) -> None:

        async def impl(arguments: Dict[str, Any]) -> Any:
            return self._tool_router.call_tool(sd.full_name, arguments or {})

        self._server.add_tool(
            impl,
            name=sd.full_name,
            description=sd.description,
        )

如上示例，impl作为回调函数注册到self._server.add_tool作为回调函数，从C语言转到python后可能会有一个疑问？

为什么impl函数在离开register_service函数作用域后依然不会被销毁？

函数是对象

def foo():
    print("hi")

bar = foo    # 函数对象赋给变量
bar()        # 调用

在python中，函数本质上是对象，意味着函数可以赋值给变量，存入列表或字典，作为参数传递，作为返回值返回，持久化引用（例如在回调系统中长期存活）。

因此：函数与普通对象无异，只要有引用，它就不会被销毁。

闭包：内部函数会”捕获”外层变量

如果在函数内部在定义函数，并且内部函数使用了外部变量，那么python会自动创建闭包（closure）。

def outer():
    x = 42
    def inner():
        print(x)
    return inner

f = outer()
f()   # 输出 42

这里 inner 捕获了外部的变量 x，即使 outer() 已经结束，x 依然存在于 inner.closure 中。

闭包让内部函数拥有独立于作用域的状态。

作用域结束 ≠ 对象销毁

python中必须要区分两类东西：

项目	是否会在函数返回后销毁？	包含内容
栈帧/作用域（Stack Frame）	✔ 会销毁	局部变量表、执行上下文
对象本身（Object）	✘ 不会销毁（只要有引用）	函数对象、闭包变量、列表、字典等

这就是为什么函数内部定义的内部函数，外部变量被闭包捕获后，不会随着外层结束而消失。

决定对象是否被销毁的只有”引用计数”，只有当引用计数变为0对象才会被销毁。