src.llm_interpreter.inputs.readers¶

CSV readers for AMMM pipeline outputs.

This module provides typed CSV readers that load CSV files into dataclass instances with graceful error handling and validation.

Author: AMMM Team Created: 2025-04-10 Last Modified: 2025-04-10

Module Contents¶

exception src.llm_interpreter.inputs.readers.CSVReadError¶

Bases: Exception

Raised when CSV reading fails.

class src.llm_interpreter.inputs.readers.CSVReader(schema_class: Type[T], csv_name: str)¶

Bases: Generic[T]

Generic CSV reader that loads CSV files into typed dataclass instances.

Features: - Graceful handling of missing columns (logs warnings) - Type coercion and validation - Column name mapping for special characters - Support for optional fields - Detailed error reporting

read(filepath: pathlib.Path | str, strict: bool = False) → list[T]¶

Read CSV file and return list of typed dataclass instances.

Parameters:

filepath – Path to CSV file
strict – If True, raise exception on missing columns. If False, log warnings.

Returns:

List of dataclass instances

Raises:

CSVReadError – If file cannot be read or validation fails

src.llm_interpreter.inputs.readers.read_csv(csv_name: str, filepath: pathlib.Path | str, strict: bool = False) → list¶

Read a CSV file using the appropriate schema.

Parameters:

csv_name – Name of CSV file (with or without .csv extension)
filepath – Path to CSV file
strict – If True, raise exception on missing columns

Returns:

List of dataclass instances

Raises:

CSVReadError – If file cannot be read

Example

>>> rows = read_csv("model_summary", "results/csv/model_summary.csv")
>>> for row in rows:
...     print(row.parameter, row.mean)

src.llm_interpreter.inputs.readers.load_all_csvs(results_dir: pathlib.Path | str, strict: bool = False) → dict[str, list]¶

Load all CSV files from the results directory.

Parameters:

results_dir – Path to results/csv directory
strict – If True, raise exception on missing columns

Returns:

Dictionary mapping CSV names to lists of dataclass instances

Example

>>> data = load_all_csvs("results/csv")
>>> model_summary = data["model_summary"]
>>> vif_summary = data["vif_summary"]

src.llm_interpreter.inputs.readers.read_stationarity_summary(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.StationarityRow]¶: Read stationarity_summary.csv.

src.llm_interpreter.inputs.readers.read_vif_summary(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.VIFRow]¶: Read vif_summary.csv.

src.llm_interpreter.inputs.readers.read_transfer_entropy_summary(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.TransferEntropyRow]¶: Read transfer_entropy_summary.csv.

src.llm_interpreter.inputs.readers.read_model_summary(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.ModelSummaryRow]¶: Read model_summary.csv.

src.llm_interpreter.inputs.readers.read_elpd_summary(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.ELPDRow]¶: Read ELPD_summary.csv.

src.llm_interpreter.inputs.readers.read_media_performance_effect(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.MediaPerformanceEffectRow]¶: Read media_performance_effect.csv.

src.llm_interpreter.inputs.readers.read_media_conversion_efficiency(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.MediaConversionEfficiencyRow]¶: Read media_conversion_efficiency.csv.

src.llm_interpreter.inputs.readers.read_media_cost_per_conversion(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.MediaCostPerConversionRow]¶: Read media_cost_per_conversion.csv.

src.llm_interpreter.inputs.readers.read_response_curve_fit(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.ResponseCurveFitRow]¶: Read response_curve_fit_combined.csv.

src.llm_interpreter.inputs.readers.read_budget_scenario_results(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.BudgetScenarioResultRow]¶: Read budget_scenario_results.csv.

src.llm_interpreter.inputs.readers.read_all_decomp(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.AllDecompRow]¶: Read all_decomp.csv.

src.llm_interpreter.inputs.readers.read_waterfall_decomposition(filepath: pathlib.Path | str) → list[src.llm_interpreter.inputs.schema_map.WaterfallDecompositionRow]¶: Read waterfall_decomposition_data.csv.