• 科学计算
• 科研
• 数据分析

本文基于 astropy 官方文档 Custom Cosmology I/O formats 展开，讲解如何向 astropy Cosmology 的统一 I/O 框架注册自定义格式：一是面向内存对象转换的to_format/from_format（convert_registry），二是面向文件读写的read/write（readwrite_registry）。读完后，你可以照着文中完整示例为任意数据形态（Row、JSON 等）实现可复用的 Cosmology 序列化/反序列化，并理解其底层注册、识别与回环验证机制。

框架概览：两套独立的注册表

astropy.cosmology.io包提供了统一接口来读取、写出和转换Cosmology对象。用户入口是Cosmology类上的四个方法（见 astropy/cosmology/io.py 的模块说明）：

• Cosmology.read/Cosmology.write：从文件读入 / 写出文件；
• Cosmology.from_format/Cosmology.to_format：在 Python 对象（如 dict、Table）与Cosmology之间转换。

这四个方法背后各挂着一套自定义注册表（而不是Table等使用的全局注册表）：

# astropy/cosmology/_src/io/connect.py readwrite_registry = io_registry.UnifiedIORegistry() # L36：read/write 用 convert_registry = io_registry.UnifiedIORegistry() # L156：from/to_format 用

在 astropy/cosmology/_src/core.py 中，它们以类变量形式绑定到Cosmology：

from_format: ClassVar = UnifiedReadWriteMethod(CosmologyFromFormat) to_format: ClassVar = UnifiedReadWriteMethod(CosmologyToFormat) read: ClassVar = UnifiedReadWriteMethod(CosmologyRead) write: ClassVar = UnifiedReadWriteMethod(CosmologyWrite)

因此，为某个格式注册自定义解析器时，需要拿到对应的注册表：

• Cosmology.from_format.registry（即convert_registry）——用于to/from_format；
• Cosmology.read.registry（即readwrite_registry）——用于read/write。

每注册一个格式，通常要提供三个函数：reader（解析为Cosmology）、writer（从Cosmology输出）、identifier（自动识别格式）。注册表本身的 API 位于 astropy/io/registry，通用框架说明见 docs/io/registry.rst。一个值得注意的机制是：CosmologyRead/CosmologyFromFormat在被子类调用时（例如FlatLambdaCDM.read(...)），会自动注入关键字参数cosmology=<该子类>，并校验用户显式传入的cosmology参数必须与该类或其__qualname__一致，否则抛出ValueError（见 astropy/cosmology/_src/io/connect.py）；整个注册调用还会包在add_enabled_units(cu)上下文中，使 astropy 宇宙学单位（如km / (Mpc s)）可用。

自定义 to/from_format 格式（convert_registry）

官方文档以astropy.table.Row为例，演示了一个完整的 |Cosmology| ↔ |Row| 转换器。文档特意只手写 "mapping" 级别的逻辑，而不直接复用内置解析器，以便展示最完整的模式。注意：真实的内置实现就在 astropy/cosmology/_src/io/builtin/row.py，其from_row的写法与下面示例几乎一致（同样是把 Row 转成 dict 后交给from_mapping）。

1. 编写 reader：Row → Cosmology

该函数应接收 1 个位置参数（行对象）和 2 个关键字参数：move_to_meta（多余字段如何处理）与cosmology（使用哪个Cosmology类）。关于元数据处理的详细说明可以通过Cosmology.from_format.help("mapping")查看。

>>> import copy >>> from astropy.cosmology import Cosmology >>> def from_table_row(row, *, move_to_meta=False, cosmology=None): ... # get name from column ... name = row['name'] if 'name' in row.columns else None ... meta = copy.deepcopy(row.meta) ... # turn row into mapping (dict of the arguments) ... mapping = dict(row) ... mapping['name'] = name ... mapping.setdefault("cosmology", meta.pop("cosmology", None)) ... mapping["meta"] = meta ... # build cosmology from map ... return Cosmology.from_format(mapping, move_to_meta=move_to_meta, ... cosmology=cosmology)

关键设计在于：reader 并不自己解析每个物理参数，而是把 Row 拼回一个 mapping，再委托给内置的"mapping"格式（Cosmology.from_format(mapping, ...)）。move_to_meta与cosmology两个参数原样透传，保证行为与内置格式一致。对照内置实现 astropy/cosmology/_src/io/builtin/mapping.py 中的from_mapping可以看到，其核心逻辑是：

