pycsamt.core.registry#

Public packer registry and serialization helpers for survey objects.

Functions

get_packer(kind)

Return the registered packer for kind or None.

list_packers()

List available packers as a mapping of kind to signatures.

pack_to_file(obj, path, *[, kind])

Serialize an object to .npz using a registered packer.

register_packer(kind, packer)

Register a serializer/deserializer pair for a given kind.

unpack_from_file(path, *[, kind])

Deserialize an object from .npz using a registered packer.

Classes

RegistryAPI(root, *[, manifest_name])

Convenience façade over Registry.
pycsamt.core.registry.register_packer(kind, packer)[source]#

Register a serializer/deserializer pair for a given kind.

Parameters:

  • kind (str) – Case-insensitive key that identifies the packer.

  • packer (tuple(callable, callable)) – Pair (pack, unpack). The first callable takes an object and returns a dict; the second takes that mapping and returns the reconstructed object.

Raises:

ValueError – If kind is empty or not a str.

Return type:

None

Notes

Registration is global to the current process via an internal dictionary. Re-registering the same kind will overwrite the previous pair.

Examples

>>> def pack(x): return {"v": x}
>>> def unpack(d): return d["v"]
>>> register_packer("toy", (pack, unpack))
>>> get_packer("toy") is not None
True

See also

get_packer, list_packers, pack_to_file, unpack_from_file

References

pycsamt.core.registry.get_packer(kind)[source]#

Return the registered packer for kind or None.

Parameters:

kind (str) – Case-insensitive key used at registration time.

Returns:

The (pack, unpack) pair if found, else None.

Return type:

tuple(callable, callable) or None

Examples

>>> _ = get_packer("missing") is None
>>> # After registering:
>>> # register_packer("toy", (pack, unpack))
>>> pk = get_packer("toy")
>>> callable(pk[0]) and callable(pk[1])
True

See also

register_packer, list_packers

pycsamt.core.registry.list_packers()[source]#

List available packers as a mapping of kind to signatures.

Returns:

Mapping kind -> "pack_name | unpack_name" for quick inspection.

Return type:

dict[str, str]

Notes

Only names available via __name__ are shown. For callables without a name, repr is used.

Examples

>>> isinstance(list_packers(), dict)
True

See also

register_packer, get_packer

pycsamt.core.registry.pack_to_file(obj, path, *, kind='bundle')[source]#

Serialize an object to .npz using a registered packer.

Parameters:

  • obj (Any) – Object to serialize. Must be supported by the chosen kind (i.e., its packer).

  • path (path-like) – Output file path. .npz is recommended.

  • kind (str, optional) – Packer name. Defaults to "bundle".

Returns:

The path actually written.

Return type:

pathlib.Path

Raises:

  • ValueError – If no packer is registered for kind.

  • OSError – For filesystem write errors.

Notes

Data are written with numpy.savez_compressed(), which stores arrays losslessly and compresses the archive.

Examples

Pack a bundle to disk:

>>> # Assume a TFBundle instance: bndl
>>> out = pack_to_file(bndl, "site001.npz", kind="bundle")
>>> out.name.endswith(".npz")
True

See also

unpack_from_file, register_packer, get_packer, numpy.savez_compressed

References

pycsamt.core.registry.unpack_from_file(path, *, kind='bundle')[source]#

Deserialize an object from .npz using a registered packer.

Parameters:

  • path (path-like) – Input NPZ path created by pack_to_file().

  • kind (str, optional) – Packer name. Defaults to "bundle". Must match the packer used during serialization.

Returns:

The reconstructed object.

Return type:

Any

Raises:

  • ValueError – If no packer is registered for kind.

  • OSError – For filesystem read errors.

  • KeyError – If expected payload keys are missing for the packer.

Notes

Loading uses allow_pickle=False for safety; payloads must be composed of basic NumPy/JSON types.

Examples

Load a previously saved bundle:

