pycsamt.agents.edi_export#

pycsamt.agents.edi_export#

EDIExportAgent — Write processed Sites back to EDI files on disk.

Allows any corrected or processed dataset (static-shift corrected, denoised, frequency-decimated) to be exported as standard SEG EDI files, making the results reusable by external tools such as Occam2D GUI, WinGLink, WALDIM, or any other MT processing software.

Wraps export() when Sites can be cast to an EDICollection, and falls back to per-item EDI writing otherwise.

Classes

EDIExportAgent(*[, api_key, model, ...])

Write processed Sites to EDI files on disk.
class pycsamt.agents.edi_export.EDIExportAgent(*, api_key=None, model=None, llm_provider='claude', file_pattern='{station}.edi', overwrite=False)[source]#

Bases: BaseAgent

Write processed Sites to EDI files on disk.

Parameters:

  • api_key (str)

  • model (str)

  • llm_provider (str)

  • file_pattern (str, optional — override default) – Filename format using {station} placeholder. Default "{station}.edi".

  • overwrite (bool, optional — override default) – Overwrite existing files (default False).

  • keys (Output data)

  • ----------

  • path (sites /)

  • output_dir (str — target directory (created if absent))

  • file_pattern

  • overwrite

  • keys

  • ----------------

  • list[str] (written_paths)

  • list[tuple[str (failed)

  • (station (str]] —)

  • message) (error)

  • int (n_failed)

  • int

  • str (output_dir)

Examples

>>> agent = EDIExportAgent()
>>> r = agent.execute({
...     "path":       "/data/WILLY_EDIs",
...     "output_dir": "/out/willy_corrected",
... })
>>> print(r["n_written"], "EDIs exported")
SYSTEM_PROMPT: str = 'You are an expert in MT data management and EDI file format conventions.\nGiven an EDI export result, write 2-3 sentences that:\n1. Confirm how many files were written and their location.\n2. Note any stations that failed and the likely cause.\n3. Recommend next steps for the exported data (e.g. inversion, external QC).\nReply in plain English.\n'#

Override in subclasses to give the LLM its domain expertise.

execute(input_data)[source]#

Run this agent on input_data and return an AgentResult.

Subclasses must implement this method. The contract:

  • Reset self._last_cost = 0.0 at the top.

  • Record wall-clock time with t0 = time.time().

  • Return AgentResult(elapsed_seconds=time.time()-t0, cost_estimate_usd=self._last_cost, ...).

Parameters:

input_data (dict[str, Any])

Return type:

AgentResult