1. 按rename映射重命名键，并通过_rename_map保证重命名后的键不与原键冲突（冲突会抛出ValueError）；
2. 从cosmology参数或 mapping 中的"cosmology"字段确定目标类（字符串会通过_COSMOLOGY_CLASSES解析为类）；
3. 用inspect.signature(cosmology)+bind_partial+apply_defaults把类签名中的默认值填进来，再逐个绑定 mapping 中出现的参数——这就是"用具体子类解析时，缺失参数会被类默认值补齐"的底层机制；
4. 剩余多余参数：若构造函数有**kwargs则收入其中，否则在move_to_meta=True时合并进meta（显式 metadata 优先），否则抛出TypeError: there are unused parameters ...。

然后注册到convert_registry：

>>> convert_registry.register_reader("astropy.row", Cosmology, from_table_row)

2. 编写 writer：Cosmology → Row

writer 只需要Cosmology对象和一个*args，用来吸收astropy.io.registry.UnifiedReadWrite（Cosmology.to_format的底层实现，见 connect.py）传递的多余信息：

>>> from astropy.table import QTable >>> def to_table_row(cosmology, *args): ... p = cosmology.to_format("mapping", cosmology_as_str=True) ... meta = p.pop("meta") ... # package parameters into lists for Table parsing ... params = {k: [v] for k, v in p.items()} ... return QTable(params, meta=meta)[0] # return row >>> convert_registry.register_writer("astropy.row", Cosmology, to_table_row)

两个细节值得注意：

• to_format("mapping", cosmology_as_str=True)会把类对象换成类名（如"FlatLambdaCDM"），使输出可以放进 Table 的str13列。参数语义可对照 mapping.py 中to_mapping的文档；
• 所有参数值被包进单元素列表（{k: [v] for ...}），因为 Table 列必须是序列；最后用QTable(...)[0]取出第一行即Row。这与内置to_row（builtin/row.py，内部调to_table后返回table[0]）的路径一致。

3. 编写 identifier 并注册

identifier 用于在用户不指定format时自动识别格式。签名是(origin, format, *args, **kwargs)：origin为"read"或"write"；对convert_registry而言，待识别对象位于args[1]（args[0]是目标类）：

>>> from astropy.cosmology import Cosmology >>> from astropy.table import Row >>> def row_identify(origin, format, *args, **kwargs): ... """Identify if object uses the Table format.""" ... if origin == "read": ... return isinstance(args[1], Row) and (format in (None, "astropy.row")) ... return False >>> convert_registry.register_identifier("astropy.row", Cosmology, row_identify)

4. 回环验证

注册完成后，即可在Cosmology.from_format/Cosmology.to_format中使用：

>>> from astropy.cosmology import Planck18 >>> row = Planck18.to_format("astropy.row") >>> row <Row index=0> cosmology name H0 Om0 Tcmb0 Neff m_nu Ob0 km / (Mpc s) K eV str13 str8 float64 float64 float64 float64 float64[3] float64 ------------- -------- ------------ ------- ------- ------- ----------- ------- FlatLambdaCDM Planck18 67.66 0.30966 2.7255 3.046 0.0 .. 0.06 0.04897 >>> cosmo = Cosmology.from_format(row) >>> cosmo == Planck18 # test it round-trips True

cosmo == Planck18为True说明整个往返链路（Cosmology → mapping → QTable/Row → mapping → Cosmology）无信息丢失。测试目录 astropy/cosmology/_src/tests/io/ 中对内置 row 格式也有等价的回环测试可以参考。

自定义 read/write 文件格式（readwrite_registry）

文件级格式与对象级格式的区别在于：reader 接收文件路径或文件对象，writer 负责落盘。官方文档给出了一个完整的 |Cosmology| ↔ JSON 序列化器示例，并且复用已注册的 "mapping" 格式来大幅简化实现——这是推荐做法。

1. 编写 reader：JSON → Cosmology

函数接收 1 个位置参数（文件对象或文件路径），其余**kwargs透传给Cosmology.from_format（由它处理元数据和目标Cosmology类，细节同Cosmology.from_format.help("mapping")）：

