diff --git a/.github/workflows/ovs-heritage-p0.yml b/.github/workflows/ovs-heritage-p0.yml new file mode 100644 index 0000000..7b0091d --- /dev/null +++ b/.github/workflows/ovs-heritage-p0.yml @@ -0,0 +1,24 @@ +name: ovs-heritage-p0 + +on: + pull_request: + paths: + - "ovs_heritage/**" + - ".github/workflows/ovs-heritage-p0.yml" + - "environment.yml" + push: + branches: ["codex/-p0"] + +jobs: + cpu-tests: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.9" + - run: python -m pip install --upgrade pip + - run: python -m pip install --extra-index-url https://download.pytorch.org/whl/cpu torch==1.12.1 numpy==1.26.4 pillow==10.4.0 pyyaml==6.0.2 pytest==8.3.3 ruff==0.6.9 + - run: python -m compileall -q ovs_heritage + - run: pytest -q ovs_heritage/tests + - run: ruff check ovs_heritage diff --git a/environment.yml b/environment.yml index cd23c68..12019b6 100644 --- a/environment.yml +++ b/environment.yml @@ -214,6 +214,8 @@ dependencies: - python-dotenv==1.0.1 - python-json-logger==2.0.7 - pytz==2023.4 + - pytest==8.3.3 + - ruff==0.6.9 - pyyaml==6.0.2 - pyzmq==26.2.0 - referencing==0.35.1 diff --git a/ovs_heritage/AUDIT.md b/ovs_heritage/AUDIT.md new file mode 100644 index 0000000..0109d97 --- /dev/null +++ b/ovs_heritage/AUDIT.md @@ -0,0 +1,71 @@ +# P0 audit of the current working tree + +This document records code observations, not assumptions. Legacy results and paths are not changed. + +## 1. Actual fine-tuning path + +**Confirmed:** `tools/finetune.py:FineTuneWrapper.forward` selects `base_model.clip_backbone` when present, calls that MaskCLIP object with `return_feat=True`, then directly calls `decode_head.cls_seg`. It does not call `LPOSS.forward`. `tools/finetune_tiled.py` imports this wrapper as `common.FineTuneWrapper` and follows the same path. Consequently this is MaskCLIP-branch fine-tuning, not full-LPOSS fine-tuning. DINO graph refinement is absent from training and from `evaluate_stitched`; the latter stitches wrapper logits. The full LPOSS path exists separately in `models/lposs/lposs.py:LPOSS.forward`. + +`configure_trainable_layers` first freezes the whole model, unfreezes the requested suffix (`depth=-1` means all) of CLIP visual transformer blocks, and unfreezes the entire decode head. This includes `decode_head.proj`; the text prototypes are buffers, not parameters. Mixers are separate trainable modules added after this configuration. + +## 2. Imbalance versus catastrophic forgetting + +Class imbalance changes the relative number/gradient contribution of supervised examples. Catastrophic forgetting changes foundation feature geometry: enabled visual blocks and the image projection (`decode_head.proj`) are optimized for the closed heritage labels, while text embeddings remain fixed. That can damage text/image alignment for unseen concepts. Class weighting or oversampling can rebalance heritage gradients, but does not constrain preservation of the original CLIP geometry and therefore cannot guarantee open-vocabulary retention. + +## 3. Softmax before cross entropy + +**Confirmed mathematical defect in the legacy training path.** `models/maskclip/maskclip.py:MaskClipHead.cls_seg` computes cosine convolution, multiplies by 100, then returns `F.softmax(...)`. `FineTuneWrapper` averages these probabilities and its direct second `cls_seg` result; `tools/finetune.py:compute_loss` passes that tensor to `F.cross_entropy`, which expects raw logits. Full `models/lposs/lposs.py:LPOSS.forward` does its own graph label-propagation scorer from normalized CLIP/DINO features rather than merely consuming the MaskCLIP probability map. P0 provides a new isolated raw scorer/loss; this does not retroactively repair checkpoints or historical measurements. + +## 4. Vocabulary-specific state + +`MaskClipHead.__init__` uses `register_buffer("class_embeddings", ...)`, so prototypes are persistent in `state_dict` and checkpoint shape/order depends on aliases and vocabulary. `class_mapping` is a plain tensor attribute, not a registered buffer, and is not saved. `class_names` is a plain Python value. `proj.weight` is persistent vocabulary-independent image projection state. P0 `PrototypeSet` is returned at runtime and `RawCosineScorer.state_dict()` is empty. + +## 5. Aliases and update_vocab + +Semicolon-separated labels are expanded in `_get_class_embeddings`: every alias produces another prototype and hence another `cls_seg` output channel. `class_mapping` is created but never used by `cls_seg`; `reduce_to_true_classes` in the LPOSS inferencer only collapses extra leading background expansion and is not a general alias reduction. `update_vocab` replaces embeddings but does not update `self.class_names`, does not move the newly created text model to CUDA, and does not explicitly preserve the caller device. `_embed_label` hard-codes prompts to `cuda`; constructor hard-codes `model.cuda()`. These choices permit CPU failure and device mismatch. List iteration preserves input order before alias expansion. P0 aggregates all prompt variants into exactly one prototype in exact runtime order. + +## 6. ADVERTISEMENTS loss risk + +**Confirmed.** `tools/finetune.py:_sanitize_targets` replaces every target outside `[0, num_classes)` with 255. With the legacy eleven-channel datasets, ID 11 is therefore silently ignored. P0 `validate_mask_ids`, dataset validation, and `supervised_cross_entropy` raise and enumerate invalid IDs; they never rewrite labels. + +## 7. Consumers of the old ontology + +| file | symbol/location | current assumption | evidence | required action | status | +|---|---|---|---|---|---| +| `tools/convert_brush_coco_to_masks.py` | `LABELS` | IDs 0..10; no advertisements | eleven literal entries | retain as historical converter; use a reviewed v2 conversion path later | legacy intentionally preserved | +| `tools/convert_brush_coco_to_masks.py` | annotation loop | overlap is input-order “last annotation wins” | unconditional `mask[ann_mask > 0] = label_id` | document/resolve overlap policy before any migration | deferred to P1/P2 | +| `mmseg/datasets/facades_train.py` and sibling facade datasets | `classes`, `palette` | eleven classes/channels | eleven literals | keep v1 meaning; new P0 adapter reads v2 source | legacy intentionally preserved | +| `segmentation/configs/_base_/datasets/facades_test.py` | `classes`, `palette` | eleven classes | eleven literals | do not use with masks containing ID 11 | legacy intentionally preserved | +| `tools/finetune.py` | metric groups | HUMAN_ACTIVITY omits advertisements | two-name set | consume v2 groups in future trainer | deferred to P1/P2 | +| `tools/compare_models_facades.py` | groups | eleven-class evaluation | local sets | re-evaluate both models on common v2 test set | deferred to P1/P2 | +| `tools/render_temporal_qualitative_grids.py` | defaults | eleven names/colors | literal lists | legacy figures remain reproducible | legacy intentionally preserved | +| `models/maskclip/maskclip.py` | head outputs | channel count follows expanded strings | embedding convolution | legacy file is unchanged; separate P0 scorer supports runtime C | legacy intentionally preserved | +| `ovs_heritage/configs/datasets/heritage_facades_v2.py` | adapter exports | twelve concepts projected to 11 main channels plus one ornament channel | values loaded from canonical source and projection | use only with explicit two-map v2 manifests | changed | +| README temporal semantics | ontology prose | text/signage combined | explicitly says combined class | update only when downstream temporal contract migrates | legacy intentionally preserved | + +The legacy/base code audited before P0 had no tracked `ADVERTISEMENTS` class in its converter or eleven-class dataset configurations. This PR now contains the canonical P0 `ADVERTISEMENTS` concept at semantic ID 11 with visualization RGB `(216, 27, 96)`, while the legacy consumers listed above remain unchanged. No annotation pixels were created, moved, or converted. + +## 8. LPOSS inference + +**Confirmed syntax/semantic defect:** `segmentation/evaluation/lposs_eval.py:LPOSS_Infrencer.forward` has a duplicated conditional expression immediately after `else i`. Python parses this as attempting to call `i` (often a Tensor) with the following parenthesized result. P0.1 removed the earlier AST test because it passed only while the bug remained and therefore encoded the defect as expected behavior. The safe legacy fix and a regression test that exercises successful inference belong with the P1 LPOSS wrapper/integration work. A later wrapper must distinguish DINO graph refinement (feature graph propagation in LPOSS), LPOSS+ pixel refinement (`pixel_refine`, CuPy Laplacian), and fallback: without CuPy pixel refinement is explicitly skipped; FAISS/CUDA availability affects graph implementations and is not equivalent to LPOSS+. + +`LPOSS_Infrencer.encode_decode` currently calls `self.model(img)` and does not reference an undefined `x`; the earlier audit statement claiming otherwise was incorrect and has been removed. The duplicated conditional-expression defect above remains separately verified. Any LPOSS fix and positive integration regression test remain outside P0. + +## 9. Historical metrics + +Values such as mIoU 0.0551→0.1676 or DAMAGE_MACRO_MIOU 0.0209→0.0802, wherever retained as experiment references, are not P0 results. Legacy single-mask metrics and v2 two-map metrics are not directly comparable. Stock and adapted models must be evaluated again on the identical 12-concept two-map test set. Future reports must distinguish `stock_repo_exact` from `stock_shared_scorer` (stock dense features with the P0 scorer). + +## 10. P0 two-map correction + +The current P0 representation supersedes the earlier single-raster v2 assumption. +Canonical concept 8 is now `ornament_region`: visible decorative geometry in an +independent binary target, not a mutually exclusive “intact ornament” class. +Legacy converters and v1 masks flatten overlaps and cannot be losslessly +reinterpreted without original overlapping annotations. The new projection, +losses, and validator do not modify legacy files or implement a two-head model. + +The repository has a temporal semantic backend registry, but no shared neutral +experiment-metadata record suitable for segmentation validation/loss settings. +P0 therefore exposes a small deterministic JSON-serializable record only; it +does not duplicate the backend registry or implement the future experiment +ledger. LPOSS model integration remains outside P0. diff --git a/ovs_heritage/README.md b/ovs_heritage/README.md new file mode 100644 index 0000000..a895bc6 --- /dev/null +++ b/ovs_heritage/README.md @@ -0,0 +1,123 @@ +# Heritage open-vocabulary foundations (P0) + +P0 defines target representation, ontology projection, runtime prototypes, raw +scoring, strict losses, validation, deterministic metadata, and CPU tests. It +does **not** implement the complete two-head model, LPOSS integration, adapter +training, retention evaluation, a general registry, or an experiment ledger; +those integrations belong to P1 or later. + +## V2: 12 concepts, two target maps + +`heritage_facades_v2_12concepts_two_heads` has stable semantic IDs 0–11, but +these are not 12 mutually exclusive model channels. Stored targets are: + +* `Y_main`: `{0,1,2,3,4,5,6,7,9,10,11,255}`. Values remain semantic IDs on + disk. Concept 8 is invalid here. +* `Y_ornament`: `{0,1,255}`. Positive `1` means visible decorative geometry + (`ornament_region`, semantic concept 8); the binary mask does not store 8. + +`ornament_region` is a pixel-level visible-geometry label independent of damage +or surface condition. Thus corrosion (`Y_main=7`) or water stain (`Y_main=5`) +may overlap `Y_ornament=1`. A completely missing ornament has `Y_ornament=0`; +its absence can be `missing_element` in `Y_main`. This mask never represents an +expected/reconstructed historical footprint. + +The legacy name `ornament_intact` was misleading. It is only a deprecated alias +requiring explicit resolution. It does not trigger dataset migration. A legacy +flattened raster cannot reveal the hidden main label under an ornament region; +without original Label Studio/COCO annotations that overlap cannot be recovered +losslessly and must not be invented. + +Legacy `heritage_facades_v1_11classes` remains an explicit, separate single-mask +schema. V1 masks are never automatically interpreted as v2. + +## Canonical output projection versus runtime vocabulary + +Canonical future supervision expects raw `main_logits [N,11,H,W]` and raw +`ornament_logits [N,1,H,W]`. Main semantic IDs map as follows: + +```text +semantic ID: 0 1 2 3 4 5 6 7 9 10 11 +main channel: 0 1 2 3 4 5 6 7 8 9 10 +semantic ID 8 -> ornament head, channel 0 +``` + +`OntologyProjection` performs semantic/channel round trips and preserves 255. +Conversion to contiguous channels occurs only at the main loss boundary; +export converts argmax channels back to semantic IDs. Ornament inference uses +sigmoid and a configurable threshold and remains a separate binary mask. The +two predictions are never flattened together. + +The stateless open-vocabulary scorer is different: it returns raw +`[N,C_main,H,W]` logits for an arbitrary reordered/subset/extended/mixed runtime +vocabulary. Runtime entries may have `semantic_id=None`. Such dynamic channels +must not be passed to the canonical supervised loss without an explicit mapping. +Prototype metadata records channel order, nullable semantic IDs, ontology hash, +prompt settings, and a deterministic specification hash. + +## Raw-logit losses + +`main_segmentation_loss` validates exactly 11 canonical channels, maps stored +semantic IDs to channel indices, and sends raw logits directly to cross entropy. +Main 255 is handled through `ignore_index`; all-ignore input returns a +differentiable zero rather than NaN. + +`ornament_region_loss` sends raw logits directly to element-wise +`binary_cross_entropy_with_logits`. It replaces ignored targets with a safe zero, +then averages only pixels where `Y_ornament != 255`; ignored pixels contribute +to neither numerator nor denominator. All-ignore input is a differentiable +zero. `combined_two_head_loss` records finite non-negative `lambda_ornament` and +optional positive `pos_weight`; P0 does not tune either value or a threshold. + +## Validation and reproducibility + +V2 manifests explicitly contain `main_mask_path`, `ornament_mask_path`, and +`facade_id` (optionally `image_path` and `source_id`). The validator checks both +files, shape equality, strict dtypes/IDs, empty splits, missing facade IDs, +facade/source-ID leakage, and reused image or mask paths. Optional `image_path` +values are opened and verified; when present, their pixel dimensions must match +both target grids. Reports count those files as `verified_image_count` rather +than treating every manifest row as a verified image. Missing advertisements is a warning. +Reports distinguish manifest rows, valid and failed samples, main and ornament +mask counts, and source counts; they do not call unchecked rows “images”. +Unknown IDs are excluded from valid statistics. + +Validation also requires explicit `schema_version` and `ontology_version` +declarations, either at the dataset-config level or through the corresponding +CLI options. V2 uses `heritage_two_map_v2`; legacy v1 uses +`heritage_single_mask_v1`. Missing, unknown, conflicting, or ontology-mismatched +declarations are errors and never trigger schema inference. Per-split reports +separate manifest rows, valid/failed sample rows, and inventory-level split +errors; error-message count is not used as a sample count. + +Legacy-v1 directory inventories remain supported. Their deterministic +fingerprint uses root-relative canonical paths plus each mask's content hash, so +content changes alter the fingerprint while moving an unchanged inventory does +not. Relative directory paths resolve files exactly once. + +Reports include the component/schema versions, ontology version/hash, complete +projection, split fingerprints, overlaps, duplicated paths, warnings/errors, +and a deterministic neutral metadata record. Hashed payloads contain no current +time. `MetadataRecord` is the intended adapter point for a future shared ledger; +no competing registry or provenance JSONL is introduced. + +```bash +python -m ovs_heritage.validate_dataset \ + --ontology ovs_heritage/configs/heritage_vocab.yaml \ + --schema-version heritage_two_map_v2 \ + --ontology-version heritage_facades_v2_12concepts_two_heads \ + --train train.csv --val val.csv --test test.csv \ + --output validation-report.json --strict +``` + +## Checks + +```bash +python -m compileall -q ovs_heritage +pytest -q ovs_heritage/tests +ruff check ovs_heritage +``` + +The CPU GitHub Actions workflow runs these commands on Python 3.9 with versions +compatible with the project environment. No model or dataset download occurs in +the unit tests. diff --git a/ovs_heritage/__init__.py b/ovs_heritage/__init__.py new file mode 100644 index 0000000..1cbb984 --- /dev/null +++ b/ovs_heritage/__init__.py @@ -0,0 +1,5 @@ +"""P0 foundations for open-vocabulary heritage-facade segmentation.""" + +from .ontology import IGNORE_INDEX, Ontology, load_ontology + +__all__ = ["IGNORE_INDEX", "Ontology", "load_ontology"] diff --git a/ovs_heritage/configs/datasets/README.md b/ovs_heritage/configs/datasets/README.md new file mode 100644 index 0000000..63e1ca0 --- /dev/null +++ b/ovs_heritage/configs/datasets/README.md @@ -0,0 +1,8 @@ +# Dataset schemas + +`heritage_facades_v2.py` exports metadata for the v2 two-map contract: stored +`Y_main` uses stable semantic IDs while stored `Y_ornament` uses binary +`0/1/255`. It defines 11 canonical main channels and one independent ornament +channel, but does not implement a model. Existing MMSeg facade configs remain +explicit legacy-v1 single-mask configurations and must not be used to infer or +migrate v2 overlaps. diff --git a/ovs_heritage/configs/datasets/heritage_facades_v2.py b/ovs_heritage/configs/datasets/heritage_facades_v2.py new file mode 100644 index 0000000..a2fd15d --- /dev/null +++ b/ovs_heritage/configs/datasets/heritage_facades_v2.py @@ -0,0 +1,20 @@ +"""Metadata adapter for the v2 two-target representation; not a model config.""" + +from ovs_heritage.ontology import load_ontology +from ovs_heritage.projection import OntologyProjection + +_ONTOLOGY = load_ontology() +_PROJECTION = OntologyProjection.from_ontology(_ONTOLOGY) +ONTOLOGY_VERSION = _ONTOLOGY.version +ONTOLOGY_HASH = _ONTOLOGY.hash +DATASET_SCHEMA_VERSION = "heritage_two_map_v2" +SEMANTIC_CONCEPTS = _ONTOLOGY.display_names +PALETTE = _ONTOLOGY.palette +MAIN_SEMANTIC_IDS = tuple(entry.semantic_id for entry in _PROJECTION.main_entries) +MAIN_CLASSES = tuple(entry.canonical_name for entry in _PROJECTION.main_entries) +MAIN_NUM_CHANNELS = _PROJECTION.main_channel_count +ORNAMENT_SEMANTIC_ID = 8 +ORNAMENT_NUM_CHANNELS = 1 +OUTPUT_MAPPING = _PROJECTION.as_dict() +EVALUATION_GROUPS = _ONTOLOGY.groups +IGNORE_INDEX = _ONTOLOGY.ignore_index diff --git a/ovs_heritage/configs/heritage_vocab.yaml b/ovs_heritage/configs/heritage_vocab.yaml new file mode 100644 index 0000000..fd2b66c --- /dev/null +++ b/ovs_heritage/configs/heritage_vocab.yaml @@ -0,0 +1,305 @@ +{ + "version": "heritage_facades_v2_12concepts_two_heads", + "ignore_index": 255, + "groups": { + "STRUCTURAL_DAMAGE": [ + "crack", + "spalling", + "delamination", + "missing_element" + ], + "SURFACE_STAIN": [ + "water_stain", + "efflorescence", + "corrosion" + ], + "HUMAN_ACTIVITY": [ + "repairs", + "text_or_images", + "advertisements" + ], + "DAMAGE_MACRO": [ + "crack", + "spalling", + "delamination", + "missing_element", + "water_stain", + "efflorescence", + "corrosion" + ], + "ORNAMENT": [ + "ornament_region" + ] + }, + "classes": [ + { + "id": 0, + "name": "background", + "display_name": "BACKGROUND", + "description": "background area of a facade", + "prompts": [ + "a building facade background" + ], + "aliases": [], + "role": "background", + "is_heritage": true, + "evaluation_groups": [], + "color": [ + 0, + 0, + 0 + ] + }, + { + "id": 1, + "name": "crack", + "display_name": "CRACK", + "description": "a visible crack in facade material", + "prompts": [ + "a crack in a building facade", + "cracked masonry on a facade" + ], + "aliases": [ + "fissure" + ], + "role": "damage", + "is_heritage": true, + "evaluation_groups": [ + "STRUCTURAL_DAMAGE", + "DAMAGE_MACRO" + ], + "color": [ + 229, + 57, + 53 + ] + }, + { + "id": 2, + "name": "spalling", + "display_name": "SPALLING", + "description": "loss or flaking of facade surface", + "prompts": [ + "spalling concrete on a facade", + "flaking masonry surface" + ], + "aliases": [], + "role": "damage", + "is_heritage": true, + "evaluation_groups": [ + "STRUCTURAL_DAMAGE", + "DAMAGE_MACRO" + ], + "color": [ + 30, + 136, + 229 + ] + }, + { + "id": 3, + "name": "delamination", + "display_name": "DELAMINATION", + "description": "separation of facade material layers", + "prompts": [ + "delamination on a building facade", + "separated facade material layers" + ], + "aliases": [], + "role": "damage", + "is_heritage": true, + "evaluation_groups": [ + "STRUCTURAL_DAMAGE", + "DAMAGE_MACRO" + ], + "color": [ + 67, + 160, + 71 + ] + }, + { + "id": 4, + "name": "missing_element", + "display_name": "MISSING_ELEMENT", + "description": "a missing architectural element", + "prompts": [ + "a missing element on a building facade", + "a lost architectural facade component" + ], + "aliases": [], + "role": "damage", + "is_heritage": true, + "evaluation_groups": [ + "STRUCTURAL_DAMAGE", + "DAMAGE_MACRO" + ], + "color": [ + 251, + 140, + 0 + ] + }, + { + "id": 5, + "name": "water_stain", + "display_name": "WATER_STAIN", + "description": "staining caused by water", + "prompts": [ + "a water stain on a building facade", + "water discoloration on masonry" + ], + "aliases": [], + "role": "damage", + "is_heritage": true, + "evaluation_groups": [ + "SURFACE_STAIN", + "DAMAGE_MACRO" + ], + "color": [ + 142, + 36, + 170 + ] + }, + { + "id": 6, + "name": "efflorescence", + "display_name": "EFFLORESCENCE", + "description": "salt deposits on a facade", + "prompts": [ + "efflorescence on masonry", + "white salt deposits on a facade" + ], + "aliases": [], + "role": "damage", + "is_heritage": true, + "evaluation_groups": [ + "SURFACE_STAIN", + "DAMAGE_MACRO" + ], + "color": [ + 253, + 216, + 53 + ] + }, + { + "id": 7, + "name": "corrosion", + "display_name": "CORROSION", + "description": "visible corrosion on facade material", + "prompts": [ + "corrosion on a building facade", + "rust corrosion on facade metal" + ], + "aliases": [ + "rust" + ], + "role": "damage", + "is_heritage": true, + "evaluation_groups": [ + "SURFACE_STAIN", + "DAMAGE_MACRO" + ], + "color": [ + 0, + 172, + 193 + ] + }, + { + "id": 8, + "name": "ornament_region", + "display_name": "ORNAMENT_REGION", + "description": "visible ornamental or decorative facade geometry, independently of damage or surface condition", + "prompts": [ + "visible ornamental geometry on a historic facade", + "a decorative architectural region on a building facade" + ], + "aliases": [ + "ornament_intact" + ], + "role": "heritage", + "is_heritage": true, + "evaluation_groups": [ + "ORNAMENT" + ], + "color": [ + 158, + 158, + 158 + ] + }, + { + "id": 9, + "name": "repairs", + "display_name": "REPAIRS", + "description": "visible repair work or patching", + "prompts": [ + "a repaired area on a building facade", + "patch repair on facade material" + ], + "aliases": [], + "role": "human_activity", + "is_heritage": true, + "evaluation_groups": [ + "HUMAN_ACTIVITY" + ], + "color": [ + 78, + 158, + 158 + ] + }, + { + "id": 10, + "name": "text_or_images", + "display_name": "TEXT_OR_IMAGES", + "description": "non-commercial writing, graffiti, imagery, or visual content", + "prompts": [ + "non-commercial writing on a building facade", + "graffiti or an image on a facade", + "non-commercial visual content on a building" + ], + "aliases": [ + "graffiti" + ], + "role": "human_activity", + "is_heritage": true, + "evaluation_groups": [ + "HUMAN_ACTIVITY" + ], + "color": [ + 142, + 126, + 71 + ] + }, + { + "id": 11, + "name": "advertisements", + "display_name": "ADVERTISEMENTS", + "description": "commercial advertising attached to or displayed on a facade", + "prompts": [ + "an advertisement attached to a building facade", + "a commercial advertising sign or banner on a facade", + "an advertising poster or billboard on a building" + ], + "aliases": [ + "advertising sign", + "commercial banner" + ], + "role": "human_activity", + "is_heritage": true, + "evaluation_groups": [ + "HUMAN_ACTIVITY" + ], + "color": [ + 216, + 27, + 96 + ] + } + ] +} diff --git a/ovs_heritage/losses.py b/ovs_heritage/losses.py new file mode 100644 index 0000000..b072623 --- /dev/null +++ b/ovs_heritage/losses.py @@ -0,0 +1,93 @@ +"""Strict raw-logit losses for canonical v2 two-head supervision.""" +from __future__ import annotations + +from dataclasses import dataclass +import math + +import torch +import torch.nn.functional as F + +from .ontology import load_ontology +from .projection import OntologyProjection + + +@dataclass(frozen=True) +class CombinedLoss: + total: torch.Tensor + main: torch.Tensor + ornament: torch.Tensor + metadata: dict[str, float | None] + + +def _validate_raw_logits(logits: torch.Tensor, label: str) -> None: + if not logits.is_floating_point(): + raise ValueError(f"{label} must be floating-point raw logits") + if torch.isfinite(logits).logical_not().any(): + raise ValueError(f"{label} contain non-finite values") + + +def main_segmentation_loss( + main_logits: torch.Tensor, y_main: torch.Tensor, + projection: OntologyProjection | None = None, +) -> torch.Tensor: + projection = projection or OntologyProjection.from_ontology(load_ontology()) + if main_logits.ndim != 4 or main_logits.shape[1] != projection.main_channel_count: + raise ValueError(f"main_logits must be [N,{projection.main_channel_count},H,W]") + _validate_raw_logits(main_logits, "main_logits") + if y_main.ndim == 4 and y_main.shape[1] == 1: + y_main = y_main[:, 0] + if y_main.ndim != 3 or main_logits.shape[0] != y_main.shape[0] or main_logits.shape[2:] != y_main.shape[1:]: + raise ValueError("main_logits and Y_main have incompatible shapes") + channels = projection.semantic_main_to_channels(y_main) + if torch.all(channels == projection.ignore_index): + return main_logits.sum() * 0.0 + return F.cross_entropy(main_logits, channels, ignore_index=projection.ignore_index) + + +def ornament_region_loss( + ornament_logits: torch.Tensor, y_ornament: torch.Tensor, + *, pos_weight: float | None = None, +) -> torch.Tensor: + if ornament_logits.ndim != 4 or ornament_logits.shape[1] != 1: + raise ValueError("ornament_logits must be [N,1,H,W]") + _validate_raw_logits(ornament_logits, "ornament_logits") + if y_ornament.ndim == 3: + y_ornament = y_ornament.unsqueeze(1) + if y_ornament.shape != ornament_logits.shape: + raise ValueError("ornament_logits and Y_ornament have incompatible shapes") + OntologyProjection._validate_integer_target(y_ornament, "Y_ornament") + found = set(torch.unique(y_ornament.detach()).cpu().tolist()) + invalid = sorted(found - {0, 1, 255}) + if invalid: + raise ValueError(f"Y_ornament contains invalid values {invalid}; allowed values are 0, 1, 255") + if pos_weight is not None and (not math.isfinite(pos_weight) or pos_weight <= 0): + raise ValueError("pos_weight must be finite and positive") + valid = y_ornament != 255 + if not valid.any(): + return ornament_logits.sum() * 0.0 + safe_target = torch.where(valid, y_ornament, torch.zeros_like(y_ornament)).to(ornament_logits.dtype) + weight_tensor = None if pos_weight is None else ornament_logits.new_tensor([pos_weight]) + elementwise = F.binary_cross_entropy_with_logits( + ornament_logits, safe_target, reduction="none", pos_weight=weight_tensor, + ) + return elementwise[valid].mean() + + +def combined_two_head_loss( + main_logits: torch.Tensor, ornament_logits: torch.Tensor, + y_main: torch.Tensor, y_ornament: torch.Tensor, + *, lambda_ornament: float = 1.0, pos_weight: float | None = None, + projection: OntologyProjection | None = None, +) -> CombinedLoss: + if not math.isfinite(lambda_ornament) or lambda_ornament < 0: + raise ValueError("lambda_ornament must be finite and non-negative") + main = main_segmentation_loss(main_logits, y_main, projection) + ornament = ornament_region_loss(ornament_logits, y_ornament, pos_weight=pos_weight) + total = main + lambda_ornament * ornament + return CombinedLoss(total, main, ornament, { + "lambda_ornament": float(lambda_ornament), + "pos_weight": None if pos_weight is None else float(pos_weight), + }) + + +supervised_cross_entropy = main_segmentation_loss diff --git a/ovs_heritage/metadata.py b/ovs_heritage/metadata.py new file mode 100644 index 0000000..b4983e8 --- /dev/null +++ b/ovs_heritage/metadata.py @@ -0,0 +1,95 @@ +"""Neutral immutable metadata records for future experiment-ledger adapters.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from hashlib import sha256 +import json +import math +from types import MappingProxyType +from typing import Any, Mapping + + +def _freeze(value: Any) -> Any: + if isinstance(value, Mapping): + if any(not isinstance(key, str) for key in value): + raise TypeError("metadata mapping keys must be strings") + return MappingProxyType({key: _freeze(item) for key, item in value.items()}) + if isinstance(value, (list, tuple)): + return tuple(_freeze(item) for item in value) + if value is None or isinstance(value, (str, bool, int)): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise ValueError("metadata numbers must be finite") + return value + raise TypeError(f"metadata value is not JSON serializable: {type(value).__name__}") + + +def _thaw(value: Any) -> Any: + if isinstance(value, Mapping): + return {key: _thaw(item) for key, item in value.items()} + if isinstance(value, tuple): + return [_thaw(item) for item in value] + return value + + +def canonical_json(payload: Mapping[str, Any]) -> str: + return json.dumps( + _thaw(payload), sort_keys=True, ensure_ascii=False, + separators=(",", ":"), allow_nan=False, + ) + + +def payload_hash(payload: Mapping[str, Any]) -> str: + return sha256(canonical_json(payload).encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class MetadataRecord: + payload: Mapping[str, Any] + _hash: str = field(init=False, repr=False) + + def __post_init__(self) -> None: + frozen = _freeze(self.payload) + object.__setattr__(self, "payload", frozen) + object.__setattr__(self, "_hash", payload_hash(frozen)) + + @property + def hash(self) -> str: + return self._hash + + def to_dict(self) -> dict[str, Any]: + return {"payload": _thaw(self.payload), "hash": self.hash} + + def to_json(self) -> str: + return json.dumps( + self.to_dict(), sort_keys=True, ensure_ascii=False, + separators=(",", ":"), allow_nan=False, + ) + + +def make_metadata( + *, component_name: str, component_version: str, ontology_version: str, + ontology_hash: str, mapping: Mapping[str, Any], validator_schema_version: str | None = None, + source_fingerprints: Mapping[str, str] | None = None, + vocabulary_specification_hash: str | None = None, + loss_settings: Mapping[str, Any] | None = None, + ornament_threshold: float | None = None, +) -> MetadataRecord: + if not component_name.strip() or not component_version.strip(): + raise ValueError("component name and version must be non-empty") + if ornament_threshold is not None and ( + not math.isfinite(ornament_threshold) or not 0 <= ornament_threshold <= 1 + ): + raise ValueError("ornament_threshold must be finite and in 0..1") + payload = { + "component": {"name": component_name, "version": component_version}, + "ontology": {"version": ontology_version, "hash": ontology_hash}, + "mapping": dict(mapping), + "validator_schema_version": validator_schema_version, + "source_fingerprints": dict(sorted((source_fingerprints or {}).items())), + "vocabulary_specification_hash": vocabulary_specification_hash, + "loss_settings": dict(loss_settings or {}), + "ornament_inference_threshold": ornament_threshold, + } + return MetadataRecord(payload) diff --git a/ovs_heritage/ontology.py b/ovs_heritage/ontology.py new file mode 100644 index 0000000..e4ff945 --- /dev/null +++ b/ovs_heritage/ontology.py @@ -0,0 +1,290 @@ +"""Typed, strictly validated ontology loaded from the single YAML source.""" +from __future__ import annotations + +from dataclasses import dataclass +from hashlib import sha256 +import json +from pathlib import Path +from typing import Any, Mapping + +import yaml +from yaml import YAMLError + + +IGNORE_INDEX = 255 +DEFAULT_ONTOLOGY = Path(__file__).parent / "configs" / "heritage_vocab.yaml" +V1_VERSION = "heritage_facades_v1_11classes" +V2_VERSION = "heritage_facades_v2_12concepts_two_heads" +V2_CLASS_NAMES = ( + "background", "crack", "spalling", "delamination", "missing_element", + "water_stain", "efflorescence", "corrosion", "ornament_region", + "repairs", "text_or_images", "advertisements", +) +V1_CLASS_NAMES = V2_CLASS_NAMES[:8] + ("ornament_intact",) + V2_CLASS_NAMES[9:-1] +VERSION_CLASS_NAMES = { + V1_VERSION: V1_CLASS_NAMES, + V2_VERSION: V2_CLASS_NAMES, +} +REQUIRED_V2_GROUPS = { + "STRUCTURAL_DAMAGE": V2_CLASS_NAMES[1:5], + "SURFACE_STAIN": V2_CLASS_NAMES[5:8], + "HUMAN_ACTIVITY": ("repairs", "text_or_images", "advertisements"), + "DAMAGE_MACRO": V2_CLASS_NAMES[1:8], +} + + +class OntologyError(ValueError): + pass + + +@dataclass(frozen=True) +class OntologyClass: + id: int + name: str + display_name: str + description: str + prompts: tuple[str, ...] + aliases: tuple[str, ...] + role: str + is_heritage: bool + evaluation_groups: tuple[str, ...] + color: tuple[int, int, int] + + +@dataclass(frozen=True) +class Ontology: + version: str + ignore_index: int + classes: tuple[OntologyClass, ...] + groups: Mapping[str, tuple[str, ...]] + hash: str + + @property + def class_names(self) -> tuple[str, ...]: + return tuple(item.name for item in self.classes) + + @property + def display_names(self) -> tuple[str, ...]: + return tuple(item.display_name for item in self.classes) + + @property + def palette(self) -> tuple[tuple[int, int, int], ...]: + return tuple(item.color for item in self.classes) + + @property + def valid_ids(self) -> frozenset[int]: + return frozenset(item.id for item in self.classes) + + def by_name(self, name: str) -> OntologyClass: + matches = [item for item in self.classes if item.name == name] + if not matches: + raise OntologyError(f"unknown canonical class name {name!r} in {self.version}") + return matches[0] + + def resolve_name(self, name: str, *, allow_deprecated_alias: bool = False) -> OntologyClass: + try: + return self.by_name(name) + except OntologyError: + if allow_deprecated_alias: + matches = [item for item in self.classes if name in item.aliases] + if len(matches) == 1: + return matches[0] + raise OntologyError( + f"unknown class name {name!r}; deprecated aliases require explicit resolution" + ) + + +def _canonical_hash(data: Mapping[str, Any]) -> str: + normalized = json.dumps(data, sort_keys=True, ensure_ascii=False, + separators=(",", ":")) + return sha256(normalized.encode("utf-8")).hexdigest() + + +def _type_error(path: str, expected: str, value: Any) -> OntologyError: + return OntologyError(f"{path} must be {expected}, got {value!r} ({type(value).__name__})") + + +def _required_string(mapping: Mapping[str, Any], key: str, path: str) -> str: + value = mapping.get(key) + field_path = f"{path}.{key}" if path else key + if not isinstance(value, str) or not value.strip(): + raise _type_error(field_path, "a non-empty string", value) + return value + + +def _string_list(mapping: Mapping[str, Any], key: str, path: str, *, non_empty: bool) -> tuple[str, ...]: + value = mapping.get(key) + field_path = f"{path}.{key}" + if not isinstance(value, list): + raise _type_error(field_path, "a list of strings", value) + if non_empty and not value: + raise OntologyError(f"{field_path} must be a non-empty list of non-empty strings") + for index, item in enumerate(value): + if not isinstance(item, str) or not item.strip(): + raise _type_error(f"{field_path}[{index}]", "a non-empty string", item) + return tuple(value) + + +def _parse_config(data: Mapping[str, Any]) -> tuple[str, int, tuple[OntologyClass, ...], dict[str, tuple[str, ...]]]: + """Strict schema boundary: validate raw values before any conversion.""" + if not isinstance(data, Mapping): + raise _type_error("root", "a mapping", data) + version = data.get("version") + supported = sorted(VERSION_CLASS_NAMES) + if not isinstance(version, str) or not version.strip(): + raise OntologyError( + f"version must be a non-empty string, got {version!r} ({type(version).__name__}); " + f"supported versions: {supported}" + ) + if version not in VERSION_CLASS_NAMES: + raise OntologyError( + f"version {version!r} is unsupported; supported versions: {supported}" + ) + ignore = data.get("ignore_index") + if type(ignore) is not int: + raise _type_error("ignore_index", "an integer", ignore) + raw_classes = data.get("classes") + if not isinstance(raw_classes, list) or not raw_classes: + raise _type_error("classes", "a non-empty list", raw_classes) + + classes = [] + for index, raw in enumerate(raw_classes): + path = f"classes[{index}]" + if not isinstance(raw, Mapping): + raise _type_error(path, "a mapping", raw) + raw_id = raw.get("id") + if type(raw_id) is not int: + raise _type_error(f"{path}.id", "an integer", raw_id) + name = _required_string(raw, "name", path) + display_name = _required_string(raw, "display_name", path) + description = _required_string(raw, "description", path) + role = _required_string(raw, "role", path) + is_heritage = raw.get("is_heritage") + if type(is_heritage) is not bool: + raise _type_error(f"{path}.is_heritage", "a boolean", is_heritage) + prompts = _string_list(raw, "prompts", path, non_empty=True) + aliases = _string_list(raw, "aliases", path, non_empty=False) + evaluation_groups = _string_list(raw, "evaluation_groups", path, non_empty=False) + color = raw.get("color") + if not isinstance(color, list) or len(color) != 3: + raise _type_error(f"{path}.color", "a list of exactly three integers", color) + for component_index, component in enumerate(color): + if type(component) is not int or not 0 <= component <= 255: + raise _type_error(f"{path}.color[{component_index}]", "an integer in 0..255", component) + classes.append(OntologyClass(raw_id, name, display_name, description, prompts, aliases, + role, is_heritage, evaluation_groups, tuple(color))) + + raw_groups = data.get("groups") + if not isinstance(raw_groups, Mapping): + raise _type_error("evaluation_groups", "a mapping", raw_groups) + groups = {} + for group_name, members in raw_groups.items(): + if not isinstance(group_name, str) or not group_name.strip(): + raise _type_error("evaluation_groups.", "a non-empty string", group_name) + group_path = f"evaluation_groups.{group_name}" + if not isinstance(members, list): + raise _type_error(group_path, "a list of strings", members) + for member_index, member in enumerate(members): + if not isinstance(member, str) or not member.strip(): + raise _type_error(f"{group_path}[{member_index}]", "a non-empty string", member) + groups[group_name] = tuple(members) + return version, ignore, tuple(classes), groups + + +def ontology_from_mapping(data: Mapping[str, Any]) -> Ontology: + version, ignore, classes, groups = _parse_config(data) + ids, names = [c.id for c in classes], [c.name for c in classes] + if len(ids) != len(set(ids)): + raise OntologyError("duplicate numeric class IDs") + if len(names) != len(set(names)): + raise OntologyError("duplicate canonical class names") + if ignore in ids: + raise OntologyError(f"ignore_index {ignore} must not be a class") + aliases = [a.casefold() for c in classes for a in c.aliases] + reserved = {n.casefold() for n in names} + if len(aliases) != len(set(aliases)) or reserved.intersection(aliases): + raise OntologyError("duplicate/conflicting aliases") + colors = [c.color for c in classes] + if len(colors) != len(set(colors)): + raise OntologyError("palette colors must be unique") + expected_names = VERSION_CLASS_NAMES[version] + if ignore != IGNORE_INDEX: + raise OntologyError(f"{version} requires ignore_index=255, got {ignore}") + if ids != list(range(len(expected_names))): + raise OntologyError(f"{version} requires ordered IDs 0..{len(expected_names) - 1}, got {ids}") + if tuple(names) != expected_names: + raise OntologyError(f"{version} requires canonical class order {list(expected_names)}, got {names}") + if len(colors) != len(expected_names): + raise OntologyError(f"{version} requires exactly {len(expected_names)} palette entries") + known = set(names) + for group, members in groups.items(): + unknown = set(members) - known + if unknown: + raise OntologyError(f"group {group} references unknown classes: {sorted(unknown)}") + for c in classes: + unknown_groups = set(c.evaluation_groups) - set(groups) + if unknown_groups: + raise OntologyError(f"class {c.name} references unknown groups: {sorted(unknown_groups)}") + for group in c.evaluation_groups: + if c.name not in groups[group]: + raise OntologyError(f"inconsistent membership for {c.name} in {group}") + class_groups = {c.name: set(c.evaluation_groups) for c in classes} + for group, members in groups.items(): + for member in members: + if group not in class_groups[member]: + raise OntologyError( + f"inconsistent membership: top-level group {group} contains {member}, " + f"but {member}.evaluation_groups omits {group}" + ) + if version == V2_VERSION: + for group, required_members in REQUIRED_V2_GROUPS.items(): + actual = groups.get(group) + if actual is None: + raise OntologyError(f"{version} requires evaluation group {group}") + if tuple(actual) != tuple(required_members): + raise OntologyError(f"{group} must be {list(required_members)}, got {list(actual)}") + ornament = groups.get("ORNAMENT") + if ornament != ("ornament_region",): + raise OntologyError("ORNAMENT must contain exactly ['ornament_region']") + return Ontology(version, ignore, tuple(classes), groups, _canonical_hash(data)) + + +def load_ontology(path: str | Path = DEFAULT_ONTOLOGY) -> Ontology: + path = Path(path) + try: + with path.open(encoding="utf-8") as stream: + data = yaml.safe_load(stream) + except YAMLError as exc: + raise OntologyError(f"{path}: malformed YAML: {exc}") from exc + return ontology_from_mapping(data) + + +def validate_mask_ids(values: Any, ontology: Ontology, source: str = "mask") -> set[int]: + found = extract_mask_ids(values, source) + unknown = found - ontology.valid_ids - {ontology.ignore_index} + if unknown: + raise OntologyError(f"{source}: unknown mask IDs {sorted(unknown)}; allowed IDs are " + f"{sorted(ontology.valid_ids)} plus ignore {ontology.ignore_index}") + return found + + +def extract_mask_ids(values: Any, source: str = "mask") -> set[int]: + """Extract IDs only after proving that a mask has a non-boolean integer dtype.""" + import numpy as np + array = np.asarray(values) + found_values = _display_unique_values(array) + if array.dtype == np.bool_ or not np.issubdtype(array.dtype, np.integer): + raise OntologyError( + f"{source}: mask dtype must be an integer dtype, got {array.dtype}; " + f"found IDs {found_values}" + ) + return set(np.unique(array).tolist()) + + +def _display_unique_values(array: Any) -> list[Any]: + """Return JSON/error-friendly unique values without coercing their type.""" + import numpy as np + try: + return np.unique(array).tolist() + except (TypeError, ValueError): + return list(dict.fromkeys(repr(value) for value in np.asarray(array).flat)) diff --git a/ovs_heritage/projection.py b/ovs_heritage/projection.py new file mode 100644 index 0000000..b37f34d --- /dev/null +++ b/ovs_heritage/projection.py @@ -0,0 +1,193 @@ +"""Canonical semantic-ID projection for the v2 two-target representation.""" +from __future__ import annotations + +from dataclasses import asdict, dataclass +import math + +import torch + +from .ontology import Ontology, V2_CLASS_NAMES, V2_VERSION + + +IGNORE_INDEX = 255 +MAIN_SEMANTIC_IDS = (0, 1, 2, 3, 4, 5, 6, 7, 9, 10, 11) +IGNORE_BEHAVIOR = "ignore value 255 is preserved and excluded from loss" +UNKNOWN_BEHAVIOR = "unknown IDs raise an error and are never remapped to ignore" + + +@dataclass(frozen=True) +class MappingEntry: + semantic_id: int + canonical_name: str + output_head: str + channel_index: int + interpretation: str + ignore_behavior: str = IGNORE_BEHAVIOR + unknown_behavior: str = UNKNOWN_BEHAVIOR + + +@dataclass(frozen=True) +class OntologyProjection: + entries: tuple[MappingEntry, ...] + ignore_index: int = IGNORE_INDEX + + @classmethod + def from_ontology(cls, ontology: Ontology) -> "OntologyProjection": + if ontology.version != V2_VERSION: + raise ValueError(f"canonical two-head projection requires {V2_VERSION}") + if ontology.by_name("ornament_region").id != 8: + raise ValueError("semantic ID 8 must be exactly ornament_region") + if tuple(sorted(ontology.valid_ids)) != tuple(range(12)): + raise ValueError("v2 projection requires semantic IDs 0..11") + main = tuple( + MappingEntry( + semantic_id, + ontology.by_name(next(item.name for item in ontology.classes if item.id == semantic_id)).name, + "main", + channel, + "multiclass_softmax", + ) + for channel, semantic_id in enumerate(MAIN_SEMANTIC_IDS) + ) + ornament = MappingEntry(8, ontology.by_name("ornament_region").name, "ornament", 0, "independent_sigmoid") + projection = cls(main + (ornament,)) + if tuple(entry.channel_index for entry in projection.main_entries) != tuple(range(11)): + raise ValueError("main channel indices must be contiguous 0..10") + return projection + + def __post_init__(self) -> None: + if type(self.ignore_index) is not int or self.ignore_index != IGNORE_INDEX: + raise ValueError("ignore_index must be the exact integer 255") + if not isinstance(self.entries, tuple) or not self.entries: + raise ValueError("projection entries must be a non-empty tuple") + known_heads = {"main": "multiclass_softmax", "ornament": "independent_sigmoid"} + for index, entry in enumerate(self.entries): + if not isinstance(entry, MappingEntry): + raise ValueError(f"entries[{index}] must be a MappingEntry") + if ( + type(entry.semantic_id) is not int + or not 0 <= entry.semantic_id < len(V2_CLASS_NAMES) + ): + raise ValueError( + f"entries[{index}].semantic_id must be an integer in 0..{len(V2_CLASS_NAMES) - 1}" + ) + if not isinstance(entry.canonical_name, str) or not entry.canonical_name.strip(): + raise ValueError(f"entries[{index}].canonical_name must be non-empty") + if entry.canonical_name != V2_CLASS_NAMES[entry.semantic_id]: + raise ValueError( + f"entries[{index}].canonical_name must be " + f"{V2_CLASS_NAMES[entry.semantic_id]!r} for semantic ID {entry.semantic_id}" + ) + if entry.output_head not in known_heads: + raise ValueError(f"entries[{index}] has unknown output head {entry.output_head!r}") + if entry.interpretation != known_heads[entry.output_head]: + raise ValueError(f"entries[{index}] has inconsistent output interpretation") + if type(entry.channel_index) is not int or entry.channel_index < 0: + raise ValueError(f"entries[{index}].channel_index must be a non-negative integer") + if entry.ignore_behavior != IGNORE_BEHAVIOR: + raise ValueError( + f"entries[{index}].ignore_behavior must be the canonical ignore policy" + ) + if entry.unknown_behavior != UNKNOWN_BEHAVIOR: + raise ValueError( + f"entries[{index}].unknown_behavior must be the canonical unknown-ID policy" + ) + semantic_ids = [entry.semantic_id for entry in self.entries] + if len(semantic_ids) != len(set(semantic_ids)): + raise ValueError("projection has duplicate semantic IDs") + head_channels = [(entry.output_head, entry.channel_index) for entry in self.entries] + if len(head_channels) != len(set(head_channels)): + raise ValueError("projection has duplicate channel indices within an output head") + main_entries = tuple(entry for entry in self.entries if entry.output_head == "main") + ornament_entries = tuple(entry for entry in self.entries if entry.output_head == "ornament") + if tuple(entry.semantic_id for entry in main_entries) != MAIN_SEMANTIC_IDS: + raise ValueError(f"main semantic IDs must be exactly {list(MAIN_SEMANTIC_IDS)}") + if tuple(entry.channel_index for entry in main_entries) != tuple(range(len(MAIN_SEMANTIC_IDS))): + raise ValueError("main channel indices must be contiguous 0..10") + if len(ornament_entries) != 1 or ( + ornament_entries[0].semantic_id, + ornament_entries[0].channel_index, + ) != (8, 0): + raise ValueError("ornament projection must be semantic ID 8 at channel 0") + + @property + def main_entries(self) -> tuple[MappingEntry, ...]: + return tuple(entry for entry in self.entries if entry.output_head == "main") + + @property + def main_channel_count(self) -> int: + return len(self.main_entries) + + def for_semantic_id(self, semantic_id: int) -> MappingEntry: + matches = [entry for entry in self.entries if entry.semantic_id == semantic_id] + if len(matches) != 1: + raise ValueError(f"semantic ID {semantic_id} has no unique canonical projection") + return matches[0] + + def for_channel(self, head: str, channel_index: int) -> MappingEntry: + matches = [ + entry for entry in self.entries + if entry.output_head == head and entry.channel_index == channel_index + ] + if len(matches) != 1: + raise ValueError(f"{head} channel {channel_index} has no unique canonical projection") + return matches[0] + + def semantic_main_to_channels(self, target: torch.Tensor) -> torch.Tensor: + self._validate_integer_target(target, "Y_main") + found = set(torch.unique(target.detach()).cpu().tolist()) + semantic_to_channel = {entry.semantic_id: entry.channel_index for entry in self.main_entries} + allowed = set(semantic_to_channel) | {self.ignore_index} + invalid = sorted(found - allowed) + if invalid: + detail = "semantic ID 8 belongs to the ornament target" if 8 in invalid else "unknown IDs" + raise ValueError(f"Y_main contains invalid semantic IDs {invalid}: {detail}") + result = torch.full_like(target, self.ignore_index, dtype=torch.long) + for semantic_id, channel_index in semantic_to_channel.items(): + result[target == semantic_id] = channel_index + return result + + def main_channels_to_semantic(self, channels: torch.Tensor) -> torch.Tensor: + self._validate_integer_target(channels, "main channel prediction") + found = set(torch.unique(channels.detach()).cpu().tolist()) + channel_to_semantic = {entry.channel_index: entry.semantic_id for entry in self.main_entries} + allowed = set(channel_to_semantic) | {self.ignore_index} + invalid = sorted(found - allowed) + if invalid: + raise ValueError(f"main channel prediction contains unknown channel indices {invalid}") + result = torch.full_like(channels, self.ignore_index, dtype=torch.long) + for channel_index, semantic_id in channel_to_semantic.items(): + result[channels == channel_index] = semantic_id + return result + + def main_logits_to_semantic(self, logits: torch.Tensor) -> torch.Tensor: + if logits.ndim != 4 or logits.shape[1] != self.main_channel_count or not logits.is_floating_point(): + raise ValueError(f"main logits must be finite floating [N,{self.main_channel_count},H,W]") + if not torch.isfinite(logits).all(): + raise ValueError("main logits contain non-finite values") + return self.main_channels_to_semantic(logits.argmax(dim=1)) + + def ornament_logits_to_binary(self, logits: torch.Tensor, *, threshold: float = 0.5) -> torch.Tensor: + if logits.ndim != 4 or logits.shape[1] != 1 or not logits.is_floating_point(): + raise ValueError("ornament logits must be floating [N,1,H,W] raw logits") + if not math.isfinite(threshold) or not 0 <= threshold <= 1: + raise ValueError("ornament threshold must be in 0..1") + if not torch.isfinite(logits).all(): + raise ValueError("ornament logits contain non-finite values") + return (torch.sigmoid(logits) >= threshold).to(torch.uint8) + + def as_dict(self) -> dict[str, object]: + return { + "ignore_index": self.ignore_index, + "entries": [asdict(entry) for entry in self.entries], + } + + @staticmethod + def _validate_integer_target(target: torch.Tensor, label: str) -> None: + integer_dtypes = {torch.uint8, torch.int8, torch.int16, torch.int32, torch.int64} + if target.dtype not in integer_dtypes: + raise ValueError(f"{label} must have an integer dtype, got {target.dtype}") + if target.ndim not in (2, 3, 4): + raise ValueError(f"{label} must be a spatial target tensor, got shape {tuple(target.shape)}") + if target.ndim == 4 and target.shape[1] != 1: + raise ValueError(f"{label} four-dimensional targets must have exactly one channel") diff --git a/ovs_heritage/scoring.py b/ovs_heritage/scoring.py new file mode 100644 index 0000000..7e6e00e --- /dev/null +++ b/ovs_heritage/scoring.py @@ -0,0 +1,52 @@ +"""Raw cosine dense scorer; intentionally contains no softmax or vocabulary state.""" +from __future__ import annotations +import math + +import torch +import torch.nn.functional as F +from torch import nn + + +class RawCosineScorer(nn.Module): + def __init__(self, scale: float = 100.0, eps: float = 1e-12): + super().__init__() + if not math.isfinite(eps) or eps <= 0: + raise ValueError("eps must be finite and positive") + self.scale = float(scale) + self.eps = eps + + def forward(self, features: torch.Tensor, prototypes: torch.Tensor, + *, scale: torch.Tensor | float | None = None, + bias: torch.Tensor | float | None = None) -> torch.Tensor: + if features.ndim not in (3, 4): + raise ValueError("features must be [D,H,W] or [N,D,H,W]") + if prototypes.ndim != 2: + raise ValueError("prototypes must be [C,D]") + if not features.is_floating_point() or not prototypes.is_floating_point(): + raise ValueError("features and prototypes must be floating-point tensors") + if not torch.isfinite(features).all() or not torch.isfinite(prototypes).all(): + raise ValueError("features and prototypes must be finite") + if torch.any(torch.linalg.vector_norm(prototypes, dim=1) <= self.eps): + raise ValueError("prototypes must have finite non-zero norms") + unbatched = features.ndim == 3 + if unbatched: + features = features.unsqueeze(0) + if features.shape[1] != prototypes.shape[1]: + raise ValueError(f"embedding dimension mismatch: features={features.shape[1]}, prototypes={prototypes.shape[1]}") + prototypes = prototypes.to(device=features.device, dtype=features.dtype) + logits = torch.einsum("ndhw,cd->nchw", F.normalize(features, dim=1, eps=self.eps), + F.normalize(prototypes, dim=1, eps=self.eps)) + scale = self.scale if scale is None else scale + scale = torch.as_tensor(scale, device=logits.device, dtype=logits.dtype) + bias = torch.as_tensor(0.0 if bias is None else bias, device=logits.device, dtype=logits.dtype) + for value, label in ((scale, "scale"), (bias, "bias")): + if value.ndim > 1 or (value.ndim == 1 and value.numel() not in (1, prototypes.shape[0])): + raise ValueError(f"{label} must be scalar or have one value per class") + if not torch.isfinite(value).all(): + raise ValueError(f"{label} must be finite") + if scale.ndim: + scale = scale.view(1, -1, 1, 1) + if bias.ndim: + bias = bias.view(1, -1, 1, 1) + logits = logits * scale + bias + return logits[0] if unbatched else logits diff --git a/ovs_heritage/tests/test_dataset_validation.py b/ovs_heritage/tests/test_dataset_validation.py new file mode 100644 index 0000000..b49e068 --- /dev/null +++ b/ovs_heritage/tests/test_dataset_validation.py @@ -0,0 +1,456 @@ +import csv +import json +from pathlib import Path + +import numpy as np +from PIL import Image +import pytest + +from ovs_heritage.ontology import V1_VERSION, load_ontology, ontology_from_mapping +from ovs_heritage.validate_dataset import ( + V1_DATASET_SCHEMA, + V2_DATASET_SCHEMA, + main, + validate_splits, +) + + +def validate_v2(sources): + ontology = load_ontology() + return validate_splits( + sources, + ontology, + schema_version=V2_DATASET_SCHEMA, + ontology_version=ontology.version, + ) + + +def make_v1_ontology(): + with open("ovs_heritage/configs/heritage_vocab.yaml", encoding="utf-8") as stream: + data = json.load(stream) + data["version"] = V1_VERSION + data["classes"] = data["classes"][:11] + data["classes"][8]["name"] = "ornament_intact" + data["classes"][8]["aliases"] = [] + data["groups"]["ORNAMENT"] = ["ornament_intact"] + data["groups"]["HUMAN_ACTIVITY"].remove("advertisements") + return ontology_from_mapping(data) + + +def save(path, values): + Image.fromarray(np.asarray(values, dtype=np.uint8)).save(path) + + +def write_v2_manifest(path, rows): + with path.open("w", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=["main_mask_path", "ornament_mask_path", "facade_id", "source_id"]) + writer.writeheader() + writer.writerows(rows) + + +def test_v2_two_maps_preserve_overlap_advertisements_and_metadata(tmp_path): + main = tmp_path / "main.png" + ornament = tmp_path / "ornament.png" + save(main, [[7, 5, 11, 255]]) + save(ornament, [[1, 1, 0, 255]]) + manifest = tmp_path / "v2.csv" + write_v2_manifest(manifest, [{"main_mask_path": main.name, "ornament_mask_path": ornament.name, + "facade_id": "facade_1", "source_id": "source_1"}]) + report = validate_v2({"test": manifest}) + assert report["valid"] + assert report["splits"]["test"]["valid_sample_count"] == 1 + assert report["splits"]["test"]["main_pixel_count"]["7"] == 1 + assert report["splits"]["test"]["ornament_pixel_count"]["1"] == 2 + assert report["reproducibility"]["hash"] + assert report["semantic_projection"]["entries"][-1]["semantic_id"] == 8 + + +def test_v2_invalid_labels_shape_missing_facade_and_empty_split(tmp_path): + cases = [ + ([[8]], [[1]], "f", "invalid Y_main"), + ([[42]], [[0]], "f", "invalid Y_main"), + ([[0]], [[2]], "f", "invalid Y_ornament"), + ([[0, 1]], [[0]], "f", "shape mismatch"), + ([[0]], [[0]], "", "requires non-empty facade_id"), + ] + for index, (main_values, ornament_values, facade, message) in enumerate(cases): + main = tmp_path / f"m{index}.png" + ornament = tmp_path / f"o{index}.png" + save(main, main_values) + save(ornament, ornament_values) + manifest = tmp_path / f"case{index}.csv" + write_v2_manifest(manifest, [{"main_mask_path": main.name, "ornament_mask_path": ornament.name, + "facade_id": facade, "source_id": str(index)}]) + report = validate_v2({"test": manifest}) + assert not report["valid"] and message in " ".join(report["errors"]) + empty = tmp_path / "empty.csv" + write_v2_manifest(empty, []) + assert "split is empty" in " ".join(validate_v2({"test": empty})["errors"]) + + +def test_facade_and_path_leakage_across_splits(tmp_path): + main = tmp_path / "main.png" + ornament = tmp_path / "ornament.png" + save(main, [[0]]) + save(ornament, [[0]]) + manifests = [] + for split in ("train", "test"): + manifest = tmp_path / f"{split}.csv" + write_v2_manifest(manifest, [{"main_mask_path": main.name, "ornament_mask_path": ornament.name, + "facade_id": "same", "source_id": "same"}]) + manifests.append(manifest) + report = validate_v2({"train": manifests[0], "test": manifests[1]}) + assert report["facade_overlaps"] and report["duplicated_paths"] and not report["valid"] + + +def test_v1_explicit_schema_rejects_id11(tmp_path): + ontology = make_v1_ontology() + mask = tmp_path / "legacy.png" + save(mask, [[11]]) + manifest = tmp_path / "legacy.csv" + with manifest.open("w", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=["mask_path", "facade_id"]) + writer.writeheader() + writer.writerow({"mask_path": mask.name, "facade_id": "f"}) + report = validate_splits( + {"test": manifest}, + ontology, + schema_version=V1_DATASET_SCHEMA, + ontology_version=ontology.version, + ) + assert not report["valid"] and "invalid legacy-v1 IDs [11]" in " ".join(report["errors"]) + + +def test_cli_writes_report_on_failure(tmp_path): + manifest = tmp_path / "empty.csv" + write_v2_manifest(manifest, []) + output = tmp_path / "report.json" + ontology = load_ontology() + assert main([ + "--test", str(manifest), + "--schema-version", V2_DATASET_SCHEMA, + "--ontology-version", ontology.version, + "--output", str(output), "--strict", + ]) == 1 + assert output.exists() and json.loads(output.read_text())["valid"] is False + + +def test_schema_and_ontology_declarations_are_required_and_must_match(tmp_path): + ontology = load_ontology() + with pytest.raises(TypeError): + validate_splits({"test": tmp_path / "missing.csv"}, ontology) + with pytest.raises(ValueError, match="unsupported dataset schema"): + validate_splits( + {}, ontology, schema_version="typo", ontology_version=ontology.version, + ) + with pytest.raises(ValueError, match="does not match loaded"): + validate_splits( + {}, ontology, schema_version=V2_DATASET_SCHEMA, + ontology_version="heritage_facades_v2_typo", + ) + with pytest.raises(ValueError, match="requires dataset schema"): + validate_splits( + {}, ontology, schema_version=V1_DATASET_SCHEMA, + ontology_version=ontology.version, + ) + + +def test_conflicting_row_declaration_and_split_statistics(tmp_path): + good_main = tmp_path / "good_main.png" + good_ornament = tmp_path / "good_ornament.png" + save(good_main, [[0]]) + save(good_ornament, [[0]]) + manifest = tmp_path / "mixed.csv" + fields = [ + "main_mask_path", "ornament_mask_path", "facade_id", + "schema_version", "ontology_version", + ] + ontology = load_ontology() + with manifest.open("w", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=fields) + writer.writeheader() + writer.writerow({ + "main_mask_path": good_main.name, + "ornament_mask_path": good_ornament.name, + "facade_id": "ok", + "schema_version": V2_DATASET_SCHEMA, + "ontology_version": ontology.version, + }) + writer.writerow({ + "main_mask_path": "missing.png", + "ornament_mask_path": good_ornament.name, + "facade_id": "bad", + "schema_version": V2_DATASET_SCHEMA, + "ontology_version": ontology.version, + }) + report = validate_v2({"test": manifest}) + split = report["splits"]["test"] + assert split["manifest_row_count"] == 2 + assert split["valid_sample_count"] == 1 + assert split["failed_sample_count"] == 1 + assert split["split_error_count"] == 0 + assert split["valid_sample_count"] + split["failed_sample_count"] == split["manifest_row_count"] + + conflict = tmp_path / "conflict.csv" + with conflict.open("w", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=fields) + writer.writeheader() + writer.writerow({ + "main_mask_path": good_main.name, + "ornament_mask_path": good_ornament.name, + "facade_id": "f", + "schema_version": "wrong", + "ontology_version": ontology.version, + }) + conflict_report = validate_v2({"test": conflict}) + conflict_split = conflict_report["splits"]["test"] + assert conflict_split["manifest_row_count"] == 1 + assert conflict_split["split_error_count"] == 0 + assert conflict_split["valid_sample_count"] == 0 + assert conflict_split["failed_sample_count"] == 1 + assert "conflicting schema_version" in conflict_split["sample_errors"][0] + + +def test_empty_and_unreadable_manifests_are_split_errors(tmp_path): + empty = tmp_path / "empty_again.csv" + write_v2_manifest(empty, []) + empty_split = validate_v2({"test": empty})["splits"]["test"] + assert empty_split["manifest_row_count"] == 0 + assert empty_split["failed_sample_count"] == 0 + assert empty_split["split_error_count"] == 1 + + unreadable = validate_v2({"test": tmp_path / "missing.csv"})["splits"]["test"] + assert unreadable["manifest_row_count"] == 0 + assert unreadable["failed_sample_count"] == 0 + assert unreadable["split_error_count"] == 1 + + +def test_two_declaration_errors_count_as_one_failed_row(tmp_path): + main_mask = tmp_path / "declaration_main.png" + ornament_mask = tmp_path / "declaration_ornament.png" + save(main_mask, [[0]]) + save(ornament_mask, [[0]]) + manifest = tmp_path / "two_conflicts.json" + manifest.write_text(json.dumps([{ + "main_mask_path": main_mask.name, + "ornament_mask_path": ornament_mask.name, + "facade_id": "facade", + "schema_version": "wrong_schema", + "ontology_version": "wrong_ontology", + }])) + split = validate_v2({"test": manifest})["splits"]["test"] + assert split["manifest_row_count"] == 1 + assert split["valid_sample_count"] == 0 + assert split["failed_sample_count"] == 1 + assert split["split_error_count"] == 0 + assert len(split["sample_errors"]) == 2 + assert "conflicting schema_version" in split["sample_errors"][0] + assert "conflicting ontology_version" in split["sample_errors"][1] + assert split["valid_sample_count"] + split["failed_sample_count"] == 1 + + +def test_multiple_invalid_rows_count_as_failed_samples(tmp_path): + ornament = tmp_path / "ornament.png" + save(ornament, [[0]]) + manifest = tmp_path / "invalid_rows.csv" + write_v2_manifest(manifest, [ + {"main_mask_path": "missing_a.png", "ornament_mask_path": ornament.name, + "facade_id": "a", "source_id": "a"}, + {"main_mask_path": "missing_b.png", "ornament_mask_path": ornament.name, + "facade_id": "b", "source_id": "b"}, + ]) + split = validate_v2({"test": manifest})["splits"]["test"] + assert split["manifest_row_count"] == 2 + assert split["valid_sample_count"] == 0 + assert split["failed_sample_count"] == 2 + assert split["split_error_count"] == 0 + + +def write_v2_manifest_with_image(path, rows): + fields = ["image_path", "main_mask_path", "ornament_mask_path", "facade_id", "source_id"] + with path.open("w", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=fields) + writer.writeheader() + writer.writerows(rows) + + +def test_source_id_and_image_path_leakage_with_different_masks(tmp_path): + image = tmp_path / "image.png" + save(image, [[0]]) + manifests = [] + for split in ("train", "test"): + main_mask = tmp_path / f"{split}_main.png" + ornament_mask = tmp_path / f"{split}_ornament.png" + save(main_mask, [[0]]) + save(ornament_mask, [[0]]) + manifest = tmp_path / f"{split}_source.csv" + write_v2_manifest_with_image(manifest, [{ + "image_path": image.name, + "main_mask_path": main_mask.name, + "ornament_mask_path": ornament_mask.name, + "facade_id": f"facade_{split}", + "source_id": "repeated_source", + }]) + manifests.append(manifest) + report = validate_v2({"train": manifests[0], "test": manifests[1]}) + assert report["source_id_overlaps"][0]["source_ids"] == ["repeated_source"] + reused = report["duplicated_paths"][0]["paths"] + assert reused[0]["roles"] == {"train": ["image_path"], "test": ["image_path"]} + assert report["splits"]["train"]["verified_image_count"] == 1 + + +def test_optional_image_must_be_readable_and_match_mask_grid(tmp_path): + main_mask = tmp_path / "main.png" + ornament_mask = tmp_path / "ornament.png" + save(main_mask, [[0]]) + save(ornament_mask, [[0]]) + for image_name, expected in (("missing.png", "missing or unreadable"), ("corrupt.png", "missing or unreadable")): + if image_name == "corrupt.png": + (tmp_path / image_name).write_text("not an image") + manifest = tmp_path / f"{image_name}.csv" + write_v2_manifest_with_image(manifest, [{ + "image_path": image_name, + "main_mask_path": main_mask.name, + "ornament_mask_path": ornament_mask.name, + "facade_id": "facade", + "source_id": image_name, + }]) + assert expected in " ".join(validate_v2({"test": manifest})["errors"]) + large_image = tmp_path / "large.png" + save(large_image, [[0, 0]]) + mismatch = tmp_path / "mismatch.csv" + write_v2_manifest_with_image(mismatch, [{ + "image_path": large_image.name, + "main_mask_path": main_mask.name, + "ornament_mask_path": ornament_mask.name, + "facade_id": "facade", + "source_id": "large", + }]) + assert "image/mask grid mismatch" in " ".join(validate_v2({"test": mismatch})["errors"]) + + +def test_facade_leakage_survives_corrupted_mask_and_whitespace_is_rejected(tmp_path): + ornament = tmp_path / "ornament.png" + good_main = tmp_path / "good.png" + save(ornament, [[0]]) + save(good_main, [[0]]) + train = tmp_path / "train_corrupt.csv" + test = tmp_path / "test_good.csv" + write_v2_manifest(train, [{ + "main_mask_path": "missing.png", "ornament_mask_path": ornament.name, + "facade_id": "shared", "source_id": "train", + }]) + write_v2_manifest(test, [{ + "main_mask_path": good_main.name, "ornament_mask_path": ornament.name, + "facade_id": "shared", "source_id": "test", + }]) + report = validate_v2({"train": train, "test": test}) + assert report["facade_overlaps"][0]["facade_ids"] == ["shared"] + + whitespace = tmp_path / "whitespace.csv" + write_v2_manifest(whitespace, [{ + "main_mask_path": good_main.name, "ornament_mask_path": ornament.name, + "facade_id": " shared", "source_id": "sample ", + }]) + errors = " ".join(validate_v2({"test": whitespace})["errors"]) + assert "surrounding whitespace" in errors + + +def test_v1_relative_directory_content_fingerprint_is_root_independent(tmp_path, monkeypatch): + ontology = make_v1_ontology() + first = tmp_path / "first" / "masks" + second = tmp_path / "second" / "masks" + first.mkdir(parents=True) + second.mkdir(parents=True) + save(first / "a.png", [[1]]) + save(second / "a.png", [[1]]) + monkeypatch.chdir(tmp_path) + + def validate_directory(path): + return validate_splits( + {"test": path}, ontology, + schema_version=V1_DATASET_SCHEMA, + ontology_version=ontology.version, + ) + + first_report = validate_directory(Path("first/masks")) + second_report = validate_directory(Path("second/masks")) + assert first_report["valid"] and second_report["valid"] + assert first_report["source_fingerprints"]["test"] == second_report["source_fingerprints"]["test"] + save(first / "a.png", [[2]]) + changed = validate_directory(Path("first/masks")) + assert changed["source_fingerprints"]["test"] != first_report["source_fingerprints"]["test"] + + +def test_physical_path_leakage_is_role_independent(tmp_path): + shared = tmp_path / "shared.png" + train_ornament = tmp_path / "train_ornament.png" + test_main = tmp_path / "test_main.png" + save(shared, [[0]]) + save(train_ornament, [[0]]) + save(test_main, [[0]]) + train = tmp_path / "train_roles.csv" + test = tmp_path / "test_roles.csv" + write_v2_manifest(train, [{ + "main_mask_path": shared.name, + "ornament_mask_path": train_ornament.name, + "facade_id": "train_facade", + "source_id": "train_source", + }]) + write_v2_manifest(test, [{ + "main_mask_path": test_main.name, + "ornament_mask_path": shared.name, + "facade_id": "test_facade", + "source_id": "test_source", + }]) + report = validate_v2({"train": train, "test": test}) + duplicate = report["duplicated_paths"][0]["paths"][0] + assert duplicate["roles"] == { + "train": ["main_mask_path"], + "test": ["ornament_mask_path"], + } + + +def test_v2_rejects_one_physical_file_in_multiple_roles(tmp_path): + shared = tmp_path / "shared_roles.png" + save(shared, [[0]]) + manifest = tmp_path / "same_row_roles.csv" + write_v2_manifest(manifest, [{ + "main_mask_path": shared.name, + "ornament_mask_path": shared.name, + "facade_id": "facade", + "source_id": "source", + }]) + errors = validate_v2({"test": manifest})["splits"]["test"]["sample_errors"] + assert "one physical file cannot serve multiple roles" in errors[0] + + +@pytest.mark.parametrize("field,value", [("facade_id", " bad"), ("source_id", 7)]) +def test_v1_optional_identifiers_are_strict(tmp_path, field, value): + ontology = make_v1_ontology() + mask = tmp_path / "legacy_id.png" + save(mask, [[0]]) + manifest = tmp_path / f"legacy_{field}.json" + row = {"mask_path": mask.name, field: value} + manifest.write_text(json.dumps([row])) + report = validate_splits( + {"test": manifest}, + ontology, + schema_version=V1_DATASET_SCHEMA, + ontology_version=ontology.version, + ) + assert report["splits"]["test"]["failed_sample_count"] == 1 + assert field in report["splits"]["test"]["sample_errors"][0] + + +def test_v1_does_not_coerce_non_string_mask_path(tmp_path): + ontology = make_v1_ontology() + manifest = tmp_path / "legacy_numeric_path.json" + manifest.write_text(json.dumps([{"mask_path": 123}])) + report = validate_splits( + {"test": manifest}, + ontology, + schema_version=V1_DATASET_SCHEMA, + ontology_version=ontology.version, + ) + assert "mask_path must be a non-empty string" in " ".join(report["errors"]) diff --git a/ovs_heritage/tests/test_end_to_end.py b/ovs_heritage/tests/test_end_to_end.py new file mode 100644 index 0000000..22f70b9 --- /dev/null +++ b/ovs_heritage/tests/test_end_to_end.py @@ -0,0 +1,57 @@ +import csv +import json + +import numpy as np +from PIL import Image +import torch + +from ovs_heritage.losses import combined_two_head_loss +from ovs_heritage.metadata import make_metadata +from ovs_heritage.ontology import load_ontology +from ovs_heritage.projection import OntologyProjection +from ovs_heritage.validate_dataset import V2_DATASET_SCHEMA, validate_splits + + +def test_cpu_two_map_p0_flow(tmp_path): + ontology = load_ontology() + projection = OntologyProjection.from_ontology(ontology) + main_path = tmp_path / "main.png" + ornament_path = tmp_path / "ornament.png" + Image.fromarray(np.array([[7, 5, 11, 255]], dtype=np.uint8)).save(main_path) + Image.fromarray(np.array([[1, 1, 0, 255]], dtype=np.uint8)).save(ornament_path) + manifest = tmp_path / "manifest.csv" + with manifest.open("w", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=["main_mask_path", "ornament_mask_path", "facade_id"]) + writer.writeheader() + writer.writerow({"main_mask_path": main_path.name, "ornament_mask_path": ornament_path.name, + "facade_id": "facade_1"}) + report = validate_splits( + {"test": manifest}, ontology, + schema_version=V2_DATASET_SCHEMA, ontology_version=ontology.version, + ) + assert report["valid"] + + y_main = torch.tensor([[[7, 5, 11, 255]]]) + y_ornament = torch.tensor([[[1, 1, 0, 255]]]) + main_logits = torch.randn(1, 11, 1, 4, requires_grad=True) + ornament_logits = torch.randn(1, 1, 1, 4, requires_grad=True) + channel_target = projection.semantic_main_to_channels(y_main) + assert channel_target.tolist() == [[[7, 5, 10, 255]]] + losses = combined_two_head_loss( + main_logits, ornament_logits, y_main, y_ornament, lambda_ornament=0.5, + ) + losses.total.backward() + semantic_prediction = projection.main_logits_to_semantic(main_logits.detach()) + ornament_prediction = projection.ornament_logits_to_binary( + ornament_logits.detach(), threshold=0.5, + ) + assert semantic_prediction.shape == y_main.shape + assert ornament_prediction.shape == ornament_logits.shape + metadata = make_metadata( + component_name="p0.synthetic_flow", component_version="1", + ontology_version=ontology.version, ontology_hash=ontology.hash, + mapping=projection.as_dict(), validator_schema_version=report["validator_schema_version"], + source_fingerprints=report["source_fingerprints"], loss_settings=losses.metadata, + ornament_threshold=0.5, + ) + assert json.loads(metadata.to_json())["hash"] == metadata.hash diff --git a/ovs_heritage/tests/test_losses.py b/ovs_heritage/tests/test_losses.py new file mode 100644 index 0000000..35fdcb6 --- /dev/null +++ b/ovs_heritage/tests/test_losses.py @@ -0,0 +1,51 @@ +import torch +import torch.nn.functional as F +import pytest + +from ovs_heritage.losses import combined_two_head_loss, main_segmentation_loss, ornament_region_loss +from ovs_heritage.ontology import load_ontology +from ovs_heritage.projection import OntologyProjection + + +def test_main_loss_maps_semantic_ids_and_uses_raw_logits(): + logits = torch.randn(1, 11, 1, 2, requires_grad=True) + semantic = torch.tensor([[[9, 255]]]) + channels = OntologyProjection.from_ontology(load_ontology()).semantic_main_to_channels(semantic) + got = main_segmentation_loss(logits, semantic) + assert torch.allclose(got, F.cross_entropy(logits, channels, ignore_index=255)) + + +def test_main_all_ignore_is_differentiable_zero_and_id8_rejected(): + logits = torch.randn(1, 11, 2, 2, requires_grad=True) + loss = main_segmentation_loss(logits, torch.full((1, 2, 2), 255)) + assert loss.item() == 0 and loss.requires_grad + with pytest.raises(ValueError, match="semantic ID 8"): + main_segmentation_loss(logits, torch.full((1, 2, 2), 8)) + + +def test_ornament_ignore_mask_and_all_ignore(): + logits = torch.tensor([[[[0.0, 10.0, -2.0]]]], requires_grad=True) + target = torch.tensor([[[[1, 255, 0]]]]) + got = ornament_region_loss(logits, target) + expected = F.binary_cross_entropy_with_logits(logits[..., [0, 2]], torch.tensor([[[[1.0, 0.0]]]])) + assert torch.allclose(got, expected) + all_ignore = ornament_region_loss(logits, torch.full_like(target, 255)) + assert all_ignore.item() == 0 and all_ignore.requires_grad + + +def test_combined_loss_settings_and_overlap_targets(): + main = torch.randn(1, 11, 1, 2) + ornament = torch.randn(1, 1, 1, 2) + y_main = torch.tensor([[[7, 5]]]) + y_ornament = torch.tensor([[[1, 1]]]) + result = combined_two_head_loss(main, ornament, y_main, y_ornament, lambda_ornament=0.25) + assert torch.allclose(result.total, result.main + 0.25 * result.ornament) + assert result.metadata == {"lambda_ornament": 0.25, "pos_weight": None} + with pytest.raises(ValueError, match="non-negative"): + combined_two_head_loss(main, ornament, y_main, y_ornament, lambda_ornament=-1) + + +def test_probability_simplex_values_are_not_used_to_guess_provenance(): + simplex_values = torch.softmax(torch.randn(1, 11, 2, 2), dim=1) + loss = main_segmentation_loss(simplex_values, torch.zeros(1, 2, 2, dtype=torch.long)) + assert torch.isfinite(loss) diff --git a/ovs_heritage/tests/test_metadata.py b/ovs_heritage/tests/test_metadata.py new file mode 100644 index 0000000..9445d49 --- /dev/null +++ b/ovs_heritage/tests/test_metadata.py @@ -0,0 +1,54 @@ +import json + +from ovs_heritage.metadata import make_metadata +from ovs_heritage.ontology import load_ontology +from ovs_heritage.projection import OntologyProjection + + +def test_metadata_is_deterministic_json_serializable(): + ontology = load_ontology() + kwargs = dict( + component_name="test", component_version="1", ontology_version=ontology.version, + ontology_hash=ontology.hash, mapping=OntologyProjection.from_ontology(load_ontology()).as_dict(), + validator_schema_version="v2", source_fingerprints={"test": "abc"}, + loss_settings={"lambda_ornament": 0.5, "pos_weight": None}, ornament_threshold=0.5, + ) + first = make_metadata(**kwargs) + second = make_metadata(**kwargs) + assert first.hash == second.hash + assert json.loads(first.to_json()) == first.to_dict() + + +def test_nested_caller_mutation_cannot_change_record_and_exports_are_copies(): + nested = {"entries": [{"semantic_id": 1}]} + record = make_metadata( + component_name="test", component_version="1", ontology_version="v", + ontology_hash="h", mapping=nested, + ) + original_hash = record.hash + nested["entries"][0]["semantic_id"] = 99 + exported = record.to_dict() + exported["payload"]["mapping"]["entries"][0]["semantic_id"] = 42 + assert record.hash == original_hash + assert record.to_dict()["payload"]["mapping"]["entries"][0]["semantic_id"] == 1 + + +def test_metadata_rejects_invalid_values_and_thresholds(): + import pytest + + for value in (float("nan"), float("inf"), -0.1, 1.1): + with pytest.raises(ValueError): + make_metadata( + component_name="test", component_version="1", ontology_version="v", + ontology_hash="h", mapping={}, ornament_threshold=value, + ) + with pytest.raises(ValueError, match="non-empty"): + make_metadata( + component_name="", component_version="1", ontology_version="v", + ontology_hash="h", mapping={}, + ) + with pytest.raises(TypeError, match="not JSON serializable"): + make_metadata( + component_name="test", component_version="1", ontology_version="v", + ontology_hash="h", mapping={"bad": object()}, + ) diff --git a/ovs_heritage/tests/test_ontology.py b/ovs_heritage/tests/test_ontology.py new file mode 100644 index 0000000..be93f19 --- /dev/null +++ b/ovs_heritage/tests/test_ontology.py @@ -0,0 +1,211 @@ +import json + +import numpy as np +import pytest + +from ovs_heritage.ontology import ( + OntologyError, + load_ontology, + ontology_from_mapping, + validate_mask_ids, +) + +CONFIG = "ovs_heritage/configs/heritage_vocab.yaml" + + +def config(): + with open(CONFIG, encoding="utf-8") as stream: + return json.load(stream) + + +def test_exact_v2_ontology_and_groups(): + ontology = load_ontology() + assert [item.id for item in ontology.classes] == list(range(12)) + assert ontology.ignore_index == 255 and 255 not in ontology.valid_ids + assert ontology.by_name("background").id == 0 + assert ontology.by_name("text_or_images").id == 10 + assert ontology.by_name("advertisements").id == 11 + assert ontology.by_name("ornament_region").id == 8 + assert "advertisements" in ontology.groups["HUMAN_ACTIVITY"] + assert "advertisements" not in ontology.groups["DAMAGE_MACRO"] + assert tuple(ontology.groups["DAMAGE_MACRO"]) == ontology.class_names[1:8] + assert len(ontology.palette) == len(set(ontology.palette)) == 12 + + +def test_hash_independent_of_mapping_key_order_and_yaml_format(tmp_path): + data = config() + reordered = {key: data[key] for key in reversed(data)} + yaml_path = tmp_path / "ontology.yaml" + import yaml + yaml_path.write_text("# comment\n" + yaml.safe_dump(reordered, sort_keys=False), encoding="utf-8") + assert load_ontology().hash == load_ontology(yaml_path).hash + + +@pytest.mark.parametrize("bad_id", [11.0, True, "11"]) +def test_ontology_ids_must_be_real_integers(bad_id): + data = config() + data["classes"][11]["id"] = bad_id + with pytest.raises(OntologyError, match=r"classes\[11\]\.id must be an integer"): + ontology_from_mapping(data) + + +@pytest.mark.parametrize("version", [ + "heritage_facades_v2_12concepts_two_head", + "arbitrary_unseen_ontology", + "", + 2, +]) +def test_unknown_empty_and_non_string_versions_are_rejected(version): + data = config() + data["version"] = version + with pytest.raises(OntologyError, match=r"supported versions:.*v1_11classes.*v2_12concepts_two_heads"): + ontology_from_mapping(data) + + +def test_unknown_version_is_rejected_before_other_corruption(): + data = config() + data["version"] = "future_unregistered_version" + data["ignore_index"] = 254 + data["classes"][9], data["classes"][11] = data["classes"][11], data["classes"][9] + with pytest.raises(OntologyError, match=r"version 'future_unregistered_version' is unsupported"): + ontology_from_mapping(data) + + +@pytest.mark.parametrize( + ("class_index", "field", "value", "error_path"), + [ + (11, "is_heritage", "false", r"classes\[11\]\.is_heritage"), + (3, "prompts", "abc", r"classes\[3\]\.prompts"), + (11, "aliases", "ad", r"classes\[11\]\.aliases"), + (11, "evaluation_groups", "HUMAN_ACTIVITY", r"classes\[11\]\.evaluation_groups"), + (2, "name", 123, r"classes\[2\]\.name"), + (4, "id", True, r"classes\[4\]\.id"), + (5, "color", [0, True, 2], r"classes\[5\]\.color\[1\]"), + ], +) +def test_class_schema_rejects_coercible_wrong_types(class_index, field, value, error_path): + data = config() + data["classes"][class_index][field] = value + with pytest.raises(OntologyError, match=error_path): + ontology_from_mapping(data) + + +def test_prompt_and_alias_lists_accept_only_schema_valid_lists(): + data = config() + data["classes"][0]["prompts"] = ["a valid non-empty prompt"] + data["classes"][0]["aliases"] = [] + ontology = ontology_from_mapping(data) + assert ontology.classes[0].prompts == ("a valid non-empty prompt",) + assert ontology.classes[0].aliases == () + + +def test_top_level_group_schema_paths_are_strict(): + data = config() + data["groups"]["HUMAN_ACTIVITY"] = "advertisements" + with pytest.raises(OntologyError, match=r"evaluation_groups\.HUMAN_ACTIVITY"): + ontology_from_mapping(data) + + +def test_duplicate_ids_names_and_aliases_are_rejected(): + mutations = ( + lambda data: data["classes"][1].__setitem__("id", 0), + lambda data: data["classes"][1].__setitem__("name", "background"), + lambda data: data["classes"][1].__setitem__("aliases", ["rust"]), + ) + for mutate in mutations: + data = config() + mutate(data) + with pytest.raises(OntologyError): + ontology_from_mapping(data) + + +def test_strict_versions_and_canonical_order(): + data = config() + data["ignore_index"] = 254 + with pytest.raises(OntologyError, match="ignore_index=255"): + ontology_from_mapping(data) + + data = config() + data["classes"][9]["name"], data["classes"][11]["name"] = ( + data["classes"][11]["name"], data["classes"][9]["name"] + ) + with pytest.raises(OntologyError, match="canonical class order"): + ontology_from_mapping(data) + + data = config() + data["classes"][11]["id"] = 12 + with pytest.raises(OntologyError, match="ordered IDs"): + ontology_from_mapping(data) + + +def test_v1_is_exactly_zero_through_ten(): + data = config() + data["version"] = "heritage_facades_v1_11classes" + data["classes"] = data["classes"][:11] + data["classes"][8]["name"] = "ornament_intact" + data["classes"][8]["aliases"] = [] + data["groups"]["ORNAMENT"] = ["ornament_intact"] + data["groups"]["HUMAN_ACTIVITY"].remove("advertisements") + v1 = ontology_from_mapping(data) + assert v1.class_names[-1] == "text_or_images" + assert v1.valid_ids == frozenset(range(11)) + + +def test_groups_are_bidirectionally_consistent(): + data = config() + data["classes"][11]["evaluation_groups"] = [] + with pytest.raises(OntologyError, match="top-level group HUMAN_ACTIVITY contains advertisements"): + ontology_from_mapping(data) + + data = config() + data["groups"]["HUMAN_ACTIVITY"].remove("advertisements") + with pytest.raises(OntologyError, match="advertisements in HUMAN_ACTIVITY"): + ontology_from_mapping(data) + + +def test_real_non_json_yaml_and_malformed_yaml(tmp_path): + yaml_path = tmp_path / "plain.yaml" + yaml_path.write_text(""" +version: heritage_facades_v1_11classes +ignore_index: 255 +groups: {} +classes: [] +""", encoding="utf-8") + with pytest.raises(OntologyError, match="non-empty list"): + load_ontology(yaml_path) # parsed as YAML, then semantically rejected + + malformed = tmp_path / "bad.yaml" + malformed.write_text("version: [unterminated", encoding="utf-8") + with pytest.raises(OntologyError, match="malformed YAML"): + load_ontology(malformed) + + +def test_mask_dtype_is_checked_before_values_are_converted(): + ontology = load_ontology() + assert validate_mask_ids(np.array([11, 255], dtype=np.uint8), ontology) == {11, 255} + for array in ( + np.array([11.5, 255.9]), + np.array([11.0, 255.0]), + np.array([True, False]), + np.array(["11", "255"]), + np.array([11], dtype=object), + ): + with pytest.raises(OntologyError, match=r"dtype.*found IDs"): + validate_mask_ids(array, ontology, "typed-mask.npy") + + +def test_unknown_ids_are_explicit(): + with pytest.raises(OntologyError, match=r"mock.png: unknown mask IDs \[17\]"): + validate_mask_ids(np.array([17], dtype=np.int16), load_ontology(), "mock.png") + + +def test_ornament_region_is_canonical_and_legacy_alias_is_explicit(): + ontology = load_ontology() + assert ontology.by_name("ornament_region").id == 8 + with pytest.raises(OntologyError, match="unknown canonical"): + ontology.by_name("ornament_intact") + with pytest.raises(OntologyError, match="explicit resolution"): + ontology.resolve_name("ornament_intact") + assert ontology.resolve_name("ornament_intact", allow_deprecated_alias=True).name == "ornament_region" + with pytest.raises(OntologyError, match="unknown canonical"): + ontology.by_name("does_not_exist") diff --git a/ovs_heritage/tests/test_projection.py b/ovs_heritage/tests/test_projection.py new file mode 100644 index 0000000..6ae1f3b --- /dev/null +++ b/ovs_heritage/tests/test_projection.py @@ -0,0 +1,107 @@ +import pytest +import torch + +from ovs_heritage.ontology import load_ontology +from ovs_heritage.projection import MAIN_SEMANTIC_IDS, MappingEntry, OntologyProjection + + +def test_exact_two_head_mapping_and_round_trip(): + projection = OntologyProjection.from_ontology(load_ontology()) + assert projection.main_channel_count == 11 + assert projection.for_semantic_id(8).output_head == "ornament" + assert projection.for_semantic_id(8).channel_index == 0 + for semantic_id in MAIN_SEMANTIC_IDS: + entry = projection.for_semantic_id(semantic_id) + assert projection.for_channel("main", entry.channel_index).semantic_id == semantic_id + semantic = torch.tensor([list(MAIN_SEMANTIC_IDS) + [255]]) + channels = projection.semantic_main_to_channels(semantic) + assert channels[0, -1].item() == 255 + assert torch.equal(projection.main_channels_to_semantic(channels), semantic) + + +def test_projection_rejects_ornament_and_unknown_in_main(): + projection = OntologyProjection.from_ontology(load_ontology()) + with pytest.raises(ValueError, match="semantic ID 8"): + projection.semantic_main_to_channels(torch.tensor([[[8]]])) + with pytest.raises(ValueError, match="99"): + projection.semantic_main_to_channels(torch.tensor([[[99]]])) + + +def test_duplicate_head_channel_is_ambiguous(): + entries = ( + MappingEntry(0, "background", "main", 0, "multiclass_softmax"), + MappingEntry(1, "crack", "main", 0, "multiclass_softmax"), + ) + with pytest.raises(ValueError, match="duplicate channel"): + OntologyProjection(entries) + + +def test_projection_rejects_ontology_name_drift_and_invalid_predictions(): + from dataclasses import replace + + ontology = load_ontology() + changed_classes = tuple( + replace(item, name="ornament_changed") if item.id == 8 else item + for item in ontology.classes + ) + with pytest.raises(ValueError, match="ornament_region"): + OntologyProjection.from_ontology(replace(ontology, classes=changed_classes)) + projection = OntologyProjection.from_ontology(ontology) + with pytest.raises(ValueError, match="non-finite"): + projection.main_logits_to_semantic(torch.full((1, 11, 1, 1), float("nan"))) + with pytest.raises(ValueError, match="threshold"): + projection.ornament_logits_to_binary(torch.zeros(1, 1, 1, 1), threshold=float("nan")) + + +def test_incomplete_and_noncontiguous_projections_fail_closed(): + canonical = OntologyProjection.from_ontology(load_ontology()).entries + with pytest.raises(ValueError, match="main semantic IDs"): + OntologyProjection(canonical[:-2] + canonical[-1:]) + changed = list(canonical) + changed[2] = MappingEntry(2, "spalling", "main", 9, "multiclass_softmax") + with pytest.raises(ValueError, match="duplicate channel|contiguous"): + OntologyProjection(tuple(changed)) + + +def test_unmapped_values_never_become_ignore(): + projection = OntologyProjection.from_ontology(load_ontology()) + with pytest.raises(ValueError, match="99"): + projection.semantic_main_to_channels(torch.tensor([[[99]]])) + with pytest.raises(ValueError, match="99"): + projection.main_channels_to_semantic(torch.tensor([[[99]]])) + assert projection.semantic_main_to_channels(torch.tensor([[[255]]])).item() == 255 + assert projection.main_channels_to_semantic(torch.tensor([[[255]]])).item() == 255 + + +def test_projection_validates_public_constructor_contract(): + canonical = OntologyProjection.from_ontology(load_ontology()).entries + with pytest.raises(ValueError, match="exact integer 255"): + OntologyProjection(canonical, ignore_index=254) + with pytest.raises(ValueError, match="exact integer 255"): + OntologyProjection(canonical, ignore_index=True) + wrong_name = list(canonical) + wrong_name[0] = MappingEntry(0, "not_background", "main", 0, "multiclass_softmax") + with pytest.raises(ValueError, match="background"): + OntologyProjection(tuple(wrong_name)) + with pytest.raises(ValueError, match="MappingEntry"): + OntologyProjection(canonical[:-1] + ({"semantic_id": 8},)) + false_policy = list(canonical) + false_policy[0] = MappingEntry( + 0, + "background", + "main", + 0, + "multiclass_softmax", + ignore_behavior="ignore values may be rewritten", + ) + with pytest.raises(ValueError, match="canonical ignore policy"): + OntologyProjection(tuple(false_policy)) + + +def test_projection_rejects_multichannel_spatial_targets(): + projection = OntologyProjection.from_ontology(load_ontology()) + target = torch.zeros((1, 2, 3, 4), dtype=torch.long) + with pytest.raises(ValueError, match="exactly one channel"): + projection.semantic_main_to_channels(target) + with pytest.raises(ValueError, match="exactly one channel"): + projection.main_channels_to_semantic(target) diff --git a/ovs_heritage/tests/test_scoring.py b/ovs_heritage/tests/test_scoring.py new file mode 100644 index 0000000..1bc4d13 --- /dev/null +++ b/ovs_heritage/tests/test_scoring.py @@ -0,0 +1,43 @@ +import pytest +import torch + +from ovs_heritage.scoring import RawCosineScorer + + +def test_cpu_raw_scorer_dynamic_channels_and_no_state(): + scorer = RawCosineScorer(scale=2) + features = torch.randn(2, 4, 3, 5) + assert scorer(features, torch.randn(7, 4)).shape == (2, 7, 3, 5) + assert scorer(features, torch.randn(2, 4)).shape == (2, 2, 3, 5) + assert scorer.state_dict() == {} + + +def test_unbatched_per_class_parameters(): + out = RawCosineScorer()( + torch.randn(4, 2, 3), + torch.randn(3, 4), + scale=torch.ones(3), + bias=torch.arange(3.0), + ) + assert out.shape == (3, 2, 3) + + +def test_dimension_and_shape_errors(): + with pytest.raises(ValueError, match="dimension mismatch"): + RawCosineScorer()(torch.randn(1, 3, 2, 2), torch.randn(2, 4)) + with pytest.raises(ValueError, match="prototypes"): + RawCosineScorer()(torch.randn(3, 2, 2), torch.randn(3)) + + +def test_scorer_rejects_non_floating_nonfinite_zero_norm_and_bad_eps(): + with pytest.raises(ValueError, match="eps"): + RawCosineScorer(eps=0) + scorer = RawCosineScorer() + with pytest.raises(ValueError, match="floating-point"): + scorer(torch.ones(1, 3, 2, 2, dtype=torch.int64), torch.ones(2, 3)) + with pytest.raises(ValueError, match="finite"): + scorer(torch.full((1, 3, 2, 2), float("nan")), torch.ones(2, 3)) + with pytest.raises(ValueError, match="non-zero"): + scorer(torch.ones(1, 3, 2, 2), torch.zeros(2, 3)) + with pytest.raises(ValueError, match="scale must be finite"): + scorer(torch.ones(1, 3, 2, 2), torch.ones(2, 3), scale=float("inf")) diff --git a/ovs_heritage/tests/test_vocabulary.py b/ovs_heritage/tests/test_vocabulary.py new file mode 100644 index 0000000..1868ffb --- /dev/null +++ b/ovs_heritage/tests/test_vocabulary.py @@ -0,0 +1,40 @@ +import torch + +from ovs_heritage.ontology import load_ontology +from ovs_heritage.scoring import RawCosineScorer +from ovs_heritage.vocabulary import RuntimeClass, build_prototypes, heritage_runtime_vocabulary + + +def encoder(prompts): + return torch.tensor([[len(prompt), sum(map(ord, prompt)) % 19 + 1, 1.0] for prompt in prompts], dtype=torch.float32) + + +def test_runtime_orders_subset_extended_mixed_and_unseen(): + ontology = load_ontology() + mixed = heritage_runtime_vocabulary(ontology, ["advertisements", "crack"]) + ( + RuntimeClass("unseen", ("an unseen thing",), semantic_id=None), + ) + result = build_prototypes(mixed, encoder, ontology_hash=ontology.hash) + assert result.channel_names == ("advertisements", "crack", "unseen") + assert result.semantic_ids == (11, 1, None) + assert result.prototypes.shape == (3, 3) + assert result.ontology_hash == ontology.hash + assert RawCosineScorer()(torch.randn(1, 3, 2, 2), result.prototypes).shape == (1, 3, 2, 2) + + +def test_prompt_settings_change_specification_hash_without_persistent_state(): + classes = (RuntimeClass("one", ("first",), ("alias",), None),) + plain = build_prototypes(classes, encoder, include_alias_prompts=False) + aliases = build_prototypes(classes, encoder, include_alias_prompts=True) + assert plain.vocabulary_specification_hash != aliases.vocabulary_specification_hash + assert plain.prompt_settings["include_alias_prompts"] is False + assert RawCosineScorer().state_dict() == {} + + +def test_prompt_settings_are_defensively_copied(): + classes = (RuntimeClass("one", ("first",), semantic_id=None),) + result = build_prototypes(classes, encoder) + settings = result.prompt_settings + import pytest + with pytest.raises(TypeError): + settings["method"] = "changed" diff --git a/ovs_heritage/validate_dataset.py b/ovs_heritage/validate_dataset.py new file mode 100644 index 0000000..42427a5 --- /dev/null +++ b/ovs_heritage/validate_dataset.py @@ -0,0 +1,433 @@ +"""Read-only validation for explicit legacy-v1 and two-map-v2 target schemas.""" +from __future__ import annotations + +import argparse +from collections import Counter +import csv +from hashlib import sha256 +import json +from pathlib import Path +from typing import Any + +import numpy as np +from PIL import Image +import yaml +from yaml import YAMLError + +from .metadata import make_metadata +from .ontology import DEFAULT_ONTOLOGY, Ontology, V1_VERSION, V2_VERSION, extract_mask_ids, load_ontology +from .projection import MAIN_SEMANTIC_IDS, OntologyProjection + +COMPONENT_NAME = "ovs_heritage.dataset_validator" +COMPONENT_VERSION = "0.2.0" +VALIDATOR_SCHEMA_VERSION = "heritage-target-validation-v2" +V1_DATASET_SCHEMA = "heritage_single_mask_v1" +V2_DATASET_SCHEMA = "heritage_two_map_v2" +V1_MASK_COLUMNS = ("mask_path", "seg_map_path", "annotation", "mask", "label_path") + + +def _file_hash(path: Path) -> str: + digest = sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _manifest_rows(path: Path) -> list[dict[str, Any]]: + if path.suffix.lower() == ".csv": + with path.open(newline="", encoding="utf-8-sig") as stream: + return list(csv.DictReader(stream)) + data = json.loads(path.read_text(encoding="utf-8")) + if isinstance(data, dict): + data = data.get("samples", data.get("items", data.get("data", []))) + if not isinstance(data, list) or any(not isinstance(row, dict) for row in data): + raise ValueError(f"{path}: manifest must contain a list of sample mappings") + return data + + +def _read_mask(path: Path) -> np.ndarray: + if not path.exists(): + raise FileNotFoundError(f"mask does not exist: {path}") + array = np.load(path, allow_pickle=False) if path.suffix.lower() == ".npy" else np.asarray(Image.open(path)) + if array.ndim != 2: + raise ValueError(f"{path}: mask must be single-channel, got shape {array.shape}") + return array + + +def _resolve_path(value: str, manifest: Path) -> Path: + path = Path(value) + return path if path.is_absolute() else manifest.parent / path + + +def _inventory( + source: Path, + ontology: Ontology, + schema_version: str, + ontology_version: str, +) -> tuple[list[dict[str, Any]], str, bool, dict[int, list[str]]]: + if source.is_dir(): + if ontology.version != V1_VERSION: + raise ValueError("v2 requires an explicit manifest with main_mask_path and ornament_mask_path") + paths = sorted(path for path in source.rglob("*") if path.suffix.lower() in {".png", ".tif", ".tiff", ".npy"}) + rows = [{"mask_path": str(path.resolve())} for path in paths] + inventory = [f"{path.relative_to(source).as_posix()}:{_file_hash(path)}" for path in paths] + fingerprint = sha256("\n".join(inventory).encode()).hexdigest() + return rows, fingerprint, False, {} + rows = _manifest_rows(source) + declaration_errors: dict[int, list[str]] = {} + for index, row in enumerate(rows): + for field, expected in ( + ("schema_version", schema_version), + ("ontology_version", ontology_version), + ): + declared = row.get(field) + if declared not in (None, "", expected): + declaration_errors.setdefault(index, []).append( + f"{source}: row {index + 1} declares conflicting {field}={declared!r}; " + f"expected {expected!r}" + ) + return rows, _file_hash(source), "source_id" in (rows[0] if rows else {}), declaration_errors + + +def _validate_v2_row(row: dict[str, Any], index: int, manifest: Path) -> dict[str, Any]: + label = f"{manifest}: row {index + 1}" + for field in ("main_mask_path", "ornament_mask_path", "facade_id"): + if not isinstance(row.get(field), str) or not row[field].strip(): + raise ValueError(f"{label} requires non-empty {field}") + if field == "facade_id" and row[field] != row[field].strip(): + raise ValueError(f"{label}: facade_id must not contain surrounding whitespace") + source_id = row.get("source_id") + if source_id not in (None, ""): + if not isinstance(source_id, str) or source_id != source_id.strip(): + raise ValueError(f"{label}: source_id must be a string without surrounding whitespace") + main_path = _resolve_path(row["main_mask_path"], manifest) + ornament_path = _resolve_path(row["ornament_mask_path"], manifest) + image_path = None + image_shape = None + if "image_path" in row: + if not isinstance(row.get("image_path"), str) or not row["image_path"].strip(): + raise ValueError(f"{label}: image_path must be a non-empty string when present") + image_path = _resolve_path(row["image_path"], manifest) + physical_paths = { + "main_mask_path": main_path.resolve(), + "ornament_mask_path": ornament_path.resolve(), + **({"image_path": image_path.resolve()} if image_path is not None else {}), + } + if len(set(physical_paths.values())) != len(physical_paths): + roles: dict[str, list[str]] = {} + for role, path in physical_paths.items(): + roles.setdefault(str(path), []).append(role) + reused = {path: names for path, names in roles.items() if len(names) > 1} + raise ValueError(f"{label}: one physical file cannot serve multiple roles: {reused}") + if image_path is not None: + try: + with Image.open(image_path) as image: + image.load() + image_shape = (image.height, image.width) + except Exception as exc: + raise ValueError(f"{image_path}: image is missing or unreadable: {exc}") from exc + main = _read_mask(main_path) + ornament = _read_mask(ornament_path) + if main.shape != ornament.shape: + raise ValueError(f"{label}: main/ornament shape mismatch {main.shape} != {ornament.shape}") + if image_shape is not None and image_shape != main.shape: + raise ValueError(f"{label}: image/mask grid mismatch {image_shape} != {main.shape}") + main_ids = extract_mask_ids(main, str(main_path)) + ornament_ids = extract_mask_ids(ornament, str(ornament_path)) + invalid_main = sorted(main_ids - set(MAIN_SEMANTIC_IDS) - {255}) + if invalid_main: + raise ValueError(f"{main_path}: invalid Y_main semantic IDs {invalid_main}") + invalid_ornament = sorted(ornament_ids - {0, 1, 255}) + if invalid_ornament: + raise ValueError(f"{ornament_path}: invalid Y_ornament values {invalid_ornament}") + return { + "facade_id": row["facade_id"], + "paths": { + "main_mask_path": str(main_path.resolve()), + "ornament_mask_path": str(ornament_path.resolve()), + "image_path": str(image_path.resolve()) if image_path is not None else None, + }, + "main": main, + "ornament": ornament, + "source_id": source_id, + "image_verified": image_path is not None, + "dimensions": {"image": image_shape, "main": main.shape, "ornament": ornament.shape}, + } + + +def _validate_v1_row(row: dict[str, Any], index: int, manifest: Path) -> dict[str, Any]: + label = f"{manifest}: row {index + 1}" + for identifier in ("facade_id", "source_id"): + if identifier in row: + value = row[identifier] + if not isinstance(value, str) or not value or value != value.strip(): + raise ValueError( + f"{label}: {identifier} must be a non-empty string without surrounding whitespace" + ) + key = next( + ( + key + for key in V1_MASK_COLUMNS + if key in row and row[key] is not None and row[key] != "" + ), + None, + ) + if key is None: + raise ValueError(f"{label} has no legacy mask path") + if not isinstance(row[key], str) or not row[key].strip(): + raise ValueError(f"{label}: {key} must be a non-empty string") + path = _resolve_path(row[key], manifest) + mask = _read_mask(path) + ids = extract_mask_ids(mask, str(path)) + invalid = sorted(ids - set(range(11)) - {255}) + if invalid: + raise ValueError(f"{path}: invalid legacy-v1 IDs {invalid}") + return { + "facade_id": row.get("facade_id"), + "paths": {"mask_path": str(path.resolve())}, + "main": mask, + "ornament": None, + "source_id": row.get("source_id"), + "image_verified": False, + "dimensions": {"mask": mask.shape}, + } + + +def validate_splits( + sources: dict[str, str | Path], + ontology: Ontology, + *, + schema_version: str, + ontology_version: str, +) -> dict[str, Any]: + supported_schemas = {V1_DATASET_SCHEMA, V2_DATASET_SCHEMA} + if schema_version not in supported_schemas: + raise ValueError(f"unsupported dataset schema {schema_version!r}; supported: {sorted(supported_schemas)}") + if ontology_version != ontology.version: + raise ValueError( + f"declared ontology_version {ontology_version!r} does not match loaded {ontology.version!r}" + ) + expected_schema = V2_DATASET_SCHEMA if ontology.version == V2_VERSION else V1_DATASET_SCHEMA + if schema_version != expected_schema: + raise ValueError(f"ontology {ontology.version} requires dataset schema {expected_schema}") + projection = OntologyProjection.from_ontology(ontology) if ontology.version == V2_VERSION else None + report: dict[str, Any] = { + "component": {"name": COMPONENT_NAME, "version": COMPONENT_VERSION}, + "validator_schema_version": VALIDATOR_SCHEMA_VERSION, + "dataset_schema_version": schema_version, + "ontology_version": ontology.version, + "ontology_hash": ontology.hash, + "semantic_projection": projection.as_dict() if projection is not None else None, + "ignore_index": ontology.ignore_index, + "sources": {name: str(value) for name, value in sources.items()}, + "source_fingerprints": {}, + "splits": {}, + "facade_overlaps": [], + "duplicated_paths": [], + "warnings": [], + "errors": [], + } + facade_sets: dict[str, set[str]] = {} + source_id_sets: dict[str, set[str]] = {} + path_roles: dict[str, dict[str, set[str]]] = {} + report["source_id_overlaps"] = [] + for split, source_value in sources.items(): + source = Path(source_value) + valid_samples: list[dict[str, Any]] = [] + sample_failures = [] + failed_row_count = 0 + split_errors = [] + inventory_read = False + try: + rows, fingerprint, uses_source_id, declaration_errors = _inventory( + source, ontology, schema_version, ontology_version, + ) + inventory_read = True + report["source_fingerprints"][split] = fingerprint + except Exception as exc: + rows, uses_source_id, declaration_errors = [], False, {} + split_errors.append(str(exc)) + if inventory_read and not rows: + split_errors.append(f"{source}: split is empty") + declared_facades = { + row["facade_id"] for row in rows + if isinstance(row.get("facade_id"), str) + and row["facade_id"] + and row["facade_id"] == row["facade_id"].strip() + } + declared_source_ids = { + row["source_id"] for row in rows + if isinstance(row.get("source_id"), str) + and row["source_id"] + and row["source_id"] == row["source_id"].strip() + } + declared_paths: dict[str, set[str]] = {} + for row in rows: + for field in ("image_path", "main_mask_path", "ornament_mask_path", *V1_MASK_COLUMNS): + value = row.get(field) + if isinstance(value, str) and value.strip(): + physical = str(_resolve_path(value, source).resolve()) + declared_paths.setdefault(physical, set()).add(field) + for index, row in enumerate(rows): + if index in declaration_errors: + sample_failures.extend(declaration_errors[index]) + failed_row_count += 1 + continue + try: + sample = ( + _validate_v2_row(row, index, source) + if ontology.version == V2_VERSION + else _validate_v1_row(row, index, source) + ) + valid_samples.append(sample) + except Exception as exc: + sample_failures.append(str(exc)) + failed_row_count += 1 + main_counts: Counter[int] = Counter() + ornament_counts: Counter[int] = Counter() + for sample in valid_samples: + for value, count in zip(*np.unique(sample["main"], return_counts=True)): + if int(value) in set(MAIN_SEMANTIC_IDS) | set(range(11)) | {255}: + main_counts[int(value)] += int(count) + if sample["ornament"] is not None: + for value, count in zip(*np.unique(sample["ornament"], return_counts=True)): + if int(value) in {0, 1, 255}: + ornament_counts[int(value)] += int(count) + facade_sets[split] = declared_facades + source_id_sets[split] = declared_source_ids + path_roles[split] = declared_paths + source_ids = {sample["source_id"] for sample in valid_samples if sample["source_id"]} + if ontology.version == V2_VERSION and main_counts[11] == 0: + report["warnings"].append(f"{split}: ADVERTISEMENTS (semantic ID 11) is absent") + report["errors"].extend(f"{split}: {failure}" for failure in split_errors + sample_failures) + report["splits"][split] = { + "manifest_row_count": len(rows), + "source_count": len(source_ids) if uses_source_id else len(valid_samples), + "valid_sample_count": len(valid_samples), + "failed_sample_count": failed_row_count, + "split_error_count": len(split_errors), + "main_mask_count": len(valid_samples), + "ornament_mask_count": len(valid_samples) if ontology.version == V2_VERSION else 0, + "verified_image_count": sum(sample["image_verified"] for sample in valid_samples), + "main_pixel_count": {str(key): main_counts[key] for key in sorted(main_counts)}, + "ornament_pixel_count": {str(key): ornament_counts[key] for key in sorted(ornament_counts)}, + "sample_errors": sample_failures, + "split_errors": split_errors, + "dimensions": [sample["dimensions"] for sample in valid_samples], + } + names = list(sources) + for index, left in enumerate(names): + for right in names[index + 1:]: + facade_overlap = sorted(facade_sets.get(left, set()) & facade_sets.get(right, set())) + source_id_overlap = sorted(source_id_sets.get(left, set()) & source_id_sets.get(right, set())) + left_paths = path_roles.get(left, {}) + right_paths = path_roles.get(right, {}) + path_overlap = sorted(set(left_paths) & set(right_paths)) + if facade_overlap: + item = {"splits": [left, right], "facade_ids": facade_overlap} + report["facade_overlaps"].append(item) + report["errors"].append(f"facade_id overlap between {left} and {right}: {facade_overlap}") + if source_id_overlap: + item = {"splits": [left, right], "source_ids": source_id_overlap} + report["source_id_overlaps"].append(item) + report["errors"].append( + f"source_id overlap between {left} and {right}: {source_id_overlap}" + ) + if path_overlap: + item = { + "splits": [left, right], + "paths": [ + { + "path": path, + "roles": { + left: sorted(left_paths[path]), + right: sorted(right_paths[path]), + }, + } + for path in path_overlap + ], + } + report["duplicated_paths"].append(item) + report["errors"].append( + f"physical paths reused between {left} and {right}: {path_overlap}" + ) + metadata = make_metadata( + component_name=COMPONENT_NAME, + component_version=COMPONENT_VERSION, + ontology_version=ontology.version, + ontology_hash=ontology.hash, + mapping=projection.as_dict() if projection is not None else {}, + validator_schema_version=VALIDATOR_SCHEMA_VERSION, + source_fingerprints=report["source_fingerprints"], + ) + report["reproducibility"] = metadata.to_dict() + report["valid"] = not report["errors"] + return report + + +def _dataset_config(path: Path) -> tuple[dict[str, str], str, str]: + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) + except YAMLError as exc: + raise ValueError(f"{path}: malformed YAML dataset config: {exc}") from exc + if not isinstance(data, dict): + raise ValueError(f"{path}: dataset config must be a mapping") + schema_version = data.get("schema_version") + ontology_version = data.get("ontology_version") + if not isinstance(schema_version, str) or not isinstance(ontology_version, str): + raise ValueError(f"{path}: dataset config requires schema_version and ontology_version") + splits = data.get("splits", {}) + sources = { + ("val" if name == "validation" else name): str( + (path.parent / value).resolve() if not Path(value).is_absolute() else Path(value) + ) + for name, value in splits.items() + if name in {"train", "val", "validation", "test"} + } + return sources, schema_version, ontology_version + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--ontology", default=str(DEFAULT_ONTOLOGY)) + parser.add_argument("--dataset-config", type=Path) + parser.add_argument("--schema-version") + parser.add_argument("--ontology-version") + for split in ("train", "val", "test"): + parser.add_argument(f"--{split}") + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--strict", action="store_true") + args = parser.parse_args(argv) + if args.dataset_config: + sources, schema_version, ontology_version = _dataset_config(args.dataset_config) + if args.schema_version and args.schema_version != schema_version: + parser.error("--schema-version conflicts with dataset config") + if args.ontology_version and args.ontology_version != ontology_version: + parser.error("--ontology-version conflicts with dataset config") + else: + sources = {} + schema_version = args.schema_version + ontology_version = args.ontology_version + sources.update({name: getattr(args, name) for name in ("train", "val", "test") if getattr(args, name)}) + if not sources: + parser.error("provide --dataset-config or at least one split source") + if not schema_version or not ontology_version: + parser.error("explicit --schema-version and --ontology-version are required") + try: + report = validate_splits( + sources, + load_ontology(args.ontology), + schema_version=schema_version, + ontology_version=ontology_version, + ) + except Exception as exc: + report = {"valid": False, "errors": [str(exc)], "warnings": [], "sources": sources} + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8") + print(json.dumps({"valid": report["valid"], "errors": len(report["errors"]), "output": str(args.output)})) + return 1 if args.strict and not report["valid"] else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ovs_heritage/vocabulary.py b/ovs_heritage/vocabulary.py new file mode 100644 index 0000000..2474657 --- /dev/null +++ b/ovs_heritage/vocabulary.py @@ -0,0 +1,135 @@ +"""Runtime logical vocabularies and one-prototype-per-class construction.""" +from __future__ import annotations + +from dataclasses import dataclass +from hashlib import sha256 +import json +from types import MappingProxyType +from typing import Callable, Iterable, Mapping + +import torch +import torch.nn.functional as F + +from .ontology import Ontology + + +def _freeze_settings(value): + if isinstance(value, dict): + return MappingProxyType({key: _freeze_settings(item) for key, item in value.items()}) + if isinstance(value, (list, tuple)): + return tuple(_freeze_settings(item) for item in value) + return value + + +@dataclass(frozen=True) +class RuntimeClass: + name: str + prompts: tuple[str, ...] + aliases: tuple[str, ...] = () + semantic_id: int | None = None + + +@dataclass(frozen=True) +class PrototypeSet: + prototypes: torch.Tensor + channel_names: tuple[str, ...] + semantic_ids: tuple[int | None, ...] + vocabulary_specification_hash: str + ontology_hash: str | None + prompt_settings: Mapping[str, object] + + def __post_init__(self) -> None: + object.__setattr__(self, "prompt_settings", _freeze_settings(dict(self.prompt_settings))) + + @property + def vocabulary_hash(self) -> str: + """Backward-compatible name for the specification hash.""" + return self.vocabulary_specification_hash + + +def heritage_runtime_vocabulary( + ontology: Ontology, names: Iterable[str] | None = None, +) -> tuple[RuntimeClass, ...]: + wanted = ontology.class_names if names is None else tuple(names) + if len(wanted) != len(set(wanted)): + raise ValueError("runtime vocabulary contains duplicate classes") + return tuple( + RuntimeClass(item.name, item.prompts, item.aliases, item.id) + for name in wanted + for item in (ontology.by_name(name),) + ) + + +def _validate(classes: tuple[RuntimeClass, ...]) -> None: + names = [item.name for item in classes] + if len(names) != len(set(names)): + raise ValueError("runtime vocabulary contains duplicate class names") + ids = [item.semantic_id for item in classes if item.semantic_id is not None] + if len(ids) != len(set(ids)): + raise ValueError("runtime vocabulary contains duplicate semantic IDs") + aliases = [alias.casefold() for item in classes for alias in item.aliases] + if len(aliases) != len(set(aliases)): + raise ValueError("runtime vocabulary contains conflicting aliases") + for item in classes: + if not item.prompts: + raise ValueError(f"runtime class {item.name!r} has no prompts") + + +def vocabulary_specification_hash( + classes: Iterable[RuntimeClass], *, include_alias_prompts: bool, + prompt_method: str = "normalize_mean_normalize", +) -> str: + payload = { + "classes": [ + { + "name": item.name, + "semantic_id": item.semantic_id, + "prompts": list(item.prompts), + "aliases": list(item.aliases), + } + for item in classes + ], + "prompt_settings": { + "include_alias_prompts": include_alias_prompts, + "method": prompt_method, + }, + } + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return sha256(encoded.encode("utf-8")).hexdigest() + + +def build_prototypes( + classes: Iterable[RuntimeClass], text_encoder: Callable[[list[str]], torch.Tensor], + *, device=None, dtype=None, include_alias_prompts: bool = False, + ontology_hash: str | None = None, eps: float = 1e-12, +) -> PrototypeSet: + classes = tuple(classes) + _validate(classes) + prototypes = [] + for item in classes: + prompts = list(item.prompts) + if include_alias_prompts: + prompts.extend(f"a {alias}" for alias in item.aliases) + encoded = text_encoder(prompts) + if not isinstance(encoded, torch.Tensor) or encoded.ndim != 2: + raise ValueError("text_encoder must return [number_of_prompts, embedding_dim]") + if encoded.shape[0] != len(prompts) or not encoded.is_floating_point(): + raise ValueError("text_encoder must return floating embeddings for every prompt") + if device is not None or dtype is not None: + encoded = encoded.to(device=device, dtype=dtype) + normalized = F.normalize(encoded, dim=-1, eps=eps) + mean = normalized.mean(dim=0) + if not torch.isfinite(mean).all() or mean.norm() <= eps: + raise ValueError(f"prototype for {item.name!r} is zero or non-finite") + prototypes.append(F.normalize(mean, dim=0, eps=eps)) + if not prototypes: + raise ValueError("runtime vocabulary is empty") + settings = {"include_alias_prompts": include_alias_prompts, "method": "normalize_mean_normalize"} + return PrototypeSet( + torch.stack(prototypes), + tuple(item.name for item in classes), + tuple(item.semantic_id for item in classes), + vocabulary_specification_hash(classes, include_alias_prompts=include_alias_prompts), + ontology_hash, + settings, + ) diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..a635c5c --- /dev/null +++ b/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +pythonpath = .