"""
Attribute schema for FalcomChain graphs.
Defines exactly what node, edge, and graph-level attributes the
hierarchical capacitated facility location algorithms read and write.
This is the single source of truth — every constructor, validator,
and documentation page references this schema.
"""
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional
[docs]
@dataclass(frozen=True)
class AttributeSpec:
"""Specification for a single attribute."""
name: str
required: bool
type: type
default: Any
purpose: str
validator: Optional[Callable[[Any], bool]] = None
# ---------------------------------------------------------------------------
# Node attributes
# ---------------------------------------------------------------------------
NODE_ATTRIBUTES: Dict[str, AttributeSpec] = {
"demand": AttributeSpec(
name="demand",
required=True,
type=float,
default=None,
purpose=(
"Demand of the geographic unit (e.g., population, workload). "
"Used for demand-balance constraints in district formation."
),
validator=lambda v: v is not None and v >= 0,
),
"candidate": AttributeSpec(
name="candidate",
required=True,
type=int,
default=0,
purpose=(
"Whether this node is a level-1 facility candidate (0 or 1). "
"Each district must contain at least one candidate."
),
validator=lambda v: v in (0, 1, True, False),
),
"super_candidate": AttributeSpec(
name="super_candidate",
required=False,
type=int,
default=0,
purpose=(
"Whether this node is a level-2 (super-) facility candidate "
"(0 or 1). The level-2 facility set F^2 is a subset of V^1 and "
"is independent of the level-1 candidate set. If a superdistrict "
"contains no super-candidates, no level-2 facility is assigned "
"for it (soft constraint)."
),
validator=lambda v: v in (0, 1, True, False),
),
"C_X": AttributeSpec(
name="C_X",
required=False,
type=float,
default=0.0,
purpose="X coordinate of the node centroid (for visualization).",
),
"C_Y": AttributeSpec(
name="C_Y",
required=False,
type=float,
default=0.0,
purpose="Y coordinate of the node centroid (for visualization).",
),
"area": AttributeSpec(
name="area",
required=False,
type=float,
default=1.0,
purpose="Geographic area of the node (used for compactness metrics).",
),
# Set by Partition.write_to_graph()
"district": AttributeSpec(
name="district",
required=False,
type=int,
default=None,
purpose=(
"District ID this node belongs to. Set by "
"``Partition.write_to_graph()``. Used by ``Partition.from_graph()`` "
"to reconstruct a partition from a saved graph."
),
),
}
# ---------------------------------------------------------------------------
# Edge attributes
# ---------------------------------------------------------------------------
EDGE_ATTRIBUTES: Dict[str, AttributeSpec] = {
"shared_perim": AttributeSpec(
name="shared_perim",
required=False,
type=float,
default=1.0,
purpose=(
"Length of the shared boundary between two adjacent nodes. "
"Used by compactness computations and rook adjacency."
),
),
}
# ---------------------------------------------------------------------------
# Graph-level attributes (graph.graph[...])
# ---------------------------------------------------------------------------
GRAPH_ATTRIBUTES: Dict[str, AttributeSpec] = {
"crs": AttributeSpec(
name="crs",
required=False,
type=str,
default=None,
purpose="Coordinate reference system (set by from_geodataframe).",
),
"teams_per_district": AttributeSpec(
name="teams_per_district",
required=False,
type=dict,
default=None,
purpose=(
"Dict mapping district ID -> team count. Set by "
"``Partition.write_to_graph()`` so a saved graph fully describes a partition."
),
),
"capacity_level": AttributeSpec(
name="capacity_level",
required=False,
type=int,
default=None,
purpose=(
"Maximum number of teams per district. Set by "
"``Partition.write_to_graph()``."
),
),
}
# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------
[docs]
class SchemaValidationError(ValueError):
"""Raised when a graph's attributes don't match the FalcomChain schema."""
[docs]
def required_node_attributes() -> list:
"""List of node attribute names that MUST be present."""
return [name for name, spec in NODE_ATTRIBUTES.items() if spec.required]
[docs]
def validate_graph(graph, strict: bool = True) -> list:
"""
Validate that a graph satisfies the FalcomChain attribute schema.
:param graph: A networkx-like Graph.
:param strict: If True, raise SchemaValidationError on failure.
If False, return a list of error strings.
:returns: List of error messages (empty if valid).
:raises SchemaValidationError: If strict=True and validation fails.
"""
errors = []
g = graph.graph if hasattr(graph, "graph") and hasattr(graph.graph, "nodes") else graph
if g.number_of_nodes() == 0:
errors.append("Graph has no nodes.")
if strict:
raise SchemaValidationError("\n".join(errors))
return errors
# Check required node attributes
for attr_name in required_node_attributes():
spec = NODE_ATTRIBUTES[attr_name]
missing = [n for n in g.nodes if attr_name not in g.nodes[n]]
if missing:
sample = missing[:5]
more = f" (and {len(missing) - 5} more)" if len(missing) > 5 else ""
errors.append(
f"Missing node attribute '{attr_name}' on nodes: {sample}{more}. "
f"{spec.purpose}"
)
continue
# Run validator on each node value
if spec.validator:
invalid = [
n for n in g.nodes
if not spec.validator(g.nodes[n][attr_name])
]
if invalid:
sample = invalid[:5]
more = f" (and {len(invalid) - 5} more)" if len(invalid) > 5 else ""
errors.append(
f"Invalid value for node attribute '{attr_name}' on nodes: "
f"{sample}{more}."
)
# Each district needs at least one candidate (warning, not error)
candidate_count = sum(
1 for n in g.nodes if g.nodes[n].get("candidate", 0)
)
if candidate_count == 0:
errors.append(
"No nodes are marked as facility candidates (all 'candidate' values "
"are 0 or False). At least one candidate is required for districting."
)
if strict and errors:
raise SchemaValidationError("\n".join(errors))
return errors
[docs]
def describe_schema() -> str:
"""Return a human-readable description of the schema."""
lines = ["FalcomChain Graph Attribute Schema", "=" * 40, ""]
lines.append("NODE ATTRIBUTES:")
for name, spec in NODE_ATTRIBUTES.items():
req = "REQUIRED" if spec.required else "optional"
default = "" if spec.default is None else f", default={spec.default}"
lines.append(f" {name} ({spec.type.__name__}, {req}{default})")
lines.append(f" {spec.purpose}")
lines.append("")
lines.append("EDGE ATTRIBUTES:")
for name, spec in EDGE_ATTRIBUTES.items():
req = "REQUIRED" if spec.required else "optional"
default = "" if spec.default is None else f", default={spec.default}"
lines.append(f" {name} ({spec.type.__name__}, {req}{default})")
lines.append(f" {spec.purpose}")
lines.append("")
lines.append("GRAPH-LEVEL ATTRIBUTES (graph.graph[...]):")
for name, spec in GRAPH_ATTRIBUTES.items():
req = "REQUIRED" if spec.required else "optional"
lines.append(f" {name} ({spec.type.__name__}, {req})")
lines.append(f" {spec.purpose}")
lines.append("")
return "\n".join(lines)