compression-theory-viz / utils /config_loader.py
rufimelo's picture
feat: add streamlit visualization app with prompts, trajectories, and outputs
d14dfc9
Raw
History Blame Contribute Delete
4.14 kB
from pathlib import Path
from typing import List
import yaml
from utils.constants import DEFAULT_COMPRESSION_LEVELS
def load_papers_config(config_path: str) -> List[dict]:
"""
Load papers configuration from YAML file.
Args:
config_path: Path to YAML configuration file
Returns:
List of paper dicts with normalized fields: path, name, token_limits
Raises:
FileNotFoundError: If config file doesn't exist
ValueError: If config is invalid or required fields missing
"""
config_file = Path(config_path)
if not config_file.exists():
raise FileNotFoundError(f"Config file not found: {config_path}")
try:
with open(config_file, "r") as f:
config = yaml.safe_load(f)
except yaml.YAMLError as e:
raise ValueError(f"Invalid YAML in {config_path}: {e}")
if not config or "papers" not in config:
raise ValueError("Config must contain 'papers' key")
papers = config["papers"]
if not isinstance(papers, list):
raise ValueError("'papers' must be a list")
normalized_papers = []
for idx, entry in enumerate(papers):
try:
normalized = _normalize_paper_entry(entry)
normalized_papers.append(normalized)
except ValueError as e:
raise ValueError(f"Invalid paper entry at index {idx}: {e}")
return normalized_papers
def _normalize_paper_entry(entry: dict) -> dict:
"""
Normalize and validate a single paper entry.
Args:
entry: Paper entry dict from YAML
Returns:
Normalized entry with all required fields and sensible defaults
Raises:
ValueError: If required fields missing or invalid
"""
if not isinstance(entry, dict):
raise ValueError("Paper entry must be a dictionary")
# Validate required field: path
if "path" not in entry:
raise ValueError("Missing required field: 'path'")
path = entry["path"]
if not isinstance(path, str):
raise ValueError("'path' must be a string")
if not Path(path).exists():
raise ValueError(f"Paper path does not exist: {path}")
# Optional field: name (default to filename)
name = entry.get("name")
if name is None:
name = Path(path).name
# Optional field: token_limits (default to DEFAULT_COMPRESSION_LEVELS)
token_limits = entry.get("token_limits")
if token_limits is None:
token_limits = DEFAULT_COMPRESSION_LEVELS
else:
if not isinstance(token_limits, list):
raise ValueError("'token_limits' must be a list")
for limit in token_limits:
if not isinstance(limit, int) or limit <= 0:
raise ValueError(
f"Token limits must be positive integers, got: {limit}"
)
# Optional field: token_tolerances (tolerance values for each token_limit)
token_tolerances = entry.get("token_tolerances")
if token_tolerances is not None:
if not isinstance(token_tolerances, list):
raise ValueError("'token_tolerances' must be a list")
for tolerance in token_tolerances:
if not isinstance(tolerance, int) or tolerance < 0:
raise ValueError(
f"Token tolerances must be non-negative integers, got: {tolerance}"
)
# Optional field: context_pdfs (list of reference PDFs)
context_pdfs = entry.get("context_pdfs")
if context_pdfs is not None:
if not isinstance(context_pdfs, list):
raise ValueError("'context_pdfs' must be a list")
for pdf_path in context_pdfs:
if not isinstance(pdf_path, str):
raise ValueError("context_pdfs must be a list of strings")
if not Path(pdf_path).exists():
raise FileNotFoundError(f"Context PDF not found: {pdf_path}")
return {
"path": path,
"name": name,
"token_limits": token_limits,
"token_tolerances": token_tolerances,
"description": entry.get("description", ""),
"context_pdfs": context_pdfs or [],
}