Fix exterior/interior pairing check: variant-signature approach, angle 21

Old check stripped paint/trim by prefix but used wrong tr- prefix
(transmission, not trim), causing false mismatches on known-good packs.
New approach uses an allowlist of shared WERS prefixes to build variant
signatures — paint, trim, and transmission codes fall out naturally.
Restricts comparison to angle-21 base records only per Ben's spec.
Updates ford_bnp.json profile config accordingly and adds 8 unit tests.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Vadym Samoilenko 2026-04-29 12:47:01 +01:00
parent 87605828aa
commit eb3da2c980
3 changed files with 423 additions and 223 deletions

View file

@ -1,221 +1,247 @@
import os
import json
from typing import Dict, Any, FrozenSet, Set, List
def run_check(config: Dict[str, Any]) -> Dict[str, Any]:
"""
Validate that every exterior base asset has a matching interior base asset with the same
feature set (excluding paint and trim codes), and vice versa.
For each exterior item (viewtype="exterior", base imagetype/layer), the set of features
minus paint codes must match at least one interior item's feature set minus trim codes.
Both directions are checked: exterior items without an interior partner, and interior items
without an exterior partner.
For MEC packs, items with experienceCondition="2d-background" are validated separately.
:param config: Configuration dictionary containing:
- working_dir: Directory where linkingrecord.json is located
- linkingrecord_filename: The linking record filename (default: 'linkingrecord.json')
- paint_code_prefix: Prefix for paint codes to exclude from exterior features (default: 'pn')
- trim_code_prefix: Prefix for trim codes to exclude from interior features (default: 'tr-')
:return: Dictionary with validation results:
- status: "passed", "failed", or "error"
- details: Breakdown of mismatches by direction and MEC condition
"""
working_dir = config.get("working_dir", "working")
linkingrecord_filename = config.get("linkingrecord_filename", "linkingrecord.json")
paint_prefix = config.get("paint_code_prefix", "pn").lower()
trim_prefix = config.get("trim_code_prefix", "tr-").lower()
linkingrecord_path = os.path.join(working_dir, linkingrecord_filename)
if not os.path.exists(linkingrecord_path):
return {
"status": "error",
"error_message": f"Linking record file '{linkingrecord_filename}' not found in {working_dir}."
}
try:
with open(linkingrecord_path, 'r', encoding='utf-8') as f:
linkingrecord = json.load(f)
except json.JSONDecodeError as e:
return {
"status": "error",
"error_message": f"Malformed JSON in '{linkingrecord_filename}': {str(e)}"
}
except Exception as e:
return {
"status": "error",
"error_message": f"Error reading '{linkingrecord_filename}': {str(e)}"
}
if not isinstance(linkingrecord, dict):
return {
"status": "error",
"error_message": "Invalid linkingrecord.json structure: root is not an object."
}
if "items" not in linkingrecord or not isinstance(linkingrecord["items"], list):
return {
"status": "error",
"error_message": "Invalid linkingrecord.json structure: 'items' missing or not a list."
}
def _is_base_item(item: dict) -> bool:
"""A base item has no imagetype (None) or imagetype='base', and layer=0 or no layer."""
conditions = item.get("conditions", {})
if not isinstance(conditions, dict):
return False
imagetype = conditions.get("imagetype")
layer = conditions.get("layer", 0)
return imagetype in (None, "base", "") and (layer == 0 or layer is None)
def _strip_paint_codes(features: List[str]) -> FrozenSet[str]:
"""Remove paint codes (pn prefix) from exterior feature list."""
return frozenset(f for f in features if isinstance(f, str) and not f.lower().startswith(paint_prefix))
def _strip_trim_codes(features: List[str]) -> FrozenSet[str]:
"""Remove trim codes (configurable prefix) from interior feature list."""
return frozenset(f for f in features if isinstance(f, str) and not f.lower().startswith(trim_prefix))
def _collect_feature_sets(items_list, viewtype_val: str, strip_fn, mec_only: bool = False) -> Set[FrozenSet[str]]:
"""Collect normalised feature sets for all base records of the given viewtype."""
result = set()
for item in items_list:
if not isinstance(item, dict):
continue
conditions = item.get("conditions", {})
if not isinstance(conditions, dict):
continue
if conditions.get("viewtype") != viewtype_val:
continue
if not _is_base_item(item):
continue
exp_condition = conditions.get("experienceCondition", "")
is_mec = exp_condition == "2d-background"
if mec_only and not is_mec:
continue
if not mec_only and is_mec:
continue
for record in item.get("records", []):
if not isinstance(record, dict):
continue
features = record.get("features", [])
if not isinstance(features, list):
continue
normalised = strip_fn(features)
if normalised:
result.add(normalised)
return result
def _run_pairing_check(items_list, mec_only: bool = False) -> dict:
"""Compare exterior and interior feature sets and return mismatch details."""
ext_sets = _collect_feature_sets(items_list, "exterior", _strip_paint_codes, mec_only)
int_sets = _collect_feature_sets(items_list, "interior", _strip_trim_codes, mec_only)
if not ext_sets and not int_sets:
return {"applicable": False}
ext_only = ext_sets - int_sets
int_only = int_sets - ext_sets
return {
"applicable": True,
"exterior_count": len(ext_sets),
"interior_count": len(int_sets),
"exterior_without_interior": len(ext_only),
"interior_without_exterior": len(int_only),
"ext_only_feature_sets": [sorted(list(s)) for s in sorted(ext_only, key=lambda x: sorted(x))],
"int_only_feature_sets": [sorted(list(s)) for s in sorted(int_only, key=lambda x: sorted(x))],
}
items = linkingrecord["items"]
# Standard pairing check (excluding 2d-background MEC items)
standard = _run_pairing_check(items, mec_only=False)
# MEC 2d-background pairing check
mec = _run_pairing_check(items, mec_only=True)
# Determine if any items had the MEC condition
has_mec_items = mec.get("applicable", False)
# Evaluate overall pass/fail
standard_pass = not standard.get("applicable") or (
standard.get("exterior_without_interior", 0) == 0 and
standard.get("interior_without_exterior", 0) == 0
)
mec_pass = not mec.get("applicable") or (
mec.get("exterior_without_interior", 0) == 0 and
mec.get("interior_without_exterior", 0) == 0
)
if not standard.get("applicable") and not has_mec_items:
return {
"status": "passed",
"details": {
"message": "No exterior or interior base items found. Check not applicable to this pack.",
"applicable": False
}
}
overall_pass = standard_pass and mec_pass
status = "passed" if overall_pass else "failed"
details: Dict[str, Any] = {}
if standard.get("applicable"):
details["standard"] = {
"exterior_base_sets": standard["exterior_count"],
"interior_base_sets": standard["interior_count"],
"exterior_without_interior_partner": standard["exterior_without_interior"],
"interior_without_exterior_partner": standard["interior_without_exterior"],
}
if standard["exterior_without_interior"] > 0:
details["standard"]["exterior_only_feature_sets"] = standard["ext_only_feature_sets"]
if standard["interior_without_exterior"] > 0:
details["standard"]["interior_only_feature_sets"] = standard["int_only_feature_sets"]
if has_mec_items:
details["mec_2d_background"] = {
"exterior_base_sets": mec["exterior_count"],
"interior_base_sets": mec["interior_count"],
"exterior_without_interior_partner": mec["exterior_without_interior"],
"interior_without_exterior_partner": mec["interior_without_exterior"],
}
if mec["exterior_without_interior"] > 0:
details["mec_2d_background"]["exterior_only_feature_sets"] = mec["ext_only_feature_sets"]
if mec["interior_without_exterior"] > 0:
details["mec_2d_background"]["interior_only_feature_sets"] = mec["int_only_feature_sets"]
if overall_pass:
msg_parts = []
if standard.get("applicable"):
msg_parts.append(f"all {standard['exterior_count']} exterior/interior base pairs matched")
if has_mec_items:
msg_parts.append(f"all {mec['exterior_count']} MEC 2d-background pairs matched")
details["message"] = "Exterior/interior pairing passed: " + "; ".join(msg_parts) + "."
else:
msg_parts = []
if standard.get("applicable") and not standard_pass:
ext_miss = standard["exterior_without_interior"]
int_miss = standard["interior_without_exterior"]
if ext_miss:
msg_parts.append(f"{ext_miss} exterior base set(s) with no interior partner")
if int_miss:
msg_parts.append(f"{int_miss} interior base set(s) with no exterior partner")
if has_mec_items and not mec_pass:
ext_miss = mec["exterior_without_interior"]
int_miss = mec["interior_without_exterior"]
if ext_miss:
msg_parts.append(f"{ext_miss} MEC exterior base set(s) with no interior partner")
if int_miss:
msg_parts.append(f"{int_miss} MEC interior base set(s) with no exterior partner")
details["message"] = "Exterior/interior pairing failed: " + "; ".join(msg_parts) + "."
details["paint_code_prefix"] = paint_prefix
details["trim_code_prefix"] = trim_prefix
return {"status": status, "details": details}
import os
import json
from typing import Dict, Any, FrozenSet, Set, List
DEFAULT_SHARED_PREFIXES = [
"abm", # ABM (CV/Ranger model series)
"acm", # ACM (PV model series)
"vs-", # Visual Selection (PV)
"se#", # Series (CV/Ranger)
"bs-", # Body style
"dr-", # Drivetrain (PV: dr-, dr--)
"dga", # Drivetrain group (CV)
"en-", # Engine
"ca#", # Camera / cabin
]
DEFAULT_ANGLE = 21
def run_check(config: Dict[str, Any]) -> Dict[str, Any]:
"""
Validate that every exterior base record (at the configured angle) has a matching
interior base record with the same variant signature, and vice versa.
A "variant signature" is the subset of a record's WERS codes whose prefix is in the
configured allowlist (`shared_code_prefixes`). Codes not in the allowlist (paint,
transmission, trim, etc.) are excluded from the signature so they don't cause false
mismatches between exterior and interior records of the same variant.
Catches the case where interior data was deleted in variant manager prior to asset
creation: exterior records exist with a given variant signature but no interior
record covers it.
For MEC packs, items with experienceCondition="2d-background" are validated separately.
:param config: Configuration dictionary containing:
- working_dir: Directory where linkingrecord.json is located
- linkingrecord_filename: The linking record filename (default: 'linkingrecord.json')
- shared_code_prefixes: List of WERS code prefixes that should match between
exterior and interior records (default: see DEFAULT_SHARED_PREFIXES)
- angle: Image angle to compare (default: 21 the showroom front angle)
:return: Dictionary with validation results:
- status: "passed", "failed", or "error"
- details: Breakdown of mismatches by direction and MEC condition
"""
working_dir = config.get("working_dir", "working")
linkingrecord_filename = config.get("linkingrecord_filename", "linkingrecord.json")
shared_prefixes_raw = config.get("shared_code_prefixes", DEFAULT_SHARED_PREFIXES)
shared_prefixes = [p.lower() for p in shared_prefixes_raw if isinstance(p, str)]
target_angle = config.get("angle", DEFAULT_ANGLE)
linkingrecord_path = os.path.join(working_dir, linkingrecord_filename)
if not os.path.exists(linkingrecord_path):
return {
"status": "error",
"error_message": f"Linking record file '{linkingrecord_filename}' not found in {working_dir}."
}
try:
with open(linkingrecord_path, 'r', encoding='utf-8') as f:
linkingrecord = json.load(f)
except json.JSONDecodeError as e:
return {
"status": "error",
"error_message": f"Malformed JSON in '{linkingrecord_filename}': {str(e)}"
}
except Exception as e:
return {
"status": "error",
"error_message": f"Error reading '{linkingrecord_filename}': {str(e)}"
}
if not isinstance(linkingrecord, dict):
return {
"status": "error",
"error_message": "Invalid linkingrecord.json structure: root is not an object."
}
if "items" not in linkingrecord or not isinstance(linkingrecord["items"], list):
return {
"status": "error",
"error_message": "Invalid linkingrecord.json structure: 'items' missing or not a list."
}
def _is_base_item(item: dict) -> bool:
"""A base item has imagetype None/'' /'base' and layer 0 or absent."""
conditions = item.get("conditions", {})
if not isinstance(conditions, dict):
return False
imagetype = conditions.get("imagetype")
layer = conditions.get("layer", 0)
return imagetype in (None, "base", "") and (layer == 0 or layer is None)
def _build_signature(features: List[Any]) -> FrozenSet[str]:
"""
Extract the subset of features whose lowercase prefix matches any allowlisted
shared prefix. Codes outside the allowlist (paint, transmission, trim) are
excluded so they don't poison the comparison.
"""
return frozenset(
f for f in features
if isinstance(f, str)
and any(f.lower().startswith(p) for p in shared_prefixes)
)
def _collect_signatures(items_list, viewtype_val: str, mec_only: bool = False) -> Set[FrozenSet[str]]:
"""Collect variant signatures for base records of the given viewtype at the target angle."""
result = set()
for item in items_list:
if not isinstance(item, dict):
continue
conditions = item.get("conditions", {})
if not isinstance(conditions, dict):
continue
if conditions.get("viewtype") != viewtype_val:
continue
if not _is_base_item(item):
continue
exp_condition = conditions.get("experienceCondition", "")
is_mec = exp_condition == "2d-background"
if mec_only and not is_mec:
continue
if not mec_only and is_mec:
continue
for record in item.get("records", []):
if not isinstance(record, dict):
continue
if record.get("angle") != target_angle:
continue
features = record.get("features", [])
if not isinstance(features, list):
continue
signature = _build_signature(features)
if signature:
result.add(signature)
return result
def _run_pairing_check(items_list, mec_only: bool = False) -> dict:
"""Compare exterior and interior variant signatures."""
ext_sets = _collect_signatures(items_list, "exterior", mec_only)
int_sets = _collect_signatures(items_list, "interior", mec_only)
if not ext_sets and not int_sets:
return {"applicable": False}
ext_only = ext_sets - int_sets
int_only = int_sets - ext_sets
return {
"applicable": True,
"exterior_count": len(ext_sets),
"interior_count": len(int_sets),
"exterior_without_interior": len(ext_only),
"interior_without_exterior": len(int_only),
"ext_only_signatures": [sorted(list(s)) for s in sorted(ext_only, key=lambda x: sorted(x))],
"int_only_signatures": [sorted(list(s)) for s in sorted(int_only, key=lambda x: sorted(x))],
}
items = linkingrecord["items"]
standard = _run_pairing_check(items, mec_only=False)
mec = _run_pairing_check(items, mec_only=True)
has_mec_items = mec.get("applicable", False)
standard_pass = not standard.get("applicable") or (
standard.get("exterior_without_interior", 0) == 0 and
standard.get("interior_without_exterior", 0) == 0
)
mec_pass = not mec.get("applicable") or (
mec.get("exterior_without_interior", 0) == 0 and
mec.get("interior_without_exterior", 0) == 0
)
if not standard.get("applicable") and not has_mec_items:
return {
"status": "passed",
"details": {
"message": (
f"No exterior or interior base records at angle {target_angle} found. "
"Check not applicable to this pack."
),
"applicable": False,
"angle": target_angle,
"shared_code_prefixes": shared_prefixes,
}
}
overall_pass = standard_pass and mec_pass
status = "passed" if overall_pass else "failed"
details: Dict[str, Any] = {
"angle": target_angle,
"shared_code_prefixes": shared_prefixes,
}
if standard.get("applicable"):
details["standard"] = {
"exterior_variants": standard["exterior_count"],
"interior_variants": standard["interior_count"],
"exterior_without_interior_partner": standard["exterior_without_interior"],
"interior_without_exterior_partner": standard["interior_without_exterior"],
}
if standard["exterior_without_interior"] > 0:
details["standard"]["exterior_only_signatures"] = standard["ext_only_signatures"]
if standard["interior_without_exterior"] > 0:
details["standard"]["interior_only_signatures"] = standard["int_only_signatures"]
if has_mec_items:
details["mec_2d_background"] = {
"exterior_variants": mec["exterior_count"],
"interior_variants": mec["interior_count"],
"exterior_without_interior_partner": mec["exterior_without_interior"],
"interior_without_exterior_partner": mec["interior_without_exterior"],
}
if mec["exterior_without_interior"] > 0:
details["mec_2d_background"]["exterior_only_signatures"] = mec["ext_only_signatures"]
if mec["interior_without_exterior"] > 0:
details["mec_2d_background"]["interior_only_signatures"] = mec["int_only_signatures"]
if overall_pass:
msg_parts = []
if standard.get("applicable"):
msg_parts.append(f"all {standard['exterior_count']} exterior/interior variant pairs matched")
if has_mec_items:
msg_parts.append(f"all {mec['exterior_count']} MEC 2d-background pairs matched")
details["message"] = "Exterior/interior pairing passed: " + "; ".join(msg_parts) + "."
else:
msg_parts = []
if standard.get("applicable") and not standard_pass:
ext_miss = standard["exterior_without_interior"]
int_miss = standard["interior_without_exterior"]
if ext_miss:
msg_parts.append(f"{ext_miss} exterior variant(s) with no interior partner")
if int_miss:
msg_parts.append(f"{int_miss} interior variant(s) with no exterior partner")
if has_mec_items and not mec_pass:
ext_miss = mec["exterior_without_interior"]
int_miss = mec["interior_without_exterior"]
if ext_miss:
msg_parts.append(f"{ext_miss} MEC exterior variant(s) with no interior partner")
if int_miss:
msg_parts.append(f"{int_miss} MEC interior variant(s) with no exterior partner")
details["message"] = "Exterior/interior pairing failed: " + "; ".join(msg_parts) + "."
return {"status": status, "details": details}

