gcubed.projection_regression

Generate and compare baseline projection regression snapshots.

Snapshots retain human-readable CSVs alongside lossless NPY numeric blocks and explicitly typed JSON labels, dtypes and metadata. Manifests hash every artifact; large exact comparisons memory-map arrays and visit bounded row blocks without keeping complete paired projection frames alive.

DEFAULT_PROJECTION_TABLE_NAMES: tuple[str, ...] = ('raw_projections', 'database_projections', 'publishable_projections', 'annotated_publishable_projections', 'charting_projections', 'exo_projections')

The projection tables captured by default for before/after regression checks.

MISSING_VALUE_SENTINEL: str = '__GCUBED_PROJECTION_REGRESSION_MISSING__'

String used in CSV snapshots to distinguish real empty strings from missing data.

MAX_REPORTED_LABELS: int = 20

Maximum number of missing or extra row/column labels included in a JSON report.

MAX_NONNUMERIC_EXAMPLES: int = 10

Maximum number of non-numeric cell differences included in a JSON report.

def parse_model_id(model_id: str) -> tuple[str, str]:

Parse a model ID of the form <VERSION>_<BUILD>.

Args: model_id: Model identifier such as 2A_197 or 6W_197.

Returns: A tuple containing the model version and build.

Raises: ValueError: Raised when model_id is not in <VERSION>_<BUILD> form.

def default_models_root() -> pathlib.Path:

Return the default sibling models repository root.

Args: None.

Returns: The first existing sibling models directory found relative to the source checkout or current working directory. If neither exists, returns the source-checkout sibling path so callers can report a useful error.

Raises: No deliberate exceptions are raised.

def configuration_file_for_model(model_id: str, models_root: pathlib.Path | None = None) -> pathlib.Path:

Return the configuration file path for a model ID.

Args: model_id: Model identifier such as 2A_197 or 6W_197. models_root: Optional root directory of the models repository. When not supplied, the sibling models repository is used.

Returns: Path to the canonical configuration file for the requested model.

Raises: ValueError: Raised when model_id is malformed. FileNotFoundError: Raised when the configuration file does not exist.

def build_baseline_projections( model_id: str, *, models_root: pathlib.Path | None = None) -> gcubed.projections.baseline_projections.BaselineProjections:

Build baseline projections for a model using the standard object path.

Args: model_id: Model identifier such as 2A_197 or 6W_197. models_root: Optional root directory of the models repository.

Returns: The generated BaselineProjections object.

Raises: ValueError: Raised when model_id is malformed. FileNotFoundError: Raised when the model configuration file cannot be found. AssertionError: Propagated from model setup or solution validation. Exception: Propagated from model loading, model solution, or baseline projection generation.

def projection_tables( baseline_projections: gcubed.projections.baseline_projections.BaselineProjections, *, table_names: Sequence[str] = ('raw_projections', 'database_projections', 'publishable_projections', 'annotated_publishable_projections', 'charting_projections', 'exo_projections')) -> dict[str, pandas.DataFrame]:

Extract named projection tables from a baseline projection object.

Args: baseline_projections: Baseline projection object to snapshot. table_names: Public projection-table property names to extract.

Returns: A dictionary mapping table names to copied dataframes.

Raises: AttributeError: Raised when a requested table property is not present. TypeError: Raised when a requested property does not return a dataframe.

def iter_projection_tables( projections: Any, *, table_names: Sequence[str] = ('raw_projections', 'database_projections', 'publishable_projections', 'annotated_publishable_projections', 'charting_projections', 'exo_projections')) -> Iterator[tuple[str, pandas.DataFrame]]:

Yield one native table at a time without retaining a copied raw oracle.

def write_projection_tables( tables: Mapping[str, pandas.DataFrame] | Iterable[tuple[str, pandas.DataFrame]], output_dir: pathlib.Path, *, metadata: Mapping[str, Any] | None = None) -> pathlib.Path:

Write projection tables and a manifest to a snapshot directory.

Args: tables: Mapping from table name to dataframe. output_dir: Directory where CSV files and manifest.json are written. metadata: Optional additional manifest metadata.

Returns: The snapshot output directory.

Raises: OSError: Propagated if directories or files cannot be written. TypeError: Raised when any table value is not a dataframe.