>>> obj = unpack_from_file("site001.npz", kind="bundle")
>>> hasattr(obj, "z") and hasattr(obj, "freq")
True

See also

pack_to_file, register_packer, get_packer

class pycsamt.core.registry.RegistryAPI(root, *, manifest_name='manifest.json')[source]#

Bases: CoreObject

Convenience façade over Registry.

This high-level helper wires a registry root, ensures a packs/ folder exists under that root, and exposes both thin proxies to low-level registry calls and ergonomic helpers for packing objects to NPZ bundles and restoring them later.

Parameters:

  • root (path-like) – Base directory that will contain the manifest and, if needed, a packs/ subdirectory for NPZ bundles.

  • manifest_name (str, optional) – Manifest filename within root. Defaults to "manifest.json".

Variables:

  • low (Registry) – The underlying registry instance that performs the core operations (load/save, add, query).

  • root (pathlib.Path) – Absolute path to the registry root directory.

  • pack_dir (pathlib.Path) – Directory where NPZ bundles are stored (root/packs).

Notes

RegistryAPI aims to be pragmatic and file-system first. It does not do locking or concurrency control. For multi- process scenarios, add your own synchronization.

The packing flow defaults to the "bundle" packer, which serializes TFBundle-compatible objects into NPZ using numpy.savez_compressed().

Examples

Create an API, add an EDI file, and list records:

>>> from pycsamt.core.registry import RegistryAPI
>>> api = RegistryAPI("data")
>>> _ = api.add_file("site001.edi", kind="edi", fmt="edi")
>>> [r.kind for r in api.list(kind="edi")]
['edi']

Pack an object to an NPZ bundle and store it in packs:

>>> class Dummy:
...     station = "S01"; station_id = 1
>>> rec = api.add_object(Dummy(), pack=True,
...                      pack_name="S01.npz")
>>> rec.kind
'bundle'

Materialize a record back into a Python object:

>>> obj = api.materialize(rec.rid)  # may be Dummy or path

Convert a record to an EDI representation:

>>> edi_like = api.to_edi(rec.rid)

See also

pycsamt.core.registry.Registry, pycsamt.core.registry.pack_to_file, pycsamt.core.registry.unpack_from_file, pycsamt.core.registry.guess_kind, pycsamt.seg.edi.EDIFile, pycsamt.zonge.avg.AVG, pycsamt.jones.j.JFile, pycsamt.core.base.to_edi

References

save()[source]#

Persist the manifest to disk.

Thin proxy to Registry.save(). Use after mutating the underlying manifest directly, though typical helpers call save for you.

Return type:

None

list(*, kind=None)[source]#

List all records, optionally filtered by kind.

Parameters:

kind (str or None, optional) – If provided, only records whose kind matches are returned.

Returns:

Shallow views of stored Record objects.

Return type:

list of Record

Examples

>>> from pycsamt.core.registry import RegistryAPI
>>> api = RegistryAPI("data")
>>> isinstance(api.list(), list)
True
get(rid)[source]#

Return the record by id or raise an error.

Parameters:

rid (str) – Record identifier returned at insertion time.

Returns:

The matching record.

Return type:

Record

Raises:

RegistryError – If the id does not exist in the manifest.

find(*, tag=None, kind=None, dataid=None)[source]#

Filter records by tag, kind, and/or data id.

Parameters:

  • tag (str or None, optional) – Require that the tag is present in Record.tags.

  • kind (str or None, optional) – Require that the record kind matches the value.

  • dataid (str or None, optional) – Require equality with Record.dataid.

Returns:

Records satisfying all given predicates.

Return type:

list of Record

Examples

>>> from pycsamt.core.registry import RegistryAPI
>>> api = RegistryAPI("data")
>>> api.find(kind="edi")
[...]  # list of records
add_file(path, *, kind=None, fmt=None, dataid=None, station_id=None, tags=None, meta=None, with_hash=True)[source]#

Register a file path as a new record.

