aat.core

30class CitableToken(BaseModel):
31    """One token belonging to a citable passage.
32
33    `id` is unique *within* `context`, not globally -- aat-model.md's own
34    worked example uses a per-context sequence like 't1', 't2', ...,
35    starting over for each new context. A caller working across multiple
36    contexts should treat (context, id) as the real compound key, not id
37    alone -- see AATGraph.by_id().
38    """
39
40    context: str = Field(description="Context reference this token belongs to, matching its source CitedPassage.")
41    id: str = Field(description="Token id, unique within `context` (e.g. 't1', 't2', ...).")
42    value: str = Field(description="The token's own string value (its surface text).")

20class CitedPassage(BaseModel):
21    """One citable passage of text: a reference identifying its context
22    (aat-model.md's example is a CTS URN, but any stable identifier a
23    caller chooses works -- nothing here assumes CTS specifically) paired
24    with the passage's own raw text."""
25
26    context: str = Field(description="Reference identifying this passage's context, e.g. a CTS URN.")
27    text: str = Field(description="This passage's raw text, exactly as written.")

 73class AATGraph(BaseModel):
 74    """A full AAT graph for one or more citable passages: an ordered list
 75    of AATNode. Node order is not semantically significant -- the
 76    convenience accessors below don't depend on it -- but callers that
 77    built the graph from a specific passage will typically find document
 78    order convenient for display."""
 79
 80    nodes: List[AATNode] = Field(default_factory=list)
 81
 82    def by_id(self, context: str, id: str) -> Optional[AATNode]:
 83        """The node with this (context, id), or None. `id` alone isn't a
 84        reliable key across contexts (CitableToken/AATNode ids are only
 85        unique *within* one context -- see CitableToken's docstring), so
 86        both are required here."""
 87        for node in self.nodes:
 88            if node.context == context and node.id == id:
 89                return node
 90        return None
 91
 92    def actions(self) -> List[AATNode]:
 93        """Every action node, in list order."""
 94        return [n for n in self.nodes if n.role == "action"]
 95
 96    def agents(self) -> List[AATNode]:
 97        """Every agent node, in list order."""
 98        return [n for n in self.nodes if n.role == "agent"]
 99
100    def targets(self) -> List[AATNode]:
101        """Every target node, in list order."""
102        return [n for n in self.nodes if n.role == "target"]
103
104    def agents_for(self, action: AATNode) -> List[AATNode]:
105        """Every agent node whose related_node points at `action`'s id,
106        within the same context."""
107        return [
108            n for n in self.nodes
109            if n.role == "agent" and n.context == action.context and n.related_node == action.id
110        ]
111
112    def targets_for(self, action: AATNode) -> List[AATNode]:
113        """Every target node whose related_node points at `action`'s id,
114        within the same context."""
115        return [
116            n for n in self.nodes
117            if n.role == "target" and n.context == action.context and n.related_node == action.id
118        ]
119
120    def governing_action(self, action: AATNode) -> Optional[AATNode]:
121        """For a dependent action node, the action node it's subordinate
122        to; None if `action` is independent (related_node is None) or if
123        related_node doesn't resolve to any node in this graph."""
124        if action.related_node is None:
125            return None
126        return self.by_id(action.context, action.related_node)

24class AATNode(BaseModel):
25    """One node in an AAT graph.
26
27    `context` and `id` identify the token this node is built from -- for a
28    single-token action, an agent, or a target, `id` is just that token's
29    own id. For a *compound* action (aat-model.md's "Actions" section --
30    e.g. "was eating"), `id` is instead the id of the *principal verb*
31    token within the compound (e.g. the token for "eating"), and `value`
32    is the space-joined text of every component token in surface order
33    (e.g. "was eating"), not just the principal verb's own text.
34
35    `related_node` is:
36      - `None`, for an *independent* action node (aat-model.md's
37        "Actions" section);
38      - another action node's `id`, for a *dependent* action node (the id
39        of the governing action it depends on);
40      - another action node's `id`, for an agent or target node (the id
41        of the action it's the agent/target of) -- agent and target nodes
42        always have a related_node; only an action node's related_node
43        can be `None`.
44    """
45
46    context: str = Field(description="Context reference this node belongs to.")
47    id: str = Field(
48        description=(
49            "Token id this node is built from -- the principal verb's id "
50            "for a compound action, otherwise the single underlying "
51            "token's id."
52        )
53    )
54    value: str = Field(
55        description=(
56            "This node's string value -- a single token's own text, or, "
57            "for a compound action, every component token's text joined "
58            "by spaces in surface order."
59        )
60    )
61    role: Role = Field(description="'agent', 'action', or 'target'.")
62    related_node: Optional[str] = Field(
63        default=None,
64        description=(
65            "For an action node: the id of the governing action node, if "
66            "this action is dependent/subordinate; None if independent. "
67            "For an agent or target node: the id of the action node it "
68            "relates to (always set)."
69        ),
70    )

