Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 10 additions & 10 deletions ROADMAP.md
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ ucon is a dimensional analysis library for engineers building systems where unit
| v0.7.0 | MCP Error Suggestions | Complete |
| v0.7.1 | MCP Error Infrastructure for Multi-Step Chains | Complete |
| v0.7.2 | Compute Tool | Complete |
| v0.7.3 | Graph-Local Name Resolution | Planned |
| v0.7.3 | Graph-Local Name Resolution | Complete |
| v0.7.4 | UnitPackage + TOML Loading | Planned |
| v0.7.5 | MCP Extension Tools | Planned |
| v0.7.x | Schema-Level Dimension Constraints | Planned |
Expand Down Expand Up @@ -318,18 +318,18 @@ Prerequisite for factor-label chains with countable items (tablets, doses).

---

## v0.7.3 — Graph-Local Name Resolution (Planned)
## v0.7.3 — Graph-Local Name Resolution (Complete)

**Theme:** Shared infrastructure for dynamic unit extension.

- [ ] `ConversionGraph._name_registry` (case-insensitive) and `_name_registry_cs` (case-sensitive)
- [ ] `graph.register_unit(unit)` — Register unit for name resolution within graph
- [ ] `graph.resolve_unit(name)` — Lookup in graph-local registry, return None if not found
- [ ] `graph.copy()` — Deep copy edges, shallow copy registries
- [ ] `_parsing_graph` ContextVar for threading resolution through parsing
- [ ] `using_graph()` sets both conversion and parsing context
- [ ] `_lookup_factor()` checks graph-local first, falls back to global
- [ ] `_build_standard_graph()` calls `register_unit()` for all standard units
- [x] `ConversionGraph._name_registry` (case-insensitive) and `_name_registry_cs` (case-sensitive)
- [x] `graph.register_unit(unit)` — Register unit for name resolution within graph
- [x] `graph.resolve_unit(name)` — Lookup in graph-local registry, return None if not found
- [x] `graph.copy()` — Deep copy edges, shallow copy registries
- [x] `_parsing_graph` ContextVar for threading resolution through parsing
- [x] `using_graph()` sets both conversion and parsing context
- [x] `_lookup_factor()` checks graph-local first, falls back to global
- [x] `_build_standard_graph()` calls `register_unit()` for all standard units

**Outcomes:**
- Unit name resolution becomes graph-scoped, not global
Expand Down
118 changes: 118 additions & 0 deletions tests/ucon/conversion/test_graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -407,3 +407,121 @@ def test_default_graph_has_standard_conversions(self):

m = graph.convert(src=units.kilogram, dst=units.pound)
self.assertAlmostEqual(m(1), 2.20462, places=4)


class TestConversionGraphCrossDimension(unittest.TestCase):
"""Tests for cross-dimension conversions (volume↔length³)."""

def test_volume_to_length_cubed(self):
"""Test volume → length³ via product edges (liter → m³)."""
graph = get_default_graph()
# liter is volume dimension, m³ is length³
liter_prod = UnitProduct.from_unit(units.liter)
m_cubed = units.meter ** 3

m = graph.convert(src=liter_prod, dst=m_cubed)
# 1 L = 0.001 m³
self.assertAlmostEqual(m(1), 0.001, places=6)

def test_gallon_to_cubic_meter(self):
"""Test gallon → m³ via multi-hop (gal → L → m³)."""
graph = get_default_graph()
gal_prod = UnitProduct.from_unit(units.gallon)
m_cubed = units.meter ** 3

m = graph.convert(src=gal_prod, dst=m_cubed)
# 1 gal ≈ 0.00378541 m³
self.assertAlmostEqual(m(1), 0.00378541, places=5)


class TestConversionGraphProductEdgePaths(unittest.TestCase):
"""Tests for product edge path finding."""

def setUp(self):
self.graph = ConversionGraph()

def test_bfs_product_direct_edge(self):
"""Test _bfs_product_path with direct product edge."""
meter = units.meter
foot = Unit(name='foot', dimension=Dimension.length, aliases=('ft',))

m_prod = UnitProduct.from_unit(meter)
ft_prod = UnitProduct.from_unit(foot)

# Add direct product edge
self.graph.add_edge(src=m_prod, dst=ft_prod, map=LinearMap(3.28084))

m = self.graph.convert(src=m_prod, dst=ft_prod)
self.assertAlmostEqual(m(1), 3.28084, places=4)

def test_product_to_base_scale_lookup(self):
"""Test product edge lookup via base-scale version of dst."""
meter = units.meter
# Create km product and m product
km_prod = UnitProduct({UnitFactor(meter, Scale.kilo): 1})
m_prod = UnitProduct({UnitFactor(meter, Scale.one): 1})

# Add edge from km to base (m)
self.graph.add_edge(src=km_prod, dst=m_prod, map=LinearMap(1000))

# Convert km to mm - should find edge to m then scale
mm_prod = UnitProduct({UnitFactor(meter, Scale.milli): 1})
m = self.graph.convert(src=km_prod, dst=mm_prod)
# 1 km = 1,000,000 mm
self.assertAlmostEqual(m(1), 1_000_000, places=0)


