From 38c142d42e94bb558c8654747461e0a882864494 Mon Sep 17 00:00:00 2001 From: Ethanfel Date: Thu, 2 Jul 2026 11:38:49 +0200 Subject: [PATCH] Use generated signatures for missing node matching --- __init__.py | 4 +- tests/test_utfcn_core_generated.py | 133 +++++++++++++++++++++++++++++ utfcn_core.py | 28 +++++- 3 files changed, 160 insertions(+), 5 deletions(-) diff --git a/__init__.py b/__init__.py index 501f651..e254366 100644 --- a/__init__.py +++ b/__init__.py @@ -39,7 +39,9 @@ _INDEX_CACHE = None def _get_ctx(refresh=False): global _CTX_CACHE if refresh or _CTX_CACHE is None: - _CTX_CACHE = utfcn_core.build_context(utfcn_core.load_rules(_DIR)) + rules = utfcn_core.load_rules(_DIR) + generated = utfcn_core.load_generated_signatures(_DIR) + _CTX_CACHE = utfcn_core.build_context(rules, generated) return _CTX_CACHE diff --git a/tests/test_utfcn_core_generated.py b/tests/test_utfcn_core_generated.py index fb1788c..d4e16bd 100644 --- a/tests/test_utfcn_core_generated.py +++ b/tests/test_utfcn_core_generated.py @@ -115,5 +115,138 @@ class GeneratedSignatureLoaderTests(unittest.TestCase): self.assertEqual({}, dict(generated["by_out"])) +class GeneratedSignatureMatchingTests(unittest.TestCase): + def _ctx(self, rules=None, generated=None): + live_sigs = { + "CoreImageSize": { + "inputs": {"image": "IMAGE"}, + "required": {"image"}, + "outputs": ["INT", "INT"], + "output_names": ["width", "height"], + }, + "CoreMaskInvert": { + "inputs": {"mask": "MASK"}, + "required": {"mask"}, + "outputs": ["MASK"], + "output_names": ["mask"], + }, + "CuratedTarget": { + "inputs": {"image": "IMAGE"}, + "required": {"image"}, + "outputs": ["INT", "INT"], + "output_names": ["width", "height"], + }, + } + sources = { + "CoreImageSize": {"source": "core", "pack": "nodes", "display": "Core Image Size"}, + "CoreMaskInvert": {"source": "core", "pack": "nodes", "display": "Core Mask Invert"}, + "CuratedTarget": {"source": "core", "pack": "nodes", "display": "Curated Target"}, + } + by_out = utfcn_core.defaultdict(list) + for name, sig in live_sigs.items(): + by_out[sig["outputs"][0]].append(name) + return { + "sources": sources, + "sigs": live_sigs, + "by_out": by_out, + "rules": rules or {}, + "generated": generated or utfcn_core._empty_generated_signatures(), + } + + def test_generated_exact_signature_matches_missing_node_as_verified(self): + generated = utfcn_core._empty_generated_signatures() + generated["sigs"]["SampleImageSize"] = { + "inputs": {"image": "IMAGE"}, + "required": {"image"}, + "outputs": ["INT", "INT"], + "output_names": ["width", "height"], + } + generated["meta"]["SampleImageSize"] = { + "source": "generated", + "pack": "sample-pack", + "display": "Sample Image Size", + "repository": "https://github.com/example/sample-pack", + "confidence": "static_exact", + } + generated["by_out"]["INT"].append("SampleImageSize") + + result = utfcn_core.match(self._ctx(generated=generated), [{"type": "SampleImageSize"}]) + + self.assertEqual("CoreImageSize", result["SampleImageSize"][0]["to"]) + self.assertEqual("exact", result["SampleImageSize"][0]["tier"]) + self.assertTrue(result["SampleImageSize"][0]["verified"]) + + def test_curated_rule_stays_first_before_generated_exact_match(self): + generated = utfcn_core._empty_generated_signatures() + generated["sigs"]["SampleImageSize"] = { + "inputs": {"image": "IMAGE"}, + "required": {"image"}, + "outputs": ["INT", "INT"], + "output_names": ["width", "height"], + } + generated["meta"]["SampleImageSize"] = { + "source": "generated", + "pack": "sample-pack", + "display": "Sample Image Size", + "repository": "https://github.com/example/sample-pack", + "confidence": "static_exact", + } + generated["by_out"]["INT"].append("SampleImageSize") + rules = { + "SampleImageSize": [ + { + "to": "CuratedTarget", + "note": "Curated replacement wins over generated exact signature.", + } + ] + } + + result = utfcn_core.match(self._ctx(rules=rules, generated=generated), [{"type": "SampleImageSize"}]) + + self.assertEqual("CuratedTarget", result["SampleImageSize"][0]["to"]) + self.assertEqual("curated", result["SampleImageSize"][0]["tier"]) + self.assertTrue(result["SampleImageSize"][0]["verified"]) + + def test_generated_partial_signature_matches_but_is_not_verified(self): + generated = utfcn_core._empty_generated_signatures() + generated["sigs"]["SampleMaskInvert"] = { + "inputs": {"masks": "MASK"}, + "required": {"masks"}, + "outputs": ["MASK"], + "output_names": ["mask"], + } + generated["meta"]["SampleMaskInvert"] = { + "source": "generated", + "pack": "sample-pack", + "display": "Sample Mask Invert", + "repository": "https://github.com/example/sample-pack", + "confidence": "static_exact", + } + generated["by_out"]["MASK"].append("SampleMaskInvert") + + result = utfcn_core.match(self._ctx(generated=generated), [{"type": "SampleMaskInvert"}]) + + self.assertEqual("CoreMaskInvert", result["SampleMaskInvert"][0]["to"]) + self.assertEqual("partial", result["SampleMaskInvert"][0]["tier"]) + self.assertFalse(result["SampleMaskInvert"][0]["verified"]) + + def test_serialized_signature_fallback_still_handles_unknown_generated_node(self): + result = utfcn_core.match( + self._ctx(), + [ + { + "type": "SerializedMaskInvert", + "inputs": {"masks": "MASK"}, + "outputs": ["MASK"], + "output_names": ["mask"], + } + ], + ) + + self.assertEqual("CoreMaskInvert", result["SerializedMaskInvert"][0]["to"]) + self.assertEqual("partial", result["SerializedMaskInvert"][0]["tier"]) + self.assertFalse(result["SerializedMaskInvert"][0]["verified"]) + + if __name__ == "__main__": unittest.main() diff --git a/utfcn_core.py b/utfcn_core.py index 6e551a9..f4fff57 100644 --- a/utfcn_core.py +++ b/utfcn_core.py @@ -231,7 +231,7 @@ def load_rules(base_dir): return merged -def build_context(rules): +def build_context(rules, generated=None): """ Snapshot the live node registry once (signatures + source of every node). @@ -258,7 +258,13 @@ def build_context(rules): for name in classes: by_out[_first_output_type(sigs[name])].append(name) - return {"sources": sources, "sigs": sigs, "by_out": by_out, "rules": rules} + return { + "sources": sources, + "sigs": sigs, + "by_out": by_out, + "rules": rules, + "generated": generated or _empty_generated_signatures(), + } def _candidates_for(src_name, src_sig, src_pack, ctx): @@ -363,15 +369,29 @@ def match(ctx, items): `items`: [ {"type": str, "inputs": {name: TYPE}, "outputs": [TYPE], "output_names": [..]} ]. Serialized nodes only carry link slots (not widget values), so 'exact' rarely - fires; curated rules (by type name) and 'partial' link-type matches do. + fires; curated rules (by type name), bundled generated signatures, and + partial link-type matches do. - Returns { type: [candidate, ...] }. + Returns a mapping from source node type to candidate list. """ out = {} + generated = ctx.get("generated") or _empty_generated_signatures() + generated_sigs = generated.get("sigs") or {} + generated_meta = generated.get("meta") or {} + for it in items: t = it.get("type") if not t or t in out: continue + + gen_sig = generated_sigs.get(t) + if gen_sig is not None: + gen_pack = (generated_meta.get(t) or {}).get("pack") + found = _candidates_for(t, gen_sig, gen_pack, ctx) + if found: + out[t] = found + continue + inputs = {k: str(v) for k, v in (it.get("inputs") or {}).items()} sig = { "inputs": inputs,