16def validate(tokens: List[CitableToken], graph: AATGraph) -> List[str]:
17    """Return a list of human-readable problem descriptions found in
18    `graph`, given the `tokens` it was built from. An empty list means no
19    referential problems were found (not that the analysis is correct).
20
21    Checks:
22      - every node's (context, id) matches a real token in `tokens`;
23      - every node's `value` is non-empty;
24      - an action node's `related_node`, if set, resolves to another
25        action node's id in the same context;
26      - an agent or target node's `related_node` is set (not None) and
27        resolves to an action node's id in the same context.
28    """
29    problems: List[str] = []
30
31    token_ids = {(t.context, t.id) for t in tokens}
32    action_ids = {(n.context, n.id) for n in graph.nodes if n.role == "action"}
33
34    for node in graph.nodes:
35        if (node.context, node.id) not in token_ids:
36            problems.append(
37                f"node ({node.context!r}, {node.id!r}, role={node.role!r}) "
38                "does not match any token in the input token list"
39            )
40        if not node.value:
41            problems.append(
42                f"node ({node.context!r}, {node.id!r}) has an empty value"
43            )
44
45        if node.role == "action":
46            if node.related_node is not None and (node.context, node.related_node) not in action_ids:
47                problems.append(
48                    f"action node ({node.context!r}, {node.id!r}) has related_node "
49                    f"{node.related_node!r}, which is not the id of any action node "
50                    "in this context"
51                )
52        else:  # agent or target
53            if node.related_node is None:
54                problems.append(
55                    f"{node.role} node ({node.context!r}, {node.id!r}) has no "
56                    "related_node -- every agent/target node must relate to an "
57                    "action node"
58                )
59            elif (node.context, node.related_node) not in action_ids:
60                problems.append(
61                    f"{node.role} node ({node.context!r}, {node.id!r}) has related_node "
62                    f"{node.related_node!r}, which is not the id of any action node "
63                    "in this context"
64                )
65
66    return problems

169def serialize_nodes(nodes: List[AATNode]) -> str:
170    """Render `nodes` as one '#!aatnodes' block, pipe-delimited, and
171    return it as a string. Does not check for '|' inside any field's own
172    value -- there is no escaping mechanism, so avoid the delimiter
173    character in `value` if you plan to round-trip through this format."""
174    lines = [AATNODES_LABEL, _AATNODES_HEADER]
175    for node in nodes:
176        related = node.related_node or ""
177        lines.append(f"{node.context}|{node.id}|{node.value}|{node.role}|{related}")
178    return "\n".join(lines) + "\n"

181def write_nodes(nodes: List[AATNode], path: str) -> None:
182    """Write serialize_nodes(nodes) to `path`."""
183    with open(path, "w", encoding="utf-8") as f:
184        f.write(serialize_nodes(nodes))

187def read_nodes(path: str) -> List[AATNode]:
188    """Read every '#!aatnodes' block in `path` and return their rows,
189    concatenated in file order, as a list of AATNode -- other block
190    types in the same file (e.g. a '#!tokens' block) are ignored, see
191    _read_blocks().
192
193    Raises ValueError (via _read_blocks()) for a malformed file -- see
194    that function's own docstring for the exact cases. Also raises
195    ValueError (not returning an empty list) if the file has no
196    '#!aatnodes' block at all, so a caller can't mistake "wrong file" for
197    "file with zero nodes".
198    """
199    rows, seen = _read_blocks(path, AATNODES_LABEL, _AATNODES_HEADER)
200    if not seen:
201        raise ValueError(f"file has no {AATNODES_LABEL!r} block")
202
203    nodes: List[AATNode] = []
204    for context, id_, value, role, related in rows:
205        nodes.append(
206            AATNode(
207                context=context,
208                id=id_,
209                value=value,
210                role=role,
211                related_node=related or None,
212            )
213        )
214    return nodes

