File size: 3,216 Bytes
babffc8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Iterable
import yaml
REPO_ROOT = Path(__file__).resolve().parents[1]
def resolve_path(path: str | Path, base: Path | None = None) -> Path:
p = Path(path).expanduser()
if p.is_absolute():
return p
return (base or REPO_ROOT) / p
def load_config(path: str | Path) -> dict[str, Any]:
config_path = resolve_path(path, Path.cwd())
if not config_path.exists():
raise FileNotFoundError(f"Config file not found: {config_path}")
with config_path.open("r", encoding="utf-8") as f:
cfg = yaml.safe_load(f) or {}
cfg["_config_path"] = str(config_path)
return cfg
def save_config(config: dict[str, Any], path: str | Path) -> None:
out = resolve_path(path, Path.cwd())
out.parent.mkdir(parents=True, exist_ok=True)
clean = {k: v for k, v in config.items() if not k.startswith("_")}
with out.open("w", encoding="utf-8") as f:
yaml.safe_dump(clean, f, sort_keys=False)
def apply_overrides(config: dict[str, Any], **overrides: Any) -> dict[str, Any]:
out = dict(config)
for key, value in overrides.items():
if value is not None:
out[key] = value
return out
def read_jsonl(path: str | Path) -> list[dict[str, Any]]:
p = resolve_path(path)
if not p.exists():
raise FileNotFoundError(f"JSONL file not found: {p}")
rows: list[dict[str, Any]] = []
with p.open("r", encoding="utf-8") as f:
for line_no, line in enumerate(f, start=1):
line = line.strip()
if not line:
continue
try:
rows.append(json.loads(line))
except json.JSONDecodeError as exc:
raise ValueError(f"Invalid JSON on {p}:{line_no}: {exc}") from exc
return rows
def write_jsonl(path: str | Path, rows: Iterable[dict[str, Any]]) -> None:
p = resolve_path(path)
p.parent.mkdir(parents=True, exist_ok=True)
with p.open("w", encoding="utf-8") as f:
for row in rows:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
def read_json(path: str | Path) -> dict[str, Any]:
p = resolve_path(path)
if not p.exists():
raise FileNotFoundError(f"JSON file not found: {p}")
with p.open("r", encoding="utf-8") as f:
return json.load(f)
def write_json(path: str | Path, obj: Any) -> None:
p = resolve_path(path)
p.parent.mkdir(parents=True, exist_ok=True)
with p.open("w", encoding="utf-8") as f:
json.dump(obj, f, ensure_ascii=False, indent=2)
f.write("\n")
def output_dir(config: dict[str, Any], override: str | Path | None = None) -> Path:
return resolve_path(override or config["output_dir"])
def sample_rows(rows: list[dict[str, Any]], max_samples: int | None) -> list[dict[str, Any]]:
if max_samples is None or max_samples <= 0:
return rows
return rows[:max_samples]
def append_jsonl(path: str | Path, row: dict[str, Any]) -> None:
p = resolve_path(path)
p.parent.mkdir(parents=True, exist_ok=True)
with p.open("a", encoding="utf-8") as f:
f.write(json.dumps(row, ensure_ascii=False) + "\n")
|