def create_projection_snapshot( model_id: str, output_dir: pathlib.Path, *, models_root: pathlib.Path | None = None, table_names: Sequence[str] = ('raw_projections', 'database_projections', 'publishable_projections', 'annotated_publishable_projections', 'charting_projections', 'exo_projections'), label: str | None = None) -> pathlib.Path:

Generate and write a baseline projection snapshot for a model.

Args: model_id: Model identifier such as 2A_197 or 6W_197. output_dir: Directory where the snapshot artifacts should be written. models_root: Optional root directory of the models repository. table_names: Projection-table property names to include. label: Optional human-readable snapshot label, such as before or after_deep_copy_removal.

Returns: The snapshot output directory.

Raises: ValueError: Raised when model_id is malformed. FileNotFoundError: Raised when the model configuration file cannot be found. OSError: Propagated if snapshot files cannot be written. AssertionError: Propagated from model setup or solution validation. Exception: Propagated from model loading, model solution, or baseline projection generation.

def sha256_file(path: pathlib.Path) -> str:

Return the SHA-256 digest of one file.

def fingerprint_records(records: Mapping[str, typing.Any]) -> str:

Return a deterministic hash for a string-keyed record mapping.

def model_input_hashes(model_directory: pathlib.Path) -> dict[str, str]:

Hash economic and generated model inputs in relative-path order.

def repository_state(repository: pathlib.Path) -> dict[str, typing.Any]:

Return revision and dirty-state hashes for a Git repository.

def validate_projection_manifest(snapshot_dir: pathlib.Path) -> dict[str, typing.Any]:

Validate required provenance fields and all recorded output hashes.

def compare_projection_snapshots( reference_dir: pathlib.Path, candidate_dir: pathlib.Path, *, report_file: pathlib.Path | None = None, rtol: float = 0.0, atol: float = 0.0) -> dict[str, typing.Any]:

Compare two baseline projection snapshot directories.

Args: reference_dir: Directory containing the pre-change snapshot. candidate_dir: Directory containing the post-change snapshot. report_file: Optional JSON file path where the comparison report should be written. rtol: Relative tolerance for numeric cell comparisons. atol: Absolute tolerance for numeric cell comparisons.

Returns: A JSON-compatible comparison report. The top-level passed value is True only when tables, labels, numeric cells, and non-numeric cells match within the supplied tolerances.

Raises: FileNotFoundError: Raised when either snapshot directory does not exist. OSError: Propagated if snapshot CSV files or the report file cannot be read or written. ValueError: Raised when either snapshot has no projection tables or a CSV file cannot be parsed by pandas.

def compare_projection_tables( *, table_name: str, reference: pandas.DataFrame, candidate: pandas.DataFrame, rtol: float = 0.0, atol: float = 0.0) -> dict[str, typing.Any]:

Compare two projection dataframes.

Args: table_name: Name of the table being compared. reference: Reference dataframe from the pre-change snapshot. candidate: Candidate dataframe from the post-change snapshot. rtol: Relative tolerance for numeric cell comparisons. atol: Absolute tolerance for numeric cell comparisons.

Returns: A JSON-compatible report for this table.

Raises: No deliberate exceptions are raised.

def main(argv: Sequence[str] | None = None) -> int:

Run the projection regression command-line interface.

Args: argv: Optional command-line argument sequence. When None, argparse reads arguments from sys.argv.

Returns: Process exit code. 0 means success; 1 means a comparison ran but found differences.

Raises: SystemExit: Raised by argparse for invalid command-line arguments. ValueError: Propagated when model IDs or numeric arguments are invalid. FileNotFoundError: Propagated when model or snapshot paths are missing. OSError: Propagated when snapshot or report files cannot be read or written. AssertionError: Propagated from model setup or solution validation. Exception: Propagated from model loading, model solution, or baseline projection generation.

def write_lossless_table( table: pandas.DataFrame, output_dir: pathlib.Path, name: str) -> pathlib.Path:

Write adjacent numeric columns as lossless NPY blocks and typed JSON.

Object and extension columns use explicitly tagged scalar values. Numeric NPY blocks need no pickle, and later comparisons can memory-map row blocks.

def read_lossless_table(path: pathlib.Path) -> pandas.DataFrame:

Round-trip one table; use block comparison for large paired snapshots.

def compare_lossless_tables( reference: pathlib.Path, candidate: pathlib.Path, *, rtol: float = 0.0, atol: float = 0.0, block_rows: int = 4096) -> dict[str, typing.Any]:

Compare every cell in bounded row blocks, with exact structural gates.