217def read_graph(path: str) -> AATGraph:
218    """Convenience wrapper: read_nodes(path) wrapped as an AATGraph."""
219    return AATGraph(nodes=read_nodes(path))

222def serialize_tokens(tokens: List[CitableToken]) -> str:
223    """Render `tokens` as one '#!tokens' block, pipe-delimited, and
224    return it as a string, in `tokens`' own order -- unlike
225    serialize_nodes(), row order here is meaningful (it's the only thing
226    that records reading order; a token's `id` alone doesn't, especially
227    for a composite sentence-spanning id like "1.14.t3" -- see this
228    module's own top docstring). Same no-escaping caveat as
229    serialize_nodes(): avoid '|' in `value` if you plan to round-trip
230    through this format."""
231    lines = [TOKENS_LABEL, _TOKENS_HEADER]
232    for token in tokens:
233        lines.append(f"{token.context}|{token.id}|{token.value}")
234    return "\n".join(lines) + "\n"

237def write_tokens(tokens: List[CitableToken], path: str) -> None:
238    """Write serialize_tokens(tokens) to `path`."""
239    with open(path, "w", encoding="utf-8") as f:
240        f.write(serialize_tokens(tokens))

243def read_tokens(path: str) -> List[CitableToken]:
244    """Read every '#!tokens' block in `path` and return their rows,
245    concatenated in file order (which is also reading order -- see
246    serialize_tokens()'s own docstring), as a list of CitableToken --
247    other block types in the same file (e.g. a '#!aatnodes' block) are
248    ignored, see _read_blocks().
249
250    Raises ValueError (via _read_blocks()) for a malformed file. Also
251    raises ValueError if the file has no '#!tokens' block at all, same
252    reasoning as read_nodes()."""
253    rows, seen = _read_blocks(path, TOKENS_LABEL, _TOKENS_HEADER)
254    if not seen:
255        raise ValueError(f"file has no {TOKENS_LABEL!r} block")
256    return [CitableToken(context=context, id=id_, value=value) for context, id_, value in rows]

259def serialize_analysis(tokens: List[CitableToken], graph: AATGraph) -> str:
260    """Render a complete, re-displayable analysis as one string: a
261    '#!tokens' block (the complete, already-tokenized input every node
262    in `graph` refers back to -- no re-tokenization needed to reload it,
263    see this module's own top docstring) followed by a '#!aatnodes'
264    block (the AAT graph an earlier, LM-dependent analysis step
265    produced). write_analysis() is a thin wrapper that writes this same
266    string to a file; call this directly instead when the text itself is
267    what's wanted -- e.g. to hand to a UI that writes the file somewhere
268    else (a chosen directory, a download, ...), or to embed the analysis
269    in something larger without touching disk here at all. read_analysis()
270    is the matching reader for a file written either way -- see that
271    function's own docstring for the intended round trip."""
272    return serialize_tokens(tokens) + "\n" + serialize_nodes(graph.nodes)

275def write_analysis(tokens: List[CitableToken], graph: AATGraph, path: str) -> None:
276    """Write serialize_analysis(tokens, graph) to `path`. See that
277    function's own docstring for exactly what gets written, and
278    read_analysis() for the matching reader."""
279    with open(path, "w", encoding="utf-8") as f:
280        f.write(serialize_analysis(tokens, graph))

