feat(magazine): record/replay cache for vision responses in OCR comparisons
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
b9e7bcda7f
commit
e60126721d
@ -6,6 +6,13 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
|
||||
---
|
||||
|
||||
## [Unreleased]
|
||||
### Added
|
||||
- **Record/replay cache for vision OCR responses** (`core/vision_cache.py`, `MAGAZINE_VISION_CACHE=live|record|replay`). Freezes the raw model response per (model, prompt, image, parameters) in a git-ignored cache directory so OCR/layout/post-processing changes can be compared on identical vision output; replay fails loudly on a miss. `bench_ocr.py --vision-cache` and `compare_magazines.py` (`--baseline-vision-run`, `--new-vision-run`) report mode and models and say whether a comparison is frozen. See architecture/magazine-ingestion.md.
|
||||
|
||||
### Changed
|
||||
- `vision_ocr_page` routes its request through the cache layer; the default `live` mode behaves exactly as before.
|
||||
### Fixed
|
||||
- **Dwarf body shape and beard no longer leak onto human characters' images.** The dwarf body lock ("one adult dwarf", short legs, thick full beard) was switched on by the word "Zwerg" anywhere in the species field, including extraction notes about other characters. In the real library this hit Marrad (a thin human boy, "Mensch (oder Zwerg, je nach Kontext …)") and Kolon and Arthag in book 02 ("Mensch (… 'Der Zwerg' wird separat erwähnt … Warte: 'Der Zwerg lächelte grimmig' …)"), whose prompt Subject also carried that sentence about another character. Species now comes only from the field's own leading assertion: long or quoting bracket notes are dropped, alternatives ("Mensch oder Zwerg") never count as a dwarf, and a *visual species* override is honoured for the body lock too. Seven of 270 records change; tests in `tests/test_character_species_isolation.py`.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -802,6 +802,28 @@ section as it's noticed) to an allowlist (`head.closest('#s-reader,
|
||||
#s-library, #s-rehearser')`) so the same bleed-through cannot recur on
|
||||
Magazines, Settings, or any future section.
|
||||
|
||||
## Frozen vision responses for OCR A/B comparisons
|
||||
|
||||
Vision OCR is not repeatable, so two imports of one PDF differ even if only post-processing changed.
|
||||
`core/vision_cache.py` freezes the raw completion text of `vision_ocr_page` (the single place the app calls a
|
||||
vision engine; the fallback chain, crops and rotation retries all go through it).
|
||||
|
||||
* `MAGAZINE_VISION_CACHE=live` (default, unchanged production behaviour) | `record` | `replay`. An invalid value raises.
|
||||
* `replay` never calls the model; a request that was not recorded raises `VisionCacheMiss` (no silent fallback).
|
||||
* Directory: `MAGAZINE_VISION_CACHE_DIR`, default `<CONFIG_DIR or ./config>/vision_cache` (git-ignored, not in the image).
|
||||
* Key: sha256 of model, prompt text hash, image-bytes hash and every other request parameter (temperature 0.0,
|
||||
max_tokens, vendor fields such as ngram/window sizes, any seed). A change to any of them is a different key,
|
||||
so it misses in replay. Endpoint URL is not part of the key. Entries store the key inputs, response, timestamp.
|
||||
Failed calls are never stored.
|
||||
* `last_run.json` in the cache dir records mode, models, hits/recorded/misses of the last process.
|
||||
|
||||
Workflow: import the PDF once with `record`, then import after each OCR/layout/post-processing change with
|
||||
`replay` (needs the same PNG renders, i.e. same DPI/render code; a changed render is a miss by design).
|
||||
`scripts/bench_ocr.py --vision-cache record|replay` does the same for the benchmark and writes the summary into `--out`.
|
||||
`scripts/compare_magazines.py` prints the vision mode/models of each side and whether the comparison is frozen;
|
||||
supply the manifests with `--baseline-vision-run FILE --new-vision-run FILE` (or a `vision_run.json` in the folder).
|
||||
Without manifests it states "NOT frozen". Tests: `tests/test_vision_cache.py`.
|
||||
|
||||
## Known limitations
|
||||
|
||||
- Sidebar detection (`mark_sidebars`) is geometry-only (an anomalously narrow
|
||||
|
||||
@ -1283,7 +1283,8 @@ def vision_ocr_page(png_path: Path, base_url: str, model: str,
|
||||
after the fact from noisy OCR variants of the same repeating line.
|
||||
"""
|
||||
try:
|
||||
image_b64 = base64.b64encode(png_path.read_bytes()).decode("ascii")
|
||||
image_bytes = png_path.read_bytes()
|
||||
image_b64 = base64.b64encode(image_bytes).decode("ascii")
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
@ -1332,13 +1333,21 @@ def vision_ocr_page(png_path: Path, base_url: str, model: str,
|
||||
"window_size": 128,
|
||||
"images_config": {"image_mode": "gundam"},
|
||||
})
|
||||
try:
|
||||
def _fetch() -> Any:
|
||||
resp = request_with_retries(
|
||||
"POST", f"{base_url.rstrip('/')}/chat/completions",
|
||||
json=payload, timeout=timeout, attempts=2,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
content = resp.json()["choices"][0]["message"]["content"]
|
||||
return resp.json()["choices"][0]["message"]["content"]
|
||||
|
||||
try:
|
||||
# MAGAZINE_VISION_CACHE=live (default) just calls _fetch(); record/replay freeze the raw
|
||||
# completion for repeatable OCR A/B comparisons (core/vision_cache.py).
|
||||
from core import vision_cache
|
||||
content = vision_cache.cached_completion(
|
||||
model, prompt, image_bytes,
|
||||
{k: v for k, v in payload.items() if k not in ("model", "messages")}, _fetch)
|
||||
except requests.exceptions.RequestException as exc:
|
||||
status = getattr(getattr(exc, "response", None), "status_code", None)
|
||||
logging.getLogger(__name__).warning("Vision OCR request failed (%s%s)",
|
||||
@ -1636,13 +1645,21 @@ def describe_visual_region(image_path: Path, region_type: str, base_url: str,
|
||||
"window_size": 128,
|
||||
"images_config": {"image_mode": "gundam"},
|
||||
})
|
||||
try:
|
||||
def _fetch() -> Any:
|
||||
resp = request_with_retries(
|
||||
"POST", f"{base_url.rstrip('/')}/chat/completions",
|
||||
json=payload, timeout=timeout, attempts=2,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
content = resp.json()["choices"][0]["message"]["content"]
|
||||
return resp.json()["choices"][0]["message"]["content"]
|
||||
|
||||
try:
|
||||
# MAGAZINE_VISION_CACHE=live (default) just calls _fetch(); record/replay freeze the raw
|
||||
# completion for repeatable OCR A/B comparisons (core/vision_cache.py).
|
||||
from core import vision_cache
|
||||
content = vision_cache.cached_completion(
|
||||
model, prompt, base64.b64decode(image_b64),
|
||||
{k: v for k, v in payload.items() if k not in ("model", "messages")}, _fetch)
|
||||
except requests.exceptions.RequestException as exc:
|
||||
status = getattr(getattr(exc, "response", None), "status_code", None)
|
||||
logging.getLogger(__name__).warning("Figure description request failed (%s%s)",
|
||||
|
||||
139
core/vision_cache.py
Normal file
139
core/vision_cache.py
Normal file
@ -0,0 +1,139 @@
|
||||
"""Record/replay cache for vision-model (OCR) responses, so OCR/layout A/B runs are comparable.
|
||||
|
||||
A live vision model is not repeatable (model swaps, server-side sampling, load-dependent output), so two
|
||||
imports of one PDF differ even when only the post-processing changed. This module freezes the raw
|
||||
completion text per request:
|
||||
|
||||
* ``live`` (default) -- no cache; behaviour is exactly as before.
|
||||
* ``record`` -- call the model, store the response, return it.
|
||||
* ``replay`` -- never call the model; a request with no stored response raises ``VisionCacheMiss``
|
||||
(fail loudly rather than silently mixing frozen and fresh output).
|
||||
|
||||
Select with ``MAGAZINE_VISION_CACHE=live|record|replay``; the directory is
|
||||
``MAGAZINE_VISION_CACHE_DIR`` (default ``<CONFIG_DIR or ./config>/vision_cache``; ``config/`` is git-ignored).
|
||||
|
||||
The key is sha256 over (model, prompt text, sha256 of the image bytes, all other request parameters such as
|
||||
temperature, max_tokens, seed and vendor fields). Endpoint URL and timeouts are not part of it (same model
|
||||
behind another address is the same model). Each entry stores the key inputs, the response and a timestamp.
|
||||
Every non-live run keeps ``last_run.json`` (mode, models, hits/recorded/misses) in the cache directory for
|
||||
comparison reports.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
MODES = ("live", "record", "replay")
|
||||
ENV_MODE = "MAGAZINE_VISION_CACHE"
|
||||
ENV_DIR = "MAGAZINE_VISION_CACHE_DIR"
|
||||
|
||||
|
||||
class VisionCacheMiss(RuntimeError):
|
||||
"""Replay mode asked for a response that was never recorded."""
|
||||
|
||||
|
||||
def get_mode() -> str:
|
||||
mode = (os.environ.get(ENV_MODE) or "live").strip().lower()
|
||||
if mode not in MODES:
|
||||
raise ValueError(f"{ENV_MODE}={mode!r} is invalid; use one of {', '.join(MODES)}")
|
||||
return mode
|
||||
|
||||
|
||||
def cache_dir() -> Path:
|
||||
env = os.environ.get(ENV_DIR)
|
||||
if env:
|
||||
return Path(env)
|
||||
base = os.environ.get("CONFIG_DIR") or str(Path(__file__).resolve().parents[1] / "config")
|
||||
return Path(base) / "vision_cache"
|
||||
|
||||
|
||||
def make_key(model: str, prompt: str, image_bytes: bytes, params: dict[str, Any]) -> tuple[str, dict[str, Any]]:
|
||||
inputs = {
|
||||
"model": model,
|
||||
"prompt_sha256": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
|
||||
"image_sha256": hashlib.sha256(image_bytes).hexdigest(),
|
||||
"params": params,
|
||||
}
|
||||
blob = json.dumps(inputs, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
|
||||
return hashlib.sha256(blob.encode("utf-8")).hexdigest(), inputs
|
||||
|
||||
|
||||
class _Stats:
|
||||
def __init__(self) -> None:
|
||||
self.lock = threading.Lock()
|
||||
self.models: set[str] = set()
|
||||
self.hits = self.recorded = self.misses = 0
|
||||
self.started = time.strftime("%Y-%m-%dT%H:%M:%S%z")
|
||||
|
||||
def snapshot(self, mode: str) -> dict[str, Any]:
|
||||
return {"mode": mode, "models": sorted(self.models), "hits": self.hits, "recorded": self.recorded,
|
||||
"misses": self.misses, "started": self.started}
|
||||
|
||||
|
||||
_stats = _Stats()
|
||||
|
||||
|
||||
def run_summary() -> dict[str, Any]:
|
||||
"""What this process did with the cache so far (for reports)."""
|
||||
with _stats.lock:
|
||||
return _stats.snapshot(get_mode())
|
||||
|
||||
|
||||
def _write_manifest(directory: Path, mode: str) -> None:
|
||||
try:
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
tmp = directory / "last_run.json.tmp"
|
||||
tmp.write_text(json.dumps(_stats.snapshot(mode), indent=1), "utf-8")
|
||||
tmp.replace(directory / "last_run.json")
|
||||
except OSError:
|
||||
pass # the manifest is a convenience; never fail an OCR call over it
|
||||
|
||||
|
||||
def cached_completion(model: str, prompt: str, image_bytes: bytes, params: dict[str, Any],
|
||||
fetch: Callable[[], str | None]) -> str | None:
|
||||
"""Return the completion text for this request according to the cache mode.
|
||||
|
||||
``fetch`` performs the real call (live/record only) and returns the text, or None on failure;
|
||||
failures are never stored.
|
||||
"""
|
||||
mode = get_mode()
|
||||
if mode == "live":
|
||||
return fetch()
|
||||
directory = cache_dir()
|
||||
key, inputs = make_key(model, prompt, image_bytes, params)
|
||||
path = directory / f"{key}.json"
|
||||
with _stats.lock:
|
||||
_stats.models.add(model)
|
||||
if mode == "replay":
|
||||
try:
|
||||
entry = json.loads(path.read_text("utf-8"))
|
||||
content = entry["response"]
|
||||
except (OSError, ValueError, KeyError):
|
||||
with _stats.lock:
|
||||
_stats.misses += 1
|
||||
_write_manifest(directory, mode)
|
||||
raise VisionCacheMiss(
|
||||
f"no recorded vision response for model {model!r} (key {key[:12]}, image "
|
||||
f"{inputs['image_sha256'][:12]}) in {directory}; record it first with {ENV_MODE}=record")
|
||||
with _stats.lock:
|
||||
_stats.hits += 1
|
||||
_write_manifest(directory, mode)
|
||||
return content
|
||||
content = fetch()
|
||||
if content is not None:
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
tmp = path.with_suffix(".tmp")
|
||||
tmp.write_text(json.dumps({"key_inputs": inputs, "response": content,
|
||||
"recorded_at": time.strftime("%Y-%m-%dT%H:%M:%S%z")},
|
||||
ensure_ascii=False, indent=1), "utf-8")
|
||||
tmp.replace(path)
|
||||
with _stats.lock:
|
||||
_stats.recorded += 1
|
||||
with _stats.lock:
|
||||
_write_manifest(directory, mode)
|
||||
return content
|
||||
@ -185,7 +185,12 @@ def main() -> int:
|
||||
ap.add_argument("--dpi", type=int, default=110, help="110 matches the app's ingest default")
|
||||
ap.add_argument("--engine", action="append", required=True)
|
||||
ap.add_argument("--out", type=Path, help="write the full results as JSON")
|
||||
ap.add_argument("--vision-cache", choices=("live", "record", "replay"),
|
||||
help="freeze vision-engine responses (sets MAGAZINE_VISION_CACHE; see core/vision_cache.py)")
|
||||
args = ap.parse_args()
|
||||
if args.vision_cache:
|
||||
import os
|
||||
os.environ["MAGAZINE_VISION_CACHE"] = args.vision_cache
|
||||
|
||||
pages = [int(p) for p in args.pages.split(",") if p.strip()] or pick_pages(args.pdf, args.count, args.min_words)
|
||||
print(f"pages: {pages}")
|
||||
@ -232,7 +237,12 @@ def main() -> int:
|
||||
f"{statistics.mean(r['seq'] for r in ok):.3f} | {statistics.median(r['sec'] for r in ok):.1f} | "
|
||||
f"{e.get('process_rss_growth_mb', 0)} |")
|
||||
if args.out:
|
||||
args.out.write_text(json.dumps({"pdf": args.pdf.name, "pages": pages, "dpi": args.dpi, "results": results},
|
||||
vision = None
|
||||
if any(e["engine"].startswith("vision:") for e in results):
|
||||
from core import vision_cache
|
||||
vision = vision_cache.run_summary()
|
||||
args.out.write_text(json.dumps({"pdf": args.pdf.name, "pages": pages, "dpi": args.dpi, "vision_cache": vision,
|
||||
"results": results},
|
||||
ensure_ascii=False, indent=1), "utf-8")
|
||||
return 0
|
||||
|
||||
|
||||
@ -42,7 +42,36 @@ def load_magazine(folder: Path) -> dict[str, Any]:
|
||||
quality: dict[str, Any] = {}
|
||||
if (folder / "quality_report.json").is_file():
|
||||
quality = json.loads((folder / "quality_report.json").read_text("utf-8"))
|
||||
return {"folder": str(folder), "layout": layout, "articles": articles, "quality": quality}
|
||||
return {"folder": str(folder), "layout": layout, "articles": articles, "quality": quality,
|
||||
"vision_run": load_vision_run(folder / "vision_run.json")}
|
||||
|
||||
|
||||
def load_vision_run(path: Path | None) -> dict[str, Any] | None:
|
||||
"""Vision cache manifest of the run that produced a folder (core/vision_cache.py ``last_run.json``,
|
||||
copied next to the folder's layout.json as ``vision_run.json`` or passed with --*-vision-run)."""
|
||||
if path and Path(path).is_file():
|
||||
try:
|
||||
return json.loads(Path(path).read_text("utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return {"error": "unreadable"}
|
||||
return None
|
||||
|
||||
|
||||
def describe_vision_run(run: dict[str, Any] | None) -> str:
|
||||
if not run:
|
||||
return "not recorded (assume live; vision output not frozen)"
|
||||
if "error" in run:
|
||||
return "unreadable manifest"
|
||||
return (f"{run.get('mode', '?')}; models: {', '.join(run.get('models') or []) or 'none'}; "
|
||||
f"hits {run.get('hits', 0)}, recorded {run.get('recorded', 0)}, misses {run.get('misses', 0)}")
|
||||
|
||||
|
||||
def comparability_note(a: dict[str, Any] | None, b: dict[str, Any] | None) -> str:
|
||||
modes = {(r or {}).get("mode", "live") for r in (a, b)}
|
||||
if modes == {"replay"} or (modes <= {"record", "replay"} and (b or {}).get("mode") == "replay"):
|
||||
same = sorted((a or {}).get("models") or []) == sorted((b or {}).get("models") or [])
|
||||
return "frozen vision responses" + ("" if same else " (WARNING: different model sets)")
|
||||
return "NOT frozen: at least one side used live vision calls, OCR differences may be model noise"
|
||||
|
||||
|
||||
def words(text: str) -> list[str]:
|
||||
@ -290,13 +319,18 @@ def compare(a: dict[str, Any], b: dict[str, Any]) -> dict[str, Any]:
|
||||
"same_page_span": sum(x["same_pages"] for x in articles["matched"]),
|
||||
"same_title": sum(x["title_a"].strip().casefold() == x["title_b"].strip().casefold() for x in articles["matched"])},
|
||||
"quality": {"baseline": quality_numbers(a), "new": quality_numbers(b)},
|
||||
"vision": {"baseline": a.get("vision_run"), "new": b.get("vision_run"),
|
||||
"comparability": comparability_note(a.get("vision_run"), b.get("vision_run"))},
|
||||
}
|
||||
return {"summary": summary, "pages": pages, "articles": articles}
|
||||
|
||||
|
||||
def markdown(report: dict[str, Any], worst: int = 15) -> str:
|
||||
s = report["summary"]
|
||||
lines = ["# Magazine comparison", "", f"* baseline: `{s['baseline']}`", f"* new: `{s['new']}`", "",
|
||||
lines = ["# Magazine comparison", "", f"* baseline: `{s['baseline']}`", f"* new: `{s['new']}`",
|
||||
f"* vision baseline: {describe_vision_run(s['vision']['baseline'])}",
|
||||
f"* vision new: {describe_vision_run(s['vision']['new'])}",
|
||||
f"* comparability: {s['vision']['comparability']}", "",
|
||||
"| measure | value |", "|---|---|",
|
||||
f"| pages | {s['pages']} |",
|
||||
f"| pages skipped as advertisement (either side) | {s['pages_skipped_as_advertisement']} |",
|
||||
@ -331,8 +365,15 @@ def main(argv: list[str] | None = None) -> int:
|
||||
parser.add_argument("new", type=Path)
|
||||
parser.add_argument("--out", type=Path, help="write the full JSON report here")
|
||||
parser.add_argument("--md", type=Path, help="write a short Markdown summary here")
|
||||
parser.add_argument("--baseline-vision-run", type=Path, help="vision cache last_run.json of the baseline import")
|
||||
parser.add_argument("--new-vision-run", type=Path, help="vision cache last_run.json of the new import")
|
||||
args = parser.parse_args(argv)
|
||||
report = compare(load_magazine(args.baseline), load_magazine(args.new))
|
||||
base, new = load_magazine(args.baseline), load_magazine(args.new)
|
||||
if args.baseline_vision_run:
|
||||
base["vision_run"] = load_vision_run(args.baseline_vision_run)
|
||||
if args.new_vision_run:
|
||||
new["vision_run"] = load_vision_run(args.new_vision_run)
|
||||
report = compare(base, new)
|
||||
if args.out:
|
||||
args.out.write_text(json.dumps(report, ensure_ascii=False, indent=1), "utf-8")
|
||||
text = markdown(report)
|
||||
|
||||
93
tests/test_vision_cache.py
Normal file
93
tests/test_vision_cache.py
Normal file
@ -0,0 +1,93 @@
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
from core import vision_cache as vc # noqa: E402
|
||||
|
||||
PARAMS = {"temperature": 0.0, "max_tokens": 4000}
|
||||
|
||||
|
||||
class FakeVision:
|
||||
def __init__(self):
|
||||
self.calls = 0
|
||||
|
||||
def __call__(self):
|
||||
self.calls += 1
|
||||
return f"text-{self.calls}"
|
||||
|
||||
|
||||
class VisionCacheTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.dir = tempfile.mkdtemp(prefix="ttvc-vc-")
|
||||
self.addCleanup(shutil.rmtree, self.dir, True)
|
||||
|
||||
def env(self, mode):
|
||||
return mock.patch.dict(os.environ, {vc.ENV_MODE: mode, vc.ENV_DIR: self.dir})
|
||||
|
||||
def call(self, fake, model="m1", prompt="p", img=b"img", params=PARAMS):
|
||||
return vc.cached_completion(model, prompt, img, dict(params), fake)
|
||||
|
||||
def test_live_default_never_touches_the_cache(self):
|
||||
fake = FakeVision()
|
||||
with mock.patch.dict(os.environ, {vc.ENV_DIR: self.dir}, clear=False):
|
||||
os.environ.pop(vc.ENV_MODE, None)
|
||||
self.assertEqual(self.call(fake), "text-1")
|
||||
self.assertEqual(self.call(fake), "text-2")
|
||||
self.assertEqual(list(Path(self.dir).iterdir()), [])
|
||||
|
||||
def test_record_then_replay_hit_returns_frozen_text_without_calling(self):
|
||||
fake = FakeVision()
|
||||
with self.env("record"):
|
||||
self.assertEqual(self.call(fake), "text-1")
|
||||
entry = json.loads(next(Path(self.dir).glob("[0-9a-f]*.json")).read_text("utf-8"))
|
||||
self.assertEqual(entry["key_inputs"]["params"], PARAMS)
|
||||
self.assertEqual(entry["key_inputs"]["model"], "m1")
|
||||
with self.env("replay"):
|
||||
self.assertEqual(self.call(fake), "text-1")
|
||||
self.assertEqual(self.call(fake), "text-1")
|
||||
self.assertEqual(vc.run_summary()["mode"], "replay")
|
||||
self.assertEqual(fake.calls, 1)
|
||||
self.assertEqual(json.loads((Path(self.dir) / "last_run.json").read_text())["mode"], "replay")
|
||||
|
||||
def test_replay_miss_fails_loudly_and_does_not_call(self):
|
||||
fake = FakeVision()
|
||||
with self.env("replay"), self.assertRaises(vc.VisionCacheMiss):
|
||||
self.call(fake)
|
||||
self.assertEqual(fake.calls, 0)
|
||||
|
||||
def test_failed_fetch_is_not_recorded(self):
|
||||
with self.env("record"):
|
||||
self.assertIsNone(vc.cached_completion("m", "p", b"i", PARAMS, lambda: None))
|
||||
self.assertEqual(list(Path(self.dir).glob("[0-9a-f]*.json")), [])
|
||||
|
||||
def test_key_is_stable_and_independent_of_dict_order(self):
|
||||
k1, _ = vc.make_key("m", "p", b"x", {"a": 1, "b": 2})
|
||||
k2, _ = vc.make_key("m", "p", b"x", {"b": 2, "a": 1})
|
||||
self.assertEqual(k1, k2)
|
||||
|
||||
def test_any_input_change_invalidates_the_key(self):
|
||||
base, _ = vc.make_key("m", "p", b"x", PARAMS)
|
||||
for args in (("m2", "p", b"x", PARAMS), ("m", "p2", b"x", PARAMS), ("m", "p", b"y", PARAMS),
|
||||
("m", "p", b"x", {**PARAMS, "temperature": 0.7}),
|
||||
("m", "p", b"x", {**PARAMS, "seed": 1})):
|
||||
self.assertNotEqual(base, vc.make_key(*args)[0], args)
|
||||
|
||||
def test_changed_parameters_miss_in_replay(self):
|
||||
with self.env("record"):
|
||||
self.call(FakeVision())
|
||||
with self.env("replay"), self.assertRaises(vc.VisionCacheMiss):
|
||||
self.call(FakeVision(), params={**PARAMS, "temperature": 0.5})
|
||||
|
||||
def test_invalid_mode_is_rejected(self):
|
||||
with self.env("bogus"), self.assertRaises(ValueError):
|
||||
vc.get_mode()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue
Block a user