pycsamt.seg.cbase#

Classes

`CBBase`([items, verbose])	Lightweight base for EDI collections.
`CoreParser`(*[, recursive, strict, on_dup, ...])	Robust multi-source EDI parser with error tracking and duplicate policies.
`ParseMixin`()	Helpers for discovering and normalizing EDI sources.

class pycsamt.seg.cbase.ParseMixin[source]#

Bases: object

Helpers for discovering and normalizing EDI sources.

The mixin accepts files, folders, and glob patterns, and yields only paths that point to .edi files. All path inputs are normalized with user-home and relative segments resolved.

Parameters:

None – This is a mixin. It does not define its own public constructor parameters.

Variables:

EDI_SUFFIXES (set of str) – Allowed filename suffixes. Defaults to {'.edi'}.
recursive (bool) – Expected to be provided by the host class. If True, directory searches use rglob.
_errors (list of tuple(Path, BaseException)) – Optional sink for discovery issues. When present, helpers record unmatched patterns or missing files.

Notes

The mixin exposes small helpers that higher-level parsers can reuse:

_as_path converts any pathish value into an absolute pathlib.Path.
_is_edi_path returns True for files whose suffix belongs to EDI_SUFFIXES.
_iter_paths normalizes a single source or a sequence of sources into absolute paths.
_iter_edi_files walks files, directories, and glob patterns and yields only existing EDI files.
_fast_station performs a cheap scan of >HEAD to locate DATAID, when available.
_push_error records discovery problems into the host _errors list if present.

Examples

Basic file enumeration:

class Finder(ParseMixin):
    recursive = True

f = Finder()
paths = list(f._iter_edi_files(
    ['data', 'logs/*.edi'], root=Path('.')
))

See also

CoreParser: High-level parser that builds EDIFile objects and aggregates errors.
EDIFile: Reader and writer for single EDI files.

References

EDI_SUFFIXES = {'.edi'}#

class pycsamt.seg.cbase.CoreParser(*, recursive=True, strict=False, on_dup='replace', verbose=0)[source]#

Bases: ParseMixin

Robust multi-source EDI parser with error tracking and duplicate policies.

The parser consumes any combination of files, folders, or glob patterns, builds EDIFile objects, and records failures. Duplicates can be handled by station id or kept as they appear.

Parameters:

recursive (bool, default True) – Recurse into subdirectories when scanning folders.
strict (bool, default False) – If True, any read error is raised. If False, errors are collected and the parse continues.
on_dup ({‘replace’, ‘keep’}, default 'replace') – Duplicate policy by station id. With 'replace', the last item wins. With 'keep', the first seen item is preserved.
verbose (int, default 0) – Verbosity passed to EDIFile.

Variables:

results (list of _ParseResult) – Structured outcomes for each discovered input. Each entry stores the path, the optional EDIFile, and the read error if one occurred.
_errors (list of tuple(Path, BaseException)) – Discovery issues, such as unmatched glob patterns or missing files. Filled by ParseMixin._push_error().

Returns:

The parsed and de-duplicated EDI objects.

Return type:

list of EDIFile

Notes

Station identity is taken from EDIFile.station when available. If missing, a fast scan of >HEAD is used as a fallback. When no station can be resolved, the file path string is used as the key.

Examples

Parse a folder and a glob, keep the first copy:

cp = CoreParser(on_dup='keep', recursive=True)
edis = cp.parse(['data/edi', 'more/*.edi'])
errs = cp.errors()

See also

ParseMixin: Source discovery utilities used by the parser.
EDIFile: Single-file reader used to load each path.

References

results: list[_ParseResult]#

parse(sources)[source]#

Parameters:: sources (str | Path | Sequence[str | Path])
Return type:: list[EDIFile]

errors()[source]#

Return type:: list[tuple[Path, BaseException]]

class pycsamt.seg.cbase.CBBase(items=None, *, verbose=0)[source]#

Bases: object

Lightweight base for EDI collections.

The class provides a minimal container over multiple EDIFile objects. It focuses on indexing by station id, fast lookup by path, and simple iteration. Subclasses can add project-specific logic, caching, or derived computations.

Parameters:

edis (sequence of EDIFile, optional) – Initial items to populate the collection. When omitted the container starts empty.
index_by ({‘station’, ‘path’}, default 'station') – Key to index items. With 'station' the EDIFile.station is used, with fallback to file path if missing. With 'path' the absolute path string is used as the key.
on_dup ({‘replace’, ‘keep’}, default 'replace') – Duplicate policy when inserting items that share the same key.
items (Iterable[EDIFile] | None)
verbose (int)

Variables:

items (dict) – Mapping from key to EDIFile.
order (list of str) – Insertion order of keys, which preserves a stable iteration sequence.
meta (dict) – Arbitrary metadata for client code. This is not used internally.

add(ed)[source]#

Insert one EDIFile respecting the duplicate policy.

Parameters:: ed (EDIFile)
Return type:: None

get(key)#: Return the stored EDIFile by its key.

__iter__()[source]#: Iterate over stored EDIFile objects in insertion order.

keys()#: Yield collection keys in insertion order.

values()#: Yield EDIFile objects in insertion order.

items()[source]#: Yield (key, EDIFile) pairs in insertion order.

Notes

The base does not perform I/O or parsing. Use CoreParser to build items, then feed them into the collection. The class aims to be small and easy to extend rather than fully featured.

Examples

Build a collection indexed by station:

edis = CoreParser().parse(['data/edi'])
coll = CBBase(edis, index_by='station')
for key, ed in coll.items():
    print(key, ed.Z.n_freq)