Skip to content

gds.query

Query engine for exploring GDSSpec structure.

Source code in packages/gds-framework/gds/query.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class SpecQuery:
    """Query engine for exploring GDSSpec structure."""

    def __init__(self, spec: GDSSpec) -> None:
        self.spec = spec

    def param_to_blocks(self) -> dict[str, list[str]]:
        """Map each parameter to the blocks that use it."""
        mapping: dict[str, list[str]] = {p: [] for p in self.spec.parameters}
        for bname, block in self.spec.blocks.items():
            if isinstance(block, HasParams):
                for param in block.params_used:
                    if param in mapping:
                        mapping[param].append(bname)
        return mapping

    def block_to_params(self) -> dict[str, list[str]]:
        """Map each block to the parameters it uses."""
        result: dict[str, list[str]] = {}
        for bname, block in self.spec.blocks.items():
            if isinstance(block, HasParams):
                result[bname] = list(block.params_used)
            else:
                result[bname] = []
        return result

    def entity_update_map(self) -> dict[str, dict[str, list[str]]]:
        """Map entity -> variable -> list of mechanisms that update it."""
        result: dict[str, dict[str, list[str]]] = {}
        for ename, entity in self.spec.entities.items():
            result[ename] = {vname: [] for vname in entity.variables}

        for bname, block in self.spec.blocks.items():
            if isinstance(block, Mechanism):
                for ename, vname in block.updates:
                    if ename in result and vname in result[ename]:
                        result[ename][vname].append(bname)
        return result

    def dependency_graph(self) -> dict[str, set[str]]:
        """Full block dependency DAG (who feeds whom) from all wirings."""
        adj: dict[str, set[str]] = defaultdict(set)
        for wiring in self.spec.wirings.values():
            for wire in wiring.wires:
                adj[wire.source].add(wire.target)
        return dict(adj)

    def blocks_by_kind(self) -> dict[str, list[str]]:
        """Group blocks by their GDS role (kind)."""
        result: dict[str, list[str]] = {
            "boundary": [],
            "control": [],
            "policy": [],
            "mechanism": [],
            "generic": [],
        }
        for bname, block in self.spec.blocks.items():
            kind = getattr(block, "kind", "generic")
            if kind in result:
                result[kind].append(bname)
            else:
                result[kind] = [bname]
        return result

    def blocks_affecting(self, entity: str, variable: str) -> list[str]:
        """Which blocks can transitively affect this variable?

        Finds all mechanisms that directly update the variable, then
        all blocks that can transitively reach those mechanisms.
        """
        direct: list[str] = []
        for bname, block in self.spec.blocks.items():
            if isinstance(block, Mechanism) and (entity, variable) in block.updates:
                direct.append(bname)

        adj: dict[str, set[str]] = defaultdict(set)
        for wiring in self.spec.wirings.values():
            for wire in wiring.wires:
                adj[wire.source].add(wire.target)

        all_affecting: set[str] = set(direct)
        for mech_name in direct:
            for bname in self.spec.blocks:
                if self._can_reach(adj, bname, mech_name):
                    all_affecting.add(bname)

        return sorted(all_affecting)

    def admissibility_dependency_map(self) -> dict[str, list[tuple[str, str]]]:
        """Map boundary block -> state variables constraining its inputs."""
        result: dict[str, list[tuple[str, str]]] = {}
        for ac in self.spec.admissibility_constraints.values():
            result.setdefault(ac.boundary_block, []).extend(ac.depends_on)
        return result

    def mechanism_read_map(self) -> dict[str, list[tuple[str, str]]]:
        """Map mechanism -> state variables it reads."""
        return {
            mname: list(ts.reads)
            for mname, ts in self.spec.transition_signatures.items()
        }

    def variable_readers(self, entity: str, variable: str) -> list[str]:
        """Which mechanisms declare reading this state variable?"""
        ref = (entity, variable)
        return [
            mname
            for mname, ts in self.spec.transition_signatures.items()
            if ref in ts.reads
        ]

    @staticmethod
    def _can_reach(adj: dict[str, set[str]], source: str, target: str) -> bool:
        """BFS reachability check."""
        if source == target:
            return False
        visited: set[str] = set()
        queue = [source]
        while queue:
            current = queue.pop(0)
            if current == target:
                return True
            if current in visited:
                continue
            visited.add(current)
            queue.extend(adj.get(current, set()))
        return False

param_to_blocks()

Map each parameter to the blocks that use it.

Source code in packages/gds-framework/gds/query.py 
24
25
26
27
28
29
30
31
32
def param_to_blocks(self) -> dict[str, list[str]]:
    """Map each parameter to the blocks that use it."""
    mapping: dict[str, list[str]] = {p: [] for p in self.spec.parameters}
    for bname, block in self.spec.blocks.items():
        if isinstance(block, HasParams):
            for param in block.params_used:
                if param in mapping:
                    mapping[param].append(bname)
    return mapping