283def read_analysis(path: str) -> Tuple[List[CitableToken], AATGraph]:
284    """Read a file written by write_analysis(): both its '#!tokens'
285    block (as a list of CitableToken, in reading order) and its
286    '#!aatnodes' block (as an AATGraph). Raises ValueError if either
287    block is missing -- see read_tokens()/read_nodes().
288
289    Unlike the earlier '#!passages'-based format this replaced, the
290    returned `tokens` are already exactly what the original analysis
291    used -- e.g. straight into aat.core.tokens_to_html(tokens,
292    graph=graph) (also re-exported as aat.english.tokens_to_html) -- no
293    re-tokenization step, and so no dependency on
294    aat.english.tokenize() (or aat.english.tokenize_corpus_by_sentence()
295    for a sentence-spanning file) at all. This whole round trip still
296    needs no LM access at any point; it just no longer needs to re-run
297    any deterministic code either.
298    """
299    tokens = read_tokens(path)
300    graph = read_graph(path)
301    return tokens, graph

119def tokens_to_html(tokens: List[CitableToken], graph: Optional[AATGraph] = None) -> str:
120    """Render `tokens` as one continuous HTML string: ordinary words get
121    a space before them, punctuation attaches per this module's own
122    docstring (left-joining by default; opening brackets and the first
123    of a quote pair right-join instead).
124
125    If `graph` is given, every token whose (context, id) matches a node
126    in `graph` is highlighted: background/text colored the same way
127    that node is colored in `aat.core.mermaid.graph_to_mermaid()`'s
128    diagram (via `aat.core.coloring.assign_action_colors()`), with a
129    role-specific border -- a box around an `action` token, a rounded
130    box around an `agent` token, an underline under a `target` token.
131    Omit `graph` (or pass `None`) for plain, unhighlighted text.
132
133    Every token's text is HTML-escaped before being emitted, span or
134    not -- real passage text can contain a literal `<`, `>`, `&`, or
135    quote character, which would otherwise be indistinguishable from
136    markup to anything that re-parses this output.
137    """
138    color_of_node: Dict[Tuple[str, str], Tuple[str, str, str]] = {}
139    role_of_node: Dict[Tuple[str, str], str] = {}
140    if graph is not None:
141        color_of_node, _warnings = assign_action_colors(graph)
142        role_of_node = {(n.context, n.id): n.role for n in graph.nodes}
143
144    quote_counts: Dict[str, int] = {}
145    pieces: List[str] = []
146    previous_class: Optional[str] = None
147
148    for token in tokens:
149        cls = _classify(token, quote_counts)
150        rendered = html.escape(token.value)
151
152        key = (token.context, token.id)
153        color = color_of_node.get(key)
154        if color is not None:
155            fill, stroke, text_color = color
156            role = role_of_node.get(key, "")
157            role_style = _ROLE_STYLE.get(role, "").format(stroke=stroke)
158            rendered = (
159                f'<span style="background-color: {fill}; color: {text_color}; {role_style}">'
160                f"{rendered}</span>"
161            )
162
163        if not pieces:
164            # Nothing precedes the first token -- never prepend a space.
165            pieces.append(rendered)
166        elif cls == _LEFT:
167            pieces.append(rendered)
168        elif cls == _RIGHT:
169            pieces.append(" " + rendered)
170        else:  # _NORMAL
171            if previous_class == _RIGHT:
172                pieces.append(rendered)
173            else:
174                pieces.append(" " + rendered)
175
176        previous_class = cls
177
178    return "".join(pieces)

 76def graph_to_mermaid(
 77    graph: AATGraph,
 78    orientation: str = "BT",
 79    color_by_action: bool = True,
 80) -> Tuple[str, List[str]]:
 81    """Build a Mermaid `graph` diagram from an AATGraph.
 82
 83    `orientation` is Mermaid's own flowchart direction code -- `BT`
 84    (bottom-to-top, the default here), `TB`/`TD` (top-down -- synonyms),
 85    `LR`, or `RL` -- used verbatim (uppercased) in the diagram's opening
 86    line (`graph BT`, etc.). Validated by aat.core.orientation (shared
 87    with graph_to_dot()); anything else raises `ValueError` naming the
 88    valid options, rather than silently producing invalid Mermaid syntax.
 89    See https://mermaid.js.org/syntax/flowchart.html#direction.
 90
 91    `color_by_action` (default True) -- see this module's own docstring.
 92    Pass False for a plain, uncolored diagram.
 93
 94    Returns (diagram_text, warnings). A node whose `related_node` doesn't
 95    resolve to any node actually present in `graph` (same context) is
 96    still drawn, but its edge is skipped and reported as a warning --
 97    normally a sign `graph` failed aat.core.validate.validate() upstream,
 98    worth checking there first. If `color_by_action` is True and the graph
 99    has more distinct actions than the palette has colors (currently 8),
100    one warning notes that colors repeat.
101    """
102    orientation = normalize_orientation(orientation)
103
104    by_key: Dict[Tuple[str, str], AATNode] = {_node_key(n): n for n in graph.nodes}
105
106    lines = [f"graph {orientation}"]
107    for node in graph.nodes:
108        open_b, close_b = _ROLE_BRACKETS.get(node.role, ("[", "]"))
109        lines.append(f'    {node.id}{open_b}"{_escape_label(node.value)}"{close_b}')
110
111    warnings: List[str] = []
112    for node in graph.nodes:
113        if node.related_node is None:
114            continue
115        target_key = (node.context, node.related_node)
116        if target_key not in by_key:
117            warnings.append(
118                f"skipped edge {node.id} -[{node.role}]-> {node.related_node}: "
119                "target is not a node in this graph"
120            )
121            continue
122        edge_label = "dependent" if node.role == "action" else node.role
123        lines.append(f"    {node.id} -->|{edge_label}| {node.related_node}")
124
125    if color_by_action:
126        color_of_node, color_warnings = assign_action_colors(graph)
127        warnings.extend(color_warnings)
128
129        if color_of_node:
130            lines.append("")
131
132            # Group by color VALUE, not by action -- two actions that
133            # cycle to the same palette color (more actions than the
134            # palette has slots) share a single classDef/class pair here
135            # rather than emitting two redundant classes with identical
136            # colors.
137            colors_in_order: List[ColorTriple] = []
138            seen_colors = set()
139            for node in graph.nodes:
140                color = color_of_node.get(_node_key(node))
141                if color is not None and color not in seen_colors:
142                    seen_colors.add(color)
143                    colors_in_order.append(color)
144
145            class_name_of_color: Dict[ColorTriple, str] = {
146                color: f"c{i}" for i, color in enumerate(colors_in_order)
147            }
148            for color in colors_in_order:
149                fill, stroke, text = color
150                lines.append(
151                    f"    classDef {class_name_of_color[color]} "
152                    f"fill:{fill},stroke:{stroke},color:{text};"
153                )
154
155            nodes_by_class: Dict[str, List[str]] = {}
156            for node in graph.nodes:
157                color = color_of_node.get(_node_key(node))
158                if color is None:
159                    continue
160                class_name = class_name_of_color[color]
161                nodes_by_class.setdefault(class_name, []).append(node.id)
162
163            for class_name, node_ids in nodes_by_class.items():
164                lines.append(f"    class {','.join(node_ids)} {class_name};")
165
166    return "\n".join(lines), warnings

