自定义宇宙学I/O格式#
自定义宇宙学至/自定义格式#
自定义表示格式也可以注册到Astropy Cosmology I/O框架中以供这些方法使用。有关该框架的详细信息，请参阅 I/O注册表 (astropy.io.registry ） 。注意事项 Cosmology to/from_format 使用自定义注册表，可在 Cosmology.<to/from>_format.registry 。
作为示例，下面是一个 Row 转换器。我们可以也应该使用内置的解析器，比如 QTable ，但为了展示一个更完整的示例，我们仅限于“映射”解析器。
我们首先定义函数来解析 Row 变成一个 Cosmology 。该函数应该接受1个位置参数、行对象和2个关键字参数，用于说明如何处理额外的元数据以及使用哪个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)
convert_registry.register_reader("astropy.row", Cosmology, from_table_row)
下一步是执行相反操作的函数：解析 Cosmology 变成一个 Row 。此函数只需要宇宙学对象和 *args 吸收传递的不需要的信息 astropy.io.registry.UnifiedReadWrite (实现|Cosmology.to_Format|)。
>>> 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)
最后，我们编写一个函数来帮助进行格式自动识别，然后将所有内容注册到 astropy.io.registry 。
>>> 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)
现在可以在|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
自定义宇宙学读者/作者#
自定义 read / write 格式可注册到Astropy Cosmology I/O框架中。有关该框架的详细信息，请参阅 I/O注册表 (astropy.io.registry ） 。注意事项 Cosmology read/write 使用自定义注册表，可在 Cosmology.<read/write>.registry 。
作为一个例子，在下面我们将完整地计算出一个 Cosmology <->JSON(反)序列化程序。请注意，我们可以使用其他注册的解析器--这里是“映射”--来使实现更加简单。
我们首先定义将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.connect 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)
下一步是编写 Cosmology 致JSON。此功能需要宇宙学对象和文件对象/路径。我们还需要布尔标志“覆盖”来设置现有文件的行为。请注意 Quantity 与JSON本身不兼容。在这两个 write 和 read 方法，我们必须创建自定义解析器。
>>> 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)
最后，我们编写一个函数来帮助进行格式自动识别，然后将所有内容注册到 astropy.io.registry 。
>>> 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)
现在可以在|Cosmology.Read|和|Cosmology.Write|中使用注册的函数了。
>>> 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
:hide:
>>> 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