RKVision开发日志(二)

一、前言

在搭建多进程的框架的时候，初始化进程需要传递一些参数。本项目在参阅blakeblackshear/frigate: NVR with realtime local object detection for IP cameras后决定将所有的数据交给Pydantic管理。而我们的后端框架fastapi，也是用pydantic管理的，模块复用了。

Pydantic的优点有很多，他像一个功能更丰富的字典。比如：

class CameraConfig(BaseModel):
    name: str = Field(title="相机名称")
    width: int = Field(title="相机采集宽度")
    height: int = Field(title="相机采集高度")
    fps: Optional[int] = Field(title="相机采集帧率")

在上述配置类中，规定好了各个参数的类型，并用title字段添加了提示信息，对编程很友好。

重要的是，该model支持对字典的无缝转换。
比如我们可以用dict字典的数据格式先创建一个配置实例，然后转换为pydantic格式：

# dict -> pydantic
camera_config = {
    "name": "MIPI",
    "width": 800,
    "height": 600,
    "fps": 30
}
data = CameraConfig(**camera_config)
print(data)
print(data.name)
# 输出:
# name='MIPI' width=800 height=600 fps=30
# MIPI

支持从pydantic格式转换为dict格式：

# pydantic -> dict
camera_config = CameraConfig(name="MIPI",width=800,height=600,fps=30)
data = camera_config.model_dump()
print(data)
print(type(data))
print(data['name'])
# 输出:
# {'name': 'MIPI', 'width': 800, 'height': 600, 'fps': 30}
# <class 'dict'>
# MIPI

支持pydantic格式和json格式互相转换等。
支持嵌套的pydantic类：

# 官方文档例程
from typing import Optional
from pydantic import BaseModel
class Foo(BaseModel):
    count: int
    size: Optional[float] = None
class Bar(BaseModel):
    apple: str = 'x'
    banana: str = 'y'
class Spam(BaseModel):
    foo: Foo
    bars: list[Bar]
m = Spam(foo={'count': 4}, bars=[{'apple': 'x1'}, {'apple': 'x2'}])
print(m)
"""
foo=Foo(count=4, size=None) bars=[Bar(apple='x1', banana='y'), Bar(apple='x2', banana='y')]
"""
print(m.model_dump())
"""
{
    'foo': {'count': 4, 'size': None},
    'bars': [{'apple': 'x1', 'banana': 'y'}, {'apple': 'x2', 'banana': 'y'}],
}
"""

pydantic还支持参数范围校验，这个我认为是比较好用的点。

在实际使用场景中，使用pydantic一般是：从config.yaml中读取配置好的参数，程序中能正常访问各个字段的参数，发给各个模块初始化；在参数服务中能定期收集参数，并存储在文件中。

# yaml -> pandantic
import yaml
from pydantic import BaseModel

class ServerConfig(BaseModel):
    host: str
    port: int

# 1. 从 YAML 读取
with open("config.yaml", "r", encoding="utf-8") as f:
    data = yaml.safe_load(f)  # 将 yaml 转为 dict

# 2. 实例化对象
config = ServerConfig(**data)

# 3. 访问字段
print(config.host)  # 直接通过点语法访问，有自动补全

from app.core import AppConfig, CameraConfig, AlgoConfig, FrontendConfig, WebConfig
import yaml

config = AppConfig(
    cameras=[
        CameraConfig(name="MIPI", width=800, height=600, fps=30)
    ],
    algo=AlgoConfig(status="idle"),
    frontend=FrontendConfig(web=WebConfig(ip="127.0.0.1", port=8080))
)

# 写入 YAML
with open("test.yaml", "w", encoding="utf-8") as f:
    yaml.dump(config.model_dump(), f)
"""
test.yaml的内容：
algo:
  status: idle
cameras:
- fps: 30
  height: 600
  name: MIPI
  width: 800
frontend:
  web:
    ip: 127.0.0.1
    port: 8080
"""

三、对Pydantic的补充

参考frigate中，用斜杠访问数据的函数get_nested_object()，我觉得可能会用到，就直接抄过来了。另外，针对我们现在的业务场景，可以适当封装一下几种转换，毕竟我英语不好，上来一个model_dump()我确实反应不过来是干啥的。
针对上面验证过的几种数据格式，我们封装几种转换，分别是

def to_dict(self) -> dict[str, Any]:
    """将当前实例转换为字典"""
@classmethod
def from_dict(cls: Type[T], data: dict[str, Any]) -> T:
    """从字典创建实例"""
def to_yaml(self) -> str:
    """将当前实例转换为 YAML 字符串"""
@classmethod
def from_yaml(cls: Type[T], yaml_str: str) -> T:
    """从 YAML 字符串创建实例"""
def to_file(self, file_path: str):
    """直接保存到文件"""
@classmethod
def from_file(cls: Type[T], file_path: str) -> T:
    """从文件加载"""

测试调用：

config = AppConfig(
    cameras=[
        CameraConfig(name="MIPI", width=1080, height=600, fps=30)
    ],
    algo=AlgoConfig(status="idle"),
    frontend=FrontendConfig(web=WebConfig(ip="127.0.0.1", port=8080))
)
config.to_file("test.yaml")
config2 = AppConfig.from_file("test.yaml")
print(config2)

没什么问题了，可以继续对多进程的框架进行搭建了。