feat(datagen): add reserve grid positions and batch codes to lab config
- introduce reserve grid positions for QA witness pattern - add batch codes to run entries in lab configuration - update schedule with run start time and track interval - enhance validation checks for run-batch mapping
This commit is contained in:
administrator
285
make_process_flow.py
Normal file
285
make_process_flow.py
Normal file
@@ -0,0 +1,285 @@
|
||||
"""Task 02 - Process flow diagrams.
|
||||
|
||||
Reads ./out/config/lab_config.yaml and writes the Mermaid documentation of
|
||||
the laboratory workflow: ./out/report/process_flow.md plus one standalone
|
||||
./out/report/diagrams/D<n>.mermaid file per diagram (D1-D6), and the
|
||||
./out/.done/02.ok completion marker. Spec: docs/specs/02_process_flow_diagrams.md.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent
|
||||
TASK_ID = "02"
|
||||
DEPENDS_ON = ["01"]
|
||||
|
||||
DIAGRAM_TYPES = ("flowchart", "sequenceDiagram", "erDiagram")
|
||||
|
||||
|
||||
def d1_coupon_assembly(cfg: dict) -> str:
|
||||
sub = cfg["material_system"]["substrate"]
|
||||
dims = "x".join(str(d) for d in sub["dimensions_mm"])
|
||||
return f"""flowchart LR
|
||||
A([{sub['material']} Base {dims} mm]) --> B[Cleaning]
|
||||
B --> B1([Rinsed in deionized water])
|
||||
B1 --> B2([Sonicated in cleaning solution])
|
||||
B --> C[Smearing Adhesive]
|
||||
C --> C1([{cfg['material_system']['adhesion_layer']['material']} adhesive coating])
|
||||
C --> D[Vapor Deposition System<br/>Kurt J. Lesker PVD 200]
|
||||
D --> D1([PtAu sputter coating, gradient])
|
||||
D --> E([Test Coupon])
|
||||
"""
|
||||
|
||||
|
||||
def d2_characterization_batch_assembly(cfg: dict) -> str:
|
||||
h = cfg["hierarchy"]
|
||||
t_lo, t_hi = cfg["material_system"]["coating"]["thickness_um_range"]
|
||||
lines = [
|
||||
"flowchart LR",
|
||||
" TC([Test Coupon]) --> S[SIMTRA simulation<br/>composition / atom energies]",
|
||||
f" S --> OP[Optical profilometry<br/>film thickness {t_lo}-{t_hi} um]",
|
||||
f" OP --> TW[Assemble Test Wafer<br/>x{h['coupons_per_wafer']} coupons]",
|
||||
f" TW --> TB[Assemble Test Batch<br/>x{h['wafers_per_batch']} wafers]",
|
||||
]
|
||||
for row in cfg["deposition_matrix"]:
|
||||
code = row["batch_code"]
|
||||
lines.append(
|
||||
f" TB --> {code}[{code}<br/>Pt {row['pt_power_W']} W / Au {row['au_power_W']} W]"
|
||||
)
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def d3_testing_tree(cfg: dict) -> str:
|
||||
v = cfg["volumes"]
|
||||
xrf = "x".join(str(g) for g in v["xrf_grid"])
|
||||
return f"""flowchart TD
|
||||
TC([Test Coupon]) --> F[Friction<br/>RAPID 6-probe tribometer]
|
||||
F --> FA([Lab Air -> COF dataset mu_normal])
|
||||
F --> FN([Dry N2 -> COF dataset mu_dry_nit])
|
||||
TC --> NI[Nanoindentation<br/>Bruker TI980]
|
||||
NI --> NIR([Hardness, reduced modulus<br/>{v['nanoindentation_indents_per_coupon']} indents per coupon])
|
||||
TC --> AF[AFM]
|
||||
AF --> AFR([Topography -> Ra / Rq])
|
||||
TC --> XR[micro-XRF<br/>Bruker M4 Tornado]
|
||||
XR --> XRR([Pt/Au wt% map, {xrf} grid])
|
||||
"""
|
||||
|
||||
|
||||
def d4_tribometer_session(cfg: dict) -> str:
|
||||
f = cfg["friction_assignment"]
|
||||
return f"""sequenceDiagram
|
||||
actor Op as Operator
|
||||
participant Prep as Sample Prep
|
||||
participant Plan as Excel Test Plan
|
||||
participant SW as Control Software
|
||||
participant TR as RAPID Tribometer
|
||||
|
||||
Op->>Prep: Request test
|
||||
Op->>Prep: Get samples
|
||||
Op->>Prep: Load samples onto plates
|
||||
Op->>Prep: Load ball holders ({f['counterfaces_per_holder']} counterfaces per holder)
|
||||
Op->>Plan: Create test plan<br/>(Sample Plate, Plate Location, Sample ID, Save Location,<br/>Folder Name, X/Y/Z Offset, Load, Iterations)
|
||||
Op->>TR: Transfer plates to tribometer
|
||||
Op->>TR: Set up platter
|
||||
Op->>SW: Activate software
|
||||
Op->>SW: Load plan
|
||||
SW->>TR: Activate tribometer
|
||||
TR-->>SW: Run reciprocating tests (execution loop, see D5)
|
||||
Op->>SW: Software stop
|
||||
SW-->>Op: Save avg files
|
||||
Op->>TR: Remove plates and platters
|
||||
Op->>SW: Close software
|
||||
Op->>TR: Pull equipment out
|
||||
"""
|
||||
|
||||
|
||||
def d5_execution_loop(cfg: dict) -> str:
|
||||
f = cfg["friction_assignment"]
|
||||
tracks_per_run = f["coupons_per_run"] * f["tracks_per_friction_coupon"]
|
||||
return f"""flowchart TD
|
||||
L[Load all counterfaces] --> P{{{f['plates_per_run']} plates}}
|
||||
P -->|for each plate| PR{{{f['probes_per_plate']} probes in parallel}}
|
||||
PR -->|for each probe| C{{{f['coupons_per_probe_square']} coupons on coupon square}}
|
||||
C -->|for each coupon| T[Draw track]
|
||||
T --> RC[Rotate counterface - fresh ball per track]
|
||||
RC -->|{f['tracks_per_friction_coupon']} tracks per coupon| T
|
||||
T --> DONE([{f['coupons_per_run']} coupons x {f['tracks_per_friction_coupon']} tracks = {tracks_per_run} tracks per run])
|
||||
"""
|
||||
|
||||
|
||||
def d6_data_hierarchy(cfg: dict) -> str:
|
||||
return """erDiagram
|
||||
DEPOSITION_RUN ||--|| SIMTRA_PROFILE : produces
|
||||
DEPOSITION_RUN ||--|| BATCH : creates
|
||||
BATCH ||--|{ WAFER : contains
|
||||
WAFER ||--|{ COUPON : contains
|
||||
COUPON ||--|| XRF_MAP : has
|
||||
COUPON ||--|| NANOINDENTATION : has
|
||||
COUPON ||--|| AFM : has
|
||||
COUPON ||--|| PROFILOMETRY : has
|
||||
COUPON ||--o{ TRACK : "friction coupons only"
|
||||
TRACK ||--|{ CYCLE : has
|
||||
TRACK ||--|{ LOOP_POINT : has
|
||||
TRACK ||--|| WEAR : has
|
||||
"""
|
||||
|
||||
|
||||
def deposition_table(cfg: dict) -> str:
|
||||
lines = [
|
||||
"| Batch | Pt:Au gun tilt | Pt power W | Au power W | Pt discharge V | Au discharge V |",
|
||||
"|---|---|---|---|---|---|",
|
||||
]
|
||||
for r in cfg["deposition_matrix"]:
|
||||
lines.append(
|
||||
f"| {r['batch_code']} | {r['pt_gun_tilt_deg']}:{r['au_gun_tilt_deg']} deg "
|
||||
f"| {r['pt_power_W']} | {r['au_power_W']} | {r['pt_discharge_V']} | {r['au_discharge_V']} |"
|
||||
)
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def build_diagrams(cfg: dict) -> list[tuple[str, str, str, str]]:
|
||||
"""Return (file_stem, title, description, mermaid_text) per diagram."""
|
||||
h = cfg["hierarchy"]
|
||||
return [
|
||||
(
|
||||
"D1",
|
||||
"Coupon Assembly",
|
||||
"Substrate preparation, adhesion layer, and gradient Pt-Au sputter deposition.",
|
||||
d1_coupon_assembly(cfg),
|
||||
),
|
||||
(
|
||||
"D2",
|
||||
"Characterization and Batch Assembly",
|
||||
f"Per-coupon simulation and thickness characterization, then assembly into wafers "
|
||||
f"(x{h['coupons_per_wafer']} coupons) and batches (x{h['wafers_per_batch']} wafers), "
|
||||
f"deposited per the parameter table below.",
|
||||
d2_characterization_batch_assembly(cfg),
|
||||
),
|
||||
(
|
||||
"D3",
|
||||
"Testing Tree",
|
||||
"The four characterization/testing paths every coupon can take.",
|
||||
d3_testing_tree(cfg),
|
||||
),
|
||||
(
|
||||
"D4",
|
||||
"Tribometer Session Sequence",
|
||||
"Operator workflow for one RAPID tribometer session, from test request to teardown.",
|
||||
d4_tribometer_session(cfg),
|
||||
),
|
||||
(
|
||||
"D5",
|
||||
"Execution Loop",
|
||||
"Nested iteration executed by the tribometer within one run.",
|
||||
d5_execution_loop(cfg),
|
||||
),
|
||||
(
|
||||
"D6",
|
||||
"Data Hierarchy",
|
||||
"Entity containment and per-entity measurement datasets (ERD-style).",
|
||||
d6_data_hierarchy(cfg),
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def check_dependencies(out_root: Path) -> None:
|
||||
for dep in DEPENDS_ON:
|
||||
marker = out_root / ".done" / f"{dep}.ok"
|
||||
if not marker.exists():
|
||||
raise FileNotFoundError(f"dependency marker missing: {marker} - run task {dep} first")
|
||||
|
||||
|
||||
def validate_diagram(stem: str, text: str) -> None:
|
||||
first = text.splitlines()[0].strip() if text.strip() else ""
|
||||
if not first.startswith(DIAGRAM_TYPES):
|
||||
raise ValueError(f"{stem}: unexpected diagram header {first!r}")
|
||||
|
||||
|
||||
def write_outputs(cfg: dict, out_root: Path) -> tuple[Path, int, int]:
|
||||
diagrams_dir = out_root / "report" / "diagrams"
|
||||
diagrams_dir.mkdir(parents=True, exist_ok=True)
|
||||
diagrams = build_diagrams(cfg)
|
||||
|
||||
md_parts = [
|
||||
"# Process Flow - Simulated Tribology Laboratory\n",
|
||||
"Generated by make_process_flow.py (task 02) from out/config/lab_config.yaml.",
|
||||
"Standalone diagram sources: ./diagrams/D1..D6.mermaid.\n",
|
||||
]
|
||||
total_diagram_bytes = 0
|
||||
for stem, title, description, text in diagrams:
|
||||
validate_diagram(stem, text)
|
||||
path = diagrams_dir / f"{stem}.mermaid"
|
||||
with open(path, "w", encoding="utf-8", newline="\n") as fh:
|
||||
fh.write(text)
|
||||
total_diagram_bytes += len(text.encode("utf-8"))
|
||||
logger.info("wrote %s (%s)", path, title)
|
||||
md_parts.append(f"## {stem} - {title}\n")
|
||||
md_parts.append(description + "\n")
|
||||
md_parts.append(f"```mermaid\n{text}```\n")
|
||||
if stem == "D2":
|
||||
md_parts.append("Deposition matrix:\n")
|
||||
md_parts.append(deposition_table(cfg))
|
||||
|
||||
md_text = "\n".join(md_parts)
|
||||
md_path = out_root / "report" / "process_flow.md"
|
||||
with open(md_path, "w", encoding="utf-8", newline="\n") as fh:
|
||||
fh.write(md_text)
|
||||
return md_path, len(md_text.encode("utf-8")), total_diagram_bytes
|
||||
|
||||
|
||||
def write_marker(marker_path: Path, md_path: Path, md_bytes: int, diagram_bytes: int) -> None:
|
||||
lines = [
|
||||
f"task: '{TASK_ID}'",
|
||||
"status: ok",
|
||||
f"process_flow_file: {md_path.as_posix()}",
|
||||
f"process_flow_bytes: {md_bytes}",
|
||||
"diagrams: 6",
|
||||
f"diagram_bytes: {diagram_bytes}",
|
||||
]
|
||||
marker_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(marker_path, "w", encoding="utf-8", newline="\n") as fh:
|
||||
fh.write("\n".join(lines) + "\n")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Task 02: write process flow diagrams")
|
||||
parser.add_argument("--log-level", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"])
|
||||
parser.add_argument("--out-root", type=Path, default=REPO_ROOT / "out", help="artifact tree root (default: ./out)")
|
||||
args = parser.parse_args()
|
||||
logging.basicConfig(
|
||||
level=args.log_level,
|
||||
stream=sys.stderr,
|
||||
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
|
||||
)
|
||||
|
||||
out_root: Path = args.out_root
|
||||
marker_path = out_root / ".done" / f"{TASK_ID}.ok"
|
||||
try:
|
||||
check_dependencies(out_root)
|
||||
if marker_path.exists():
|
||||
logger.info("re-run: removing stale marker %s", marker_path)
|
||||
marker_path.unlink()
|
||||
with open(out_root / "config" / "lab_config.yaml", encoding="utf-8") as fh:
|
||||
cfg = yaml.safe_load(fh)
|
||||
md_path, md_bytes, diagram_bytes = write_outputs(cfg, out_root)
|
||||
write_marker(marker_path, md_path, md_bytes, diagram_bytes)
|
||||
except Exception:
|
||||
logger.critical("task %s failed", TASK_ID, exc_info=True)
|
||||
return 1
|
||||
|
||||
print(f"task 02 ok: {md_path} ({md_bytes} bytes)")
|
||||
print(f"diagrams: 6 files under {out_root / 'report' / 'diagrams'} ({diagram_bytes} bytes)")
|
||||
print(f"marker: {marker_path}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user