View file

@ -124,8 +124,8 @@
"config": {
"working_dir": "__WORKING_DIR__",
"linkingrecord_filename": "linkingrecord.json",
"paint_code_prefix": "pn",
"trim_code_prefix": "tr-"
"shared_code_prefixes": ["abm","acm","vs-","se#","bs-","dr-","dga","en-","ca#"],
"angle": 21
}
}
]

View file

@ -0,0 +1,174 @@
import os
import sys
import json
import tempfile
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from checks.exterior_interior_pairing_check import run_check
def _write_tmp(data: dict) -> str:
d = tempfile.mkdtemp()
with open(os.path.join(d, "linkingrecord.json"), "w") as f:
json.dump(data, f)
return d
def _config(working_dir: str) -> dict:
return {"working_dir": working_dir, "linkingrecord_filename": "linkingrecord.json"}
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _item(viewtype: str, features: list, angle: int = 21,
imagetype=None, layer=0, exp_condition=None) -> dict:
conditions = {"viewtype": viewtype}
if imagetype is not None:
conditions["imagetype"] = imagetype
if layer:
conditions["layer"] = layer
if exp_condition is not None:
conditions["experienceCondition"] = exp_condition
return {
"conditions": conditions,
"records": [{"angle": angle, "features": features}],
}
# ---------------------------------------------------------------------------
# Core passing/failing cases
# ---------------------------------------------------------------------------
def test_matched_ext_int_pairs_pass():
"""Transmission on ext side and trim on int side must NOT cause a failure."""
data = {"items": [
_item("exterior", ["acmra", "vs-kz", "bs-bh", "tr-eu", "pna7"]),
_item("interior", ["acmra", "vs-kz", "bs-bh", "12yzh"]),
]}
result = run_check(_config(_write_tmp(data)))
print(f"\ntest_matched_ext_int_pairs_pass: {result['status']}")
assert result["status"] == "passed", result
details = result["details"]
assert details["standard"]["exterior_variants"] == 1
assert details["standard"]["interior_variants"] == 1
assert details["standard"]["exterior_without_interior_partner"] == 0
assert details["standard"]["interior_without_exterior_partner"] == 0
def test_ext_without_interior_fails():
"""An exterior signature at angle 21 with no matching interior should fail."""
data = {"items": [
_item("exterior", ["acmra", "vs-kz", "bs-bh"]),
_item("exterior", ["acmrj", "vs-le", "bs-bh"]),
_item("interior", ["acmra", "vs-kz", "bs-bh"]), # acmrj/vs-le missing on interior
]}
result = run_check(_config(_write_tmp(data)))
print(f"\ntest_ext_without_interior_fails: {result['status']}")
assert result["status"] == "failed", result
details = result["details"]["standard"]
assert details["exterior_without_interior_partner"] == 1
assert details["interior_without_exterior_partner"] == 0
sigs = details.get("exterior_only_signatures", [])
assert any("acmrj" in str(s) and "vs-le" in str(s) for s in sigs), sigs
def test_int_without_exterior_fails():
"""An interior signature at angle 21 with no matching exterior should fail."""
data = {"items": [
_item("exterior", ["acmra", "vs-kz", "bs-bh"]),
_item("interior", ["acmra", "vs-kz", "bs-bh"]),
_item("interior", ["acmrj", "vs-le", "bs-bh"]), # acmrj/vs-le missing on exterior
]}
result = run_check(_config(_write_tmp(data)))
print(f"\ntest_int_without_exterior_fails: {result['status']}")
assert result["status"] == "failed", result
details = result["details"]["standard"]
assert details["interior_without_exterior_partner"] == 1
assert details["exterior_without_interior_partner"] == 0
sigs = details.get("interior_only_signatures", [])
assert any("acmrj" in str(s) and "vs-le" in str(s) for s in sigs), sigs
def test_non_angle21_records_ignored():
"""Records at angles other than 21 must not influence the result."""
data = {"items": [
_item("exterior", ["acmra", "vs-kz"], angle=21),
_item("interior", ["acmra", "vs-kz"], angle=21),
# These mismatches at angle 23 must be ignored
_item("exterior", ["acmrj", "vs-le"], angle=23),
]}
result = run_check(_config(_write_tmp(data)))
print(f"\ntest_non_angle21_records_ignored: {result['status']}")
assert result["status"] == "passed", result
def test_paint_only_difference_passes():
"""Records that differ only in paint code (pn*) should still pair correctly."""
data = {"items": [
_item("exterior", ["acmra", "vs-kz", "pna7"]),
_item("interior", ["acmra", "vs-kz", "pnb1"]), # different paint, same shared codes
]}
result = run_check(_config(_write_tmp(data)))
print(f"\ntest_paint_only_difference_passes: {result['status']}")
assert result["status"] == "passed", result
def test_mec_2d_background_validated_separately():
"""MEC items (experienceCondition='2d-background') are checked in their own bucket."""
data = {"items": [
# Standard items — fully paired
_item("exterior", ["acmra", "vs-kz"]),
_item("interior", ["acmra", "vs-kz"]),
# MEC exterior with no matching MEC interior
_item("exterior", ["acmra", "vs-kz"], exp_condition="2d-background"),
]}
result = run_check(_config(_write_tmp(data)))
print(f"\ntest_mec_2d_background_validated_separately: {result['status']}")
assert result["status"] == "failed", result
assert "mec_2d_background" in result["details"], result["details"]
mec = result["details"]["mec_2d_background"]
assert mec["exterior_without_interior_partner"] == 1
# Standard bucket must still pass
standard = result["details"]["standard"]
assert standard["exterior_without_interior_partner"] == 0
assert standard["interior_without_exterior_partner"] == 0
def test_no_records_at_angle_passes_silently():
"""No exterior or interior base records at angle 21 → not applicable, passes."""
data = {"items": [
_item("exterior", ["acmra", "vs-kz"], angle=23),
_item("interior", ["acmra", "vs-kz"], angle=23),
]}
result = run_check(_config(_write_tmp(data)))
print(f"\ntest_no_records_at_angle_passes_silently: {result['status']}")
assert result["status"] == "passed", result
assert result["details"].get("applicable") is False, result["details"]
def test_layer_nonzero_excluded():
"""Items with a non-zero layer should not be treated as base records."""
data = {"items": [
_item("exterior", ["acmra", "vs-kz"]),
_item("interior", ["acmra", "vs-kz"]),
# layer=1 — this is NOT a base record; should not affect the check
_item("exterior", ["acmrj", "vs-le"], layer=1),
]}
result = run_check(_config(_write_tmp(data)))
print(f"\ntest_layer_nonzero_excluded: {result['status']}")
assert result["status"] == "passed", result
if __name__ == "__main__":
test_matched_ext_int_pairs_pass()
test_ext_without_interior_fails()
test_int_without_exterior_fails()
test_non_angle21_records_ignored()
test_paint_only_difference_passes()
test_mec_2d_background_validated_separately()
test_no_records_at_angle_passes_silently()
test_layer_nonzero_excluded()
print("\nAll tests passed.")