>>> import json, os >>> import astropy.units as u >>> from astropy.cosmology import Cosmology >>> from astropy.cosmology.io import readwrite_registry >>> def read_json(filename, **kwargs): ... # read file, from path-like or file-like ... if isinstance(filename, (str, bytes, os.PathLike)): ... with open(filename, "r") as file: ... data = file.read() ... else: # file-like : this also handles errors in dumping ... data = filename.read() ... mapping = json.loads(data) # parse json mappable to dict ... # deserialize Quantity ... for k, v in mapping.items(): ... if isinstance(v, dict) and "value" in v and "unit" in v: ... mapping[k] = u.Quantity(v["value"], v["unit"]) ... for k, v in mapping.get("meta", {}).items(): # also the metadata ... if isinstance(v, dict) and "value" in v and "unit" in v: ... mapping["meta"][k] = u.Quantity(v["value"], v["unit"]) ... return Cosmology.from_format(mapping, **kwargs) >>> readwrite_registry.register_reader("json", Cosmology, read_json)

这段代码的三个要点：

• 兼容路径与文件对象两种输入；文件对象分支的注释指出它"顺便处理写出时 dump 的错误"；
• u.Quantity不兼容原生 JSON，反序列化约定为{"value": ..., "unit": "..."}结构，主参数和meta中的 Quantity 都要还原；
• 解析出 dict 后不再自行构造Cosmology，而是交给Cosmology.from_format(mapping, **kwargs)——**kwargs中的cosmology/move_to_meta/rename等参数由 mapping 格式统一处理。

2. 编写 writer：Cosmology → JSON

writer 需要Cosmology对象与文件对象/路径，并需要一个布尔标志overwrite来定义覆盖已有文件的行为：

>>> def write_json(cosmology, file, *, overwrite=False, **kwargs): ... data = cosmology.to_format("mapping", cosmology_as_str=True) # start by turning into dict ... # serialize Quantity ... for k, v in data.items(): ... if isinstance(v, u.Quantity): ... data[k] = {"value": v.value.tolist(), "unit": str(v.unit)} ... for k, v in data.get("meta", {}).items(): # also serialize the metadata ... if isinstance(v, u.Quantity): ... data["meta"][k] = {"value": v.value.tolist(), "unit": str(v.unit)} ... ... if isinstance(file, (str, bytes, os.PathLike)): ... # check that file exists and whether to overwrite. ... if os.path.exists(file) and not overwrite: ... raise IOError(f"{file} exists. Set 'overwrite' to write over.") ... with open(file, "w") as write_file: ... json.dump(data, write_file) ... else: ... json.dump(data, file) >>> readwrite_registry.register_writer("json", Cosmology, write_json)

注意v.value.tolist()：把 ndarray 转成纯 Python 列表，避免json.dump对 numpy 类型报错；overwrite=False（默认）时若文件已存在则抛错，这是文件级 writer 的常见契约。

仓库中有一份与文档示例几乎同构、且在 CI 中持续运行的实现——单元测试 astropy/cosmology/_src/tests/io/test_json.py：它的read_json/write_json/json_identify三个函数与文档示例一一对应（测试版用Path与OSError，并把 Quantity 解析包进u.add_enabled_units(cu.redshift)上下文以支持红移单位），还额外验证了"用写入时同一子类读取部分信息文件"时缺失参数会由类默认值补齐（test_readwrite_json_subclass_partial_info）。这可以作为自己实现文件级格式时的参考模板。

3. identifier 与回环验证

>>> def json_identify(origin, filepath, fileobj, *args, **kwargs): ... """Identify if object uses the JSON format.""" ... return filepath is not None and filepath.endswith(".json") >>> readwrite_registry.register_identifier("json", Cosmology, json_identify)

文件级 identifier 的签名是(origin, filepath, fileobj, *args, **kwargs)，按 astropy/io/registry/base.py 中register_identifier的说明，filepath与fileobj任一可能为None；两者都为None时 identifier 只能依赖args。按文件扩展名判断是最简单的可靠策略。

注册后即可完成文件级读写回环：

