pycsamt.session#

Workflow sessions and normalization helpers for reproducible conversion work.

The pycsamt.session module provides a small provenance layer around common file-conversion workflows. It is useful when a script, notebook, or agent workflow produces intermediate objects that should be recorded in a manifest without turning the session itself into the science API.

Use sessions for workflow bookkeeping. Use the science modules for scientific work:

load EDI survey data with pycsamt.emtools.ensure_sites();
run quality control with pycsamt.emtools.qc.build_qc_table();
estimate and apply static shift with pycsamt.emtools.ss;
use pycsamt.agents when an agent should orchestrate a workflow.

Public Objects#

Session normalization and context management for reproducible workflows.

class pycsamt.session.Session(root, *, manifest_name='manifest.json', auto_capture=True, capture_children=False, max_children=256)#

Bases: CoreObject

Context-managed capture of key transformation outputs.

A Session wraps selected conversion utilities and transformers so that their results are automatically registered in a small on-disk registry. This enables light provenance tracking of what you produced during a workflow (e.g., the output from to_edi() or from AVGtoEDI / JtoEDI).

When the session starts, it temporarily monkey-patches the following call sites to intercept and register outputs:

pycsamt.core.base.to_edi()
pycsamt.core.transformers.AVGtoEDI.transform()
pycsamt.core.transformers.JtoEDI.transform()

On exit, the original functions are restored and the manifest is saved.

Parameters:

root (path-like) – Working directory that will hold a registry manifest and any artifacts you add explicitly via the underlying RegistryAPI.
manifest_name (str, optional) – Manifest filename under root. Defaults to "manifest.json".
auto_capture (bool, optional) – If True (default), wrap the functions above so that their outputs are registered with tags describing the source call (e.g., "AVGtoEDI.transform").
capture_children (bool, optional) – If True, when a captured output is iterable (e.g., a collection), the session attempts to also register each child item, up to max_children. Defaults to False.
max_children (int, optional) – Maximum number of iterable children to capture when capture_children=True. Defaults to 256.

Variables:

root (pathlib.Path) – Absolute path to the working directory.
reg (pycsamt.core.registry.RegistryAPI) – High-level registry façade used to record artifacts.

Notes

The session captures results, not inputs. To store inputs explicitly, call reg.add_file() or reg.add_object() on the session’s registry.

The capture is process-local and uses monkey-patching. In multi-threaded or multi-process settings, consider isolate usage or add explicit locking to your code.

Examples

Basic usage with explicit object registration:

>>> from pycsamt.session import Session
>>> from pycsamt.seg.edi import EDIFile
>>> with Session("work") as ses:
...     edi = EDIFile.from_file("data/site001.edi")
...     _ = ses.reg.add_object(edi, tags=["raw"])
...     out = ses.reg.list()
>>> isinstance(out, list)
True

Automatic capture from transformers:

>>> from pycsamt.session import Session
>>> from pycsamt.zonge.avg import AVG
>>> from pycsamt.core.transformers import AVGtoEDI
>>> with Session("work", auto_capture=True) as ses:
...     avg = AVG.from_file("data/S01.AVG")
...     edi = AVGtoEDI().transform(avg)
...     # The result of transform is recorded automatically.
...     got = ses.reg.find(tag="AVGtoEDI.transform")

Automatic capture from to_edi():

>>> from pycsamt.session import Session
>>> from pycsamt.core.base import to_edi
>>> from pycsamt.jones.j import JFile
>>> with Session("work", auto_capture=True) as ses:
...     j = JFile.from_file("data/site001.j")
...     edi_like = to_edi(j)
...     hits = ses.reg.find(tag="to_edi")

Capture children from iterable outputs (e.g., collections):

>>> from pycsamt import session as pcs  # alias form
>>> from pycsamt.seg.collection import EDICollection
>>> with pcs.Session("work",
...                  auto_capture=True,
...                  capture_children=True,
...                  max_children=8) as ses:
...     # Suppose `coll` is an iterable of EDI-like items.
...     coll = EDICollection([])  # example only
...     ses.reg.add_object(coll, tags=["batch"])
...     # Children of `coll` would be recorded when captured.

Inspect what was recorded and persist:

>>> with Session("work") as ses:
...     _ = ses.reg.list()
...     # Saving also happens on context exit.
>>> # The manifest `work/manifest.json` now reflects updates.

See also

pycsamt.core.registry.RegistryAPI, pycsamt.core.base.to_edi, pycsamt.core.transformers.AVGtoEDI, pycsamt.core.transformers.JtoEDI, pycsamt.seg.edi.EDIFile, pycsamt.seg.collection.EDICollection, pycsamt.zonge.avg.AVG, pycsamt.jones.j.JFile

References

pycsamt.session.work_session(root, *, manifest_name='manifest.json', auto_capture=True, capture_children=False, max_children=256)#

Create and return a Session with common defaults.

