Use generated signatures for missing node matching

This commit is contained in:
2026-07-02 11:38:49 +02:00
parent 8c45016c00
commit 38c142d42e
3 changed files with 160 additions and 5 deletions
+3 -1
View File
@@ -39,7 +39,9 @@ _INDEX_CACHE = None
def _get_ctx(refresh=False): def _get_ctx(refresh=False):
global _CTX_CACHE global _CTX_CACHE
if refresh or _CTX_CACHE is None: 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 return _CTX_CACHE
+133
View File
@@ -115,5 +115,138 @@ class GeneratedSignatureLoaderTests(unittest.TestCase):
self.assertEqual({}, dict(generated["by_out"])) 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__": if __name__ == "__main__":
unittest.main() unittest.main()
+24 -4
View File
@@ -231,7 +231,7 @@ def load_rules(base_dir):
return merged return merged
def build_context(rules): def build_context(rules, generated=None):
""" """
Snapshot the live node registry once (signatures + source of every node). Snapshot the live node registry once (signatures + source of every node).
@@ -258,7 +258,13 @@ def build_context(rules):
for name in classes: for name in classes:
by_out[_first_output_type(sigs[name])].append(name) 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): 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": [..]} ]. `items`: [ {"type": str, "inputs": {name: TYPE}, "outputs": [TYPE], "output_names": [..]} ].
Serialized nodes only carry link slots (not widget values), so 'exact' rarely 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 = {} 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: for it in items:
t = it.get("type") t = it.get("type")
if not t or t in out: if not t or t in out:
continue 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()} inputs = {k: str(v) for k, v in (it.get("inputs") or {}).items()}
sig = { sig = {
"inputs": inputs, "inputs": inputs,