- 嵌入式
- 系统编程
【免费下载链接】fprime
F´ - A flight software and embedded systems framework
本文围绕 F´(F Prime)飞行软件框架 GDS(地面数据系统)集成测试 API 中的History(历史)存储机制展开,系统梳理 GDS 内现有的全部历史实现:抽象基类History以及三个具体子类ChronologicalHistory、TestHistory、RamHistory。文章完整继承 histories.md 的接口文档内容,并结合 IntegrationTestAPI 用户指南 与 StandardPipeline 文档,讲解这些历史对象如何注册到解码器、如何按 FSW 时间或接收顺序排序、如何通过谓词过滤与检索,以及它们在测试用例中的实际用法。读完本文,你将掌握 GDS History 的接口契约、三种实现的差异与选型依据,以及基于start/谓词/时间戳做历史检索和测试标记的完整实战方案。
一、GDS 中的 History 概览:为解码器服务的顺序化对象存储
在 F´ 的 GDS 架构中,History 承担着一个关键职责:与解码器(decoder)注册绑定,接收流入的数据对象并按顺序保存,供上层工具随时检索。它本质上是一个有顺序的对象容器——无论是遥测通道更新(ChData)、事件消息(EventData)还是命令记录,都由各自的解码器解码后通过回调推入历史。
histories.md明确指出,当前 GDS 内共有以下几种历史实现,全部继承自同一个超类:
| Classes | 排序依据 | 主要用途 |
|---|---|---|
| History | 入队顺序(默认) | 定义所有历史应具备的接口契约 |
| ChronologicalHistory | FSW 时间 | Test API 的主历史,自动按飞行软件时间重排 |
| TestHistory | 接收顺序 | IntegrationTestAPI 的可选历史,支持谓词检索 |
| RamHistory | 入队顺序 | Standard Pipeline 使用,多个 GDS 工具的默认历史 |
这些类的模块级文档均以 Sphinxautomodule指令生成,对应的 RST 源文件位于 docs/UsersGuide/dev/testAPI/sphinx/source/histories.rst。文档中的类路径统一为fprime_gds.common.history.*下的history、chrono、test、ram四个模块。
一个值得注意的设计原则(来自超类文档):默认行为是保持对象“入队顺序”;如果某个子历史想要维护不同的顺序,必须向用户说明清楚,并且仍然要支持History类定义的全部调用接口。这意味着“顺序策略”与“接口契约”是解耦的——调用方始终通过同一套方法访问历史,至于内部如何排序,由具体实现决定。
二、History 抽象基类:接口契约(superclass)
History是全部历史实现的超类,定义了 GDS 内一个历史应有的接口。它被描述为"An ordered history that defines what interfaces a history should have within the GDS",即有序历史 + 接口规范的双重角色。作者:koran。
从源码结构看,所有具体历史都直接或间接Bases: fprime_gds.common.history.history.History,因此基类的方法签名就是整个 History 家族统一的 API 面。六个核心接口如下:
__init__()
构造器,用于完成历史的初始化设置。
data_callback(data_object)
数据回调,用于把对象压入历史。这正是历史与解码器的连接点:解码器解码出数据对象后调用该回调,历史决定是否存储、如何排序。
- 参数
data_object:要存储的对象。
retrieve(start=None)
从历史中取回对象。如果不指定起始点,返回全部对象;如果指定起始点,则返回从该起始点开始、一直到最新对象(按历史的顺序)的子列表。
- 参数
start:历史顺序中的一个位置。文档强调:start应当始终可以用索引(int)来指定。 - 返回:一个有序的对象列表。
retrieve_new()
取回自上次调用retrieve或retrieve_new以来新入队的对象,以有序列表形式返回。这是“增量消费”的关键接口——测试代码先记录当前位置,等待一段时间后再取新增数据,从而避免重复处理已读对象。
clear(start=None)
从历史中清除对象。指定起始点的清除,会使得清除后start成为历史中(相对于该历史顺序的)最早元素;即清除“从最老到 start 之前(不含 start)”的所有对象,start 及其之后的对象被保留。
- 参数
start:历史顺序中的一个位置,应当始终可以用索引(int)指定。
size()
历史中对象数量的访问器。
这六个方法构成了所有历史的公共契约。其中retrieve/retrieve_new/clear三者在语义上相互关联:retrieve用位置取对象、retrieve_new用“上次访问点”取增量、clear用位置做截断清理,三者共同支撑起测试场景中"记录位置 → 发送命令 → 检索新数据"的典型流程。
三、ChronologicalHistory:按 FSW 时间排序的主历史
ChronologicalHistory是Test API 的主历史(primary history),定义于fprime_gds.common.history.chrono模块,作者:koran。它最核心的特性是:基于 FSW(飞行软件)时间对对象重新排序。也就是说,即使数据对象到达 GDS 的顺序与飞行软件产生它们的顺序不一致,历史内部也会把对象排列成正确的时序——这对遥测/事件断序到达的场景至关重要。
类文档描述为:"A chronologically-ordered history that relies on predicates to provide filtering, searching, and retrieval operations. This history will re-order itself based on FSW time." 同时它相对基类新增了两项能力:用谓词指定start和支持 Python 的方括号(bracket)访问。
__init__(filter_pred=None)
构造器。若传入过滤谓词(filter predicate),历史会丢弃(drop)所有不满足该谓词的对象。
- 参数
filter_pred:可选,用于过滤流入data_object的谓词。
这一机制使得"只关注某类事件/通道"的子历史可以天然地从源头就进行裁剪,而不是先全量存储再检索。
data_callback(data_object)
数据回调。与基类不同的是,只有满足过滤谓词的对象才会被加入历史。
retrieve(start=None)
取回对象。若指定起始点,返回从start到最新对象(按时间序)的子列表。
- 参数
start:可选的第一个取回对象,可以是索引(int)或谓词(predicate)。 - 返回:按时间序排列的对象列表。
- 边界行为(文档明确注明):如果没有对象满足起始谓词,或者索引大于历史长度,返回空列表(而不是报错或返回全部)。
retrieve_new(repeats=False)
按时间序取回尚未通过retrieve或retrieve_new访问过的对象。
- 值得注意:该方法的签名比基类多了一个可选参数
repeats(默认False),这是ChronologicalHistory对增量读取接口的扩展,可用于控制对已读对象的处理策略。 - 返回:按时间序的对象列表。
clear(start=None)
清除对象。指定起始点后,start成为清除后历史中最早的元素;若start是谓词,则以最早满足该谓词的对象作为起始点。
- 参数
start:可选,指示第一个要移除的对象,可以是谓词、TimeType 或顺序中的索引。 - 边界行为:若没有对象满足起始谓词,或索引大于历史长度,则清除全部对象(与
retrieve的空列表行为形成对应:一个是取回失败返回空,一个是清除失败全清空)。
size()与__len__()
两者都是历史对象数量的访问器,返回 int。__len__的存在意味着历史可以直接配合 Python 的len(history)使用。
__getitem__(index)
__getitem__是 Python 的特殊方法,支持方括号访问。文档示例:
item = history[2] # 返回历史中的第二个元素该能力让历史对象具备了类似列表的随机访问体验,方便在调试和断言中直接按位置取值。
私有辅助方法(排序与定位的核心实现)
ChronologicalHistory的实现核心是三个私有方法,它们揭示了“按 FSW 时间排序”的具体机制:
_ChronologicalHistory__insert_chrono(data_object, ordered):从现有顺序的尾部向前遍历,把数据对象插入到正确的时间位置,并返回插入位置的索引。这是"重排"的基础原语——每次入队都做一次插入排序,从而维持整体时间序。其前置条件是数据对象必须提供get_time()方法(文档明确标注:Must have aget_time()method),这解释了为什么 GDS 的数据对象(如 ChData/EventData)都携带时间戳访问器。_ChronologicalHistory__get_index(start, ordered):把start(谓词、TimeType 时间戳或索引)解析为给定列表中的具体索引。_ChronologicalHistory__clear_list(start, ordered):同样把start解析为索引,供clear使用。
从这三个辅助方法的签名可以推断:ChronologicalHistory支持三种start指定方式——谓词、TimeType 时间戳、整数索引,且统一通过__get_index归一化为索引后操作,这是它能同时支撑"按事件标记位置""按时间标记位置""按序号标记位置"的根本原因。
四、TestHistory:保持接收顺序的可选历史
TestHistory是IntegrationTestAPI 的可选历史,定义于fprime_gds.common.history.test模块,作者:koran。它的定位与ChronologicalHistory形成鲜明对照:保持对象入队时的顺序(receive order),不做时间重排,同时支持谓词检索与方括号访问。
模块文档:"A receive-ordered history that relies on predicates to provide filtering, searching, and retrieval operations",以及"maintains the object order it was enqueued with"——即维护的是接收顺序而非 FSW 时间顺序。
__init__(filter_pred=None)/data_callback(data_object)
与ChronologicalHistory完全同构:构造时可传入过滤谓词,回调只接收满足谓词的对象。
retrieve(start=None)
取回对象,start可以是索引(int)或谓词;若没有对象满足起始谓词或索引越界,返回空列表。注意:与ChronologicalHistory不同,TestHistory的start不支持 TimeType 时间戳——因为接收顺序历史没有可重排的时间轴,位置只能由序号或谓词刻画。
retrieve_new()
取回自上次retrieve/retrieve_new以来新入队且未被访问过的对象(接收顺序)。
clear(start=None)
清除start之前的所有对象,start可以是索引或谓词;若无可满足项则全部清除。
size()/__len__()/__getitem__(index)
数量访问器与方括号随机访问,行为与ChronologicalHistory一致。
_TestHistory__get_index(start)
把start(谓词或索引)解析为顺序中的索引。对比ChronologicalHistory的__get_index(start, ordered),这里没有 TimeType 分支、也不需要传入ordered列表(接收序即存储序),印证了两者在“定位起点”能力上的差异。
选型启示:如果你希望断言"数据按飞行软件产生的时间顺序排列"(例如验证事件发生的先后),应使用ChronologicalHistory;如果你只关心"收到数据的先后"(例如验证命令触发的即时响应链),TestHistory更直观且开销更低。
五、RamHistory:Standard Pipeline 的默认内存历史
RamHistory定义于fprime_gds.common.history.ram模块,作者:lestarch。它的定位与前面两者都不同——它不是为 Test API 设计的,而是被 Standard Pipeline 使用,是多个 GDS 工具的默认历史。
模块文档原文:"Ram history is used by the Standard Pipeline making it the default history for several GDS Tools." 其实现本质是一个简单的、驻留 RAM 的历史:文档明确承认它"isn't exactly robust nor persistent"(既不健壮也不持久),并且"it is in the RAM, it is driven from the decoders object, which should run off the middle-ware layer"——即它由解码器对象驱动,而解码器运行在中间件(middleware)层之上。类级文档则将其描述为"Chronological variant of history",与模块级"简单内存实现"的措辞略有差异,但从其接口看,它仍以简单顺序存储为主。
__init__()
构造器,用于设置内存存储。
data_callback(data_object)
数据回调,存储对象。
retrieve(start=None)
取回对象。与其他实现不同,其参数说明为"return all objects newer than given start time"——即start以起始时间(TimeType 语义)作为边界,返回所有比该时间更新的对象。这符合其"GDS 工具默认历史"的定位:工具层通常以时间窗口拉取数据。
retrieve_new()
取回自上次retrieve/retrieve_new以来未被访问过的对象(时间序)。
clear(start=None)
清除对象。指定起始点后,start 索引处的元素成为清除后历史中最早的元素;start为顺序中的位置(int)。
size()
对象数量访问器。
局限与权衡:RamHistory的"内存驻留"特性决定了它适合 GDS 工具链的轻量默认场景——简单、零依赖、随解码器生命周期运行;代价是进程退出即丢失、无法承载持久化需求。histories.md的已知问题部分也提到,Test API 的get_latest_time()之所以不够精确,正是因为它假设历史按入队顺序即创建顺序排列;而其中一个被提出的修复方向,就是把 GDS 中的 Ram 历史替换为ChronologicalHistory(详见第八节)。
六、三种实现的横向对比
| 维度 | History(基类) | ChronologicalHistory | TestHistory | RamHistory |
|---|---|---|---|---|
| 模块 | fprime_gds.common.history.history | fprime_gds.common.history.chrono | fprime_gds.common.history.test | fprime_gds.common.history.ram |
| 顺序策略 | 入队顺序(默认约定) | 按 FSW 时间重排(插入排序) | 按接收/入队顺序 | 入队顺序为主 |
| 过滤谓词 | 无 | filter_pred支持 | filter_pred支持 | 无 |
start支持 | 索引(int) | 索引 / 谓词 / TimeType | 索引 / 谓词 | 时间(TimeType)/ 索引 |
| 方括号访问 | 未定义 | __getitem__ | __getitem__ | 未定义 |
retrieve_new签名 | retrieve_new() | retrieve_new(repeats=False) | retrieve_new() | retrieve_new() |
| 空/越界边界 | 未特别注明 | retrieve 返回空列表;clear 全清 | 同左 | 未特别注明 |
| 典型使用者 | 接口契约 | Test API 主历史 | Test API 可选历史 | Standard Pipeline 默认 |
七、与 IntegrationTestAPI / StandardPipeline 的协作实战
测试历史(Test History)的获取与排序开关
IntegrationTestAPI 通过get_command_test_history、get_telemetry_test_history、get_event_test_history向用户暴露测试历史。用户指南(user_guide.md)指出:API 的构造器与子历史创建函数都带有一个fsw_order参数(默认按 FSW 时间排序,即ChronologicalHistory语义);当fsw_order=False时,历史退化为接收顺序(TestHistory语义)。这解释了本文第四节的选型启示在 API 层如何落地——同一个 API,通过一个布尔开关切换两种历史实现。
用start标记测试位置:时间序 vs 接收序
用户指南明确给出了两种历史下的“记录点”写法,是理解 History 接口最直接的实战案例:
- 时间序历史(默认):用
get_latest_time()取得当前最新 FSW 时间戳作为标记,发送命令后以该时间戳作为start检索:
fsw_start = self.api.get_latest_time() self.api.send_command("TEST_CMD_1") results = self.api.assert_telemetry("Counter", start=fsw_start)- 接收序历史:用历史当前大小作为索引标记(即调用
size()得到的位置):
ro_start = self.api.get_telemetry_test_history().size() self.api.send_command("TEST_CMD_1") results = self.api.assert_telemetry("Counter", start=ro_start)这两种写法分别对应ChronologicalHistory(start 支持 TimeType)与TestHistory(start 支持索引)的能力差异。此外,用户指南还强调:由于 Test API 的历史支持重排,TimeType 时间戳是start最可靠的标记方式;谓词也可作为start,例如"从收到某个特定 EVR 之后才开始搜索"。
子历史(sub-history)的创建、注册与注销
API 提供get_event_subhistory/get_telemetry_subhistory系列方法创建子历史,其中fsw_order=False可创建按接收序(ERT 序)排列的子历史。子历史创建后会自动注册到 GDS,从其对应解码器自动接收数据对象(这与本文开头"History 注册到解码器"的机制完全对应);但不会被 Test API 管理——测试用例结束时不会自动清除或注销它。移除子历史是永久性的:remove_event_subhistory会将其从 GDS 取消订阅,之后不再接收新对象。
Standard Pipeline 中的历史注册链路
standard_pipeline.md展示了RamHistory在 GDS 工具链中的装配方式。StandardPipeline 的生命周期为 setup → register → run → terminate,其中与历史相关的关键方法包括:
setup_history():创建一组历史对象,用于存储解码器的输出——这是 Ram 历史被装配进流水线的入口;register_event_consumer(history)/register_telemetry_consumer(history)/register_command_consumer(history):把历史注册为相应解码器的消费者;remove_event_consumer(history)/remove_telemetry_consumer(history)/remove_command_consumer(history):注销消费者(若未注册会报错,并返回布尔值表示是否移除成功);get_event_history()/get_channel_history()/get_command_history():对外提供历史访问器。
这套"注册消费者(History)→ 解码器回调data_callback→ 历史存储 → 工具按需retrieve"的链路,正是全部 History 实现统一接口设计的价值所在:解码器无需关心下游是 Ram 还是 Chronological 实现,只需调用统一的数据回调。
八、已知问题与演进方向
get_latest_time()不精确
用户指南的 Known bugs 部分指出:因为数据对象到达 GDS 的接收顺序可能与 FSW 创建顺序不一致,get_latest_time()基于"入队顺序即创建顺序"的假设不再成立,其返回值只是最新时间的近似值。文档给出了两个修复方向:
- 将 GDS 中的 Ram 历史替换为
ChronologicalHistory(即用时间重排取代入队序假设); - 让 Test API 订阅全部数据对象,在对象入队时实时计算最新时间。
ERT 排序的落地计划
用户指南 Idiosyncrasies 部分给出了在ChronologicalHistory中支持 ERT(地面接收时间)排序的完整建议:
- 为数据对象类增加 TimeType 字段与
get_ert_time()访问器; - 让 GDS 在某个环节记录 ERT;
- 保留 Test API 构造器与子历史函数中的
fsw_order参数,并将其透传给ChronologicalHistory构造器; - 修改
ChronologicalHistory,在清除、__insert_chrono()与__get_index()三个位置根据模式选择使用get_time()还是get_ert_time()。
更好的历史标记:get_current_marker()
作为 ERT 工作的一部分,文档建议各历史实现增加get_current_marker()方法,由各实现自行决定最合适的标记方式:ChronologicalHistory应使用 TimeType,Ram 与 Test History 应使用索引。这与第七节"时间序用时间戳、接收序用索引"的实战约定完全一致,是从接口层统一"记录位置"语义的演进方向。
九、相关文档与进一步阅读
| Quick Links | 说明 |
|---|---|
| Integration Test API User Guide | Test API 总览、搜索作用域、使用模式与反模式 |
| GDS Overview | F´ 项目与 GDS 的顶层说明 |
| Integration Test API | IntegrationTestAPI 类的 Sphinx 生成文档 |
| Histories | 本文对应文档(History 家族全部实现) |
| Predicates | 谓词库:过滤、搜索与断言的基础设施 |
| Test Logger | 测试日志(.xlsx 输出)实现 |
| Standard Pipeline | 历史注册与解码器消费的标准装配链路 |
| TimeType Serializable | 时间戳类型,ChronologicalHistory排序的依据 |
| Sphinx 源码:histories.rst | 四个历史模块的automodule文档源文件 |
如果需要实际运行层面的参考,仓库中的 Ref App 集成测试 与 RPI 集成测试 展示了 Test API 在真实部署上的完整用法;contents.md 则提供了全部 Sphinx 生成文档的索引入口。
- 嵌入式
- 系统编程
【免费下载链接】fprime
F´ - A flight software and embedded systems framework
相关推荐
F´ GDS 测试 API 中的 Histories 架构:History、ChronologicalHistory、TestHistory 与 RamHistory 全解析
F´ GDS 测试 API 中的 Histories 架构:History、ChronologicalHistory、TestHistory 与 RamHist
嵌入式系统编程F´ GDS Histories 深度解析:Chronological / Test / Ram 三种历史实现的机制、接口与集成测试实战
F´ GDS Histories 深度解析:Chronological / Test / Ram 三种历史实现的机制、接口与集成测试实战 F´(F Prime)
嵌入式系统编程F´ GDS Framing Plugin 深入指南:帧封装与解帧协议的定制实现
F´ GDS Framing Plugin 深入指南:帧封装与解帧协议的定制实现 导读 F´(F Prime)飞行软件与地面数据系统(GDS)之间通过字节流、缓
嵌入式系统编程
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考