This is a convenience wrapper around Session mirroring its constructor. See Session for details on automatic capture and child registration.

Parameters:

root (path-like) – Working directory for the session. Created if missing.
manifest_name (str, optional) – Registry manifest filename. Defaults to "manifest.json".
auto_capture (bool, optional) – Enable interception of selected transforms to register their outputs. Defaults to True.
capture_children (bool, optional) – When True, iterable outputs are traversed and child items are registered up to max_children. Defaults to False.
max_children (int, optional) – Max number of children to register from one iterable output. Defaults to 256.

Returns:

A context-manageable session instance.

Return type:

Session

Examples

Use the helper to run a small workflow:

>>> from pycsamt.session import work_session
>>> with work_session("work") as ses:
...     # Use ses.reg.* to add artifacts
...     items = ses.reg.list()

Alias style:

>>> import pycsamt as pcs
>>> with pcs.work_session("work") as ses:
...     _ = ses.reg.list()

class pycsamt.session.Normalize(root, *, manifest_name='manifest.json', topo_src=None, auto_register=True)#

Bases: CoreObject

Normalize heterogeneous EM sources into a common form.

Normalize provides a small, file-system-friendly facade that tries to convert various inputs (files and objects) into a standard, iterable representation, favoring EDICollection when possible.

It accepts AVG files or objects, Jones J files, single EDI files, and already-built collections. When provided, a topo_src is attached to AVG sources before conversion.

Optionally, normalized outputs are registered in a small manifest via RegistryAPI.

Parameters:

root (path-like) – Working directory where a manifest is stored.
manifest_name (str, optional) – Manifest filename under root. Defaults to "manifest.json".
topo_src (Any or None, optional) – Optional topography source to enrich AVG data before conversion. It may be an object accepted by AVG.add_topography or another object exposing a frame attribute to assign as avg.topo.
auto_register (bool, optional) – If True (default), normalized outputs are recorded in the registry with the tag "normalized".

Variables:

root (pathlib.Path) – Absolute path to the working directory.
reg (pycsamt.core.registry.RegistryAPI) – Registry used to persist normalized outputs.
topo_src (Any or None) – Topography source passed at construction.
auto_register (bool) – Whether load() registers outputs automatically.

Notes

The normalization rules are:

If the source is an EDICollection, return it.
If the source is an EDIFile, wrap it into a one-item EDICollection.
If the source is an AVG (or AVG file path), try to attach topo_src then transform with AVGtoEDI.
If the source is a JFile (or J file path), transform with JtoEDI.
If the source is a string or path, try loading an EDIFile and wrap into a collection.
Otherwise, delegate to to_edi() and return an EDI- like object or a collection when possible.

Errors while attaching topography or performing transforms are handled defensively; the method falls back to the next strategy or returns the best-effort result.

Examples

Normalize from an AVG file into an EDI collection:

>>> from pycsamt.session import Normalize
>>> with Normalize("work") as nz:
...     coll = nz.load("data/S01.AVG")
...     # 'coll' is typically an EDICollection

Attach topography before AVG→EDI:

>>> topo = object()  # placeholder for a real topo source
>>> with Normalize("work", topo_src=topo) as nz:
...     coll = nz.load("data/S02.AVG")

Normalize from a J file path:

>>> with Normalize("work") as nz:
...     coll = nz.load("data/site001.j")

Already-normalized inputs pass through unchanged:

>>> from pycsamt.seg.collection import EDICollection
>>> with Normalize("work") as nz:
...     empty = EDICollection([])               # example
...     same = nz.load(empty)
>>> isinstance(same, EDICollection)
True

Alias form via the package facade:

>>> import pycsamt as pcs
>>> with pcs.Normalize("work") as nz:
...     out = nz.load("data/site001.edi")

References

load(source)#

Normalize source into a common, iterable EM representation.

This method attempts to coerce the input into an EDICollection when feasible. If the class was constructed with auto_register=True, the resulting object is registered in the manifest with the tag "normalized".

Parameters:: source (Any) – Input to normalize. May be an EDICollection, EDIFile, AVG, JFile, a file path to one of those, or another EM object supported by to_edi().
Returns:: Usually an EDICollection. In fallback cases, an EDI-like object produced by to_edi().
Return type:: Any

Notes

Topography is attached to AVG inputs when possible, using topo_src. Failures are ignored, and the method moves on to the next strategy.

Examples

Basic usage returning an EDI collection:

>>> from pycsamt.session import Normalize
>>> with Normalize("work") as nz:
...     coll = nz.load("data/site001.edi")

Normalization from AVG with auto-registration:

>>> from pycsamt.session import Normalize
>>> with Normalize("work", auto_register=True) as nz:
...     coll = nz.load("data/S03.AVG")
...     hits = nz.reg.find(tag="normalized")

Alias style via the package facade:

>>> import pycsamt as pcs
>>> with pcs.Normalize("work") as nz:
...     out = nz.load("data/site001.j")

