Spaces:
Running on Zero
Running on Zero
| """ | |
| stub_runner.py — A CPU, torch-free fake VLM for tests and dry runs. | |
| `StubVLMRunner` produces deterministic outputs so the whole orchestrator — | |
| scoring, durability, leaderboard, verdict — can be exercised with no GPU and no | |
| torch import. Three behaviours: | |
| perfect : schema-valid output that MATCHES the ground truth (high accuracy) and | |
| is emitted as clean bare JSON (json_robust = True). | |
| fragile : the SAME content, but wrapped in a ```json fence so it only parses | |
| after repair (json_robust = False) — used to prove that the | |
| labeler_score penalizes fragile JSON even at equal accuracy. | |
| random : emits invalid / off output for a fraction of samples, to exercise the | |
| reject sidecar and the schema_valid_rate path. | |
| Note: "fragile" uses a PROSE wrapper (structural repair), not a markdown fence — | |
| fence-stripping is benign/deterministic and no longer counts against robustness, | |
| so a fenced output would (correctly) still score as robust. | |
| It is a test double, so it is allowed to peek at the ground truth (a real runner | |
| never does). | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from typing import Optional | |
| from .coords import XYWH, XYXY, CoordSpace, from_canonical, to_canonical | |
| from .runner_types import VLMResult | |
| from .tasks_vision import VisionTaskSpec, resolved_system_prompt | |
| def _default_value(fs): | |
| """A schema-valid placeholder for one field spec.""" | |
| if fs.cardinality == "list": | |
| return [] # lists always validate (default_factory=list) | |
| if fs.nested_fields: | |
| return {f.name: _default_value(f) for f in fs.nested_fields} | |
| if fs.vocabulary == "closed": | |
| return fs.closed_values[0] | |
| vk = fs.value_kind | |
| if vk == "number": | |
| return 0.0 | |
| if vk == "integer": | |
| return 0 | |
| if vk == "bbox": | |
| return [0.0, 0.0, 0.0, 0.0] | |
| if vk == "point": | |
| return [0.0, 0.0] | |
| return "x" | |
| def _synthesize_valid(spec: VisionTaskSpec) -> dict: | |
| return {name: _default_value(fs) for name, fs in spec.fields.items()} | |
| def _gt_to_prediction(spec: VisionTaskSpec, gt, image_size) -> Optional[dict]: | |
| """Build a GT-matching, schema-valid prediction for the pilot categories. | |
| Returns None if the category has no GT-driven construction (use synthesize).""" | |
| cat = spec.category | |
| if cat == "image_classification" and isinstance(gt, dict): | |
| label = (gt.get("labels") or [gt.get("label", "x")])[0] | |
| return {"label": label, "confidence": 0.95, | |
| "top5": [{"label": label, "score": 0.95}]} | |
| if cat == "bbox_grounding" and isinstance(gt, dict): | |
| dets = [] | |
| for b in gt.get("boxes", []): | |
| canon = to_canonical(b["bbox"], CoordSpace.PIXEL_ABS, image_size, fmt=b.get("fmt", XYWH)) | |
| box = from_canonical(canon, spec.coord_space, image_size, fmt=XYXY) | |
| dets.append({"label": b.get("label", "x"), "box": [round(c, 2) for c in box], "score": 0.95}) | |
| return {"detections": dets, "count": len(dets)} | |
| if cat == "ocr_text" and isinstance(gt, dict): | |
| text = gt.get("text", "") | |
| return {"full_text": text, "lines": [{"text": text}]} | |
| return None | |
| class StubVLMRunner: | |
| """Drop-in fake runner. Matches the VLMRunner.generate signature.""" | |
| def __init__(self, model_id: str = "stub", behavior: str = "perfect", | |
| reasoning: str = "instruct", **_kwargs): | |
| self.model_id = model_id | |
| self.behavior = behavior | |
| self.reasoning = reasoning | |
| self._n = 0 | |
| def close(self) -> None: # symmetry with VLMRunner | |
| pass | |
| def generate(self, spec: VisionTaskSpec, image, mode: str, *, | |
| image_id: str = "", image_size=(64, 64), gt=None, user_prompt=None) -> VLMResult: | |
| self._n += 1 | |
| # Build content | |
| pred = _gt_to_prediction(spec, gt, image_size) | |
| if pred is None: | |
| pred = _synthesize_valid(spec) | |
| if self.behavior == "random" and (self._n % 4 == 0): | |
| raw = "I cannot answer that." # invalid → exercises reject sidecar | |
| return VLMResult(mode, raw, "stub", 8, 6, 0.001, image_id, grammar_conformant=False) | |
| body = json.dumps(pred) | |
| if self.behavior == "fragile": | |
| # prose wrapper = STRUCTURAL repair (not a benign fence) → not robust | |
| raw = f"Sure! Here is the structured result you requested: {body} — hope that helps!" | |
| else: | |
| raw = body # clean bare JSON | |
| # touch the resolved prompt so prompt wiring is exercised | |
| _ = resolved_system_prompt(spec) | |
| grammar = (mode == "constrained") | |
| return VLMResult(mode, raw, "stub", 12, max(1, len(body) // 4), 0.001, | |
| image_id, grammar_conformant=grammar) | |