169def save_mermaid(
170    graph: AATGraph,
171    path: str,
172    orientation: str = "BT",
173    color_by_action: bool = True,
174) -> List[str]:
175    """Write the diagram to `path` (e.g. 'analysis.mmd') and return any
176    warnings from graph_to_mermaid(). `orientation` is validated the same
177    way -- see graph_to_mermaid()'s own docstring."""
178    diagram, warnings = graph_to_mermaid(graph, orientation=orientation, color_by_action=color_by_action)
179    with open(path, "w", encoding="utf-8") as f:
180        f.write(diagram + "\n")
181    return warnings

 82def graph_to_dot(
 83    graph: AATGraph,
 84    orientation: str = "BT",
 85    color_by_action: bool = True,
 86) -> Tuple[str, List[str]]:
 87    """Build a Graphviz DOT digraph from an AATGraph -- see this module's
 88    own docstring for the node-shape/edge/coloring mapping, which mirrors
 89    aat.core.mermaid.graph_to_mermaid() exactly except for the target
 90    syntax.
 91
 92    `orientation` is validated exactly like graph_to_mermaid()'s (see
 93    aat.core.orientation) -- `BT` (bottom-to-top, the default here),
 94    `TB`/`TD` (top-down -- synonyms), `LR`, or `RL`. Written out as
 95    Graphviz's own `rankdir` graph attribute; since `rankdir` has no
 96    "TD" of its own, "TD" is mapped to "TB" here (same direction, just
 97    Graphviz's own name for it) -- every other value is used verbatim.
 98    Anything outside that vocabulary raises `ValueError` naming the
 99    valid options, rather than silently producing invalid DOT syntax.
100
101    `color_by_action` (default True) -- see this module's own docstring.
102    Pass False for a plain, uncolored digraph.
103
104    Returns (dot_text, warnings) -- same warning cases as
105    graph_to_mermaid(): a node whose `related_node` doesn't resolve to
106    another node actually present in `graph` (same context) is still
107    drawn, but its edge is skipped and reported as a warning; and, if
108    `color_by_action` is True and the graph has more distinct actions
109    than the palette has colors (currently 8), one warning notes that
110    colors repeat.
111    """
112    orientation = normalize_orientation(orientation)
113    rankdir = "TB" if orientation == "TD" else orientation
114
115    by_key: Dict[Tuple[str, str], AATNode] = {_node_key(n): n for n in graph.nodes}
116
117    color_of_node: Dict[Tuple[str, str], ColorTriple] = {}
118    warnings: List[str] = []
119    if color_by_action:
120        color_of_node, color_warnings = assign_action_colors(graph)
121        warnings.extend(color_warnings)
122
123    lines = ["digraph aat {", f"    rankdir={rankdir};"]
124
125    for node in graph.nodes:
126        attrs = [f"shape={_ROLE_SHAPE.get(node.role, 'box')}"]
127        styles = list(_ROLE_EXTRA_STYLES.get(node.role, []))
128
129        color = color_of_node.get(_node_key(node))
130        if color is not None:
131            fill, stroke, text = color
132            styles.append("filled")
133            attrs.append(f'fillcolor="{fill}"')
134            attrs.append(f'color="{stroke}"')
135            attrs.append(f'fontcolor="{text}"')
136
137        if styles:
138            attrs.append(f'style="{",".join(styles)}"')
139        attrs.append(f'label="{_escape_label(node.value)}"')
140
141        lines.append(f'    {node.id} [{", ".join(attrs)}];')
142
143    for node in graph.nodes:
144        if node.related_node is None:
145            continue
146        target_key = (node.context, node.related_node)
147        if target_key not in by_key:
148            warnings.append(
149                f"skipped edge {node.id} -[{node.role}]-> {node.related_node}: "
150                "target is not a node in this graph"
151            )
152            continue
153        edge_label = "dependent" if node.role == "action" else node.role
154        lines.append(f'    {node.id} -> {node.related_node} [label="{edge_label}"];')
155
156    lines.append("}")
157    return "\n".join(lines), warnings