block_to_params()

Map each block to the parameters it uses.

Source code in packages/gds-framework/gds/query.py 
34
35
36
37
38
39
40
41
42
def block_to_params(self) -> dict[str, list[str]]:
    """Map each block to the parameters it uses."""
    result: dict[str, list[str]] = {}
    for bname, block in self.spec.blocks.items():
        if isinstance(block, HasParams):
            result[bname] = list(block.params_used)
        else:
            result[bname] = []
    return result

entity_update_map()

Map entity -> variable -> list of mechanisms that update it.

Source code in packages/gds-framework/gds/query.py 
44
45
46
47
48
49
50
51
52
53
54
55
def entity_update_map(self) -> dict[str, dict[str, list[str]]]:
    """Map entity -> variable -> list of mechanisms that update it."""
    result: dict[str, dict[str, list[str]]] = {}
    for ename, entity in self.spec.entities.items():
        result[ename] = {vname: [] for vname in entity.variables}

    for bname, block in self.spec.blocks.items():
        if isinstance(block, Mechanism):
            for ename, vname in block.updates:
                if ename in result and vname in result[ename]:
                    result[ename][vname].append(bname)
    return result

dependency_graph()

Full block dependency DAG (who feeds whom) from all wirings.

Source code in packages/gds-framework/gds/query.py 
57
58
59
60
61
62
63
def dependency_graph(self) -> dict[str, set[str]]:
    """Full block dependency DAG (who feeds whom) from all wirings."""
    adj: dict[str, set[str]] = defaultdict(set)
    for wiring in self.spec.wirings.values():
        for wire in wiring.wires:
            adj[wire.source].add(wire.target)
    return dict(adj)

blocks_by_kind()

Group blocks by their GDS role (kind).

Source code in packages/gds-framework/gds/query.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def blocks_by_kind(self) -> dict[str, list[str]]:
    """Group blocks by their GDS role (kind)."""
    result: dict[str, list[str]] = {
        "boundary": [],
        "control": [],
        "policy": [],
        "mechanism": [],
        "generic": [],
    }
    for bname, block in self.spec.blocks.items():
        kind = getattr(block, "kind", "generic")
        if kind in result:
            result[kind].append(bname)
        else:
            result[kind] = [bname]
    return result

blocks_affecting(entity, variable)

Which blocks can transitively affect this variable?

Finds all mechanisms that directly update the variable, then all blocks that can transitively reach those mechanisms.

Source code in packages/gds-framework/gds/query.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def blocks_affecting(self, entity: str, variable: str) -> list[str]:
    """Which blocks can transitively affect this variable?

    Finds all mechanisms that directly update the variable, then
    all blocks that can transitively reach those mechanisms.
    """
    direct: list[str] = []
    for bname, block in self.spec.blocks.items():
        if isinstance(block, Mechanism) and (entity, variable) in block.updates:
            direct.append(bname)

    adj: dict[str, set[str]] = defaultdict(set)
    for wiring in self.spec.wirings.values():
        for wire in wiring.wires:
            adj[wire.source].add(wire.target)

    all_affecting: set[str] = set(direct)
    for mech_name in direct:
        for bname in self.spec.blocks:
            if self._can_reach(adj, bname, mech_name):
                all_affecting.add(bname)

    return sorted(all_affecting)

admissibility_dependency_map()

Map boundary block -> state variables constraining its inputs.

Source code in packages/gds-framework/gds/query.py 
106
107
108
109
110
111
def admissibility_dependency_map(self) -> dict[str, list[tuple[str, str]]]:
    """Map boundary block -> state variables constraining its inputs."""
    result: dict[str, list[tuple[str, str]]] = {}
    for ac in self.spec.admissibility_constraints.values():
        result.setdefault(ac.boundary_block, []).extend(ac.depends_on)
    return result

mechanism_read_map()

Map mechanism -> state variables it reads.

Source code in packages/gds-framework/gds/query.py 
113
114
115
116
117
118
def mechanism_read_map(self) -> dict[str, list[tuple[str, str]]]:
    """Map mechanism -> state variables it reads."""
    return {
        mname: list(ts.reads)
        for mname, ts in self.spec.transition_signatures.items()
    }

variable_readers(entity, variable)

Which mechanisms declare reading this state variable?

Source code in packages/gds-framework/gds/query.py 
120
121
122
123
124
125
126
127
def variable_readers(self, entity: str, variable: str) -> list[str]:
    """Which mechanisms declare reading this state variable?"""
    ref = (entity, variable)
    return [
        mname
        for mname, ts in self.spec.transition_signatures.items()
        if ref in ts.reads
    ]