>>> import tempfile >>> from astropy.cosmology import Planck18 >>> >>> file = tempfile.NamedTemporaryFile() >>> Planck18.write(file.name, format="json", overwrite=True) >>> with open(file.name) as f: f.readlines() ['{"cosmology": "FlatLambdaCDM", "name": "Planck18", "H0": {"value": 67.66, "unit": "km / (Mpc s)"}, "Om0": 0.30966, ... >>> >>> cosmo = Cosmology.read(file.name, format="json") >>> file.close() >>> cosmo == Planck18 # test it round-trips True

输出的 JSON 中可以看到cosmology以字符串类名出现、H0被展开为{"value": 67.66, "unit": "km / (Mpc s)"}，与 writer 的序列化约定完全对应。

注册表 API 细节与清理

注册与解除注册

从 astropy/io/registry/core.py 可以看到register_reader/register_writer的完整签名：

registry.register_reader(data_format, data_class, function, force=False, priority=0) registry.register_writer(data_format, data_class, function, force=False, priority=0) registry.register_identifier(data_format, data_class, identifier, force=False) # base.py L213

• 重复注册同一(format, class)组合且force=False时，会抛出IORegistryError（"already defined"）；
• priority用于在多个候选格式都匹配时择优，数值越高越优先；
• unregister_reader/unregister_writer/unregister_identifier用于移除注册；若格式不存在同样抛IORegistryError；
• get_reader(format, class)可查询当前注册的函数，未命中时错误信息中会附带该类的可用格式表（_get_format_table_str）。

官方文档示例在结尾特意清理了注册状态，避免示例污染 doctest 环境：

>>> from astropy.io.registry import IORegistryError >>> readwrite_registry.unregister_reader("json", Cosmology) >>> readwrite_registry.unregister_writer("json", Cosmology) >>> readwrite_registry.unregister_identifier("json", Cosmology) >>> try: ... readwrite_registry.get_reader("json", Cosmology) ... except IORegistryError: ... pass

这一"注册—测试—注销"的三段式与 test_json.py 中 fixture 的 setup/teardown 写法（注册时加force=True，yield后逐一unregister_*）一致，是写测试时的最佳实践。

查看可用格式

注册表会把注册结果反映到方法的help()与list_formats()中。当前内置格式（摘自 docs/cosmology/io/index.rst）：

>>> Cosmology.read.list_formats() Format Read Write Auto-identify ---------- ---- ----- ------------- ascii.ecsv Yes Yes Yes ascii.html Yes Yes Yes ascii.mrt Yes Yes Yes >>> Cosmology.write.list_formats() Format Read Write Auto-identify ----------- ---- ----- ------------- ascii.ecsv Yes Yes Yes ascii.html Yes Yes Yes ascii.latex No Yes Yes ascii.mrt Yes Yes Yes >>> Cosmology.to_format.list_formats() Format Read Write Auto-identify ----------------- ---- ----- ------------- astropy.cosmology Yes Yes Yes astropy.model Yes Yes Yes astropy.row Yes Yes Yes astropy.table Yes Yes Yes mapping Yes Yes Yes yaml Yes Yes No

"Auto-identify" 列即为是否注册了 identifier；如yaml只有 Yes/Yes/No，说明它必须显式传format="yaml"。你注册的第三方格式会同样出现在这两张表中。

实践要点小结

1. 先想好复用哪一层：能用内置"mapping"格式做"胶水"的，就不要直接手写参数解析——两个官方示例（Row、JSON）都是这样实现的，行为自动与move_to_meta、cosmology、rename等语义保持一致；
2. 两套注册表不要混用：对象级格式注册到convert_registry（Cosmology.from_format.registry），文件级格式注册到readwrite_registry（Cosmology.read.registry），注册错表会导致list_formats中出现却调不到；
3. identifier 决定自动识别能力：文件级建议用扩展名等强信号（JSON 例），对象级建议用isinstance判断（Row 例），并在format已显式指定时放行；
4. Quantity 需要自定义编解码：u.Quantity与 JSON 等文本格式不兼容，{"value": ..., "unit": ...}约定 + 遍历meta是两个官方示例共同采用的方案；
5. 以回环断言收尾：Cosmology.from_format(cosmo.to_format(...)) == cosmo（或文件级的Cosmology.read）是验证实现正确性的最简方式，官方示例与 astropy/cosmology/_src/tests/io/ 下的测试均以此为标准动作。

更多背景：I/O 总览见 docs/cosmology/io/index.rst，内置格式（ascii.ecsv、astropy.table、mapping、yaml等）的参数细节见 docs/cosmology/io/builtin.rst，字段重命名与子类读取行为见 docs/cosmology/io/details.rst，通用 I/O 注册框架见 docs/io/registry.rst。

• 科学计算
• 科研
• 数据分析