☰
F´ GDS History 机制深度解析:ChronologicalHistory、TestHistory 与 RamHistory 的设计与实战
2026/9/25 3:44:26 网站建设 项目流程
  • 嵌入式
  • 系统编程

【免费下载链接】fprime

F´ - A flight software and embedded systems framework

项目地址:https://gitcode.com/gh_mirrors/fpri/fprime
点击查看免费下载

本文围绕 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入队顺序(默认)定义所有历史应具备的接口契约
ChronologicalHistoryFSW 时间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(基类)ChronologicalHistoryTestHistoryRamHistory
模块fprime_gds.common.history.historyfprime_gds.common.history.chronofprime_gds.common.history.testfprime_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()基于"入队顺序即创建顺序"的假设不再成立,其返回值只是最新时间的近似值。文档给出了两个修复方向:

  1. 将 GDS 中的 Ram 历史替换为ChronologicalHistory(即用时间重排取代入队序假设);
  2. 让 Test API 订阅全部数据对象,在对象入队时实时计算最新时间。

ERT 排序的落地计划

用户指南 Idiosyncrasies 部分给出了在ChronologicalHistory中支持 ERT(地面接收时间)排序的完整建议:

  1. 为数据对象类增加 TimeType 字段与get_ert_time()访问器;
  2. 让 GDS 在某个环节记录 ERT;
  3. 保留 Test API 构造器与子历史函数中的fsw_order参数,并将其透传给ChronologicalHistory构造器;
  4. 修改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 GuideTest API 总览、搜索作用域、使用模式与反模式
GDS OverviewF´ 项目与 GDS 的顶层说明
Integration Test APIIntegrationTestAPI 类的 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

项目地址:https://gitcode.com/gh_mirrors/fpri/fprime
点击查看免费下载
上一篇:ReShade资源管理:缓冲区、纹理和着色器的创建与优化终极指南
下一篇:Obsidian模板终极指南:3分钟打造专业笔记系统

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询