160def save_dot(
161    graph: AATGraph,
162    path: str,
163    orientation: str = "BT",
164    color_by_action: bool = True,
165) -> List[str]:
166    """Write the digraph to `path` (e.g. 'analysis.dot') and return any
167    warnings from graph_to_dot(). `orientation` is validated the same
168    way -- see graph_to_dot()'s own docstring."""
169    dot_text, warnings = graph_to_dot(graph, orientation=orientation, color_by_action=color_by_action)
170    with open(path, "w", encoding="utf-8") as f:
171        f.write(dot_text + "\n")
172    return warnings

48def assign_action_colors(graph: AATGraph) -> Tuple[Dict[NodeKey, ColorTriple], List[str]]:
49    """Return (color_of_node, warnings).
50
51    `color_of_node` maps every node's (context, id) to a (fill, stroke,
52    text) hex triple, for every node that clusters with some action: an
53    action node itself (dependent or not -- see this module's own
54    docstring), or an agent/target node whose `related_node` resolves to
55    an action node in the same context. A node with no resolvable cluster
56    (e.g. an agent/target with a broken related_node -- shouldn't happen
57    in a graph that's passed aat.core.validate.validate()) is simply
58    absent from the mapping.
59
60    Colors are assigned to actions in the order they first appear in
61    `graph.nodes`, cycling through the palette (currently 8 colors) if
62    there are more distinct actions than that -- in which case two
63    distinct actions (and their own agent/target nodes) end up sharing the
64    identical color. `warnings` has one entry naming this when it happens;
65    otherwise it's empty.
66    """
67    actions = [n for n in graph.nodes if n.role == "action"]
68
69    warnings: List[str] = []
70    if len(actions) > len(_ACTION_COLOR_PALETTE):
71        warnings.append(
72            f"{len(actions)} actions but only {len(_ACTION_COLOR_PALETTE)} palette "
73            "colors -- colors repeat and some distinct actions will share a color"
74        )
75
76    action_index: Dict[NodeKey, int] = {_node_key(a): i for i, a in enumerate(actions)}
77
78    by_key: Dict[NodeKey, AATNode] = {_node_key(n): n for n in graph.nodes}
79    cluster_of_node: Dict[NodeKey, NodeKey] = {}
80    for node in graph.nodes:
81        key = _node_key(node)
82        if node.role == "action":
83            cluster_of_node[key] = key
84        elif node.related_node is not None:
85            related_key = (node.context, node.related_node)
86            if related_key in by_key:
87                cluster_of_node[key] = related_key
88
89    color_of_node: Dict[NodeKey, ColorTriple] = {}
90    for node in graph.nodes:
91        key = _node_key(node)
92        cluster_key = cluster_of_node.get(key)
93        if cluster_key is None or cluster_key not in action_index:
94            continue
95        color_of_node[key] = _ACTION_COLOR_PALETTE[action_index[cluster_key] % len(_ACTION_COLOR_PALETTE)]
96
97    return color_of_node, warnings

