rework ergonomics for standalone use
This commit is contained in:
@@ -98,6 +98,38 @@ def nodes() -> list[str]:
|
||||
return sorted(NODE_CLASS_MAPPINGS.keys())
|
||||
|
||||
|
||||
def describe(class_name: str) -> dict:
|
||||
"""Return a dict describing a node: inputs, outputs, description, keywords, category.
|
||||
|
||||
Thin wrapper around :func:`backend.node_registry.get_node_info`.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
class_name : str
|
||||
The node class name, e.g. ``"GaussianFilter"``.
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict
|
||||
Keys include ``name``, ``display_name``, ``category``, ``input``,
|
||||
``input_order``, ``output``, ``output_name``, ``description``,
|
||||
``keywords``, and more.
|
||||
|
||||
Raises
|
||||
------
|
||||
KeyError
|
||||
If no node with that name is registered.
|
||||
"""
|
||||
from backend.node_registry import NODE_CLASS_MAPPINGS, get_node_info
|
||||
_ensure_registry()
|
||||
if class_name not in NODE_CLASS_MAPPINGS:
|
||||
raise KeyError(
|
||||
f"Unknown node {class_name!r}. "
|
||||
f"Use tono.nodes() to list available nodes."
|
||||
)
|
||||
return get_node_info(class_name)
|
||||
|
||||
|
||||
def get_node(class_name: str) -> Any:
|
||||
"""Return a fresh instance of the named node class.
|
||||
|
||||
@@ -167,10 +199,12 @@ def apply(class_name: str, *args: Any, **kwargs: Any) -> Any:
|
||||
cls = NODE_CLASS_MAPPINGS[class_name]
|
||||
instance = cls()
|
||||
|
||||
input_types = cls.INPUT_TYPES()
|
||||
required = input_types.get("required", {})
|
||||
required_names = list(required.keys())
|
||||
|
||||
# Map positional args to required input names in declaration order
|
||||
if args:
|
||||
input_types = cls.INPUT_TYPES()
|
||||
required_names = list(input_types.get("required", {}).keys())
|
||||
for i, arg in enumerate(args):
|
||||
if i >= len(required_names):
|
||||
raise TypeError(
|
||||
@@ -185,6 +219,18 @@ def apply(class_name: str, *args: Any, **kwargs: Any) -> Any:
|
||||
)
|
||||
kwargs[name] = arg
|
||||
|
||||
# Fill in defaults from INPUT_TYPES metadata for any missing required inputs.
|
||||
# The metadata dict is the single source of truth for default values — the
|
||||
# GUI pre-populates widgets from it, and we mirror that behaviour here so
|
||||
# tono.GaussianFilter(field) works without repeating the default at the call site.
|
||||
for name, spec in required.items():
|
||||
if name in kwargs:
|
||||
continue
|
||||
if isinstance(spec, (list, tuple)) and len(spec) >= 2 and isinstance(spec[1], dict):
|
||||
meta = spec[1]
|
||||
if "default" in meta:
|
||||
kwargs[name] = meta["default"]
|
||||
|
||||
func = getattr(instance, cls.FUNCTION)
|
||||
|
||||
# Run inside an execution context so emit_* calls are no-ops
|
||||
|
||||
@@ -19,15 +19,35 @@ import tono
|
||||
fields = tono.load("scan.gwy")
|
||||
height = fields[0]
|
||||
|
||||
# Apply a processing node
|
||||
leveled = tono.apply("PlaneLevelField", height)
|
||||
filtered = tono.apply("GaussianFilter", leveled, sigma=2.0)
|
||||
# Every registered node is available as a top-level callable
|
||||
leveled = tono.PlaneLevelField(height)
|
||||
filtered = tono.GaussianFilter(leveled, sigma=2.0)
|
||||
|
||||
# Access the raw numpy array
|
||||
print(filtered.data.shape) # (256, 256)
|
||||
print(filtered.data.mean()) # height in metres
|
||||
```
|
||||
|
||||
## Discovering node signatures
|
||||
|
||||
Every node carries a real `inspect.Signature` synthesised from its input
|
||||
declarations, so `help()`, Jupyter's `?`, and IPython tab-completion all show
|
||||
the correct parameters, types, defaults, and enum choices:
|
||||
|
||||
```python
|
||||
>>> help(tono.EdgeDetect)
|
||||
EdgeDetect(
|
||||
field: DataField,
|
||||
method: Literal['sobel', 'prewitt', 'laplacian', 'log'],
|
||||
sigma: float = 1.0,
|
||||
) -> DataField
|
||||
Detect edges using Sobel, Prewitt, Laplacian, or LoG operators.
|
||||
...
|
||||
```
|
||||
|
||||
`dir(tono)` lists every registered node — useful for tab-completion and
|
||||
programmatic discovery.
|
||||
|
||||
## API reference
|
||||
|
||||
### Loading data
|
||||
@@ -55,19 +75,48 @@ List all supported file extensions.
|
||||
|
||||
### Processing
|
||||
|
||||
#### `tono.apply(node_name, *args, **kwargs)`
|
||||
There are two equivalent ways to invoke a node. Both return a single value
|
||||
when the node has one output, or a tuple when it has multiple.
|
||||
|
||||
Run a processing node. Positional arguments are mapped to required inputs in declaration order. Returns a single output if the node has one output, or a tuple if it has multiple.
|
||||
#### `tono.NodeName(*args, **kwargs)` — typed call syntax
|
||||
|
||||
Recommended for scripts and notebooks. Each node is exposed as a top-level
|
||||
callable with a real `inspect.Signature`, so your editor, `help()`, and Jupyter
|
||||
all know the parameters and defaults:
|
||||
|
||||
```python
|
||||
# Positional: first required input is `field`
|
||||
result = tono.apply("GaussianFilter", my_field, sigma=3.0)
|
||||
# Positional arguments map to required inputs in declaration order
|
||||
result = tono.GaussianFilter(my_field, sigma=3.0)
|
||||
|
||||
# All keyword arguments
|
||||
result = tono.apply("GaussianFilter", field=my_field, sigma=3.0)
|
||||
# Fully keyword — order-independent
|
||||
result = tono.GaussianFilter(field=my_field, sigma=3.0)
|
||||
|
||||
# Nodes with multiple outputs return a tuple
|
||||
field_out, mask_out = tono.apply("ThresholdMask", my_field, method="otsu")
|
||||
# Defaults declared in the node's INPUT_TYPES metadata are auto-filled
|
||||
result = tono.GaussianFilter(my_field) # uses sigma=1.0 from metadata
|
||||
|
||||
# Multi-output nodes return a tuple
|
||||
log_mag, mag, phase, psdf = tono.FFT2D(my_field, windowing="hann", level="mean")
|
||||
```
|
||||
|
||||
#### `tono.apply(node_name, *args, **kwargs)` — string-based dispatch
|
||||
|
||||
Use this when the node name is only known at runtime (e.g. a user-selected
|
||||
pipeline). Same arg conventions, same default-filling behaviour.
|
||||
|
||||
```python
|
||||
name = choose_node_from_config()
|
||||
result = tono.apply(name, my_field, sigma=3.0)
|
||||
```
|
||||
|
||||
#### `tono.describe(name) -> dict`
|
||||
|
||||
Return a dict describing a node's inputs, outputs, description, keywords, and
|
||||
category. Thin wrapper around the registry metadata used by the web UI.
|
||||
|
||||
```python
|
||||
info = tono.describe("EdgeDetect")
|
||||
print(info["input"]["required"].keys()) # dict_keys(['field', 'method', 'sigma'])
|
||||
print(info["output_name"]) # ['edges']
|
||||
```
|
||||
|
||||
#### `tono.get_node(name) -> node_instance`
|
||||
@@ -144,12 +193,11 @@ for path in input_dir.glob("*.gwy"):
|
||||
height = fields[0]
|
||||
|
||||
# Standard processing pipeline
|
||||
leveled = tono.apply("PlaneLevelField", height)
|
||||
filtered = tono.apply("GaussianFilter", leveled, sigma=1.5)
|
||||
leveled = tono.PlaneLevelField(height)
|
||||
filtered = tono.GaussianFilter(leveled, sigma=1.5)
|
||||
|
||||
# Extract statistics
|
||||
stats_node = tono.get_node("Statistics")
|
||||
(table,) = stats_node.process(field=filtered)
|
||||
# Scalar measurements
|
||||
table = tono.Statistics(filtered)
|
||||
results[path.name] = table
|
||||
|
||||
print(f"{path.name}: processed {height.xres}x{height.yres} "
|
||||
@@ -176,11 +224,11 @@ axes[0].set_title("Raw")
|
||||
axes[0].set_xlabel("x (um)")
|
||||
axes[0].set_ylabel("y (um)")
|
||||
|
||||
leveled = tono.apply("PlaneLevelField", field)
|
||||
leveled = tono.PlaneLevelField(field)
|
||||
axes[1].imshow(leveled.data * 1e9, extent=extent, cmap="afmhot")
|
||||
axes[1].set_title("Leveled")
|
||||
|
||||
filtered = tono.apply("GaussianFilter", leveled, sigma=2.0)
|
||||
filtered = tono.GaussianFilter(leveled, sigma=2.0)
|
||||
axes[2].imshow(filtered.data * 1e9, extent=extent, cmap="afmhot")
|
||||
axes[2].set_title("Filtered")
|
||||
|
||||
|
||||
@@ -12,8 +12,7 @@ import tono
|
||||
# ── 1. Generate a synthetic surface ──────────────────────────────────
|
||||
|
||||
print("Generating synthetic surface...")
|
||||
surface = tono.apply(
|
||||
"SyntheticSurface",
|
||||
surface = tono.SyntheticSurface(
|
||||
pattern="fbm",
|
||||
xres=256,
|
||||
yres=256,
|
||||
@@ -29,27 +28,26 @@ print(f" Height range: {np.ptp(surface.data)*1e9:.1f} nm")
|
||||
# ── 2. Level the surface ─────────────────────────────────────────────
|
||||
|
||||
print("\nLeveling...")
|
||||
leveled = tono.apply("PlaneLevelField", surface)
|
||||
leveled = tono.PlaneLevelField(surface)
|
||||
print(f" Mean after leveling: {leveled.data.mean()*1e9:.4f} nm")
|
||||
|
||||
# ── 3. Apply a Gaussian filter ───────────────────────────────────────
|
||||
|
||||
print("\nFiltering...")
|
||||
filtered = tono.apply("GaussianFilter", leveled, sigma=2.0)
|
||||
filtered = tono.GaussianFilter(leveled, sigma=2.0)
|
||||
print(f" Height range after filtering: {np.ptp(filtered.data)*1e9:.1f} nm")
|
||||
|
||||
# ── 4. Compute statistics ────────────────────────────────────────────
|
||||
|
||||
print("\nStatistics:")
|
||||
stats_node = tono.get_node("Statistics")
|
||||
(table,) = stats_node.process(field=filtered)
|
||||
table = tono.Statistics(filtered)
|
||||
for row in table:
|
||||
print(f" {row['quantity']}: {row['value']:.6g} {row.get('unit', '')}")
|
||||
|
||||
# ── 5. Edge detection ────────────────────────────────────────────────
|
||||
|
||||
print("\nEdge detection...")
|
||||
edges = tono.apply("EdgeDetect", filtered, method="sobel", sigma=1.0)
|
||||
edges = tono.EdgeDetect(filtered, method="sobel", sigma=1.0)
|
||||
print(f" Edge map range: [{edges.data.min():.2f}, {edges.data.max():.2f}]")
|
||||
|
||||
# ── 6. Create a DataField from a numpy array ─────────────────────────
|
||||
@@ -64,7 +62,7 @@ print(f" Unit: {custom_field.si_unit_z}")
|
||||
# ── 7. FFT analysis ──────────────────────────────────────────────────
|
||||
|
||||
print("\nFFT of the filtered surface...")
|
||||
fft_log_mag, fft_mag, fft_phase, fft_psdf = tono.apply("FFT2D", filtered, windowing="hann", level="mean")
|
||||
fft_log_mag, fft_mag, fft_phase, fft_psdf = tono.FFT2D(filtered, windowing="hann", level="mean")
|
||||
print(f" FFT shape: {fft_mag.data.shape}")
|
||||
print(f" Domain: {fft_mag.domain}")
|
||||
|
||||
|
||||
@@ -39,8 +39,8 @@ def test_parse_ibw_note():
|
||||
|
||||
|
||||
def test_note_load_no_file():
|
||||
from backend.nodes.note import Note
|
||||
node = Note()
|
||||
from backend.nodes.note import IgorNote
|
||||
node = IgorNote()
|
||||
try:
|
||||
node.load(filename="")
|
||||
assert False, "Expected ValueError for empty filename"
|
||||
@@ -49,8 +49,8 @@ def test_note_load_no_file():
|
||||
|
||||
|
||||
def test_note_load_file_not_found():
|
||||
from backend.nodes.note import Note
|
||||
node = Note()
|
||||
from backend.nodes.note import IgorNote
|
||||
node = IgorNote()
|
||||
try:
|
||||
node.load(filename="/nonexistent/path/file.ibw")
|
||||
assert False, "Expected FileNotFoundError"
|
||||
@@ -59,8 +59,8 @@ def test_note_load_file_not_found():
|
||||
|
||||
|
||||
def test_note_load_wrong_extension():
|
||||
from backend.nodes.note import Note
|
||||
node = Note()
|
||||
from backend.nodes.note import IgorNote
|
||||
node = IgorNote()
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
txt_path = os.path.join(tmpdir, "notes.txt")
|
||||
Path(txt_path).write_text("Key=Value")
|
||||
@@ -73,11 +73,11 @@ def test_note_load_wrong_extension():
|
||||
|
||||
def test_note_load_empty_note():
|
||||
"""An .ibw file with no parseable note entries raises ValueError."""
|
||||
from backend.nodes.note import Note
|
||||
from backend.nodes.note import IgorNote
|
||||
from unittest.mock import patch
|
||||
import numpy as np
|
||||
|
||||
node = Note()
|
||||
node = IgorNote()
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
ibw_path = os.path.join(tmpdir, "empty_note.ibw")
|
||||
Path(ibw_path).write_bytes(b"fake_ibw")
|
||||
@@ -94,7 +94,7 @@ def test_note_load_empty_note():
|
||||
|
||||
def test_note_load_ibw():
|
||||
"""Load from a real .ibw file if available in the demo directory."""
|
||||
from backend.nodes.note import Note
|
||||
from backend.nodes.note import IgorNote
|
||||
from backend.data_types import DataTable
|
||||
|
||||
demo_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "demo"))
|
||||
@@ -107,7 +107,7 @@ def test_note_load_ibw():
|
||||
if ibw_path is None:
|
||||
return
|
||||
|
||||
node = Note()
|
||||
node = IgorNote()
|
||||
try:
|
||||
result, = node.load(filename=ibw_path)
|
||||
assert isinstance(result, DataTable)
|
||||
|
||||
154
tests/test_tono_api.py
Normal file
154
tests/test_tono_api.py
Normal file
@@ -0,0 +1,154 @@
|
||||
"""Tests for the public ``tono`` library API surface."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
from typing import Literal, get_args, get_origin
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
import tono
|
||||
from backend.data_types import DataField, RecordTable
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_field() -> DataField:
|
||||
"""A small deterministic field for fast tests."""
|
||||
rng = np.random.default_rng(0)
|
||||
data = rng.standard_normal((32, 32)).astype(np.float64)
|
||||
return tono.field(data, xreal=1e-6, yreal=1e-6)
|
||||
|
||||
|
||||
# ── Typed call syntax ───────────────────────────────────────────────────
|
||||
|
||||
def test_gaussian_filter_call(sample_field):
|
||||
result = tono.GaussianFilter(sample_field, sigma=2.0)
|
||||
assert isinstance(result, DataField)
|
||||
assert result.data.shape == sample_field.data.shape
|
||||
|
||||
|
||||
def test_default_filling_from_metadata(sample_field):
|
||||
# sigma has a metadata default of 1.0 — call should succeed without it.
|
||||
result = tono.GaussianFilter(sample_field)
|
||||
assert isinstance(result, DataField)
|
||||
|
||||
|
||||
def test_keyword_only_call(sample_field):
|
||||
result = tono.GaussianFilter(field=sample_field, sigma=0.5)
|
||||
assert isinstance(result, DataField)
|
||||
|
||||
|
||||
def test_enum_input_call(sample_field):
|
||||
result = tono.EdgeDetect(sample_field, method="sobel", sigma=1.0)
|
||||
assert isinstance(result, DataField)
|
||||
|
||||
|
||||
def test_multi_output_returns_tuple(sample_field):
|
||||
out = tono.FFT2D(sample_field, windowing="hann", level="mean")
|
||||
assert isinstance(out, tuple)
|
||||
assert len(out) == 4
|
||||
assert all(isinstance(x, DataField) for x in out)
|
||||
|
||||
|
||||
def test_awkward_required_order(sample_field):
|
||||
# ThresholdMask has ``threshold`` (default 0.0) followed by ``direction``
|
||||
# (no default). The wrapper must handle this without raising, and the
|
||||
# metadata default for threshold must still fire at call time.
|
||||
mask, record = tono.ThresholdMask(
|
||||
sample_field, method="otsu", direction="above"
|
||||
)
|
||||
assert mask.shape == sample_field.data.shape
|
||||
assert isinstance(record, RecordTable)
|
||||
|
||||
|
||||
# ── Signature introspection ─────────────────────────────────────────────
|
||||
|
||||
def test_signature_has_expected_params():
|
||||
sig = inspect.signature(tono.GaussianFilter)
|
||||
assert list(sig.parameters) == ["field", "sigma"]
|
||||
assert sig.parameters["sigma"].default == 1.0
|
||||
assert sig.parameters["field"].annotation is DataField
|
||||
|
||||
|
||||
def test_signature_enum_uses_literal():
|
||||
sig = inspect.signature(tono.EdgeDetect)
|
||||
annotation = sig.parameters["method"].annotation
|
||||
assert get_origin(annotation) is Literal
|
||||
assert set(get_args(annotation)) == {"sobel", "prewitt", "laplacian", "log"}
|
||||
|
||||
|
||||
def test_signature_return_annotation_single_output():
|
||||
sig = inspect.signature(tono.GaussianFilter)
|
||||
assert sig.return_annotation is DataField
|
||||
|
||||
|
||||
def test_signature_return_annotation_multi_output():
|
||||
sig = inspect.signature(tono.FFT2D)
|
||||
assert sig.return_annotation is tuple
|
||||
|
||||
|
||||
def test_docstring_contains_description_and_params():
|
||||
doc = tono.EdgeDetect.__doc__ or ""
|
||||
assert "Sobel" in doc or "sobel" in doc
|
||||
assert "field" in doc
|
||||
assert "method" in doc
|
||||
assert "sigma" in doc
|
||||
|
||||
|
||||
# ── Module-level dunder behaviour ───────────────────────────────────────
|
||||
|
||||
def test_dir_lists_nodes():
|
||||
entries = dir(tono)
|
||||
assert "GaussianFilter" in entries
|
||||
assert "EdgeDetect" in entries
|
||||
assert "apply" in entries # existing top-level export
|
||||
assert "describe" in entries
|
||||
|
||||
|
||||
def test_unknown_attribute_raises_attribute_error():
|
||||
with pytest.raises(AttributeError, match="NotARealNode"):
|
||||
tono.NotARealNode # noqa: B018
|
||||
|
||||
|
||||
def test_dunder_lookup_raises_immediately():
|
||||
# __getattr__ should short-circuit dunder lookups so that e.g. copy/pickle
|
||||
# machinery doesn't trigger registry loading.
|
||||
with pytest.raises(AttributeError):
|
||||
tono.__some_dunder__ # noqa: B018
|
||||
|
||||
|
||||
def test_wrappers_are_cached():
|
||||
first = tono.GaussianFilter
|
||||
second = tono.GaussianFilter
|
||||
assert first is second
|
||||
|
||||
|
||||
# ── describe() ──────────────────────────────────────────────────────────
|
||||
|
||||
def test_describe_returns_expected_keys():
|
||||
info = tono.describe("EdgeDetect")
|
||||
assert info["name"] == "EdgeDetect"
|
||||
assert "input" in info
|
||||
assert "output" in info
|
||||
assert "description" in info
|
||||
assert "field" in info["input"]["required"]
|
||||
|
||||
|
||||
def test_describe_unknown_node_raises():
|
||||
with pytest.raises(KeyError):
|
||||
tono.describe("NotARealNode")
|
||||
|
||||
|
||||
# ── apply() still works and matches wrapper semantics ───────────────────
|
||||
|
||||
def test_apply_default_filling(sample_field):
|
||||
# apply() should also fill metadata defaults now.
|
||||
result = tono.apply("GaussianFilter", sample_field)
|
||||
assert isinstance(result, DataField)
|
||||
|
||||
|
||||
def test_apply_and_wrapper_are_equivalent(sample_field):
|
||||
via_apply = tono.apply("GaussianFilter", sample_field, sigma=1.5)
|
||||
via_wrapper = tono.GaussianFilter(sample_field, sigma=1.5)
|
||||
np.testing.assert_array_equal(via_apply.data, via_wrapper.data)
|
||||
190
tono.py
190
tono.py
@@ -12,7 +12,10 @@ Quick start::
|
||||
# Load SPM data
|
||||
fields = tono.load("scan.gwy")
|
||||
|
||||
# Process
|
||||
# Typed call syntax — help(tono.GaussianFilter) shows the signature
|
||||
result = tono.GaussianFilter(fields[0], sigma=3.0)
|
||||
|
||||
# String-based dispatch for dynamic cases
|
||||
result = tono.apply("GaussianFilter", fields[0], sigma=3.0)
|
||||
|
||||
# Create a field from a numpy array
|
||||
@@ -22,15 +25,22 @@ Quick start::
|
||||
See ``backend.api`` for the full API documentation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
from typing import Any, Callable, Literal
|
||||
|
||||
from backend.api import ( # noqa: F401
|
||||
apply,
|
||||
channel_names,
|
||||
describe,
|
||||
field,
|
||||
get_node,
|
||||
load,
|
||||
nodes,
|
||||
supported_formats,
|
||||
)
|
||||
from backend.api import _ensure_registry as _ensure_registry
|
||||
from backend.data_types import ( # noqa: F401
|
||||
DataField,
|
||||
DataTable,
|
||||
@@ -38,3 +48,181 @@ from backend.data_types import ( # noqa: F401
|
||||
MeshModel,
|
||||
RecordTable,
|
||||
)
|
||||
|
||||
|
||||
# ── Runtime node wrappers ─────────────────────────────────────────────
|
||||
#
|
||||
# PEP 562 module __getattr__: expose every registered node as a
|
||||
# top-level callable so users can write ``tono.GaussianFilter(field, sigma=2)``
|
||||
# instead of ``tono.apply("GaussianFilter", field, sigma=2)``. Each wrapper
|
||||
# carries a real inspect.Signature synthesised from the node's INPUT_TYPES,
|
||||
# so help(tono.GaussianFilter), Jupyter's ``?``, and IPython tab-completion
|
||||
# all show the correct parameters, defaults, and enum choices.
|
||||
|
||||
_NODE_WRAPPER_CACHE: dict[str, Callable[..., Any]] = {}
|
||||
|
||||
|
||||
# Map tono type-system names to Python runtime types used in annotations.
|
||||
# Unknown names fall back to ``typing.Any``.
|
||||
_TONO_TYPE_MAP: dict[str, Any] = {
|
||||
"DATA_FIELD": DataField,
|
||||
"IMAGE": DataField,
|
||||
"FLOAT": float,
|
||||
"INT": int,
|
||||
"STRING": str,
|
||||
"BOOL": bool,
|
||||
"LINE": LineData,
|
||||
"RECORD_TABLE": RecordTable,
|
||||
"TABLE": RecordTable,
|
||||
"DATA_TABLE": DataTable,
|
||||
"MESH": MeshModel,
|
||||
"COORD": tuple,
|
||||
}
|
||||
|
||||
|
||||
def _tono_type_to_python(spec: Any) -> Any:
|
||||
"""Resolve a tono input/output type spec to a Python annotation.
|
||||
|
||||
``spec`` is the first element of an INPUT_TYPES tuple — either a string
|
||||
type name (``"FLOAT"``) or a list of choices (``["sobel", "prewitt"]``).
|
||||
"""
|
||||
if isinstance(spec, (list, tuple)) and all(isinstance(x, str) for x in spec):
|
||||
# Enum input — use typing.Literal so help() shows the exact choices.
|
||||
if len(spec) == 0:
|
||||
return str
|
||||
return Literal[tuple(spec)] # type: ignore[valid-type]
|
||||
if isinstance(spec, str):
|
||||
return _TONO_TYPE_MAP.get(spec, Any)
|
||||
return Any
|
||||
|
||||
|
||||
def _build_node_wrapper(name: str, cls: type) -> Callable[..., Any]:
|
||||
"""Build a callable wrapper with a synthesised inspect.Signature for ``cls``."""
|
||||
input_types = cls.INPUT_TYPES()
|
||||
required = input_types.get("required", {})
|
||||
optional = input_types.get("optional", {})
|
||||
|
||||
params: list[inspect.Parameter] = []
|
||||
param_doc_lines: list[str] = []
|
||||
EMPTY = inspect.Parameter.empty
|
||||
|
||||
def _param(arg_name: str, raw_spec: Any, force_optional: bool) -> None:
|
||||
# raw_spec is a tuple like ("FLOAT", {...}) or (["sobel","prewitt"],)
|
||||
type_token = raw_spec[0] if isinstance(raw_spec, (list, tuple)) and raw_spec else raw_spec
|
||||
meta: dict = {}
|
||||
if isinstance(raw_spec, (list, tuple)) and len(raw_spec) >= 2 and isinstance(raw_spec[1], dict):
|
||||
meta = raw_spec[1]
|
||||
|
||||
annotation = _tono_type_to_python(type_token)
|
||||
if force_optional:
|
||||
default: Any = None
|
||||
elif "default" in meta:
|
||||
default = meta["default"]
|
||||
else:
|
||||
default = EMPTY
|
||||
|
||||
params.append(
|
||||
inspect.Parameter(
|
||||
arg_name,
|
||||
inspect.Parameter.POSITIONAL_OR_KEYWORD,
|
||||
default=default,
|
||||
annotation=annotation,
|
||||
)
|
||||
)
|
||||
|
||||
# Docstring line
|
||||
if isinstance(type_token, (list, tuple)):
|
||||
type_desc = "{" + ", ".join(repr(c) for c in type_token) + "}"
|
||||
elif isinstance(type_token, str):
|
||||
type_desc = type_token
|
||||
else:
|
||||
type_desc = "Any"
|
||||
default_desc = "" if default is EMPTY else f" (default: {default!r})"
|
||||
param_doc_lines.append(f" {arg_name} : {type_desc}{default_desc}")
|
||||
|
||||
for arg_name, raw_spec in required.items():
|
||||
_param(arg_name, raw_spec, force_optional=False)
|
||||
|
||||
# Python signatures require that any parameter with a default is followed
|
||||
# only by parameters that also have defaults. Some nodes declare required
|
||||
# inputs in an order that violates this (e.g. ThresholdMask has ``threshold``
|
||||
# with a default followed by ``direction`` without). Walk the required
|
||||
# params in reverse and clear defaults that would break ordering — the
|
||||
# default still fires at runtime via apply()'s metadata default-filling,
|
||||
# it just isn't displayed in the signature for that slot.
|
||||
_seen_no_default = False
|
||||
for i in range(len(params) - 1, -1, -1):
|
||||
if params[i].default is EMPTY:
|
||||
_seen_no_default = True
|
||||
elif _seen_no_default:
|
||||
params[i] = params[i].replace(default=EMPTY)
|
||||
|
||||
for arg_name, raw_spec in optional.items():
|
||||
_param(arg_name, raw_spec, force_optional=True)
|
||||
|
||||
# Return annotation: single-output nodes return the unwrapped Python type;
|
||||
# multi-output nodes return a tuple.
|
||||
outputs = getattr(cls, "OUTPUTS", ())
|
||||
if len(outputs) == 1:
|
||||
return_annotation: Any = _tono_type_to_python(outputs[0][0])
|
||||
else:
|
||||
return_annotation = tuple
|
||||
|
||||
signature = inspect.Signature(params, return_annotation=return_annotation)
|
||||
|
||||
description = (getattr(cls, "DESCRIPTION", "") or "").strip()
|
||||
output_lines = [
|
||||
f" {out_name} : {out_type}"
|
||||
for out_type, out_name, *_ in outputs
|
||||
]
|
||||
|
||||
doc_parts: list[str] = []
|
||||
if description:
|
||||
doc_parts.append(description)
|
||||
doc_parts.append("")
|
||||
doc_parts.append("Parameters")
|
||||
doc_parts.append("----------")
|
||||
doc_parts.extend(param_doc_lines if param_doc_lines else [" (none)"])
|
||||
if output_lines:
|
||||
doc_parts.append("")
|
||||
doc_parts.append("Returns")
|
||||
doc_parts.append("-------")
|
||||
doc_parts.extend(output_lines)
|
||||
docstring = "\n".join(doc_parts)
|
||||
|
||||
def wrapper(*args: Any, **kwargs: Any) -> Any:
|
||||
return apply(name, *args, **kwargs)
|
||||
|
||||
wrapper.__name__ = name
|
||||
wrapper.__qualname__ = f"tono.{name}"
|
||||
wrapper.__module__ = "tono"
|
||||
wrapper.__doc__ = docstring
|
||||
wrapper.__signature__ = signature # type: ignore[attr-defined]
|
||||
wrapper.__wrapped_node__ = name # type: ignore[attr-defined]
|
||||
return wrapper
|
||||
|
||||
|
||||
def __getattr__(name: str) -> Any:
|
||||
# Skip dunder lookups so imports, pickling, and debuggers don't trigger
|
||||
# registry loading or spurious AttributeError chains.
|
||||
if name.startswith("__") and name.endswith("__"):
|
||||
raise AttributeError(f"module 'tono' has no attribute {name!r}")
|
||||
|
||||
from backend.node_registry import NODE_CLASS_MAPPINGS
|
||||
|
||||
_ensure_registry()
|
||||
if name not in NODE_CLASS_MAPPINGS:
|
||||
raise AttributeError(f"module 'tono' has no attribute {name!r}")
|
||||
|
||||
cached = _NODE_WRAPPER_CACHE.get(name)
|
||||
if cached is None:
|
||||
cached = _build_node_wrapper(name, NODE_CLASS_MAPPINGS[name])
|
||||
_NODE_WRAPPER_CACHE[name] = cached
|
||||
return cached
|
||||
|
||||
|
||||
def __dir__() -> list[str]:
|
||||
from backend.node_registry import NODE_CLASS_MAPPINGS
|
||||
|
||||
_ensure_registry()
|
||||
return sorted({*globals().keys(), *NODE_CLASS_MAPPINGS.keys()})
|
||||
|
||||
Reference in New Issue
Block a user