Parameters:

  • path (path-like) – File path. Relative paths are resolved under root.

  • kind (str or None, optional) – Logical kind (e.g. "edi", "avg"). If omitted, defaults to "meta".

  • fmt (str or None, optional) – Short format hint (e.g. "edi").

  • dataid (str or None, optional) – External data or survey id.

  • station_id (str or int or None, optional) – Station identifier. Cast to string if provided.

  • tags (list[str] or None, optional) – User-defined labels to ease later queries.

  • meta (dict[str, Any] or None, optional) – Extra JSON-serializable metadata.

  • with_hash (bool, optional) – If True and the file exists, compute and store a SHA-256 checksum. Defaults to True.

Returns:

The created and persisted record.

Return type:

Record

Notes

This method saves the manifest on success.

Examples

>>> from pycsamt.core.registry import RegistryAPI
>>> api = RegistryAPI("data")
>>> rec = api.add_file("site001.edi",
...                    kind="edi", fmt="edi")
>>> rec.kind, rec.fmt
('edi', 'edi')
add_object(obj, *, kind=None, tags=None, meta=None, pack=False, pack_name=None)[source]#

Register an object and, optionally, pack it to an NPZ bundle.

Parameters:

  • obj (Any) – Object to register. If kind is None, a coarse label is inferred via guess_kind().

  • kind (str or None, optional) – Explicit kind override (e.g. "edi", "bundle").

  • tags (list[str] or None, optional) – User-defined labels.

  • meta (dict[str, Any] or None, optional) – Extra JSON-serializable metadata for the record.

  • pack (bool, optional) – If True, serialize obj with the "bundle" packer into pack_dir and rewrite the record to refer to that NPZ on disk. Defaults to False.

  • pack_name (str or None, optional) – Output NPZ filename when pack=True. Defaults to "<rid>.npz".

Returns:

The created record. When pack=True, the record kind becomes "bundle" and path points to the NPZ file inside pack_dir.

Return type:

Record

Notes

Packing uses pack_to_file() with kind="bundle". The manifest is saved after packing.

Examples

>>> from pycsamt.core.registry import RegistryAPI
>>> api = RegistryAPI("data")
>>> class Obj: station="S02"; station_id=2
>>> rec = api.add_object(Obj(), tags=["demo"])
>>> isinstance(rec.rid, str)
True

Pack immediately to NPZ:

>>> rec2 = api.add_object(Obj(), pack=True,
...                       pack_name="S02.npz")
>>> rec2.kind
'bundle'

See also

pycsamt.core.registry.pack_to_file, pycsamt.core.registry.guess_kind

materialize(rid)[source]#

Load a record into a concrete Python object when possible.

Parameters:

rid (str) – Record id to resolve.

Returns:

If the record is a "bundle" with a valid path, returns the unpacked object via unpack_from_file(). For known kinds with readable files this returns:

Otherwise, returns the string path (or None) of the record.

Return type:

Any

Notes

Errors during import or read are swallowed and the path is returned instead. This keeps the method side-effect free with respect to the registry while remaining robust.

Examples

>>> from pycsamt.core.registry import RegistryAPI
>>> api = RegistryAPI("data")
>>> # Assuming a valid record id:
>>> # obj = api.materialize(rid)
to_edi(rid, **kw)[source]#

Convert a record’s materialized object into an EDI-like form.

Parameters:

  • rid (str) – Record id to resolve and convert.

  • **kw (Any) – Forwarded to to_edi(), which performs the actual conversion.

Returns:

An object produced by to_edi(). The concrete type depends on the available converters and inputs.

Return type:

Any

Notes

This is a convenience wrapper that first calls materialize(), then delegates to pycsamt.core.base.to_edi().

Examples

>>> from pycsamt.core.registry import RegistryAPI
>>> api = RegistryAPI("data")
>>> # Convert an existing record to an EDI representation
>>> edi_like = api.to_edi("some-record-id")

See also

pycsamt.core.base.to_edi, pycsamt.seg.edi.EDIFile