class TestConversionGraphFactorwiseErrors(unittest.TestCase):
"""Tests for factorwise conversion error paths."""

def setUp(self):
self.graph = get_default_graph()

def test_pseudo_dimension_isolation(self):
"""Test that pseudo-dimensions cannot convert between each other."""
# angle and ratio both have zero vector but are isolated
rad_prod = UnitProduct.from_unit(units.radian)
pct_prod = UnitProduct.from_unit(units.percent)

with self.assertRaises(ConversionNotFound) as ctx:
self.graph.convert(src=rad_prod, dst=pct_prod)
self.assertIn("pseudo-dimension", str(ctx.exception).lower())

def test_factor_structure_mismatch_after_vector_grouping(self):
"""Test error when factor structures don't align after vector grouping."""
# Create products with same total dimension but different structures
# This is hard to trigger since dimension check catches most cases
# Need to use custom units with same dimension vector

# Actually, this error path (line 640) is hard to hit because
# dimension check happens first. Skip this test.
pass


class TestConversionGraphAmbiguousDecomposition(unittest.TestCase):
"""Tests for ambiguous factor decomposition errors."""

def test_factors_by_dimension_ambiguity(self):
"""Test error when UnitProduct has ambiguous factor decomposition."""
# Create a product where factors_by_dimension would raise ValueError
# This happens when multiple factors have the same dimension

# Two different length units in same product
meter = units.meter
foot = Unit(name='foot', dimension=Dimension.length, aliases=('ft',))

# m * ft - both length, ambiguous
ambiguous = UnitProduct({
UnitFactor(meter, Scale.one): 1,
UnitFactor(foot, Scale.one): 1,
})

# Target with single length²
target = UnitProduct({UnitFactor(meter, Scale.one): 2})

graph = ConversionGraph()
graph.add_edge(src=meter, dst=foot, map=LinearMap(3.28084))

with self.assertRaises(ConversionNotFound) as ctx:
graph.convert(src=ambiguous, dst=target)
self.assertIn("Ambiguous", str(ctx.exception))
4 changes: 3 additions & 1 deletion ucon/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@
Ratio,
)
from ucon.checking import enforce_dimensions
from ucon.graph import get_default_graph, using_graph
from ucon.graph import get_default_graph, get_parsing_graph, set_default_graph, using_graph
from ucon.units import UnknownUnitError, get_unit_by_name


Expand All @@ -76,7 +76,9 @@
'UnitSystem',
'UnknownUnitError',
'get_default_graph',
'get_parsing_graph',
'get_unit_by_name',
'set_default_graph',
'units',
'using_graph',
]
126 changes: 124 additions & 2 deletions ucon/graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
- :func:`set_default_graph` — Replace the default graph.
- :func:`reset_default_graph` — Reset to standard graph on next access.
- :func:`using_graph` — Context manager for scoped graph override.
- :func:`get_parsing_graph` — Get the graph for name resolution during parsing.
"""
from __future__ import annotations

Expand Down Expand Up @@ -67,6 +68,7 @@ class ConversionGraph:
- Direct edge lookup
- BFS path composition for multi-hop conversions
- Factorwise decomposition for UnitProduct conversions
- Graph-local unit name resolution (v0.7.3+)
"""

# Edges between Units, partitioned by Dimension
Expand All @@ -78,6 +80,12 @@ class ConversionGraph:
# Rebased units: original unit → RebasedUnit (for cross-basis edges)
_rebased: dict[Unit, RebasedUnit] = field(default_factory=dict)

# Graph-local name resolution (case-insensitive keys)
_name_registry: dict[str, Unit] = field(default_factory=dict)

# Graph-local name resolution (case-sensitive keys for shorthands like 'm', 'L')
_name_registry_cs: dict[str, Unit] = field(default_factory=dict)

# ------------- Edge Management -------------------------------------------