39def parse_cex_ctsdata(text: str, delimiter: str = "#") -> List[CitedPassage]:
40    """Parse every '#!ctsdata' block in `text` (a CEX file's own full
41    contents, already read into a string) and return its rows,
42    concatenated in file order, as a list of CitedPassage -- `context` is
43    each row's own CTS URN, `text` is its citable text. Every other CEX
44    block type is ignored entirely, whether it appears before, between,
45    or after the '#!ctsdata' block(s); multiple '#!ctsdata' blocks in one
46    file are concatenated, in file order, same as
47    aat.core.serialization's own multi-block handling.
48
49    Raises ValueError, naming the offending line, for a '#!ctsdata' row
50    that has no `delimiter` in it at all -- most often a sign `delimiter`
51    is wrong for this particular file. Raises ValueError (not returning
52    an empty list) if the file has no '#!ctsdata' block at all, so a
53    caller can't mistake "wrong file" for "file with zero passages" --
54    same reasoning as aat.core.serialization.read_nodes()/read_tokens().
55    """
56    passages: List[CitedPassage] = []
57    in_ctsdata = False
58    seen = False
59
60    for line_no, raw_line in enumerate(text.splitlines(), start=1):
61        line = raw_line.strip("\r\n")
62        if line == "" or line.startswith("//"):
63            continue
64
65        if line.startswith("#!"):
66            in_ctsdata = line == CTSDATA_LABEL
67            if in_ctsdata:
68                seen = True
69            continue
70
71        if not in_ctsdata:
72            continue
73
74        urn, sep, passage_text = line.partition(delimiter)
75        if not sep:
76            raise ValueError(
77                f"line {line_no}: no {delimiter!r} delimiter found in a "
78                f"{CTSDATA_LABEL!r} row -- wrong --delimiter for this file? {line!r}"
79            )
80        passages.append(CitedPassage(context=urn, text=passage_text))
81
82    if not seen:
83        raise ValueError(f"file has no {CTSDATA_LABEL!r} block")
84
85    return passages

88def read_cex_passages(path: str, delimiter: str = "#") -> List[CitedPassage]:
89    """Read `path` as a CEX file and return parse_cex_ctsdata() of its
90    contents -- see that function's own docstring for the exact format
91    and error cases."""
92    with open(path, "r", encoding="utf-8") as f:
93        text = f.read()
94    return parse_cex_ctsdata(text, delimiter=delimiter)