Color rxp_phart DAG nodes by pipeline group

[Copilot]

Add sub-pipeline coloring to rxp_phart using _rixpress/dag.json metadata, defaulting to uncolored labels when no pipeline color is present.

• Pipeline color injection

  • rxp_phart now reads pipeline colors from dag.json and applies ANSI color codes directly to DOT labels before rendering.
  • Missing or invalid pipeline color metadata falls back to original label text.

• Shared DOT handling

  • Centralized DOT file validation in a helper for consistent error handling.

• Tests

  • Added a focused test that validates ANSI-colored labels are injected in the DOT parsing path.
  • Test output includes a deterministic ASCII preview so the visualization appears in GitHub Actions logs.

Example (colored label injection):

"alpha" [label="\033[38;2;230;159;0malpha\033[0m"];

Original prompt

Absolutely. Here's a concise, technical implementation document for adding sub-pipeline coloring to rxp_phart in ryxpress. I’ve written it as a professional spec suitable for your engineers.

---

Implementation Spec: Sub-Pipeline Coloring for rxp_phart

Objective

Enhance rxp_phart to display pipeline derivation nodes colored by sub-pipeline group, leveraging metadata stored in _rixpress/dag.json. This aligns with existing visualization patterns in rixpress (e.g., rxp_ggdag(color_by="pipeline")) and improves readability of large pipelines in ASCII DAGs.

---

Background

• rxp_phart(dot_path: str) renders a DOT file as ASCII using phart, showing node labels.

• Nodes generated by rxp_pipeline() retain metadata:

  • pipeline_group: descriptive name of the sub-pipeline.
  • pipeline_color: assigned color (CSS/hex) for the sub-pipeline.

• maybe_color(node_name: str, suffix: str = '') can wrap labels in ANSI color codes.

By injecting pipeline_color into node labels before rendering, rxp_phart can display nodes with colors corresponding to their sub-pipeline.

---

Design

1. Data Flow

dag.json → extract node colors → modify DOT labels → rxp_phart → colored ASCII DAG

• Read _rixpress/dag.json using get_nodes_edges().
• Map node names → pipeline_color.
• Read DOT file generated by rxp_dag_for_ci().
• Replace node labels in DOT with colored labels using maybe_color().
• Pass the modified DOT to rxp_phart() for rendering.

---

2. Function Signature

def rxp_phart_by_pipeline(dot_path: str = "_rixpress/dag.dot") -> None:
    """
    Render a DOT graph file as an ASCII diagram, coloring nodes by sub-pipeline.

    Args:
        dot_path (str): Path to the DOT file (default "_rixpress/dag.dot").
    """

---

3. Steps

1. Load DAG metadata

from ryxpress import get_nodes_edges

dag = get_nodes_edges("_rixpress/dag.json")
nodes = dag["nodes"]
node_colors = {n["name"]: n.get("pipeline_color", "") for n in nodes}

2. Read DOT file

with open(dot_path, "r") as f:
    lines = f.readlines()

3. Inject ANSI colors

from ryxpress import maybe_color

colored_lines = []
for line in lines:
    for node_name, color in node_colors.items():
        if node_name in line and color:
            line = line.replace(node_name, maybe_color(node_name, color))
    colored_lines.append(line)

4. Write back modified DOT

with open(dot_path, "w") as f:
    f.writelines(colored_lines)

5. Render ASCII DAG

from ryxpress import rxp_phart

rxp_phart(dot_path)

---

4. Considerations

• ANSI support: Ensure the terminal supports ANSI codes.
• Color conflicts: If pipeline_color is missing, fallback to default maybe_color() behavior.
• Non-intrusive: Do not alter original DOT files permanently; optionally, work on a temporary copy.
• Performance: For very large pipelines, consider building a regex-based replacement for efficiency.

---

5. Optional Enhancements

• Dual encoding: Keep node type coloring in addition to pipeline border (requires DOT manipulation with label and style attributes).

• CLI flag: Add --color-by-pipeline flag to rxp_phart for backward compatibility.

• Unit tests: Test with:

  • Nodes with/without pipeline_color
  • Multiple sub-pipelines
  • Nested/linear DAGs

---

6. Deliverables

• rxp_phart_by_pipeline(dot_path: str = "_rixpress/dag.dot")
• Unit tests verifying correct coloring
• Optional CLI integration

---

This spec ensures engineers can implement sub-pipeline coloring cleanly, with minimal disruption to the existing rxp_phart pipeline.

---

I can also draft a ready-to-commit Python implementation for this spec that could go straight into ryxpress if you want. Do you want me to do that?

---

💬 We'd love your input! Share your thoughts on Copilot coding agent in our 2 minute survey.

[b-rodrigues]

@copilot would it be possibl eto see the output of the test to view the visualisation of the pipeline in the github actions view?

[Copilot]

@copilot would it be possibl eto see the output of the test to view the visualisation of the pipeline in the github actions view?

I added a deterministic ASCII preview to the rxp_phart coloring test and ensured it prints to the test log so it shows up in GitHub Actions output. You can see it in the test logs from commit ed7709f (colored alpha with a beta child).