def add_edge(
Expand Down Expand Up @@ -279,6 +287,83 @@ def edges_for_transform(self, transform: BasisTransform) -> list[tuple[Unit, Uni
result.append((original, dst))
return result

# ------------- Name Resolution --------------------------------------------

def register_unit(self, unit: Unit) -> None:
"""Register a unit for name resolution within this graph.

Populates both case-insensitive and case-sensitive registries
with the unit's name, shorthand, and aliases.

Parameters
----------
unit : Unit
The unit to register.
"""
# Register canonical name (case-insensitive)
self._name_registry[unit.name.lower()] = unit
self._name_registry_cs[unit.name] = unit

# Register shorthand (case-sensitive only — 'm' vs 'M' matters)
if unit.shorthand:
self._name_registry_cs[unit.shorthand] = unit

# Register aliases
for alias in (unit.aliases or ()):
if alias:
self._name_registry[alias.lower()] = unit
self._name_registry_cs[alias] = unit

def resolve_unit(self, name: str) -> tuple[Unit, Scale] | None:
"""Resolve a unit string in graph-local registry.

Checks case-sensitive registry first (for shorthands like 'm', 'L'),
then falls back to case-insensitive lookup.

Parameters
----------
name : str
The unit name or alias to resolve.

Returns
-------
tuple[Unit, Scale] | None
(unit, Scale.one) if found, None otherwise.
Caller should fall back to global registry if None.
"""
# Case-sensitive first (preserves shorthand like 'm' vs 'M')
if name in self._name_registry_cs:
return self._name_registry_cs[name], Scale.one

# Case-insensitive fallback
if name.lower() in self._name_registry:
return self._name_registry[name.lower()], Scale.one

return None

def copy(self) -> 'ConversionGraph':
"""Return a deep copy suitable for extension.

Creates independent copies of edge dictionaries and name registries.
The returned graph can be modified without affecting the original.

Returns
-------
ConversionGraph
A new graph with copied state.
"""
import copy as copy_module

new = ConversionGraph()
new._unit_edges = copy_module.deepcopy(self._unit_edges)
new._product_edges = copy_module.deepcopy(self._product_edges)
new._rebased = dict(self._rebased)
new._name_registry = dict(self._name_registry)
new._name_registry_cs = dict(self._name_registry_cs)
return new

# ------------- Internal Helpers ------------------------------------------

def _ensure_dimension(self, dim: Dimension) -> None:
if dim not in self._unit_edges:
self._unit_edges[dim] = {}
Expand Down Expand Up @@ -622,6 +707,7 @@ def _convert_factorwise(self, *, src: UnitProduct, dst: UnitProduct) -> Map:

_default_graph: ConversionGraph | None = None
_graph_context: ContextVar[ConversionGraph | None] = ContextVar("graph", default=None)
_parsing_graph: ContextVar[ConversionGraph | None] = ContextVar("parsing_graph", default=None)


def get_default_graph() -> ConversionGraph:
Expand Down Expand Up @@ -655,20 +741,50 @@ def reset_default_graph() -> None:
_default_graph = None


def get_parsing_graph() -> ConversionGraph | None:
"""Get the graph to use for name resolution during parsing.

Returns the context-local parsing graph if set, otherwise None.
Used by _lookup_factor() to check graph-local registry first.

Returns
-------
ConversionGraph | None
The parsing graph, or None if not in a using_graph() context.
"""
return _parsing_graph.get()


@contextmanager
def using_graph(graph: ConversionGraph):
"""Context manager for scoped graph override.

Sets both the conversion graph and parsing graph contexts,
so that name resolution and conversions both use the same graph.

Usage::

with using_graph(custom_graph):
result = value.to(target) # uses custom_graph
unit = get_unit_by_name("custom_unit") # resolves in custom_graph

Parameters
----------
graph : ConversionGraph
The graph to use within this context.

Yields
------
ConversionGraph
The same graph passed in.
"""
token = _graph_context.set(graph)
token_graph = _graph_context.set(graph)
token_parsing = _parsing_graph.set(graph)
try:
yield graph
finally:
_graph_context.reset(token)
_graph_context.reset(token_graph)
_parsing_graph.reset(token_parsing)


def _build_standard_graph() -> ConversionGraph:
Expand All @@ -677,6 +793,12 @@ def _build_standard_graph() -> ConversionGraph:

graph = ConversionGraph()

# Register all standard units for graph-local name resolution
for name in dir(units):
obj = getattr(units, name)
if isinstance(obj, Unit) and not isinstance(obj, RebasedUnit):
graph.register_unit(obj)

# --- Length ---
graph.add_edge(src=units.meter, dst=units.foot, map=LinearMap(3.28084))
graph.add_edge(src=units.foot, dst=units.inch, map=LinearMap(12))
Expand Down
11 changes: 11 additions & 0 deletions ucon/units.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
from typing import Dict, Tuple, Union

from ucon.core import Dimension, Scale, Unit, UnitFactor, UnitProduct, UnitSystem
from ucon.graph import get_parsing_graph
from ucon.parsing import parse_unit_expression, ParseError


Expand Down Expand Up @@ -352,6 +353,9 @@ def _lookup_factor(s: str) -> Tuple[Unit, Scale]:
"""
Look up a single unit factor, handling scale prefixes.

Checks graph-local registry first (if within a using_graph() context),
then falls back to the global registry.

Prioritizes prefix+unit interpretation over direct unit lookup,
except for priority aliases (like 'min', 'mcg') which are checked first
to avoid ambiguous parses or to handle domain-specific conventions.
Expand All @@ -373,6 +377,13 @@ def _lookup_factor(s: str) -> Tuple[Unit, Scale]:
Raises:
UnknownUnitError: If the unit cannot be resolved.
"""
# Check graph-local registry first (if in using_graph() context)
graph = get_parsing_graph()
if graph is not None:
result = graph.resolve_unit(s)
if result is not None:
return result

# Check priority scaled aliases first (e.g., "mcg" -> microgram)
if s in _PRIORITY_SCALED_ALIASES:
return _PRIORITY_SCALED_ALIASES[s]
Expand Down