See also

pycsamt.core.transformers.AVGtoEDI, pycsamt.core.transformers.JtoEDI, pycsamt.core.base.to_edi, pycsamt.seg.collection.EDICollection

pycsamt.session.normalize_session(root, *, manifest_name='manifest.json', topo_src=None, auto_register=True)#

Create and return a Normalize with common defaults.

This is a convenience wrapper around Normalize, mirroring its constructor.

Parameters:

root (path-like) – Working directory.
manifest_name (str, optional) – Manifest filename under root. Defaults to "manifest.json".
topo_src (Any or None, optional) – Optional topography source for AVG inputs.
auto_register (bool, optional) – Register outputs from Normalize.load(). Defaults to True.

Returns:

A context-manageable normalizer instance.

Return type:

Normalize

Examples

Use the helper to normalize a file quickly:

>>> from pycsamt.session import normalize_session
>>> with normalize_session("work") as nz:
...     coll = nz.load("data/site001.edi")

Alias form:

>>> import pycsamt as pcs
>>> with pcs.normalize_session("work") as nz:
...     coll = nz.load("data/S01.AVG")

Concepts#

Session: Context manager for recording workflow outputs in a registry manifest. It can automatically capture selected conversion results from to_edi and the AVG/J-to-EDI transformers.
work_session: Convenience constructor for pycsamt.session.Session.
Normalize: Context manager that accepts heterogeneous sources such as AVG, Jones J, EDI files, EDI-like objects, and collections, then returns an EDI-like collection whenever possible.
normalize_session: Convenience constructor for pycsamt.session.Normalize.

Recommended Pattern#

Use Normalize at the boundary of a workflow, then pass the normalized EDI output to the real processing API.

from pycsamt.session import normalize_session
from pycsamt.emtools import ensure_sites
from pycsamt.emtools.qc import build_qc_table

input_path = "data/AMT/WILLY_DATA/L18"
work_dir = "work/willy_l18"

with normalize_session(work_dir) as nz:
    edi_collection = nz.load(input_path)

sites = ensure_sites(edi_collection, recursive=True, verbose=0)
qc = build_qc_table(sites)

This keeps responsibilities clear: normalization and provenance live in the session layer, while survey loading and QC live in the EM tools.

Record Conversion Outputs#

Use Session when you want conversion outputs to be registered automatically.

from pycsamt.session import work_session
from pycsamt.zonge.avg import AVG
from pycsamt.transformers.jedi import AVGtoEDI

with work_session("work/avg_conversion") as ses:
    avg = AVG.from_file("raw/S01.AVG")
    edi = AVGtoEDI().transform(avg)

    records = ses.reg.find(tag="AVGtoEDI.transform")

The registry manifest is saved when the context exits. The captured result is the transformer output, not the original input. If the raw input also needs to be tracked, register it explicitly through ses.reg.

Explicit Registry Entries#

Automatic capture is helpful, but explicit registration is clearer when the object is important to the workflow.

from pycsamt.session import Session
from pycsamt.seg.edi import EDIFile

with Session("work/manual_registry", auto_capture=False) as ses:
    edi = EDIFile.from_file("data/site001.edi")
    ses.reg.add_object(edi, tags=["raw", "edi"])
    ses.reg.save()

Disable automatic capture when a script should control exactly what appears in the manifest.

Capture Children From Collections#

By default a collection result is registered as one object. Set capture_children=True to also register iterable children up to max_children.

from pycsamt.session import work_session
from pycsamt.transformers.jedi import JtoEDI
from pycsamt.jones.collection import JCollection

with work_session(
    "work/j_batch",
    capture_children=True,
    max_children=32,
) as ses:
    j_files = JCollection("raw/jones")
    edi_collection = JtoEDI().transform(j_files)

    batch_records = ses.reg.find(tag="JtoEDI.transform")

Use this mode when downstream audit needs station-level conversion records. Keep max_children bounded for large surveys.

Topography During Normalization#

When converting AVG sources, Normalize can attach a topography source before conversion.

from pycsamt.session import normalize_session

topo = "data/topography.csv"

with normalize_session("work/with_topography", topo_src=topo) as nz:
    edi_collection = nz.load("raw/line18.avg")

If the AVG object supports add_topography, that method is used. Otherwise, the normalizer attempts a conservative fallback for objects exposing a frame attribute.

Package Root Shortcuts#

The session helpers are also available lazily from the package root:

import pycsamt as pcs

with pcs.work_session("work/root_shortcut") as ses:
    records = ses.reg.list()

This import style is convenient in notebooks, while direct imports from pycsamt.session are preferred in library code.

When Not To Use It#

Do not use Session as a replacement for:

a Sites object;
an EDI collection;
the desktop or web application session state;
project memory in pycsamt.assistant.memory;
workflow agents in pycsamt.agents.

The session module is intentionally small. It records and normalizes workflow artifacts; it does not own electromagnetic interpretation.