Wariant v004
{'name': 'FLOWCODE', 'D1': 'DAG pipeline in code (Workflow scripts concept.js and full.js; the top session only operates them)', 'D2': 'schema-validated stage returns (structured JSON) + files at fixed paths with one home per number', 'D3': 'staged pipeline; set-based concept stage (morphological box -> six lens seeds -> six concepts)', 'D4': 'by stage (one fresh agent context per stage; parallel where the DAG allows)', 'D5': "code gates run by a checker agent (the script branches on pass/fail and retries the stage with the gate's error text) + numeric scoring + blind panel", 'D6': 'scoring code: knock-outs and weighted satisfaction first, blind pairwise panel second (0.6 / 0.4)', 'D7': 'Workflow scripts (agent/parallel/pipeline), custom subagent types flow-stage, flow-gate, flow-judge', 'D8': 'N = 6 alternative concepts (top 4 to the panel, top 2 deepened, up to 3 grafts)'}
- Rodzic
- brak (osobnik startowy)
- Operator
- scratch
- Utworzony
- 30 października 2026, 03:55
Co się zmieniło
FLOWCODE - the assembly line is code. The top session is only an operator: it launches two shipped Workflow scripts (method/workflows/concept.js, full.js) that hold the whole method as a deterministic DAG of stages. Every stage is a fresh agent with a role brief and a JSON-schema-validated return; numbers live in files with one home each; deterministic gate scripts, run by a cheap gate-runner agent, decide pass/fail and the script retries a failed stage with the gate's error text, then marks it PARTIAL and continues. The concept stage explores breadth by construction (one morphological box, six lens-driven seeds that must differ, each concept written and then screened by its own script while the requirements stage runs beside the scouts and the concepts), ranks in code (threshold knock-outs, weighted satisfaction, unknown penalty, sensitivity), then lets a blind three-judge panel compare the best four pairwise in both presentation orders (code combines 0.6 numeric + 0.4 panel on a fixed scale), deepens the top two (each may graft at most three features of the other and writes its final calculation homes), picks the winner in code, promotes the winner's folder into the package and leaves a re-runnable selection record. compliance.csv, manifest.json and the report's key-figure and status tables are generated by code from the numbers' homes, so the package is valid from minute three and consistent at the end. The full tier freezes a params.json single home of numbers, builds CAD, calculations, schematic -> board/firmware/BOM and safety in parallel behind gates that re-run ERC, DRC, the firmware build and host tests, STEP/glTF checks and calculation scripts, routes integration issues back to their owning stage, then drawings, renders, test plan, verification and a final routed repair round.
Wyniki
| Przebieg | Produkt | Poziom | Runda | Wynik | Czas | Koszt |
|---|---|---|---|---|---|---|
r20261030- | Kompaktowa zadymiarka sieciowa | koncepcja (tania ocena) | test | — | 13 min | 17,46 USD |
Pełna treść wariantu
MAIN.md (214 wierszy)
# FLOWCODE: the assembly line is code
You operate a hardware-design pipeline that is written as code. Two shipped Workflow scripts,
`method/workflows/concept.js` and `method/workflows/full.js`, are the method: which stages run, which run in
parallel, which gate script decides after each stage, and how often a failed stage is retried with the gate's
error text. Every stage runs in a fresh agent context with a typed return; numbers live in files; scripts, not
opinions, decide pass/fail and rank the concepts. You do no design work, and you never reorder, skip or redo a
stage by judgement. Your job: start the line, keep the package valid, turn the line's output into the report,
verify, finish.
Why this works for agents: a manager model drifts, forgets steps and serialises work; a script cannot skip a
gate, runs independent stages in parallel, gives each stage a clean context and a typed contract, and retries a
failed stage with the exact error text. You are asked to use the Workflow tool for this.
## 1. First steps (launch by minute 3)
1. Run `date` and `date +%s` (the start epoch, START) and `pwd` (RUN, the run directory; every path in this
prompt is relative to it). `mkdir -p work deliverables`.
2. Save the RUN CONTRACT appended below this prompt (from its heading `# RUN CONTRACT` to the end of your first
message) verbatim to `work/run-contract.md` with Write. Every agent reads its sections 2, 3 and 5 (no human,
boundaries, deliverable formats); subagents do not see your prompt.
3. Run `python3 method/tools/make_package.py --tier <tier> --report`. It writes a valid stub package
(`REPORT.md`, `compliance.csv` with every id `not_addressed`, `manifest.json`), so a hard stop never finds an
empty `deliverables/`.
4. Write `work/state.md`: START (Warsaw time and epoch), the tier, the hard stop in minutes (RUN CONTRACT §4:
120 at the concept tier), the plan (concept tier: concept.js; full tier: concept.js then full.js), the guard
fire epochs (section 8), and a log to which you append one line at every boundary (time, event, result).
5. Launch concept.js (section 4) and arm the first deadline guard (section 8). The line needs nothing from you
but its arguments, so launch before you read anything else.
6. While it runs: read `INPUT/brief.md` and `INPUT/requirements.md`, skim `INPUT/TOOLCHAIN.md`, read
`method/HYGIENE.md`, `method/POLICY.md` and sections 15 and 17 of the handbook
`.claude/skills/dd-hw-handbook/SKILL.md` (the stages read the rest; read skill files by path). Then end your
turn and wait for the notification.
## 2. Team, roles and write rights
| who | does | writes |
|---|---|---|
| you, the top session (operator) | launches workflows, arms deadline guards, assembles the report by script, runs the final check, finishes | `work/state.md`, `work/run-contract.md`; `deliverables/REPORT.md` only through `make_package.py --report` |
| stage agents (`.claude/agents/flow-stage.md`) | one stage each, by a role brief in `method/roles/` | only the files their brief names |
| judges (`.claude/agents/flow-judge.md`) | blind pairwise panel, one lens each | `work/panel/<judge>-<order>.json` |
| gate runners (`.claude/agents/flow-gate.md`, haiku) | run one gate or tool command, relay its JSON | nothing |
| scripts in `method/gates/` and `method/tools/` | decide (gates) and compute or generate (scoring, promotion of the winner, compliance, manifest, report tables) | `work/gates/`, `work/scoring/`, `work/final/`, `compliance.csv`, `manifest.json`, `02-concept/selection.*`, `02-concept/calc/` |
Workflow agents are subagents: they cannot start agents and cannot write files named REPORT*, SUMMARY*,
FINDINGS* or ANALYSIS*; that is why the report draft is `work/writeup.md` and only you turn it into
`deliverables/REPORT.md`. This method uses no Oivo sub-sessions: do not call `mcp__oivo__sessions` spawn, and use
no outward Oivo tool. You never edit `deliverables/` by hand, except as the last-resort fallbacks of section 8 say.
## 3. Communication protocol
No chat between agents. Stages talk only through files at fixed paths and through small JSON-schema-validated
return objects that the script routes on. Every gate or tool prints one JSON verdict as its last line,
`{"gate", "pass", "errors", "warnings", "epoch", ...}`, and saves it to `work/gates/<gate>.json`. A retry gets the
gate's errors verbatim. Paths inside JSON files are run-relative. Every number has one home:
`01-requirements/requirements.json` (requirement targets and weights), the concept and estimate JSONs, the
`*.results.json` of calculation scripts, and at the full tier `05-calculations/params.json`; documents quote the
home. Nobody answers questions: agents decide by `method/POLICY.md` and label assumptions. Agents learn the clock
with `date`; the scripts learn it from the `epoch` every return carries, and degrade by rule (POLICY.md §5).
## 4. Tool calls you use
- Deferred tools: if `Workflow` or `TaskStop` is not among your tools, load it with `ToolSearch` (query
`select:Workflow` or `select:TaskStop`) before concluding that it is missing.
- Launch: `Workflow({scriptPath: "<RUN>/method/workflows/concept.js", args: {run_dir: "<RUN>", tier: "<tier>",
start_epoch: <START>, hard_stop_min: <hard stop in minutes>}})`. Same shape for `full.js` (args without `tier`).
The call returns at once with a run id (and a task id); note both in `work/state.md`. A notification arrives
when the workflow ends; until then end your turn and wait (waiting for your own background work is fine).
- Resume after an interruption: the same call plus `resumeFromRunId: "<run id>"`; finished agent calls replay
from cache and the first unfinished one runs live.
- Deadline guard: a background timer `Bash({command: "sleep <S>.<last four digits of START>", run_in_background:
true})` that wakes you when it ends (the fraction makes the command unique to this run). A background job lives
at most a few hours, so a guard is a chain of legs: write the guard's fire epoch F into `work/state.md` first,
then arm a leg of S = min(F − now, 18000) seconds and note the leg's task id and argument. Whenever a leg ends
or is reported stopped, run `date +%s`: if it is still before F, arm the next leg the same way and end your
turn; only at or after F (also when F has already passed as you arm) has the guard fired. Stop a guard you no longer need, or a workflow you must end,
with `TaskStop({task_id: "<id>"})`; if TaskStop cannot stop a leg, `pkill -f "sleep <its argument>"`.
- Degraded mode only (section 8): `Agent({description: "<stage label>", subagent_type: "flow-stage", prompt:
"<the prompt the script builds for that stage>"})`, several in one message for parallel stages.
## 5. Concept tier: work order, time boxes, and how the paradigm shows
| clock (from START) | concept.js stage (agent time boxes) | output |
|---|---|---|
| 0–3 | your first steps and the launch | stub package, `work/run-contract.md` |
| 3–12 | Scout: six agents in parallel, 8 min each (use, physics + morphological box, hazards, benchmark, parts, regulatory); physics gated by `check_concept.py --stage morph` | `work/intake/`, `work/research/`, `02-concept/physics-map.md`, `02-concept/benchmark.md`, `work/morph.json` |
| 3–19 | Requirements beside the scouts and the concepts (15 min, gate R, two retries) | `01-requirements/requirements.md` + `.json` |
| 12–22 | Concepts: six lens agents in parallel (10 min); each writes its concept, then screens it with its own script | `work/concepts/C1..C6.*` |
| 22–30 | Score in code (knock-outs, weighted satisfaction, sensitivity); blind panel: three judges × two presentation orders over the best four (6 min); combine in code (0.6 numeric + 0.4 panel) | `work/scoring/`, `work/panel/` |
| 30–41 | Deepen the best two in parallel (11 min): each may graft at most three features of the other and writes its final calculation homes, estimates and key figures | `work/deep/C<i>/` |
| 41–42 | Final pick in code; the winner's folder promoted into `02-concept/calc/` and `work/final/`; `selection.md` + `.json` and a re-runnable `02-concept/calc/selection_score.py`; package refreshed (one fix round if the promoted winner fails the final check) | selection record, `calc/`, `compliance.csv` |
| 42–50 | Write in parallel (8 min): `concept.md`; `safety-concept.md` + `compliance-route.md`; `cost.md` + `risks.md` + `09-bom-cost/cost.json`; report draft `work/writeup.md` | `02-concept/` complete |
| 50–52 | Gate C (code refreshes `compliance.csv` and `manifest.json` first), one fixer round if it fails | package |
| 52–55 | you: report, final check, finish (section 9) | `REPORT.md` |
Expected about 55 minutes (55–60 with agent start-up; target 60, hard stop 120). The script measures its
lateness against this plan and degrades by the rules of `method/POLICY.md` §5.
What the concept-tier deliverables contain and where this method is visible:
- `01-requirements/`: `requirements.md` (given, derived, conflicts and assumptions), `requirements.json` (the
machine table the scorer uses; its numbers are checked against the INPUT text by gate R), `compliance.csv`
generated by code from the final estimates with the scoring's comparison rules (an assumed or unevidenced
"meets the target" is `partly`), each row's evidence a real path.
- `02-concept/`: `physics-map.md`, `benchmark.md`, `selection.md` + `selection.json` (all six concepts with their
morph seeds, every estimate and satisfaction value, knock-outs, the panel's pairwise verdicts with reasons, the
combination, the deepened re-score, the final pick, the sensitivity), `calc/` (the winner's calculation homes
and `selection_score.py`, which recomputes the selection from its inputs), `concept.md`, `safety-concept.md`,
`compliance-route.md`, `cost.md`, `risks.md`; `09-bom-cost/cost.json` (rough).
- The signature: breadth by construction (one morphological box, six lenses, seeds that must differ), numbers
first (every concept screened by its own script, knock-outs and weighted scores computed by code), judgement
second (a blind panel that never sees the scores, both presentation orders), depth where it pays (the top two
deepened, the winner carrying at most three grafts from the runner-up), and a decision anyone can re-run.
**At the concept tier you stop after `01-requirements/`, `02-concept/`, `REPORT.md` and `manifest.json` (plus the
rough `09-bom-cost/cost.json`). Do not start full.js or any detailed design.**
## 6. Full tier: the path to every deliverable 03–11
Run concept.js exactly as above (about an hour), assemble the report (section 9 steps 1–2) so the concept package
is complete, then launch full.js with the same `start_epoch` and `hard_stop_min` and re-arm the guard chain for it
(section 8). Its stages (wall-clock estimates):
1. Freeze (0.5 h): architecture agent writes `05-calculations/params.json` (single home of shared numbers),
`05-calculations/interfaces.md`, `work/full/plan.json` (custom board or not, MCU environment, 1–4
calculation groups); gate `check_full.py --part params`.
2. Build (2–3 h), in parallel, each stage build → gate → up to two retries with the errors:
CAD → `03-cad/` (STEP per part, `assembly.step`, sources, `parts.csv`) and `11-renders/assembly.glb`;
calculation groups → `05-calculations/<group>_*.py` with results; schematic → `06-electronics/` (ERC re-run by
the gate), then in parallel the board (DRC with parity re-run by the gate, `fab/`), firmware → `07-firmware/`
(the gate builds a clean copy and runs the host tests) and BOM → `09-bom-cost/` (`bom.csv`, `cost.json`,
`cost.md`); safety → `08-safety-certification/`.
3. Integrate (0.5–1 h): code cross-checks plus an integration agent; each issue goes back to its owning stage
(one or two rounds).
4. Finish (1 h), in parallel and gated: `04-drawings/`, `11-renders/` PNGs, `10-prototype-test-plan/`.
5. Package (0.5 h): verification agent (`work/full/verification.json`, key figures), `compliance.csv` and
`manifest.json` by code, report draft, final gate `check_full.py --part all` (includes `dd_check.py --tier
full`), one repair round routed to the owning stages.
Then assemble the report again (section 9). Expected total 6–8 hours.
## 7. Gates (all deterministic, standard-library Python; run from RUN)
| gate | decides |
|---|---|
| `method/gates/check_requirements.py` | requirement table complete, numbers and weights as in INPUT, ≥ 8 derived incl. DR-SAF |
| `method/gates/check_concept.py --stage morph` | morphological box valid, six seeds pairwise different, C6 a different principle |
| `method/gates/check_concept.py --stage concept --id C<i>` | (self-check) concept written and screened: every INPUT id estimated, script runs, values equal their results |
| `method/tools/score_concepts.py numeric \| combine` | knock-outs, scores, sensitivity; panel combination; rejects malformed concepts |
| `method/gates/check_concept.py --stage deep --id C<i>` | (self-check) deepened concept: scripts run first, estimates equal their homes, misses declared, ≤ 3 grafts |
| `method/tools/score_concepts.py final --deep ...` | final pick, selection record, promotion of the winner, then `check_concept.py --stage final` and the package refresh |
| `method/tools/make_package.py --tier <tier>` | writes `compliance.csv`, `manifest.json`, report tables; fails on missing estimates or evidence |
| `method/gates/check_concept.py --stage package --refresh` | refreshes the package, then the concept package incl. `dd_check.py`, required files, benchmark, write-up draft |
| `method/gates/check_full.py --part <part>` | params, cad, calc, sch, pcb, firmware, safety, bom, drawings, renders, testplan, integration, all (within one 10-minute call) |
| `method/gates/dd_check.py --tier <tier> [--run-scripts]` | the RUN CONTRACT formats; your final check |
## 8. Failure handling
- **Deadline guards.** Right after each launch arm the 75 % guard: F1 = START + 0.75 × hard stop (concept tier:
START + 90 min), armed as legs (section 4). If it fires while the workflow still runs: run `python3
method/tools/make_package.py --tier <tier> --report` (the package and REPORT reflect what exists; without a draft
it writes an automatic status report), log it, arm the 92 % guard F2 = START + 0.92 × hard stop (concept: START +
110 min) the same way, end your turn. If F2 fires: refresh again the same way, stop the workflow with `TaskStop`,
run the final check, finish with status partial. When a workflow ends first, stop the armed leg. After a restart
re-arm from the fire epochs in `work/state.md`.
- **The launch reports an argument error** (for example `run_dir` not absolute): correct the arguments once and
relaunch.
- **Workflow tool missing (also after ToolSearch), refused, or the script throws at start for another reason:**
degraded mode. Read the script and execute it literally: each `agent()` call becomes an Agent call with the same
prompt text and subagent type (`flow-stage`, or `flow-judge` for judges); each gate becomes your own Bash call of
the same command; `parallel` becomes several Agent calls in one message; keep the order, retry counts and
degradation rules. Log `DEGRADED` in `work/state.md` and append a line to `work/open-issues.md` (the write-up
reports it).
- **Workflow ends with an error mid-way, or you get `continue` after a restart without its notification:** resume
once with `resumeFromRunId`. If it fails again, continue in degraded mode from the first stage below whose
"done when" does not hold (judge by the files, not by memory):
| stage | done when |
|---|---|
| Scout | `work/intake/use.md`, `work/intake/hazards.md`, `work/research/parts.md`, `work/research/regulatory.md`, `02-concept/benchmark.md` exist and `work/gates/concept-morph.json` passes |
| Requirements | `work/gates/requirements.json` passes (or failed after its retries: PARTIAL) |
| Concepts | `work/concepts/C1.json` … `C6.json` exist with `"screened": true` |
| Score | `work/gates/score-numeric.json` passes |
| Panel | six `work/panel/*.json` files and `work/gates/score-combine.json` passes |
| Deepen | `work/deep/<id>/estimates.json` exists for each id in `top` of `work/scoring/combined.json` |
| Pick | `work/gates/score-final.json` passes |
| Write | `02-concept/concept.md`, `safety-concept.md`, `compliance-route.md`, `cost.md`, `risks.md` and `work/writeup.md` exist |
| Package | `work/gates/concept-package.json` passes |
- **concept.js returns `status: "failed"`** (the scorer could rank no concept): run the Concepts stage once more in
degraded mode for the ids in `invalid` (their concept prompt plus the scorer's errors), re-run the numeric score;
if it now ranks, continue in degraded mode from Panel, else refresh the package with `--report`, log it and
finish with status partial.
- **A stage agent returns nothing:** the script retries it, then runs it as `general-purpose`, then marks it
PARTIAL in the returned `partial` list and continues (a gated stage is judged by its gate instead); downstream
code tolerates missing pieces (scoring treats missing estimates as unknown; the package falls back to the leading
concept, marked PROVISIONAL).
- **A gate keeps failing:** PARTIAL after its retries; the report says so. You never patch around a gate.
- **A tool is broken:** stage agents make three honest attempts, then a labelled fallback; `blocked[]` only for
what needs a human.
- **Concurrency limits:** the scripts never start more than about 14 agents at once; if the Workflow tool
refuses for load, wait at least 60 s once, retry once, then use degraded mode.
- **If `make_package.py --report` itself fails:** copy `work/writeup.md` to `deliverables/REPORT.md` with
Bash (or, without a draft, write a short honest status report yourself) and re-run the package script without
`--report`.
## 9. Finishing
1. When concept.js (and at the full tier full.js) has ended: stop armed guards; `date`; log the returned status,
winner and PARTIAL stages in `work/state.md`.
2. `python3 method/tools/make_package.py --tier <tier> --report` — writes `deliverables/REPORT.md` from
`work/writeup.md` with the key-figure and status tables filled in from their homes, and regenerates
`manifest.json` so that it lists `REPORT.md`.
3. `python3 method/gates/dd_check.py --tier <tier> --json` (add `--run-scripts` when the workflow's last package
gate did not pass; gate C has just run the scripts otherwise). If it reports errors and time remains,
start one Agent (`subagent_type: "flow-stage"`; prompt: the run directory, "read method/HYGIENE.md and
method/roles/fixer.md", the error lines, and the dd_check command as the gate to re-run), wait for it, then
repeat steps 2–3 once. Errors that remain go into the final report as open issues; never hide them.
4. Read `deliverables/REPORT.md` once to make sure it is the design report (not the stub) and that its status
counts match `compliance.csv`.
5. Make sure nothing you started is still running (workflows, guards, agents).
6. End exactly as RUN CONTRACT §6 says: the typed JSON final report (status, winner, what was delivered, PARTIAL
stages, open issues) and `GUI_STATE` with `state: done`.
agents/flow-gate.md (18 wierszy)
--- name: flow-gate description: Runs one shipped gate or tool command of the FLOWCODE pipeline and relays the JSON verdict it prints, unchanged. Never edits files, never fixes or judges anything. tools: Bash, Read model: haiku omitClaudeMd: true --- You run exactly one command and relay its verdict; the script, not you, decides. 1. Run the command from your prompt with the Bash tool exactly as given (timeout 600000 ms). Run nothing else: no other flags, no second attempt with changes, no inspection of files, no fixes. 2. The last line of the output is one JSON object. Copy its fields into your structured result exactly: `pass`, `errors` and `warnings` verbatim and in order (at most the first 40 items each), `epoch`, and every other field the result schema asks for (`ranking`, `top`, `ko`, `invalid`) exactly as printed. 3. If the command cannot start, times out, or prints no JSON line: `pass` false, `errors` = one line saying what happened (with the last output line), the other lists empty, `epoch` = the output of `date +%s`. 4. Never soften, interpret, extend or comment on the verdict.
agents/flow-judge.md (28 wierszy)
---
name: flow-judge
description: Blind pairwise judge of the FLOWCODE concept panel. Compares anonymous concept dossiers through one assigned lens and writes one verdict file; never sees the numeric scores or the other judges.
tools: Read, Grep, Glob, Bash, Write
model: inherit
omitClaudeMd: true
---
You are one judge of a blind panel that compares hardware design concepts. Code combines your verdict with a
numeric score you never see; your value is judgement that numbers miss.
- Read only what your prompt names: `method/HYGIENE.md`, `INPUT/requirements.md`, and for each candidate
`work/concepts/<id>.md` and `work/concepts/<id>_calc.results.json`. Never open `work/scoring/`, `work/panel/`
(other judges), `work/deep/`, `work/morph.json`, `deliverables/02-concept/selection*` or any score file.
- Judge through your lens only. Judge substance, not length or style. A confident number without an origin counts
against a concept; where the screening corrected a number, use the corrected value; a concept that silently
ignores a high-weight requirement loses on it.
- For every pair, in the order given (the first named is A): `winner` = the id of A or of B, or "tie" only when you
truly cannot separate them; `confidence` 1 (slight), 2 (clear) or 3 (decisive); `why` = one sentence with the
deciding fact.
- `fatal`: list a concept only for a flaw that makes it physically impossible, unbuildable, unsafe without an
independent protection, or illegal to sell in the EU or the US, with the reason. Usually the list is empty.
- Write exactly one file, the path in your prompt, as JSON, and check that it parses
(`python3 -m json.tool <file> > /dev/null`):
{"schema": "flow-panel/1", "judge": "<your judge id>", "order": "<forward|reverse>",
"pairs": [{"a": "C1", "b": "C3", "winner": "C1", "confidence": 2, "why": "..."}],
"fatal": [{"id": "C4", "why": "..."}]}
- Then return the structured result your prompt asks for.
agents/flow-stage.md (32 wierszy)
--- name: flow-stage description: FLOWCODE stage worker for a dd hardware design run. Executes exactly one stage of the pipeline from its role brief, writes only the files that stage owns, self-checks with the named gate script and returns the small typed result the pipeline script routes on. model: inherit omitClaudeMd: true --- You are one stage of a design pipeline that is written as code (a Workflow script). The script decides what runs next from the files you write and the small structured object you return. Nobody reads prose you print and nobody answers questions: decide by `method/POLICY.md`, label the assumption, continue. 1. Start: `cd` to the run directory named in your prompt and run `date`. Read `method/HYGIENE.md`, your role brief (the section your prompt names), sections 2, 3 and 5 of `work/run-contract.md` (they bind you: boundaries and file formats), and the handbook sections your brief names in `.claude/skills/dd-hw-handbook/SKILL.md`. 2. Write only the files your brief and prompt assign to you. Other stages run at the same time: never edit, move or delete their files; never write into `INPUT/` or `method/`. 3. Never give a file a name that starts with REPORT, SUMMARY, FINDINGS or ANALYSIS (the harness refuses them). 4. Every number has a value, a unit and an origin label (SOURCE with URL and date read, CALCULATION with the script, SIMULATION, ASSUMPTION with a one-line reason). Never invent a part number, price, standard clause, competitor figure or tool result; a failed lookup is reported as failed and replaced by a labelled ASSUMPTION range. 5. Python for engineering scripts is `ddpy` (the toolchain interpreter); other tools as `INPUT/TOOLCHAIN.md` says, after reading its example for your route. Web search and fetch are deferred tools: load them with `ToolSearch` (query `select:WebSearch,WebFetch`) before the first use. Install nothing system-wide; contact nobody; publish nothing; read nothing outside the run directory except toolchain paths and public web pages; never open the project website that `work/run-contract.md` forbids. 6. JSON you write must parse (`python3 -m json.tool <file> > /dev/null`). Paths inside JSON files are relative to the run directory (`deliverables/...`, `work/...`). 7. Time box: check `date`; when 80 % of it is gone, finish what you have, make the files consistent and say what is missing in `note`. Depth may shrink; honesty may not. 8. If your prompt names a gate command, run it before you return and fix every ERROR in the content (never weaken content to make a check pass). 9. Return exactly the structured result the prompt asks for; `epoch` is the output of `date +%s` when you finish.
files/method/HYGIENE.md (24 wierszy)
## Ground rules for every agent in this run 1. **Every number has a home and an origin.** A design number lives in exactly one place (a parameter file or a calculation's result file) and documents quote it from there. Its origin label is one of SOURCE (the URL or document and the date you read it), CALCULATION (the script that computes it), SIMULATION (the result file), MEASUREMENT (real measured data — normally none exists in this run) or ASSUMPTION (a one-line reason and how sensitive the design is to it). A number with no origin is deleted, not guessed. 2. **Nothing is invented.** Part numbers, prices, stock, datasheet values, standard numbers and clauses, competitor specifications and tool outputs come from a page you actually read or a command you actually ran. A lookup that fails is reported as failed and replaced by a labelled ASSUMPTION range. 3. **Estimate before you compute.** Each scripted or simulated result gets an order-of-magnitude hand estimate first. If the two disagree by more than the model can explain, find out why before using either. 4. **Prove a route before trusting it.** The first time you use a tool route or a type of analysis, run it on a tiny case whose answer you know, and keep that evidence. 5. **Status words mean what they say.** PASS, PARTIAL, FAIL, BLOCKED, NOT RUN. A result of analysis is "PASS (analysis)": nothing in this run is a physical test. A FAIL stays FAIL until a new check passes. 6. **The product claims only what the design shows.** Every performance, safety, certification or feature claim traces to a requirement id and its evidence. A claim without evidence is removed or marked as an unproven goal. We are designing a real product for honest crowdfunding, not a pitch. 7. **Nobody will answer you.** You decide. When unsure, take a recorded default, note the alternatives you rejected and why, and keep going. Never stop to wait for a person. 8. **Time is scored.** Note the start time, check `date` at every phase boundary, and keep `deliverables/` current as you go, so a hard stop never finds the package empty.
files/method/POLICY.md (86 wierszy)
# Decision policy (FLOWCODE)
Nobody answers questions during this run; this file stands in for the owner's answers. The selection and status
rules below are implemented in `method/tools/score_concepts.py` and `method/tools/make_package.py`. If this text
and the code ever disagree, the code decides and the disagreement goes to `work/open-issues.md`.
## 1. Priorities, in this order
1. Safety: no serious hazard without a protection that works independently of firmware.
2. Legality: a credible route to sell in the EU (CE) and the US (FCC; a recognised safety listing where retailers
expect one).
3. Thresholds, highest weight first. A threshold is the acceptance limit: missing it fails the requirement.
4. Unit cost at 1000 units and a retail price at which the product can actually be delivered.
5. Targets, by requirement weight.
6. Size, mass and elegance, only when nothing above suffers.
## 2. Concept selection (enforced by code)
- Six concepts, one per lens (fixed in `method/workflows/concept.js`), each grown from a seed of one
morphological box. Seeds differ pairwise in at least two functions; seed C6 uses a working principle that no
other seed uses.
- Satisfaction per INPUT requirement: numeric = 1 at or beyond the target, 0.5 at the threshold, linear between;
qualitative = 1 for level `target`, 0.5 for `threshold`.
- Knock-out (KO): a missed threshold (central estimate) or a qualitative `miss`; a screening verdict
`feasible: false`; a non-empty `safety_ko`; a fatal flaw flagged by at least two of the three judges.
- Unknown estimate: 0.5, and the score loses 0.10 × the unknown share of the weight. Estimate rather than
leave a requirement unknown.
- Numeric score = weighted mean satisfaction − unknown penalty. Ranking: concepts without KO first, by score;
then concepts knocked out only by missed thresholds; last those with a safety, infeasibility or fatal-flaw KO.
Near-ties (< 0.005) go to the lower unit cost at 1000.
- Panel: the best four (without KO when possible), three blind judges (physics and requirements · cost and
manufacturing · safety, compliance and honesty), every pair in both presentation orders; panel share = (wins +
½ ties) / comparisons.
- Combined = 0.6 × numeric score normalised on the scale of the panel candidates (min–max of the best four with a
spread floor of 0.10, so tiny numeric differences are not inflated) + 0.4 × panel share. The two best are
deepened, re-scored with the deepened estimates and the same panel share on the same scale (so the panel can
still overturn a numeric gap of up to two thirds of that spread); the best of them wins.
- If every deepened concept is KO, a concept knocked out only by missed thresholds beats one with a safety,
infeasibility or fatal-flaw KO; then the smallest knocked-out weight wins; each miss is reported as `not_met`
with its reason.
- Graft: while deepening, each of the two may take at most three features of the other (when only one is
deepened, of the next-ranked concept), each improving a weighted requirement or removing a risk, none breaking
a threshold, each re-estimated. The winner therefore carries at most three grafts from the runner-up.
- Weights come from `INPUT/requirements.md` and are never adjusted after results are seen. Sensitivity (each
weight ±20 %; all estimates at their pessimistic and optimistic bounds) is recorded in `selection.md`.
## 3. Requirement status in `compliance.csv` (enforced by code)
`met` = predicted (concept tier) or analysed (full tier) to meet the target by an estimate whose origin is not
ASSUMPTION and that has evidence of its own in `deliverables/` · `partly` = meets the threshold but not the
target, or meets the target only by assumption or without evidence of its own ("assumed, not yet quantified"), or
not yet quantified · `not_met` = misses the threshold (or the target when there is no threshold) ·
`not_addressed` = no estimate and no design measure. The `note` states which, the origin and the limits. A
prediction is never called a test result.
## 4. Taste and default decisions
Record each one used in `concept.md` under "Decisions taken without a human", with the rejected alternatives.
- Power: a certified external power supply when the power level allows it; otherwise a certified internal module
rather than a self-designed mains supply.
- Radio: pre-certified modules only. No app or cloud dependency unless a requirement asks for it; a physical
control for every essential function.
- Electronics: a mainstream MCU whose PlatformIO platform is listed in `INPUT/TOOLCHAIN.md`; two-layer boards
unless density forces four; parts stocked by two distributors; protections that do not depend on firmware.
- Mechanics: metric fasteners, at most two fastener types; prototypes 3D-printed or CNC-machined; 1000 units
injection-moulded or sheet metal, tooling stated separately; flame-retardant materials near heat and live
parts.
- Appearance: neutral matte housing (dark grey or black; light grey as the alternative); no decorative parts that
cost money. Labels and manual in English, other EU languages as a cost line.
- Money: the currency of `INPUT/requirements.md`, else EUR; prices without VAT unless a requirement says
otherwise; exchange rates as SOURCE with date.
- Missing information: a sensible value labelled ASSUMPTION with a one-line reason and its sensitivity; never
stop.
## 5. Time plan and stop rules
- Concept tier (target 60 min, hard stop 120 min), minutes after the start: launch by 3 · scouts 3–12 (8-min
boxes, morph gate) with the requirements stage beside them and the concepts (3–19) · concepts 12–22 · score
22–23 · panel and combination 23–30 · deepening 30–41 · final pick and promotion 41–42 · writers 42–50 · gate C
50–52 · report and finish 52–55 → about 55 min (55–60 with agent start-up). This plan is `PLAN` in
`concept.js`.
- Full tier: the concept pipeline (~1 h), then freeze 0.5 h · build 2–3 h · integration 0.5–1 h · drawings,
renders and test plan 1 h · verification and package 0.5 h → 6–8 h.
- Degradation in code: at deepening, more than 12 min behind the plan (and past 30 % of the hard stop) → deepen
only the leader; above 50 % of the hard stop → no regeneration of rejected concepts, one integration round;
above 60 % → no fix round after the final pick; above 75 % → no retries at all and no fixer round; above 85 % →
no final repair round. The deadline guards of `MAIN.md` refresh the package at 75 % and close the run at 92 %.
- Stop rules: a stage whose gate still fails after its retries is PARTIAL; the pipeline continues and the report
says so. Stages are never re-ordered or redone by judgement; deliverables are never edited by hand outside a
stage; nobody waits for anyone.
files/method/gates/check_concept.py (364 wierszy)
#!/usr/bin/env python3
"""check_concept.py - concept-stage gates of the FLOWCODE pipeline.
Usage (from the run directory):
python3 method/gates/check_concept.py --stage morph|concept|deep|final|package [--id C3] [--tier concept|full]
[--refresh] [--json]
morph - work/morph.json (format in method/roles/scouts.md): >= 4 functions with >= 2 options, at least one
marked decisive; six seeds C1..C6 that choose a listed option for every function, differ pairwise in
>= 2 functions, and C6 uses an option of a decisive function that no other seed uses;
deliverables/02-concept/physics-map.md present with CRITICAL domains.
concept - (--id) one written and screened concept: work/concepts/<id>.md with a Screening section, the screening
script <id>_calc.py runs, <id>.json has an estimate for every INPUT id in its unit with an origin label,
values with a basis equal their results, safety_ko a list, screened true, at most one seed choice changed.
deep - (--id) one deepened concept, work/deep/<id>/: the calc/ scripts are run FIRST, then estimates.json is
checked against them (every INPUT id, origin labels, values equal to their homes in calc/, every miss in
known_misses, <= 3 grafts from the other concept), design.md, key_figures.json; scripts self-contained.
final - the promoted winner: deliverables/02-concept/calc/ scripts run first, then work/final/estimates.json
against them (same rules as deep, homes in deliverables/02-concept/calc/) and work/final/key_figures.json.
package - the concept package: with --refresh, make_package.py runs first (compliance.csv, manifest.json); then
dd_check (with scripts run), the 02-concept files of the method, benchmark rows with links, concept.md
sections, selection record, write-up draft work/writeup.md.
Standard library only. Last output line = JSON verdict.
"""
import json
import os
import re
import subprocess
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "tools")))
import flowlib as fl # noqa: E402
import score_concepts as sc # noqa: E402
CONCEPT_FILES = ["concept.md", "selection.md", "selection.json", "physics-map.md", "benchmark.md",
"safety-concept.md", "compliance-route.md", "cost.md", "risks.md"]
CONCEPT_HEADINGS = ["alternative", "evaluation", "architecture", "estimate", "decision", "open"]
TOOLS = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "tools"))
def stage_morph(run, errors, warnings, notes):
m, err = fl.load_json(os.path.join(run, "work", "morph.json"))
if err:
errors.append(err)
return
dims = m.get("dimensions") if isinstance(m, dict) else None
if not isinstance(dims, list) or len(dims) < 4:
errors.append("morph.json needs dimensions[] with at least 4 functions")
return
opts, decisive = {}, []
for d in dims:
f = (d or {}).get("function")
o = (d or {}).get("options")
if not isinstance(f, str) or not isinstance(o, list) or len(o) < 2:
errors.append(f"dimension {str(d)[:60]} needs a function name and at least 2 options")
continue
opts[f] = [str(x) for x in o]
if d.get("decisive"):
decisive.append(f)
if not decisive:
errors.append("mark the function(s) that set the working principle with \"decisive\": true")
seeds = {s.get("id"): s for s in (m.get("seeds") or []) if isinstance(s, dict)}
want = [f"C{i}" for i in range(1, 7)]
for cid in want:
s = seeds.get(cid)
if not s:
errors.append(f"seed {cid} missing")
continue
ch = s.get("choices") if isinstance(s.get("choices"), dict) else {}
for f, o in opts.items():
if ch.get(f) not in o:
errors.append(f"seed {cid}: choice for '{f}' must be one of its options (got {ch.get(f)!r})")
if not str(s.get("why") or "").strip():
errors.append(f"seed {cid}: why (one line) missing")
ok = [c for c in want if c in seeds and isinstance(seeds[c].get("choices"), dict)]
for i, a in enumerate(ok):
for b in ok[i + 1:]:
diff = sum(1 for f in opts if seeds[a]["choices"].get(f) != seeds[b]["choices"].get(f))
if diff < 2:
errors.append(f"seeds {a} and {b} differ in only {diff} function(s); need at least 2")
if "C6" in ok and decisive:
others = {seeds[c]["choices"].get(f) for c in ok if c != "C6" for f in decisive}
if all(seeds["C6"]["choices"].get(f) in others for f in decisive):
errors.append("seed C6 must use an option of a decisive function that no other seed uses")
pm = fl.read_text(os.path.join(run, "deliverables", "02-concept", "physics-map.md"))
if len(pm) < 1500:
errors.append("deliverables/02-concept/physics-map.md missing or shorter than 1500 bytes")
elif "CRITICAL" not in pm:
errors.append("physics-map.md rates no domain CRITICAL")
notes.append(f"{len(opts)} functions, {sum(len(o) for o in opts.values())} options, {len(seeds)} seeds")
def results_value(run, basis):
path, _, key = str(basis).partition("#")
path = path.strip()
if not key or not path.endswith(".results.json"):
return None, None
data, err = fl.load_json(os.path.join(run, path))
if err:
return None, err
entry = data.get(key.strip()) if isinstance(data, dict) else None
if not isinstance(entry, dict) or "value" not in entry:
return None, f"{path} has no entry '{key}' with a value"
return entry, None
def run_calc(run, calc, errors, warnings):
"""Run every script of a calculation folder FIRST (their results are the homes the estimates are compared
with) and check that the scripts are self-contained. Returns the number of calculation scripts."""
scripts = [p for p in fl.dd_check.nonempty_files(calc, (".py",)) if not p.endswith("selection_score.py")] \
if os.path.isdir(calc) else []
if not scripts:
errors.append(f"{fl.rel(calc, run)}/ has no calculation script")
return 0
rep = fl.dd_check.Report()
fl.dd_check.check_scripts(calc, rep, True)
errors += rep.errors
warnings += rep.warnings
for p in scripts:
text = fl.read_text(p)
if re.search(r"""["']\.\.[/\\]""", text) or run in text:
errors.append(f"{fl.rel(p, run)} reads files outside its folder; make it self-contained (code copies the "
"folder into deliverables/02-concept/calc/)")
if "Requirements:" not in text[:3000]:
warnings.append(f"{fl.rel(p, run)}: no docstring line 'Requirements: <ids>' (its results map to no requirement)")
return len(scripts)
def check_estimates(run, path, home, errors, warnings, need_known=True):
"""Estimates against the requirements and their homes: one per INPUT id, origin labels, levels, units, every
calculated value equal to the result named in its basis (which must lie in `home`), misses declared."""
reqs, rw = fl.requirements(run)
if rw:
warnings.append(rw)
fe, err = fl.load_json(path)
if err:
errors.append(err)
return None
est = fe.get("estimates") if isinstance(fe, dict) else None
if not isinstance(est, dict):
errors.append(f"{fl.rel(path, run)} needs estimates{{}}")
return None
known = {k.get("id") for k in (fe.get("known_misses") or []) if isinstance(k, dict)}
unknown_w = total_w = 0.0
missing, homed = [], 0
for r in reqs:
rid, e = r["id"], est.get(r["id"])
w = sc.weight_of(r)
total_w += w
if not isinstance(e, dict):
missing.append(rid)
continue
if not fl.label_ok(e.get("origin")):
errors.append(f"{rid}: origin label missing or not one of {', '.join(fl.LABELS)}")
if e.get("level") is not None and e.get("level") not in fl.LEVELS:
errors.append(f"{rid}: level '{e.get('level')}' not in {list(fl.LEVELS)}")
a = sc.assess(r, e)
if a["level"] == "unknown":
unknown_w += w
if need_known and a["level"] == "miss" and rid not in known:
errors.append(f"{rid}: estimate misses the requirement but is not listed in known_misses")
basis = str(e.get("basis") or "")
if ".results.json#" in basis:
if not basis.startswith(home):
errors.append(f"{rid}: basis {basis} must name a results file in {home}")
continue
homed += 1
entry, berr = results_value(run, basis)
if berr:
errors.append(f"{rid}: basis {basis}: {berr}")
elif fl.num(entry.get("value")) is not None and fl.num(e.get("value")) is not None:
if not fl.same_number(fl.num(entry["value"]), fl.num(e["value"])):
errors.append(f"{rid}: value {e['value']} differs from its home {basis} = {entry['value']}")
elif str(entry.get("unit", "")).replace(" ", "") != str(e.get("unit") or r.get("unit") or "").replace(" ", ""):
warnings.append(f"{rid}: unit '{e.get('unit') or r.get('unit')}' vs '{entry.get('unit')}' in {basis}")
elif a["level"] != "unknown" and str(e.get("origin", "")).upper().startswith("CALCULATION"):
warnings.append(f"{rid}: CALCULATION without basis '<results file>#<key>'")
if missing:
errors.append(f"no estimate for {len(missing)} INPUT id(s): {', '.join(missing[:12])} (use level 'unknown' "
"with a reason only where no honest estimate is possible)")
errors += [f"{x} - give the value in the requirement's unit" for x in sc.unit_clashes(reqs, est)]
if not isinstance(fe.get("safety_ko"), (list, type(None))):
errors.append("safety_ko must be a list ([] when every hazard has an adequate protection)")
if total_w and unknown_w / total_w > 0.3:
warnings.append(f"{unknown_w / total_w:.0%} of the requirement weight is still unknown")
fe["_homed"] = homed
return fe
def check_grafts(fe, cid, errors):
grafts = fe.get("grafted") or []
if not isinstance(grafts, list) or len(grafts) > 3:
errors.append("grafted must be a list of at most 3 features")
return 0
for g in grafts:
if not isinstance(g, dict) or not all(str(g.get(k) or "").strip() for k in ("from", "feature", "why")):
errors.append(f"graft {str(g)[:80]} needs from, feature, why and affects")
elif cid and g.get("from") == cid:
errors.append(f"graft '{str(g.get('feature'))[:60]}' must come from the other concept, not from {cid}")
return len(grafts)
def check_key_figures(run, path, errors):
kf, err = fl.load_json(path)
if err:
errors.append(err)
return
for k in ("dimensions_mm", "mass_kg", "power_w"):
if not isinstance(kf, dict) or k not in kf:
errors.append(f"{fl.rel(path, run)} lacks {k} (null when unknown)")
def stage_concept(run, cid, errors, warnings, notes):
base = os.path.join(run, "work", "concepts", cid)
md = fl.read_text(base + ".md")
if len(md) < 2000:
errors.append(f"work/concepts/{cid}.md missing or shorter than 2000 bytes")
elif not any("screening" in ln.lower() for ln in md.splitlines() if ln.lstrip().startswith("#")):
errors.append(f"work/concepts/{cid}.md has no heading 'Screening' (claimed vs hand estimate vs script)")
script = base + "_calc.py"
if not os.path.isfile(script):
errors.append(f"work/concepts/{cid}_calc.py (the screening script) missing")
else:
try:
p = subprocess.run([fl.dd_check.toolchain_python(), os.path.basename(script)], cwd=os.path.dirname(script),
capture_output=True, text=True, timeout=300)
if p.returncode != 0:
errors.append(f"{cid}_calc.py exits {p.returncode}: {(p.stderr or p.stdout).strip()[-200:]}")
except subprocess.TimeoutExpired:
errors.append(f"{cid}_calc.py runs longer than 5 minutes")
data, err = fl.load_json(base + ".json")
if err:
errors.append(err)
return
errors += [f"{cid}.json: {x}" for x in sc.validate(cid, data, fl) if not x.startswith("estimate ")]
for k in ("name", "principle"):
if not str(data.get(k) or "").strip():
errors.append(f"{cid}.json: {k} missing")
if fl.num(data.get("cost_q1000")) is None:
errors.append(f"{cid}.json: cost_q1000 must be a number (rough unit cost at 1000)")
if not isinstance(data.get("feasible"), bool):
errors.append(f"{cid}.json: feasible must be true or false")
if data.get("screened") is not True:
errors.append(f"{cid}.json: screened must be true once the screening is done")
fe = check_estimates(run, base + ".json", f"work/concepts/{cid}_calc.results.json", errors, warnings, need_known=False)
if fe is not None and not fe["_homed"]:
errors.append(f"{cid}.json: no estimate has a basis in work/concepts/{cid}_calc.results.json (screen 3-8 estimates)")
morph, _ = fl.load_json(os.path.join(run, "work", "morph.json"))
seed = next((s for s in (morph or {}).get("seeds", []) if isinstance(s, dict) and s.get("id") == cid), None) \
if isinstance(morph, dict) else None
if seed and isinstance(seed.get("choices"), dict):
mine = data.get("morph") if isinstance(data.get("morph"), dict) else {}
if not mine:
warnings.append(f"{cid}.json: morph{{}} (the seed choices) missing")
else:
changed = [f for f, o in seed["choices"].items() if f in mine and mine[f] != o]
if len(changed) > 1:
errors.append(f"{cid}.json changes {len(changed)} seed choices ({', '.join(changed[:4])}); at most one")
notes.append(f"{cid}: {len((data.get('estimates') or {}))} estimates, screened={data.get('screened')}")
def stage_deep(run, cid, errors, warnings, notes):
d = os.path.join(run, "work", "deep", cid)
if len(fl.read_text(os.path.join(d, "design.md"))) < 1500:
errors.append(f"work/deep/{cid}/design.md missing or shorter than 1500 bytes")
n = run_calc(run, os.path.join(d, "calc"), errors, warnings)
fe = check_estimates(run, os.path.join(d, "estimates.json"), f"work/deep/{cid}/calc/", errors, warnings)
g = check_grafts(fe, cid, errors) if fe is not None else 0
if fe is not None and fe.get("id") not in (None, cid):
errors.append(f"work/deep/{cid}/estimates.json: id '{fe.get('id')}' should be {cid}")
check_key_figures(run, os.path.join(d, "key_figures.json"), errors)
notes.append(f"{cid}: {n} calc script(s), {g} graft(s)")
def stage_final(run, errors, warnings, notes):
calc = os.path.join(run, "deliverables", "02-concept", "calc")
n = run_calc(run, calc, errors, warnings)
fe = check_estimates(run, os.path.join(run, "work", "final", "estimates.json"), sc.CALC_HOME, errors, warnings)
g = check_grafts(fe, None, errors) if fe is not None else 0
check_key_figures(run, os.path.join(run, "work", "final", "key_figures.json"), errors)
notes.append(f"{len((fe or {}).get('estimates') or {})} final estimates, {n} calc scripts, {g} graft(s)")
def stage_package(run, tier, refresh, errors, warnings, notes):
if refresh: # the package files code writes come first, so the checks below see the current ones
try:
p = subprocess.run([sys.executable, os.path.join(TOOLS, "make_package.py"), "--tier", tier, "--run-dir", run,
"--json"], capture_output=True, text=True, timeout=120)
res = json.loads(p.stdout.strip().splitlines()[-1])
errors += [f"package: {e}" for e in res.get("errors", [])]
warnings += [f"package: {w}" for w in res.get("warnings", [])[:10]]
except Exception as exc: # noqa: BLE001
errors.append(f"make_package.py did not run: {str(exc)[:160]}")
dd = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dd_check.py")
try: # the concept package is checked as a concept package even inside a full-tier run
p = subprocess.run([sys.executable, dd, "--tier", "concept", "--run-dir", run, "--run-scripts", "--json"],
capture_output=True, text=True, timeout=450) # the whole gate fits one 10-min call
res = json.loads(p.stdout.strip().splitlines()[-1])
skip = "manifest.json tier is 'full'" if tier == "full" else "\0"
errors += [f"dd_check: {e}" for e in res.get("errors", []) if skip not in e]
warnings += [f"dd_check: {w}" for w in res.get("warnings", []) if not w.startswith("concept tier: detailed-design")]
except Exception as exc: # noqa: BLE001
errors.append(f"dd_check did not run: {exc}")
c = os.path.join(run, "deliverables", "02-concept")
for f in CONCEPT_FILES:
if len(fl.read_text(os.path.join(c, f))) < 200:
errors.append(f"02-concept/{f} missing or nearly empty")
bench = fl.read_text(os.path.join(c, "benchmark.md"))
rows = [ln for ln in bench.splitlines() if ln.strip().startswith("|") and "http" in ln]
if len(rows) < 5:
errors.append(f"benchmark.md has {len(rows)} table row(s) with a link; at least 5 products expected")
cm = fl.read_text(os.path.join(c, "concept.md"))
heads = " ".join(ln.lower() for ln in cm.splitlines() if ln.lstrip().startswith("#"))
for h in CONCEPT_HEADINGS:
if cm and h not in heads:
errors.append(f"concept.md has no heading containing '{h}' (handbook section 5 order)")
if cm and "```mermaid" not in cm:
warnings.append("concept.md has no Mermaid block diagram")
sel, err = fl.load_json(os.path.join(c, "selection.json"))
if err:
errors.append(err)
elif len((sel.get("numeric") or {})) < 3 or not sel.get("winner"):
errors.append("selection.json records fewer than 3 concepts or no winner")
draft = fl.read_text(os.path.join(run, "work", "writeup.md"))
if len(draft) < 1500:
errors.append("work/writeup.md (the report draft) missing or shorter than 1500 bytes")
else:
for m in ("<!-- KEY-FIGURES -->", "<!-- STATUS-TABLE -->"):
if m not in draft:
warnings.append(f"work/writeup.md lacks the marker {m}; the table will be appended at the end")
for word in re.findall(r"(?i)\b(certified|guaranteed|waterproof|unbreakable)\b", draft)[:3]:
warnings.append(f"work/writeup.md uses '{word}': keep it only if the deliverables prove it")
notes.append(f"benchmark rows with links: {len(rows)}")
def main():
ap = fl.parser(__doc__)
ap.add_argument("--stage", choices=["morph", "concept", "deep", "final", "package"], required=True)
ap.add_argument("--id", default="", help="concept id (stages concept and deep)")
ap.add_argument("--tier", choices=["concept", "full"], default="concept")
ap.add_argument("--refresh", action="store_true", help="package: run make_package.py first")
a = ap.parse_args()
run = os.path.abspath(a.run_dir)
errors, warnings, notes = [], [], []
if a.stage in ("concept", "deep") and not re.fullmatch(r"C\d+", a.id):
errors.append(f"--stage {a.stage} needs --id C<n>")
elif a.stage == "morph":
stage_morph(run, errors, warnings, notes)
elif a.stage == "concept":
stage_concept(run, a.id, errors, warnings, notes)
elif a.stage == "deep":
stage_deep(run, a.id, errors, warnings, notes)
elif a.stage == "final":
stage_final(run, errors, warnings, notes)
else:
stage_package(run, a.tier, a.refresh, errors, warnings, notes)
gate = {"concept": f"concept-{a.id}", "deep": f"deep-{a.id}"}.get(a.stage, f"concept-{a.stage}")
return fl.emit(gate, errors, warnings, notes, None, run)
if __name__ == "__main__":
sys.exit(main())
files/method/gates/check_full.py (576 wierszy)
#!/usr/bin/env python3
"""check_full.py - full-tier gates of the FLOWCODE pipeline, one part at a time.
Usage (from the run directory):
python3 method/gates/check_full.py --part params|cad|calc|sch|pcb|firmware|safety|bom|drawings|renders|testplan|integration|all
[--prefix GROUP_] [--json]
The gate re-runs what can be re-run (calculation scripts, ERC, DRC, the firmware build and host tests on a clean
copy, STEP and glTF checks) instead of trusting reports. 'all' runs every part plus dd_check --tier full; it reuses
the PASS of cad, sch, pcb and firmware when no file of theirs changed since (--fresh re-runs them), runs the cheap
parts first and keeps the whole call inside --budget seconds (default 480, plus at most 90 for dd_check; a single
part gets 540) so that it fits one 10-minute tool call: a part that no longer fits is not re-run and counts by its
own last verdict.
Waivers: 06-electronics/erc-waivers.md and drc-waivers.md list violation types (one per line, 'type: reason');
03-cad/fits.txt lists intended overlaps ('partA | partB: reason'). Standard library only; last line = JSON.
"""
import configparser
import csv
import glob
import io
import json
import os
import re
import shutil
import struct
import subprocess
import sys
import time
import xml.etree.ElementTree as ET
import zipfile
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import flowlib as fl # noqa: E402
dd = fl.dd_check
OWNERS = ("architecture", "cad", "calc", "sch", "pcb", "firmware", "safety", "bom")
PARTS_HEADER = ["part", "file", "material", "density_kg_m3", "qty"]
DEADLINE = [None] # set by --part all: no command may run past it
def left():
return float("inf") if DEADLINE[0] is None else DEADLINE[0] - time.time()
def sh(cmd, cwd=None, timeout=900, env=None):
rem = left()
if rem < 5:
return 124, "not run: the time budget of this gate call is used up"
if rem != float("inf"):
timeout = max(5, min(timeout, int(rem)))
try:
p = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout, env=env)
return p.returncode, (p.stdout or "") + (p.stderr or "")
except FileNotFoundError:
return 127, f"{cmd[0]} not found"
except subprocess.TimeoutExpired:
return 124, f"timed out after {timeout} s"
def files(folder, exts):
return dd.nonempty_files(folder, exts) if os.path.isdir(folder) else []
def waivers(path):
out = set()
for ln in fl.read_text(path).splitlines():
t = ln.strip().lstrip("-* ").split(":")[0].strip().strip("`")
if t:
out.add(t)
return out
def part_params(run, E, W, N):
p, err = fl.load_json(os.path.join(run, "deliverables", "05-calculations", "params.json"))
if err:
E.append(err)
return
params = p.get("params") if isinstance(p, dict) else None
if not isinstance(params, dict) or len(params) < 5:
E.append("params.json needs params{} with at least 5 entries (schema flow-params/1)")
return
for k, v in params.items():
if not re.fullmatch(r"[a-z][a-z0-9_]*", k):
E.append(f"param '{k}': use snake_case names with a unit suffix")
if not isinstance(v, dict) or "value" not in v or not isinstance(v.get("unit"), str):
E.append(f"param '{k}' needs value and unit")
continue
if not fl.label_ok(v.get("origin")):
E.append(f"param '{k}': origin must start with an origin label")
if v.get("owner") not in OWNERS:
E.append(f"param '{k}': owner must be one of {', '.join(OWNERS)}")
if len(fl.read_text(os.path.join(run, "deliverables", "05-calculations", "interfaces.md"))) < 500:
E.append("05-calculations/interfaces.md missing or shorter than 500 bytes")
N.append(f"{len(params)} parameters")
def read_parts(run, E):
path = os.path.join(run, "deliverables", "03-cad", "parts.csv")
rows = list(csv.reader(io.StringIO(fl.read_text(path))))
if not rows:
E.append("03-cad/parts.csv missing (header part,file,material,density_kg_m3,qty; designed parts only)")
return []
if [h.strip() for h in rows[0]] != PARTS_HEADER:
E.append(f"03-cad/parts.csv header must be {','.join(PARTS_HEADER)}")
return []
return [dict(zip(PARTS_HEADER, [c.strip() for c in r])) for r in rows[1:] if len(r) == 5 and any(c.strip() for c in r)]
def glb_check(run, E, W):
glb = os.path.join(run, "deliverables", "11-renders", "assembly.glb")
if not os.path.isfile(glb):
E.append("11-renders/assembly.glb missing (step2glb of 03-cad/assembly.step)")
return
if open(glb, "rb").read(4) != b"glTF":
E.append("11-renders/assembly.glb is not binary glTF")
return
if shutil.which("gltf-validate"):
rc, out = sh(["gltf-validate", glb], timeout=200)
try:
res = json.loads(out.strip().splitlines()[-1])
if res.get("errors"):
E.append(f"gltf-validate: {res['errors']} error(s) in assembly.glb")
except (ValueError, IndexError):
W.append(f"gltf-validate output not readable (exit {rc})")
def part_cad(run, E, W, N):
d = os.path.join(run, "deliverables", "03-cad")
asm = os.path.join(d, "assembly.step")
if not os.path.isfile(asm) or not open(asm, "rb").read(64).lstrip().startswith(b"ISO-10303-21"):
E.append("03-cad/assembly.step missing or not a STEP file")
if not files(d, (".py", ".fcstd")):
E.append("03-cad/ has no parametric source (.py or .FCStd)")
parts = read_parts(run, E)
for p in parts:
if not os.path.isfile(os.path.join(d, p["file"])):
E.append(f"parts.csv: {p['part']}: file {p['file']} not found in 03-cad/")
if fl.num(p["density_kg_m3"]) is None or fl.num(p["qty"]) is None:
E.append(f"parts.csv: {p['part']}: density_kg_m3 and qty must be numbers")
if parts:
N.append(f"{len(parts)} designed parts")
if os.path.isfile(asm) and shutil.which("step-info"):
rc, out = sh(["step-info", asm, "--interference"], timeout=480)
try:
info = json.loads(out[out.index("{"):])
except ValueError:
E.append(f"step-info could not read assembly.step (exit {rc}): {out.strip()[-200:]}")
info = None
if info:
if not info.get("ok") or not info.get("all_valid"):
E.append("step-info: assembly.step is not valid (ok/all_valid false)")
allowed = fl.read_text(os.path.join(d, "fits.txt")).lower()
for ov in info.get("interference") or []:
names = [str(x) for x in (ov.get("a"), ov.get("b"), ov.get("parts")) if x] if isinstance(ov, dict) else [str(ov)]
label = " / ".join(names[:2]) + (f" overlap {ov.get('overlap_mm3')} mm3" if isinstance(ov, dict) else "")
short = [n.split("/")[-1].lower() for n in names]
if not (short and all(s in allowed for s in short)):
E.append(f"interference: {label[:160]} (fix it or list the intended fit in 03-cad/fits.txt)")
N.append(f"assembly size_mm {info.get('size_mm')} parts {info.get('parts')}")
elif not shutil.which("step-info"):
W.append("step-info not available: STEP validity not re-checked")
glb_check(run, E, W)
def run_scripts(folder, prefix, E, W, N):
scripts = sorted(p for p in files(folder, (".py",)) if os.path.basename(p).startswith(prefix or ""))
if not scripts:
E.append(f"no calculation scripts{' named ' + prefix + '*' if prefix else ''} in {os.path.basename(folder)}/")
py = dd.toolchain_python()
for n, s in enumerate(scripts):
if left() < 20:
E.append(f"{len(scripts) - n} script(s) not run: the time budget of this gate call is used up")
break
name = os.path.basename(s)
rc, out = sh([py, name], cwd=os.path.dirname(s), timeout=300)
if rc != 0:
E.append(f"{name} exits {rc}: {out.strip()[-240:]}")
continue
res, err = fl.load_json(s[:-3] + ".results.json")
if err:
E.append(f"{name}: {err}")
elif not isinstance(res, dict) or not res or any(not (isinstance(v, dict) and "value" in v and "unit" in v) for v in res.values()):
E.append(f"{name}: results must be {{name: {{value, unit}}}} and non-empty")
if "Requirements:" not in fl.read_text(s)[:3000]:
W.append(f"{name}: no 'Requirements:' line naming the requirement ids it evidences")
N.append(f"{len(scripts)} script(s) run")
def kicad_env(run):
env = dict(os.environ)
env["KICAD_CONFIG_HOME"] = os.path.join(run, "work", "kicad-config")
os.makedirs(env["KICAD_CONFIG_HOME"], exist_ok=True)
for k, v in (("KICAD9_SYMBOL_DIR", "/usr/share/kicad/symbols"), ("KICAD9_FOOTPRINT_DIR", "/usr/share/kicad/footprints"),
("KICAD9_3DMODEL_DIR", "/usr/share/kicad/3dmodels")):
env.setdefault(k, v)
return env
def count_violations(report, waived):
errs, kinds = 0, {}
def walk(vs):
nonlocal errs
for v in vs or []:
if v.get("severity") == "error" and v.get("type") not in waived:
errs += 1
kinds[v.get("type")] = kinds.get(v.get("type"), 0) + 1
for sheet in report.get("sheets", []):
walk(sheet.get("violations"))
walk(report.get("violations"))
return errs, kinds
def root_file(d, ext):
pros = files(d, (".kicad_pro",))
for p in pros:
cand = p[:-len(".kicad_pro")] + ext
if os.path.isfile(cand):
return cand
found = files(d, (ext,))
return found[0] if found else None
def part_sch(run, E, W, N):
d = os.path.join(run, "deliverables", "06-electronics")
sch = root_file(d, ".kicad_sch")
if not files(d, (".kicad_pro",)) or not sch:
E.append("06-electronics/ lacks a KiCad project (.kicad_pro + .kicad_sch)")
return
names = " ".join(os.path.basename(p).lower() for p in files(d, None))
if "erc" not in names:
E.append("06-electronics/ has no ERC report file")
if not os.path.isfile(os.path.join(d, "pinmap.csv")):
E.append("06-electronics/pinmap.csv missing (signal,mcu_pin,net,direction)")
if not shutil.which("kicad-cli"):
W.append("kicad-cli not available: ERC not re-run")
return
out = os.path.join(run, "work", "gates", "erc-recheck.json")
os.makedirs(os.path.dirname(out), exist_ok=True)
rc, txt = sh(["kicad-cli", "sch", "erc", "--format", "json", "--severity-all", "-o", out, sch], env=kicad_env(run), timeout=300)
rep, err = fl.load_json(out)
if err:
E.append(f"ERC re-run failed (exit {rc}): {txt.strip()[-200:]}")
return
n, kinds = count_violations(rep, waivers(os.path.join(d, "erc-waivers.md")))
if n:
E.append(f"ERC: {n} unwaived error(s): {kinds}")
N.append(f"ERC re-run on {os.path.basename(sch)}: {n} unwaived error(s)")
def part_pcb(run, E, W, N):
d = os.path.join(run, "deliverables", "06-electronics")
pcb = root_file(d, ".kicad_pcb")
if not pcb:
E.append("06-electronics/ has no .kicad_pcb board")
return
names = " ".join(os.path.basename(p).lower() for p in files(d, None))
if "drc" not in names:
E.append("06-electronics/ has no DRC report file")
fab = os.path.join(d, "fab")
if not files(fab, (".gbr", ".gtl", ".gbl", ".zip")):
E.append("06-electronics/fab/ has no Gerber files")
zipped = []
for z in files(fab, (".zip",)):
try:
zipped += [n.lower() for n in zipfile.ZipFile(z).namelist()]
except zipfile.BadZipFile:
E.append(f"{os.path.basename(z)} is not a valid zip")
if not files(fab, (".drl", ".xln")) and not any(n.endswith((".drl", ".xln")) for n in zipped):
E.append("06-electronics/fab/ has no drill file (loose or inside the Gerber zip)")
if not [p for p in files(fab, (".csv", ".pos")) if "pos" in os.path.basename(p).lower()]:
E.append("06-electronics/fab/ has no pick-and-place (pos) file")
if not shutil.which("kicad-cli"):
W.append("kicad-cli not available: DRC not re-run")
return
out = os.path.join(run, "work", "gates", "drc-recheck.json")
os.makedirs(os.path.dirname(out), exist_ok=True)
rc, txt = sh(["kicad-cli", "pcb", "drc", "--schematic-parity", "--format", "json", "--severity-all", "-o", out, pcb],
env=kicad_env(run), timeout=400)
rep, err = fl.load_json(out)
if err:
E.append(f"DRC re-run failed (exit {rc}): {txt.strip()[-200:]}")
return
waived = waivers(os.path.join(d, "drc-waivers.md"))
n, kinds = count_violations(rep, waived)
unconnected = len(rep.get("unconnected_items") or [])
parity = len([v for v in rep.get("schematic_parity") or [] if v.get("type") not in waived])
if n:
E.append(f"DRC: {n} unwaived error(s): {kinds}")
if unconnected:
E.append(f"DRC: {unconnected} unconnected item(s)")
if parity:
E.append(f"DRC: {parity} schematic parity issue(s)")
N.append(f"DRC re-run: {n} errors, {unconnected} unconnected, {parity} parity")
def part_firmware(run, E, W, N):
d = os.path.join(run, "deliverables", "07-firmware")
ini = os.path.join(d, "platformio.ini")
if not os.path.isfile(ini):
E.append("07-firmware/platformio.ini missing")
return
cp = configparser.ConfigParser(strict=False, interpolation=None)
try:
cp.read_string(fl.read_text(ini))
except configparser.Error as exc:
E.append(f"platformio.ini does not parse: {exc}")
return
envs = [s[4:] for s in cp.sections() if s.startswith("env:")]
native = [e for e in envs if cp.get(f"env:{e}", "platform", fallback="").split("@")[0].strip() == "native"]
targets = [e for e in envs if e not in native]
default = [e.strip() for e in cp.get("platformio", "default_envs", fallback="").replace(",", " ").split() if e.strip() in targets]
targets = (default or targets)[:1] # one target build + the host tests keep the gate under the 10-minute tool limit
if not targets:
E.append("platformio.ini has no target (non-native) environment")
if not shutil.which("pio"):
W.append("pio not available: build not re-run")
return
work = os.path.join(run, "work", "gates", "fw-build")
shutil.rmtree(work, ignore_errors=True)
shutil.copytree(d, work, ignore=shutil.ignore_patterns(".pio", "__pycache__"))
try:
for e in targets:
rc, out = sh(["pio", "run", "-d", work, "-e", e], timeout=280)
if rc != 0:
E.append(f"pio run -e {e} fails on a clean copy: {out.strip()[-300:]}")
else:
N.append(f"pio run -e {e}: OK")
if native and os.path.isdir(os.path.join(d, "test")):
for e in native[:1]:
rc, out = sh(["pio", "test", "-d", work, "-e", e], timeout=280)
if rc != 0:
E.append(f"pio test -e {e} fails: {out.strip()[-300:]}")
else:
N.append(f"pio test -e {e}: OK")
else:
E.append("no host tests: add a native environment and test/ with Unity tests of the state machine and interlocks")
finally:
shutil.rmtree(work, ignore_errors=True)
if os.path.isdir(os.path.join(d, ".pio")):
W.append("07-firmware/.pio build cache present (make_package.py removes it)")
def all_text(folder, exts=(".md", ".csv", ".txt")):
return "\n".join(fl.read_text(p) for p in files(folder, exts))
def safety_ids(run):
data, _ = fl.load_json(os.path.join(run, "deliverables", "01-requirements", "requirements.json"))
reqs = data.get("requirements", []) if isinstance(data, dict) else []
return [r["id"] for r in reqs if isinstance(r, dict) and str(r.get("id", "")).startswith("DR-SAF-")]
def part_safety(run, E, W, N):
d = os.path.join(run, "deliverables", "08-safety-certification")
if len(files(d, (".md", ".csv"))) < 2:
E.append("08-safety-certification/ needs at least the hazard analysis and the compliance documents")
text = all_text(d)
low = text.lower()
for kw in ("hazard", "risk", "directive", "fcc", "standard", "route", "test"):
if kw not in low:
E.append(f"08-safety-certification/ never mentions '{kw}'")
nh = len(set(re.findall(r"\bH-?\d{1,3}\b", text)))
if nh < 5:
E.append(f"hazard analysis has {nh} numbered hazard(s) (H-01 ...); at least 5 expected")
miss = [i for i in safety_ids(run) if i not in text]
if miss:
E.append(f"safety requirements not traced in 08-safety-certification/: {', '.join(miss[:10])}")
if low.count("http") < 3:
W.append("fewer than 3 links: give each regulatory item its SOURCE")
N.append(f"{nh} hazards")
def part_bom(run, E, W, N):
deliv = os.path.join(run, "deliverables")
rep = dd.Report()
cost = dd.check_cost_json(deliv, rep, required=True)
dd.check_bom(deliv, cost, rep)
E += rep.errors
W += rep.warnings
N += rep.notes
if not os.path.isfile(os.path.join(deliv, "09-bom-cost", "cost.md")):
E.append("09-bom-cost/cost.md missing")
rows = list(csv.reader(io.StringIO(fl.read_text(os.path.join(deliv, "09-bom-cost", "bom.csv")))))
undated = [r[0] for r in rows[1:] if len(r) == 13 and not re.search(r"\d{4}-\d{2}-\d{2}|ASSUMPTION", r[12])]
if undated:
E.append(f"bom.csv: {len(undated)} line(s) without a read date (YYYY-MM-DD) or ASSUMPTION in notes, e.g. items {', '.join(undated[:5])}")
def png_size(path):
with open(path, "rb") as fh:
head = fh.read(24)
if head[:8] != b"\x89PNG\r\n\x1a\n":
return None
return struct.unpack(">II", head[16:24])
def part_drawings(run, E, W, N):
d = os.path.join(run, "deliverables", "04-drawings")
docs = files(d, (".pdf", ".svg"))
if not docs:
E.append("04-drawings/ has no PDF or SVG drawing")
return
names = [os.path.basename(p).lower() for p in docs]
for p in read_parts(run, []):
keys = {p["part"].lower(), os.path.splitext(os.path.basename(p["file"]))[0].lower()}
if not any(k and k in n for k in keys for n in names):
E.append(f"no drawing for designed part '{p['part']}' (file name must contain the part name)")
if not any("assembly" in n or "asm" in n for n in names):
E.append("no assembly drawing (file name containing 'assembly')")
for p in docs:
if p.lower().endswith(".pdf"):
ok = open(p, "rb").read(5) == b"%PDF-"
if ok and shutil.which("pdfinfo"):
ok = sh(["pdfinfo", p], timeout=60)[0] == 0
if not ok:
E.append(f"{os.path.basename(p)} does not open as PDF")
else:
try:
ET.parse(p)
except ET.ParseError as exc:
E.append(f"{os.path.basename(p)} is not valid SVG: {exc}")
N.append(f"{len(docs)} drawing file(s)")
def part_renders(run, E, W, N):
d = os.path.join(run, "deliverables", "11-renders")
pngs = files(d, (".png",))
if len(pngs) < 3:
E.append(f"11-renders/ has {len(pngs)} PNG(s); product, exploded and user-interface views expected")
for p in pngs:
size = png_size(p)
if not size:
E.append(f"{os.path.basename(p)} is not a PNG")
elif size[0] < 1024 or size[1] < 640:
E.append(f"{os.path.basename(p)} is {size[0]}x{size[1]}; at least 1024x640 expected")
elif os.path.getsize(p) < 30000:
W.append(f"{os.path.basename(p)} is under 30 kB - check that the object fills the frame")
glb_check(run, E, W)
def part_testplan(run, E, W, N):
d = os.path.join(run, "deliverables", "10-prototype-test-plan")
text = all_text(d)
if len(text) < 2000:
E.append("10-prototype-test-plan/ missing or shorter than 2000 bytes")
return
miss = [i for i in fl.input_ids(run) + safety_ids(run) if i not in text]
if miss:
E.append(f"test plan has no test for {len(miss)} id(s): {', '.join(miss[:12])}")
low = text.lower()
if "bring-up" not in low and "bring up" not in low:
E.append("test plan has no bring-up section")
if "pass" not in low:
W.append("test plan never states pass criteria")
def part_integration(run, E, W, N):
deliv = os.path.join(run, "deliverables")
p, _ = fl.load_json(os.path.join(deliv, "05-calculations", "params.json"))
params = p.get("params", {}) if isinstance(p, dict) else {}
for rf in files(os.path.join(deliv, "05-calculations"), (".results.json",)):
res, err = fl.load_json(rf)
if err or not isinstance(res, dict):
continue
for k, v in res.items():
pv = params.get(k)
if isinstance(pv, dict) and isinstance(v, dict) and fl.num(pv.get("value")) is not None and fl.num(v.get("value")) is not None \
and str(pv.get("unit")) == str(v.get("unit")) and not fl.same_number(fl.num(pv["value"]), fl.num(v["value"])):
E.append(f"params.json {k} = {pv['value']} {pv['unit']} but {os.path.basename(rf)} computes {v['value']}")
bom = fl.read_text(os.path.join(deliv, "09-bom-cost", "bom.csv")).lower()
for part in read_parts(run, []):
if bom and part["part"].lower() not in bom and os.path.splitext(part["file"])[0].lower() not in bom:
W.append(f"designed part '{part['part']}' not found in bom.csv")
pin = list(csv.reader(io.StringIO(fl.read_text(os.path.join(deliv, "06-electronics", "pinmap.csv")))))
fw = all_text(os.path.join(deliv, "07-firmware"), (".c", ".cpp", ".h", ".hpp", ".ino"))
if pin and fw:
for row in pin[1:]:
if row and row[0].strip() and row[0].strip() not in fw:
W.append(f"pinmap signal '{row[0].strip()}' does not occur in the firmware sources")
N.append(f"{len(params)} params cross-checked against calculation results")
WATCH = {"cad": ["03-cad", "11-renders/assembly.glb"], "sch": ["06-electronics"], "pcb": ["06-electronics"],
"firmware": ["07-firmware"]}
def reusable(run, name):
"""'all' reuses an expensive part's PASS when no watched file changed since that verdict."""
g, err = fl.load_json(os.path.join(run, "work", "gates", f"full-{name}.json"))
if err or not g.get("pass"):
return False
newest = 0.0
for w in WATCH.get(name, []):
p = os.path.join(run, "deliverables", w)
if os.path.isfile(p):
newest = max(newest, os.path.getmtime(p))
for root, _d, names in os.walk(p):
for n in names:
newest = max(newest, os.path.getmtime(os.path.join(root, n)))
return newest and newest < g.get("epoch", 0) - 1
PARTS = {"params": part_params, "cad": part_cad, "sch": part_sch, "pcb": part_pcb, "firmware": part_firmware,
"safety": part_safety, "bom": part_bom, "drawings": part_drawings, "renders": part_renders,
"testplan": part_testplan, "integration": part_integration}
ALL_ORDER = ("params", "safety", "bom", "testplan", "integration", "drawings", "renders", "calc", "cad", "sch", "pcb", "firmware")
NEED_S = {"renders": 60, "calc": 60, "cad": 150, "sch": 60, "pcb": 90, "firmware": 150} # least time worth starting with
def last_verdicts(run, name):
"""The part's own latest gate verdicts (calc: one per group prefix)."""
pat = "full-calc*.json" if name == "calc" else f"full-{name}.json"
out = []
for p in sorted(glob.glob(os.path.join(run, "work", "gates", pat))):
g, err = fl.load_json(p)
if not err and isinstance(g, dict):
out.append(g)
return out
def main():
ap = fl.parser(__doc__)
ap.add_argument("--part", choices=sorted(list(PARTS) + ["calc", "all"]), required=True)
ap.add_argument("--prefix", default="", help="calc: only scripts whose name starts with this prefix")
ap.add_argument("--fresh", action="store_true", help="all: re-run every part even if unchanged since it passed")
ap.add_argument("--budget", type=int, default=0, help="seconds for the checks (default: 540 for one part; 480 for "
"all, whose dd_check gets at most 90 more)")
a = ap.parse_args()
run = os.path.abspath(a.run_dir)
E, W, N = [], [], []
DEADLINE[0] = time.time() + max(60, a.budget or (480 if a.part == "all" else 540))
if a.part == "calc":
run_scripts(os.path.join(run, "deliverables", "05-calculations"), a.prefix, E, W, N)
elif a.part in PARTS:
PARTS[a.part](run, E, W, N)
else:
plan, _ = fl.load_json(os.path.join(run, "work", "full", "plan.json"))
no_board = isinstance(plan, dict) and plan.get("board") is False and not root_file(
os.path.join(run, "deliverables", "06-electronics"), ".kicad_pcb")
for name in ALL_ORDER:
e, w, n = [], [], []
if name == "pcb" and no_board:
W.append("pcb: skipped (work/full/plan.json says the design has no custom board)")
continue
if name in WATCH and not a.fresh and reusable(run, name):
N.append(f"{name}: PASS reused (no file changed since its own gate passed)")
continue
if left() < NEED_S.get(name, 15):
own = last_verdicts(run, name)
if own and all(g.get("pass") for g in own):
W.append(f"{name}: not re-run in this call (time budget); its own gate passed")
else:
E.append(f"{name}: not re-run in this call (time budget) and its own last gate did not pass - "
f"run check_full.py --part {name}")
continue
if name == "calc":
run_scripts(os.path.join(run, "deliverables", "05-calculations"), "", e, w, n)
else:
PARTS[name](run, e, w, n)
E += [f"{name}: {x}" for x in e]
W += [f"{name}: {x}" for x in w]
DEADLINE[0] = None # dd_check (no scripts) is quick and always runs
dd_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dd_check.py")
rc, out = sh([sys.executable, dd_path, "--tier", "full", "--run-dir", run, "--json"], timeout=90)
try:
res = json.loads(out.strip().splitlines()[-1])
E += [f"dd_check: {x}" for x in res.get("errors", [])]
W += [f"dd_check: {x}" for x in res.get("warnings", [])]
except (ValueError, IndexError):
E.append(f"dd_check did not run (exit {rc})")
gate = f"full-{a.part}" + (f"-{a.prefix.rstrip('_')}" if a.prefix else "")
return fl.emit(gate, E, W, N, None, run)
if __name__ == "__main__":
sys.exit(main())
files/method/gates/check_requirements.py (166 wierszy)
#!/usr/bin/env python3
"""check_requirements.py - Gate R: deliverables/01-requirements/ against INPUT/requirements.md.
Usage (from the run directory): python3 method/gates/check_requirements.py [--json]
Checks requirements.md (every INPUT id, the Derived and Conflicts parts, every derived id) and the machine
table requirements.json (format of method/templates/requirements.json): one INPUT entry per INPUT id; numeric
entries with direction, target and threshold copied from the INPUT text in its own unit (the numbers must
occur in that requirement's section), consistent limits, the INPUT priority/weight as weight; derived entries
DR-<AREA>-<nn> with parents, statement, verification method and an origin label; at least 8 derived entries,
at least one DR-SAF-*. Standard library only. Last output line = JSON verdict.
"""
import os
import re
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import flowlib as fl # noqa: E402
VERIFY = {"analysis", "inspection", "demonstration", "test"}
DR_ID = re.compile(r"^DR-[A-Z0-9]{2,8}-\d{1,3}$")
PRIO = re.compile(r"(?im)^\W*priority\b[^0-9\n]{0,12}([0-9]+(?:[.,][0-9]+)?)")
WEIGHT_LINE = re.compile(r"(?im)^\W*(?:weight|importance)\s*[:=]\s*([0-9]+(?:[.,][0-9]+)?)\s*(?:/\s*[0-9]+)?\s*$")
def target_direction(block):
"""Direction stated on the INPUT 'Target:' line, or None when it cannot be read."""
m = re.search(r"(?im)^\W*target\s*[:=]([^\n]{0,60})", block)
if not m:
return None
s = m.group(1).lower()
le = re.search(r"≤|<=|⩽|\bat most\b|\bno more than\b|\bmaximum\b|\bup to\b", s)
ge = re.search(r"≥|>=|⩾|\bat least\b|\bno less than\b|\bminimum\b", s)
if le and (not ge or le.start() < ge.start()):
return "<="
if ge:
return ">="
return None
def check_given(r, ids, blocks, errors, warnings):
rid = r["id"]
if rid not in ids:
errors.append(f"requirements.json: '{rid}' has source INPUT but is not an id of INPUT/requirements.md")
return
w = fl.num(r.get("weight"))
if w is None or w <= 0:
errors.append(f"{rid}: weight must be a number > 0")
block = blocks.get(rid, "")
m = PRIO.search(block) or WEIGHT_LINE.search(block)
if m and w is not None and not fl.same_number(w, float(m.group(1).replace(",", ".")), 1e-6):
errors.append(f"{rid}: weight {w:g} differs from the INPUT priority/weight {m.group(1)}")
kind = r.get("kind")
if kind not in ("numeric", "qualitative"):
errors.append(f"{rid}: kind must be numeric or qualitative")
return
if kind == "qualitative":
if not str(r.get("target_text") or "").strip():
errors.append(f"{rid}: qualitative entry needs target_text (the INPUT target in words)")
return
d = r.get("direction")
t, thr = fl.num(r.get("target")), fl.num(r.get("threshold"))
if d not in ("<=", ">="):
errors.append(f"{rid}: numeric entry needs direction '<=' or '>='")
if t is None:
errors.append(f"{rid}: numeric entry needs a numeric target")
if r.get("threshold") is not None and thr is None:
errors.append(f"{rid}: threshold must be a number or null")
if None in (t, thr) or d not in ("<=", ">="):
pass
elif (d == "<=" and thr < t) or (d == ">=" and thr > t):
errors.append(f"{rid}: threshold {thr:g} is stricter than target {t:g} for direction {d}")
if not isinstance(r.get("unit"), str):
errors.append(f"{rid}: unit must be a string (the INPUT unit)")
if block:
found = fl.numbers_in(block)
for name, v in (("target", t), ("threshold", thr)):
if v is not None and not any(fl.same_number(v, x, 1e-6) for x in found):
errors.append(f"{rid}: {name} {v:g} does not occur in the INPUT text of {rid} (copy the number in the INPUT unit)")
bd = target_direction(block)
if bd and d in ("<=", ">=") and bd != d:
errors.append(f"{rid}: direction {d} contradicts the INPUT target ({bd})")
else:
warnings.append(f"{rid}: no INPUT section found; numbers not cross-checked")
def check_derived(r, md_text, errors):
rid = r["id"]
if not DR_ID.match(rid):
errors.append(f"derived id '{rid}' must look like DR-<AREA>-<nn> (e.g. DR-SAF-01)")
par = r.get("parents")
if not (isinstance(par, list) and par and all(isinstance(p, str) and p.strip() for p in par)):
errors.append(f"{rid}: parents must be a non-empty list (requirement ids, 'brief', 'regulation' or 'manufacturing')")
if not str(r.get("statement") or "").strip():
errors.append(f"{rid}: statement missing")
ver = {v.strip().lower() for v in re.split(r"[+,/;&]| and ", str(r.get("verification") or "")) if v.strip()}
if not ver or not ver <= VERIFY:
errors.append(f"{rid}: verification must be Analysis, Inspection, Demonstration and/or Test (got '{r.get('verification')}')")
if not fl.label_ok(r.get("origin")):
errors.append(f"{rid}: origin must start with SOURCE, CALCULATION, SIMULATION, MEASUREMENT or ASSUMPTION")
if fl.num(r.get("weight")) is None:
errors.append(f"{rid}: weight missing")
if r.get("kind") == "numeric" and (r.get("direction") not in ("<=", ">=") or fl.num(r.get("target")) is None):
errors.append(f"{rid}: numeric derived entry needs direction and numeric target")
if rid not in md_text:
errors.append(f"{rid}: in requirements.json but not in requirements.md")
def main():
a = fl.parser(__doc__).parse_args()
run = os.path.abspath(a.run_dir)
errors, warnings, notes = [], [], []
ids, blocks = fl.input_ids(run), fl.input_blocks(run)
if not ids:
errors.append("no requirement ids found in INPUT/requirements.md")
d = os.path.join(run, "deliverables", "01-requirements")
md = fl.read_text(os.path.join(d, "requirements.md"))
if not md.strip():
errors.append("deliverables/01-requirements/requirements.md missing or empty")
else:
missing = [i for i in ids if i not in md]
if missing:
errors.append(f"requirements.md lacks {len(missing)} INPUT id(s): {', '.join(missing[:12])}")
heads = [ln.lower() for ln in md.splitlines() if ln.lstrip().startswith("#")]
for word in ("derived", "conflict"):
if not any(word in h for h in heads):
errors.append(f"requirements.md has no heading containing '{word}' (handbook section 2 parts)")
if not any("assumption" in ln.lower() for ln in md.splitlines()):
warnings.append("requirements.md has no assumptions register")
data, err = fl.load_json(os.path.join(d, "requirements.json"))
if err:
errors.append(err)
elif not isinstance(data, dict) or not isinstance(data.get("requirements"), list):
errors.append("requirements.json must be an object with requirements[] (see method/templates/requirements.json)")
else:
seen, given, derived = set(), [], []
for r in data["requirements"]:
if not isinstance(r, dict) or not isinstance(r.get("id"), str):
errors.append(f"requirements.json entry without a string id: {str(r)[:80]}")
continue
if r["id"] in seen:
errors.append(f"requirements.json: duplicate id {r['id']}")
seen.add(r["id"])
(given if r.get("source") == "INPUT" else derived).append(r)
if r.get("source") not in ("INPUT", "derived"):
errors.append(f"{r['id']}: source must be INPUT or derived")
have = {r["id"] for r in given}
miss = [i for i in ids if i not in have]
if miss:
errors.append(f"requirements.json lacks {len(miss)} INPUT id(s) with source INPUT: {', '.join(miss[:12])}")
for r in given:
check_given(r, ids, blocks, errors, warnings)
for r in derived:
if r.get("source") == "derived":
check_derived(r, md, errors)
if len(derived) < 8:
errors.append(f"only {len(derived)} derived requirement(s); walk every area of handbook section 2 (at least 8 expected)")
if not any(r["id"].startswith("DR-SAF-") for r in derived):
errors.append("no DR-SAF-* derived safety requirement")
notes.append(f"{len(given)} INPUT entries for {len(ids)} INPUT ids, {len(derived)} derived")
return fl.emit("requirements", errors, warnings, notes, None, run)
if __name__ == "__main__":
sys.exit(main())
files/method/gates/dd_check.py (431 wierszy)
#!/usr/bin/env python3
"""dd_check.py - check deliverables/ against the RUN CONTRACT layout (contract v1.x).
Usage (from the run directory):
python3 method/gates/dd_check.py --tier concept|full [--run-scripts] [--json]
Checks the formats the RUN CONTRACT fixes (compliance.csv, manifest.json, bom.csv, cost.json, the
folder layout of the tier, evidence paths) and prints ERROR / WARN lines. Exit code 0 = no errors.
It checks form, not engineering quality. Standard library only.
"""
import argparse
import csv
import io
import json
import os
import re
import shutil
import subprocess
import sys
COMPLIANCE_HEADER = ["req_id", "status", "value", "unit", "evidence", "note"]
COMPLIANCE_STATUS = {"met", "partly", "not_met", "not_addressed"}
BOM_HEADER = ["item", "qty", "description", "manufacturer", "mpn", "supplier", "supplier_part", "url",
"unit_price_1", "unit_price_100", "unit_price_1000", "currency", "notes"]
HEADING_ID = re.compile(r"^#{1,6}\s+([A-Z][A-Z0-9]{1,7}-[A-Z0-9]{1,7}-\d{1,4})\b")
LOOSE_ID = re.compile(r"\b([A-Z][A-Z0-9]{1,7}-[A-Z]{2,7}-\d{2,4})\b")
FULL_DIRS = ["03-cad", "04-drawings", "05-calculations", "06-electronics", "07-firmware",
"08-safety-certification", "09-bom-cost", "10-prototype-test-plan", "11-renders"]
class Report:
def __init__(self):
self.errors, self.warnings, self.notes = [], [], []
def error(self, msg):
self.errors.append(msg)
def warn(self, msg):
self.warnings.append(msg)
def note(self, msg):
self.notes.append(msg)
def requirement_ids(path):
"""Ids of INPUT/requirements.md: headings first, loose tokens as a fallback."""
if not os.path.isfile(path):
return []
with open(path, encoding="utf-8", errors="replace") as fh:
text = fh.read()
ids = []
for line in text.splitlines():
m = HEADING_ID.match(line.strip())
if m and m.group(1) not in ids:
ids.append(m.group(1))
if not ids:
for m in LOOSE_ID.finditer(text):
if m.group(1) not in ids:
ids.append(m.group(1))
return ids
def nonempty_files(folder, exts=None):
out = []
for root, _dirs, files in os.walk(folder):
for name in files:
p = os.path.join(root, name)
if os.path.getsize(p) == 0:
continue
if exts and not name.lower().endswith(tuple(exts)):
continue
out.append(p)
return out
def split_paths(field):
parts = re.split(r"[;|\n]+|\s+(?=[\w./-]+/)", field or "")
return [p.strip().strip("`'\"") for p in parts if p.strip()]
def evidence_exists(deliv, ref):
ref = ref.split("#", 1)[0].split("::", 1)[0].strip()
if not ref:
return True
if ref.startswith("deliverables/"):
ref = ref[len("deliverables/"):]
ref = ref.lstrip("./")
m = re.match(r"^(.*?)(:\d+(-\d+)?)$", ref)
if m:
ref = m.group(1)
return os.path.exists(os.path.join(deliv, ref))
def check_compliance(deliv, ids, rep):
path = os.path.join(deliv, "01-requirements", "compliance.csv")
if not os.path.isfile(path):
rep.error("01-requirements/compliance.csv missing")
return
raw = open(path, encoding="utf-8-sig", errors="replace").read()
rows = list(csv.reader(io.StringIO(raw)))
if not rows:
rep.error("compliance.csv is empty")
return
header = [h.strip() for h in rows[0]]
if header != COMPLIANCE_HEADER:
rep.error(f"compliance.csv header is {header}, expected {COMPLIANCE_HEADER}")
return
seen = {}
for n, row in enumerate(rows[1:], start=2):
if not any(c.strip() for c in row):
continue
if len(row) != len(COMPLIANCE_HEADER):
rep.error(f"compliance.csv line {n}: {len(row)} fields, expected {len(COMPLIANCE_HEADER)} (quote fields with commas)")
continue
rid, status, _value, _unit, evidence, _note = [c.strip() for c in row]
if rid in seen:
rep.error(f"compliance.csv line {n}: duplicate id {rid} (first at line {seen[rid]})")
seen[rid] = n
if status not in COMPLIANCE_STATUS:
rep.error(f"compliance.csv line {n}: status '{status}' not in {sorted(COMPLIANCE_STATUS)}")
if status != "not_addressed" and not evidence:
rep.error(f"compliance.csv line {n} ({rid}): status {status} without evidence path")
for ref in split_paths(evidence):
if not evidence_exists(deliv, ref):
rep.error(f"compliance.csv line {n} ({rid}): evidence path not found: {ref}")
if ids:
missing = [i for i in ids if i not in seen]
extra = [i for i in seen if i not in ids]
if missing:
rep.error(f"compliance.csv lacks {len(missing)} requirement id(s): {', '.join(missing)}")
if extra:
rep.warn(f"compliance.csv has ids not in INPUT/requirements.md: {', '.join(extra)}")
rep.note(f"compliance.csv: {len(seen)} rows for {len(ids)} requirement ids")
else:
rep.warn("no requirement ids found in INPUT/requirements.md; row coverage not checked")
def check_ids_in_requirements(deliv, ids, rep):
folder = os.path.join(deliv, "01-requirements")
texts = ""
for p in nonempty_files(folder, (".md", ".yaml", ".yml", ".txt")):
texts += open(p, encoding="utf-8", errors="replace").read() + "\n"
if not texts:
rep.error("01-requirements/ has no requirements document (.md/.yaml) besides compliance.csv")
return
absent = [i for i in ids if i not in texts]
if absent:
rep.warn(f"requirement ids not mentioned in the 01-requirements documents: {', '.join(absent)}")
def num_or_null(x):
return x is None or isinstance(x, (int, float)) and not isinstance(x, bool)
def check_manifest(deliv, tier, rep):
path = os.path.join(deliv, "manifest.json")
if not os.path.isfile(path):
rep.error("manifest.json missing")
return
try:
man = json.load(open(path, encoding="utf-8"))
except Exception as exc: # noqa: BLE001
rep.error(f"manifest.json does not parse: {exc}")
return
if man.get("tier") != tier:
rep.error(f"manifest.json tier is {man.get('tier')!r}, expected {tier!r}")
files = man.get("files")
if not isinstance(files, list) or not files:
rep.error("manifest.json files[] missing or empty")
else:
listed = set()
for f in files:
if not isinstance(f, dict) or not {"path", "kind", "desc"} <= set(f):
rep.error(f"manifest.json files[] entry lacks path/kind/desc: {f}")
continue
listed.add(f["path"].lstrip("./"))
if not evidence_exists(deliv, f["path"]):
rep.error(f"manifest.json lists a file that does not exist: {f['path']}")
actual = set()
for p in nonempty_files(deliv):
rel = os.path.relpath(p, deliv)
if rel != "manifest.json":
actual.add(rel)
unlisted = sorted(a for a in actual if a not in listed and not any(a.startswith(l.rstrip("/") + "/") for l in listed))
if unlisted:
rep.warn(f"{len(unlisted)} deliverable file(s) not listed in manifest.json, e.g. {unlisted[:5]}")
kf = man.get("key_figures")
if not isinstance(kf, dict):
rep.error("manifest.json key_figures missing")
else:
for key in ("dimensions_mm", "mass_kg", "power_w", "unit_cost", "retail_price"):
if key not in kf:
rep.error(f"manifest.json key_figures.{key} missing (use null when unknown)")
dims = kf.get("dimensions_mm")
if dims is not None and not (isinstance(dims, list) and len(dims) == 3 and all(num_or_null(d) for d in dims)):
rep.error("key_figures.dimensions_mm must be [L, W, H] numbers or null")
for key in ("mass_kg", "power_w"):
if key in kf and not num_or_null(kf[key]):
rep.error(f"key_figures.{key} must be a number or null")
uc = kf.get("unit_cost")
if uc is not None:
if not isinstance(uc, dict) or not {"currency", "q1", "q100", "q1000"} <= set(uc):
rep.error("key_figures.unit_cost needs currency, q1, q100, q1000")
elif not all(num_or_null(uc[q]) for q in ("q1", "q100", "q1000")):
rep.error("key_figures.unit_cost q1/q100/q1000 must be numbers or null")
rp = kf.get("retail_price")
if rp is not None and (not isinstance(rp, dict) or not {"currency", "value"} <= set(rp) or not num_or_null(rp.get("value"))):
rep.error("key_figures.retail_price needs currency and a numeric value (or null)")
blocked = man.get("blocked")
if not isinstance(blocked, list):
rep.error("manifest.json blocked[] missing (use [] when nothing is blocked)")
else:
for b in blocked:
if not isinstance(b, dict) or not {"what", "why"} <= set(b):
rep.error(f"manifest.json blocked[] entry needs what and why: {b}")
def check_cost_json(deliv, rep, required):
path = os.path.join(deliv, "09-bom-cost", "cost.json")
if not os.path.isfile(path):
if required:
rep.error("09-bom-cost/cost.json missing")
return None
try:
cost = json.load(open(path, encoding="utf-8"))
except Exception as exc: # noqa: BLE001
rep.error(f"cost.json does not parse: {exc}")
return None
for key in ("currency", "bom", "unit_cost"):
if key not in cost:
rep.error(f"cost.json lacks {key}")
for key in ("bom", "unit_cost"):
block = cost.get(key)
if isinstance(block, dict) and not all(num_or_null(block.get(q)) for q in ("q1", "q100", "q1000")):
rep.error(f"cost.json {key} needs numeric q1, q100, q1000")
other = cost.get("other", {})
if not isinstance(other, dict):
rep.error("cost.json other must be an object of {name: {q1, q100, q1000}}")
else:
try:
for q in ("q1", "q100", "q1000"):
total = float(cost["bom"][q]) + sum(float(v.get(q) or 0) for v in other.values())
unit = float(cost["unit_cost"][q])
if unit > 0 and abs(total - unit) / unit > 0.02:
rep.warn(f"cost.json unit_cost.{q}={unit} differs from bom+other={total:.2f} by >2 %")
except (KeyError, TypeError, ValueError):
pass
return cost
def as_float(text):
try:
return float(str(text).replace(",", ".").strip())
except ValueError:
return None
def check_bom(deliv, cost, rep):
path = os.path.join(deliv, "09-bom-cost", "bom.csv")
if not os.path.isfile(path):
rep.error("09-bom-cost/bom.csv missing")
return
rows = list(csv.reader(io.StringIO(open(path, encoding="utf-8-sig", errors="replace").read())))
if not rows or [h.strip() for h in rows[0]] != BOM_HEADER:
rep.error(f"bom.csv header must be exactly {','.join(BOM_HEADER)}")
return
totals = {"unit_price_1": 0.0, "unit_price_100": 0.0, "unit_price_1000": 0.0}
currencies = set()
lines = 0
for n, row in enumerate(rows[1:], start=2):
if not any(c.strip() for c in row):
continue
if len(row) != len(BOM_HEADER):
rep.error(f"bom.csv line {n}: {len(row)} fields, expected {len(BOM_HEADER)}")
continue
rec = dict(zip(BOM_HEADER, [c.strip() for c in row]))
lines += 1
qty = as_float(rec["qty"])
if qty is None:
rep.error(f"bom.csv line {n}: qty '{rec['qty']}' is not a number")
continue
if not rec["mpn"] and not rec["supplier_part"]:
rep.warn(f"bom.csv line {n}: neither mpn nor supplier_part ({rec['description'][:40]})")
if rec["url"] and not rec["url"].startswith(("http://", "https://")):
rep.warn(f"bom.csv line {n}: url is not http(s)")
currencies.add(rec["currency"])
for col in totals:
price = as_float(rec[col])
if price is None:
rep.warn(f"bom.csv line {n}: {col} '{rec[col]}' is not a number")
else:
totals[col] += qty * price
rep.note(f"bom.csv: {lines} lines; sums 1/100/1000 = {totals['unit_price_1']:.2f} / "
f"{totals['unit_price_100']:.2f} / {totals['unit_price_1000']:.2f} {','.join(sorted(currencies))}")
if len(currencies) > 1:
rep.warn(f"bom.csv mixes currencies {sorted(currencies)}; totals not comparable")
elif cost and isinstance(cost.get("bom"), dict):
for q, col in (("q1", "unit_price_1"), ("q100", "unit_price_100"), ("q1000", "unit_price_1000")):
ref = as_float(cost["bom"].get(q))
if ref and abs(ref - totals[col]) / ref > 0.02:
rep.warn(f"cost.json bom.{q}={ref} differs from the bom.csv sum {totals[col]:.2f} by >2 %")
def toolchain_python():
"""The toolchain Python the RUN CONTRACT names (ddpy), else the venv, else this interpreter."""
for cand in (shutil.which("ddpy"), "/opt/dd/venv/bin/python"):
if cand and os.path.exists(cand):
return cand
return sys.executable
def check_scripts(folder, rep, run):
scripts = [p for p in nonempty_files(folder, (".py",))]
for script in scripts:
stem = os.path.splitext(script)[0]
results = stem + ".results.json"
if run:
try:
proc = subprocess.run([toolchain_python(), os.path.basename(script)], cwd=os.path.dirname(script),
capture_output=True, text=True, timeout=300)
if proc.returncode != 0:
rep.error(f"{os.path.relpath(script)} exits {proc.returncode}: {proc.stderr.strip()[-200:]}")
except subprocess.TimeoutExpired:
rep.error(f"{os.path.relpath(script)} runs longer than 5 minutes")
if not os.path.isfile(results):
rep.warn(f"{os.path.relpath(script)} has no {os.path.basename(results)}")
continue
try:
data = json.load(open(results, encoding="utf-8"))
bad = [k for k, v in data.items() if not (isinstance(v, dict) and "value" in v and "unit" in v)]
if bad:
rep.warn(f"{os.path.relpath(results)}: entries without value/unit: {bad[:5]}")
except Exception as exc: # noqa: BLE001
rep.error(f"{os.path.relpath(results)} does not parse: {exc}")
def check_full(deliv, rep, run):
for d in FULL_DIRS:
if not nonempty_files(os.path.join(deliv, d)):
rep.error(f"{d}/ missing or empty")
cad = os.path.join(deliv, "03-cad")
asm = os.path.join(cad, "assembly.step")
if not os.path.isfile(asm):
rep.error("03-cad/assembly.step missing")
else:
if not open(asm, "rb").read(64).lstrip().startswith(b"ISO-10303-21"):
rep.error("03-cad/assembly.step is not a STEP file (no ISO-10303-21 header)")
steps = [p for p in nonempty_files(cad, (".step", ".stp")) if not p.endswith("assembly.step")]
if not steps:
rep.error("03-cad/ has no per-part STEP files")
if not nonempty_files(cad, (".py", ".fcstd")):
rep.error("03-cad/ has no parametric source (.py or .FCStd)")
if not nonempty_files(os.path.join(deliv, "04-drawings"), (".pdf", ".svg")):
rep.error("04-drawings/ has no PDF or SVG drawing")
check_scripts(os.path.join(deliv, "05-calculations"), rep, run)
ele = os.path.join(deliv, "06-electronics")
if not nonempty_files(ele, (".kicad_pro",)) or not nonempty_files(ele, (".kicad_sch",)):
rep.error("06-electronics/ lacks a KiCad project (.kicad_pro + .kicad_sch)")
names = " ".join(os.path.basename(p).lower() for p in nonempty_files(ele))
if "erc" not in names:
rep.warn("06-electronics/ has no file named like an ERC report")
if nonempty_files(ele, (".kicad_pcb",)) and "drc" not in names:
rep.warn("06-electronics/ has a board but no file named like a DRC report")
if not os.path.isfile(os.path.join(deliv, "07-firmware", "platformio.ini")):
rep.error("07-firmware/platformio.ini missing")
cost = check_cost_json(deliv, rep, required=True)
check_bom(deliv, cost, rep)
if not os.path.isfile(os.path.join(deliv, "09-bom-cost", "cost.md")):
rep.error("09-bom-cost/cost.md missing")
ren = os.path.join(deliv, "11-renders")
if not nonempty_files(ren, (".png",)):
rep.error("11-renders/ has no PNG render")
glb = os.path.join(ren, "assembly.glb")
if not os.path.isfile(glb):
rep.error("11-renders/assembly.glb missing")
elif open(glb, "rb").read(4) != b"glTF":
rep.error("11-renders/assembly.glb is not a binary glTF file")
def main():
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--tier", choices=["concept", "full"], required=True)
ap.add_argument("--run-dir", default=".", help="run directory (default: current directory)")
ap.add_argument("--run-scripts", action="store_true", help="execute calculation scripts (≤5 min each)")
ap.add_argument("--json", action="store_true", help="print a JSON summary as the last line")
args = ap.parse_args()
run_dir = os.path.abspath(args.run_dir)
deliv = os.path.join(run_dir, "deliverables")
rep = Report()
if not os.path.isdir(deliv):
rep.error("deliverables/ missing")
else:
ids = requirement_ids(os.path.join(run_dir, "INPUT", "requirements.md"))
if not os.path.isdir(os.path.join(deliv, "01-requirements")):
rep.error("01-requirements/ missing")
else:
check_compliance(deliv, ids, rep)
check_ids_in_requirements(deliv, ids, rep)
if not nonempty_files(os.path.join(deliv, "02-concept"), (".md",)):
rep.error("02-concept/ has no markdown document")
report = os.path.join(deliv, "REPORT.md")
if not os.path.isfile(report):
rep.error("REPORT.md missing")
elif os.path.getsize(report) < 1500:
rep.warn("REPORT.md is shorter than 1500 bytes")
check_manifest(deliv, args.tier, rep)
check_scripts(os.path.join(deliv, "02-concept"), rep, args.run_scripts)
if args.tier == "concept":
check_cost_json(deliv, rep, required=False)
started = [d for d in FULL_DIRS if d != "09-bom-cost" and nonempty_files(os.path.join(deliv, d))]
if started:
rep.warn(f"concept tier: detailed-design folders present ({', '.join(started)}); they are not judged at this tier")
else:
check_full(deliv, rep, args.run_scripts)
for msg in rep.notes:
print(f"NOTE {msg}")
for msg in rep.warnings:
print(f"WARN {msg}")
for msg in rep.errors:
print(f"ERROR {msg}")
print(f"dd_check: tier={args.tier} errors={len(rep.errors)} warnings={len(rep.warnings)} -> "
f"{'PASS' if not rep.errors else 'FAIL'}")
if args.json:
print(json.dumps({"tier": args.tier, "pass": not rep.errors, "errors": rep.errors,
"warnings": rep.warnings, "notes": rep.notes}))
return 0 if not rep.errors else 1
if __name__ == "__main__":
sys.exit(main())
files/method/gates/flowlib.py (184 wierszy)
#!/usr/bin/env python3
"""flowlib.py - shared helpers of the FLOWCODE gates and tools (standard library only).
Every gate or tool prints NOTE / WARN / ERROR lines and, as its LAST line, one JSON verdict
{"gate", "pass", "errors", "warnings", "epoch", ...}; the same object is saved to work/gates/<gate>.json.
Paths inside the method's JSON files are relative to the run directory.
"""
import argparse
import json
import os
import re
import sys
import time
GATES_DIR = os.path.dirname(os.path.abspath(__file__))
if GATES_DIR not in sys.path:
sys.path.insert(0, GATES_DIR)
import dd_check # noqa: E402 shared block: requirement ids and RUN CONTRACT format checks
LABELS = ("SOURCE", "CALCULATION", "SIMULATION", "MEASUREMENT", "ASSUMPTION")
LEVELS = ("target", "threshold", "miss", "unknown")
_NUM = re.compile(r"(?<![\w.])[-+]?\d+(?:[ ]\d{3})*(?:[.,]\d+)?(?!\d)")
def parser(doc):
ap = argparse.ArgumentParser(description=doc.strip().splitlines()[0])
ap.add_argument("--run-dir", default=".", help="run directory (default: current directory)")
ap.add_argument("--json", action="store_true", help="accepted for symmetry; the JSON verdict is always printed last")
return ap
def load_json(path):
"""(data, error_text); never raises."""
if not os.path.isfile(path):
return None, f"{rel(path)} missing"
try:
with open(path, encoding="utf-8") as fh:
return json.load(fh), None
except Exception as exc: # noqa: BLE001
return None, f"{rel(path)} does not parse: {exc}"
def write_json(path, data):
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(data, fh, indent=1, ensure_ascii=False)
fh.write("\n")
os.replace(tmp, path)
def read_text(path):
try:
with open(path, encoding="utf-8", errors="replace") as fh:
return fh.read()
except OSError:
return ""
def rel(path, start=None):
try:
return os.path.relpath(path, start or os.getcwd())
except ValueError:
return path
def input_ids(run):
return dd_check.requirement_ids(os.path.join(run, "INPUT", "requirements.md"))
def input_blocks(run):
"""{requirement id: the text of its section in INPUT/requirements.md}."""
lines = read_text(os.path.join(run, "INPUT", "requirements.md")).splitlines()
blocks, cur = {}, None
for line in lines:
s = line.strip()
m = dd_check.HEADING_ID.match(s)
if m:
cur = m.group(1)
blocks.setdefault(cur, []).append(s)
elif s.startswith("#"):
cur = None
elif cur:
blocks[cur].append(s)
if not blocks: # loose format: a block runs from a line with an id to the next line with another id
for line in lines:
ids = dd_check.LOOSE_ID.findall(line)
if ids:
cur = ids[0]
if cur:
blocks.setdefault(cur, []).append(line.strip())
return {k: "\n".join(v) for k, v in blocks.items()}
def numbers_in(text):
out = []
for tok in _NUM.findall(text or ""):
t = re.sub(r"[ ]", "", tok)
cands = [t.replace(",", ".")]
if re.fullmatch(r"[-+]?\d{1,3}(,\d{3})+", t):
cands.append(t.replace(",", ""))
for c in cands:
try:
out.append(float(c))
except ValueError:
pass
return out + [abs(x) for x in out if x < 0]
def same_number(a, b, rel_tol=0.01):
return abs(a - b) <= 1e-9 + rel_tol * max(abs(a), abs(b))
def num(x):
if isinstance(x, bool):
return None
if isinstance(x, (int, float)):
return float(x)
if isinstance(x, str):
try:
return float(x.strip().replace(",", "."))
except ValueError:
return None
return None
def label_ok(origin):
return isinstance(origin, str) and origin.strip().split(" ")[0].split(":")[0].upper() in LABELS
def requirements(run):
"""(INPUT requirements in INPUT order as dicts, warning or None).
Entries come from deliverables/01-requirements/requirements.json; an INPUT id missing there becomes a
qualitative requirement of weight 1 so that scoring and packaging still run."""
data, err = load_json(os.path.join(run, "deliverables", "01-requirements", "requirements.json"))
table = {}
if isinstance(data, dict) and isinstance(data.get("requirements"), list):
for r in data["requirements"]:
if isinstance(r, dict) and isinstance(r.get("id"), str):
table.setdefault(r["id"], r)
ids = input_ids(run) or [k for k, r in table.items() if r.get("source") == "INPUT"]
reqs, missing = [], []
for i in ids:
r = table.get(i)
if r is None:
missing.append(i)
r = {"id": i, "source": "INPUT", "kind": "qualitative", "weight": 1}
reqs.append(r)
warn = None
if err:
warn = f"{err}; every requirement scored as qualitative with weight 1"
elif missing:
warn = f"requirements.json lacks {len(missing)} INPUT id(s) (scored as qualitative, weight 1): {', '.join(missing[:8])}"
return reqs, warn
def cap(items, n=40):
items = list(items)
return items if len(items) <= n else items[:n] + [f"... and {len(items) - n} more (run the command to see all)"]
def emit(gate, errors, warnings=(), notes=(), extra=None, run=None):
"""Print the verdict (JSON last), save it to work/gates/<gate>.json, return the exit code."""
errors, warnings = list(errors), list(warnings)
for n in notes:
print(f"NOTE {n}")
for w in warnings:
print(f"WARN {w}")
for e in errors:
print(f"ERROR {e}")
res = {"gate": gate, "pass": not errors, "errors": cap(errors), "warnings": cap(warnings),
"epoch": int(time.time())}
if extra:
res.update(extra)
if run:
try:
write_json(os.path.join(run, "work", "gates", f"{gate}.json"), res)
except OSError:
pass
print(f"{gate}: errors={len(errors)} warnings={len(warnings)} -> {'PASS' if not errors else 'FAIL'}")
print(json.dumps(res, ensure_ascii=False))
return 0 if not errors else 1
files/method/roles/architecture.md (24 wierszy)
# Architecture freeze (stage P, full tier)
Inputs: `deliverables/02-concept/*` (concept, selection, `calc/`), `deliverables/01-requirements/*`, `work/final/*`,
`work/deep/<winner>/design.md`. Handbook §5–§12.
Write:
1. `deliverables/05-calculations/params.json`, the single home of every number that later stages share:
```
{"schema": "flow-params/1",
"params": {"<snake_case_name_with_unit>": {"value": x, "unit": "...", "origin": "<LABEL> <path or reason>",
"owner": "architecture|cad|calc|sch|pcb|firmware|safety|bom", "note": "..."}}}
```
At least: the envelope and wall thickness, mounting points, the PCB outline and its mounting holes, connector
positions, supply voltage and current limits, the power budget per operating mode, thermal limits, the MCU and
its PlatformIO environment, key component MPNs, materials and densities, the target unit cost. Later stages
read numbers from here and never retype them; a stage may add the params it owns.
2. `deliverables/05-calculations/interfaces.md`: every interface between subsystems (mechanical, electrical,
thermal, data, user) with the param names it uses.
3. `work/full/plan.json`: `{"board": true|false, "mcu_env": "...", "calc_groups": [{"name": "<one lowercase
word>", "questions": ["..."]}]}` with one to four groups of calculations detailed design must run (the
concept's key estimates, every requirement verified by calculation, every safety limit). Return the same plan.
Gate: `python3 method/gates/check_full.py --part params`
files/method/roles/bom.md (9 wierszy)
# BOM and cost stage (full tier): `deliverables/09-bom-cost/` Handbook §12. `bom.csv` with exactly the RUN CONTRACT header; every line a real part (offline catalogue first, then distributor pages, read-only) with the date read in `notes`, or an ASSUMPTION with its range; designed parts from `03-cad/parts.csv` with their process cost; the KiCad BOM when it exists. `cost.json` (bom = the bom.csv sums; other lines; unit_cost) and `cost.md` (roll-up at 1/100/1000, tooling and certification amortisation, retail price logic); `work/full/key_figures-cost.json` with `unit_cost` and `retail_price`. Gate: `python3 method/gates/check_full.py --part bom`
files/method/roles/cad.md (11 wierszy)
# CAD stage (full tier): `deliverables/03-cad/` and `deliverables/11-renders/assembly.glb` Handbook §7. One build123d or CadQuery script per designed part plus an assembly script, parameters read from `params.json`; a STEP per designed part and `assembly.step`; purchased parts as labelled simplified bodies with datasheet envelopes; `parts.csv` (`part,file,material,density_kg_m3,qty`: designed parts only, each file name containing the part name). Check every STEP with `step-info` and the assembly with `--interference`; list intended fits in `03-cad/fits.txt` (`partA | partB: reason`). Mass from volume × density; bounding box against the size requirements; write both to `work/full/cad-figures.json`. Export `deliverables/11-renders/assembly.glb` with `step2glb`. Gate: `python3 method/gates/check_full.py --part cad`
files/method/roles/calc.md (8 wierszy)
# Calculation stage (full tier): `deliverables/05-calculations/<group>_<question>.py` Handbook §6. One script per question of your group, with a docstring line `Requirements: <ids>`; inputs read from `params.json` with their origin; `pint` units; a known-answer check for any new analysis route; runs as `ddpy <script>` from that folder, offline, under five minutes; writes `<stem>.results.json`; states value, requirement, margin and status. Name a result like a param only when it computes that param. Gate: `python3 method/gates/check_full.py --part calc --prefix <group>_`
files/method/roles/concepts.md (40 wierszy)
# Concepts (stages E and F): six lenses in parallel; each agent writes its concept, then screens it with its own script
Inputs: `work/morph.json` (your seed), `deliverables/02-concept/physics-map.md`,
`deliverables/02-concept/benchmark.md`, `work/research/parts.md`, `work/intake/*.md`, and `INPUT/requirements.md`
for ids, targets, thresholds and weights (`01-requirements/` is being written at the same time). Handbook §5 and §6.
Spend about 60 % of your time box writing and 40 % screening.
## 1. Write
Grow your seed into the concept that best serves your lens while respecting every threshold you can. You may
change at most one seed choice; say which and why. Do not mention the lens or the other concepts in the `.md`: a
blind panel reads it.
Write `work/concepts/C<i>.md` (about 900 words at most): name; the principle in one sentence; how it works; a
Mermaid block diagram; the morph choices; key components with real candidates from `parts.md` (MPN, price, date);
first-order numbers for the highest-weighted requirements, each with its origin and a short hand calculation;
hazards and their protections (say which act without firmware); rough unit cost at 1000 with its basis; the top
three risks.
Write `work/concepts/C<i>.json` in the format of `method/templates/concept.json`: `morph` = your choices; an
estimate for EVERY INPUT requirement id, in that requirement's own unit, each with an `origin`; numeric
requirements get `value` plus `low`/`high`; qualitative ones get `level` (`target`, `threshold`, `miss`,
`unknown`) and optionally a `value` text. Use `unknown` only when no honest estimate is possible. `safety_ko` is a
list of the hazards you could not give an adequate protection (`[]` when none). Set `cost_q1000`, `currency`,
`feasible: true`, `screened: false`.
## 2. Screen (you measure your own numbers; you do not improve them)
1. Choose the 3–8 estimates that decide the ranking: the highest-weighted numeric requirements, every estimate
within 20 % of its threshold, and any number that looks optimistic.
2. Write `work/concepts/C<i>_calc.py` (run with `ddpy` from `work/concepts/`, `pint` units, every input labelled
with its origin at the top) that recomputes them from first principles and writes
`work/concepts/C<i>_calc.results.json` (`{"<key>": {"value": x, "unit": "..."}}`). Make a hand estimate first
and explain any gap.
3. Update `C<i>.json` in place: the script's value for each screened estimate with `origin: "CALCULATION"`,
`basis: "work/concepts/C<i>_calc.results.json#<key>"` and `low`/`high` from the script's input ranges;
`feasible: false` only when physics rules the concept out; `flags`: one line per problem found;
`screened: true`. A screened number that came out worse stays worse: that is the point of the screening.
4. Append a section "Screening" to `C<i>.md`: a table claimed vs hand estimate vs script, and the verdict.
Self-check: `python3 method/gates/check_concept.py --stage concept --id C<i>` (it runs your script).
files/method/roles/deepen.md (42 wierszy)
# Deepen (stage H): one of the two leading concepts, in parallel with the other; grafts and final calculation homes
Inputs: your concept's `work/concepts/C<i>.md`, `.json` and `_calc.*`; the graft source named in your prompt (its
screened dossier `work/concepts/C<j>.md` and `.json` only: it is being deepened at the same time);
`deliverables/02-concept/physics-map.md`, `deliverables/02-concept/benchmark.md`, `work/research/*.md`,
`work/intake/*.md`, `deliverables/01-requirements/requirements.json`. Handbook §5, §6, §11, §12;
`method/POLICY.md` §2. Keep the concept's working principle: deepen it, do not redesign it. Never edit
`work/concepts/` or another `work/deep/` folder. After you finish, code picks the winner from the estimates and
copies the winner's folder into the package (`calc/` into `deliverables/02-concept/calc/`, the estimates and key
figures into `work/final/`): write everything as final.
Write into `work/deep/C<i>/` only:
1. `design.md`: architecture and subsystems (make or buy) with a Mermaid block diagram; the interfaces
(mechanical, electrical, thermal, data) with numbers; hazards with their protections (firmware-independent
where harm is serious); a compliance route sketch; a rough BOM of the major items (at most 15 lines) with real
parts (MPN, supplier, URL, price at 1/100/1000, date read) or labelled ASSUMPTION ranges where a price could
not be read; unit cost at 1/100/1000 including the non-BOM lines; the top five risks, each with the analysis or
test that would retire it; the grafts you took and why.
2. Grafts: take zero to three features of the graft source. Accept one only if it improves a weighted requirement
or removes a risk, breaks no threshold and fits your architecture; re-estimate every requirement it affects.
3. `calc/`: the calculation homes of your key estimates, one script per question (start from your concept's
screening script; never make a result nicer than it is): inputs labelled with their origin at the top, a
docstring whose first line says what it computes and a line `Requirements: <ids>`, `pint` units, runnable as
`ddpy <script>` from `calc/`, offline, under five minutes, writing `<stem>.results.json`
(`{"<key>": {"value": x, "unit": "..."}}`). Self-contained: a script reads only files inside `calc/` (the
folder is copied). Run every script.
4. `estimates.json` (format `method/templates/concept.json`, `id` = your concept): one estimate per INPUT
requirement id, in its own unit, each with an `origin`; resolve unknowns where you can; every calculated value
has `origin: "CALCULATION"`, `basis: "work/deep/C<i>/calc/<stem>.results.json#<key>"` and equals that result;
`cost_q1000` from your BOM; `safety_ko` (a list, `[]` when every hazard has an adequate protection);
`grafted: [{"from", "feature", "why", "affects"}]` (at most three, from the graft source);
`known_misses: [{"id", "why"}]` for every requirement predicted to miss its threshold (or its target when it
has no threshold). A level that rests only on judgement is an ASSUMPTION, and `compliance.csv` reports an
assumed "meets the target" as `partly`: show the level by a source or a calculation where you can.
5. `key_figures.json`: `{"dimensions_mm": [L, W, H] or null, "mass_kg": x or null, "power_w": x or null,
"origins": {"<key>": "<label and path>"}}`.
Self-check: `python3 method/gates/check_concept.py --stage deep --id C<i>` (it runs your scripts first).
Fix round (only when your prompt says FIX ROUND): the final check rejected the copy of your folder in the package;
fix the causes in `work/deep/C<i>/` (never in `deliverables/`), keep the design, re-run the self-check.
files/method/roles/drawings.md (9 wierszy)
# Drawings stage (full tier): `deliverables/04-drawings/` Handbook §8. A drawing for every designed part in `03-cad/parts.csv` (the file name contains the part name) and an assembly drawing (file name contains "assembly") with balloons matching the `bom.csv` item numbers; views, dimensions, a general tolerance class plus explicit tolerances on functional features, material, finish, quantity, revision. Route A (build123d projection → SVG → PDF) or route B (FreeCAD TechDraw), see `INPUT/TOOLCHAIN.md`. Render a PNG preview of each drawing and check that it is not empty. Gate: `python3 method/gates/check_full.py --part drawings`
files/method/roles/electronics.md (21 wierszy)
# Electronics stages (full tier): `deliverables/06-electronics/` Two stages use this brief: "sch" first, then "pcb" (beside firmware and BOM). ## sch: `deliverables/06-electronics/` (schematic) Handbook §9. Block diagram and power budget over all modes first, in `06-electronics/design.md`. A KiCad 9 schematic written headlessly by the route of `INPUT/TOOLCHAIN.md` and its KiCad example (own `KICAD_CONFIG_HOME=work/kicad-config`; project `sym-lib-table` and `fp-lib-table`); real parts with MPNs and footprints; protections (fuse or current limit, reverse polarity, ESD and transients on external connectors, inrush, a hardware thermal cut-off for hot parts). ERC as JSON in `erc.json` with zero errors (waivers by violation type with reasons in `erc-waivers.md`); schematic PDF; BOM export; `pinmap.csv` (`signal,mcu_pin,net,direction`) for the firmware. Gate: `python3 method/gates/check_full.py --part sch` ## pcb: `deliverables/06-electronics/` (board) Handbook §9. Outline and mounting holes from `params.json`; placement, routing (freerouting), zones filled; DRC as JSON in `drc.json` with schematic parity, zero errors and zero unconnected items (waivers by type in `drc-waivers.md`); `fab/` with Gerbers (a zip is fine), drill files and the pick-and-place file (name containing `pos`); a board STEP for CAD, and the `pcb_*` params you own. Gate: `python3 method/gates/check_full.py --part pcb`
files/method/roles/firmware.md (9 wierszy)
# Firmware stage (full tier): `deliverables/07-firmware/` Handbook §10. A PlatformIO project for the schematic's MCU, platform version pinned exactly as `INPUT/TOOLCHAIN.md` lists; pins from `06-electronics/pinmap.csv` with the signal names as identifiers; a state machine with explicit fault states, a safe default, a watchdog and debounced inputs; constants from `params.json`; an `[env:native]` with Unity tests in `test/` for the state machine and the interlocks (write the failing test first). Never upload. Leave no `.pio/` folder behind. Gate: `python3 method/gates/check_full.py --part firmware`
files/method/roles/fixer.md (11 wierszy)
# Fixer: repairs what a gate rejected, without changing the design Input: the gate errors in your prompt. Fix form and consistency only: missing or misnamed files, broken paths, JSON or schema errors, a number quoted differently from its home (take the home's value), missing requirement ids, missing headings or sections whose content already exists elsewhere in the package. Never change a design decision, a requirement, a requirement's numbers or a home value just to pass; never delete content to silence a check. When an error needs real engineering or new research, do the smallest honest piece that resolves it within your time box; otherwise append one line per unresolved error to `work/open-issues.md`. Re-run the gate command from your prompt, then return.
files/method/roles/full-common.md (14 wierszy)
# Rules for every full-tier stage
- Numbers come from `deliverables/05-calculations/params.json`, the single home of shared numbers: add the params
you own, never retype another owner's number. Read `deliverables/05-calculations/interfaces.md` and
`deliverables/02-concept/concept.md` first.
- Tool commands and working examples are in `INPUT/TOOLCHAIN.md`: read the example of your route before writing
your own. A broken tool route gets three honest attempts with new information each time, then a labelled
fallback; put an item in `work/blocked/<stage>.json` (`[{"what": "...", "why": "..."}]`) only when a human is
truly needed (a purchase, a physical test, a credential).
- Write only inside your own folders. Record the requirement ids your files evidence in
`work/full/evidence-<stage>.json` (`{"<id>": ["deliverables/..."]}`) and describe your files in
`work/manifest-desc/<stage>.json` (`{"deliverables/<path>": "<one line>"}`).
- When your prompt lists RETRY, INTEGRATION ISSUES or FINAL-GATE ERRORS, fix exactly those in your own files.
files/method/roles/integration.md (15 wierszy)
# Integration (stage R): find cross-discipline inconsistencies; never fix them yourself
Run `python3 method/gates/check_full.py --part integration` and read its output (params against calculation
results, designed parts against the BOM, pin map against the firmware). Then check by reading the files:
- the CAD envelope and mounting against the PCB outline, holes and connector positions;
- `bom.csv` against `03-cad/parts.csv` and the KiCad BOM;
- the power budget against the supply rating and the thermal calculations;
- `pinmap.csv` against the schematic and the firmware sources;
- `params.json` against every consumer (retyped numbers, stale copies);
- the safety measures claimed in `08-safety-certification/` against what CAD, electronics and firmware implement.
Write `work/full/integration-<round>.json`: `{"issues": [{"owner": "cad|calc|sch|pcb|firmware|safety|bom",
"issue": "<what is inconsistent, where (paths), and the concrete fix>"}]}`. List only real inconsistencies with a
concrete fix, most important first, at most twelve, and return the same list. An empty list ends the loop.
files/method/roles/renders.md (8 wierszy)
# Renders stage (full tier): `deliverables/11-renders/` Handbook §14. From `deliverables/11-renders/assembly.glb` with Blender in background mode (example in `INPUT/TOOLCHAIN.md`): a product view, an exploded view and a view of the user interface, at least 1600 × 1000 pixels; a scale check against a known dimension; `renders.md` captions them as renders of a digital design. Measure (object fills the frame, part names present) instead of trusting a glance. Never use generative image models. Gate: `python3 method/gates/check_full.py --part renders`
files/method/roles/requirements.md (33 wierszy)
# Requirements (stage C): runs beside the six scouts and the six concept agents
Inputs: `INPUT/requirements.md`, `INPUT/brief.md`, and the scout files `work/intake/use.md`,
`work/intake/hazards.md`, `work/research/regulatory.md`, `deliverables/02-concept/benchmark.md` as they appear (the
scouts start with you and finish after about eight minutes). Handbook §2. `method/POLICY.md` §1 for conflicts.
Order of work: first the given part of both files (it needs only INPUT, and the scorer reads `requirements.json`
as soon as the concepts are done); then, once the scout files exist, the derived requirements and the conflicts.
Write only these two files:
1. `deliverables/01-requirements/requirements.md` with three parts under these headings:
- "Given requirements": a table with one row per INPUT id, the id spelled exactly: id, the statement in one
short line (keep every number), your interpretation (the exact condition under which it will be verified),
the design consequence. Keep the cells short.
- "Derived requirements": ids `DR-<AREA>-<nn>`, each with parents, statement, metric, target, threshold,
verification method (Analysis, Inspection, Demonstration, Test), rationale with origin label, priority.
Walk every area listed in handbook §2 and write either the requirement or one line on why the area does
not apply. Safety requirements from `work/intake/hazards.md` become `DR-SAF-*`.
- "Conflicts and decisions": the requirement pairs that pull against each other and how POLICY.md resolves
them; then the assumptions register (each ASSUMPTION with reason and sensitivity).
2. `deliverables/01-requirements/requirements.json` in the format of `method/templates/requirements.json`:
- one entry per INPUT id, `source: "INPUT"`, same spelling;
- `kind: "numeric"` only when the INPUT target is a number with a direction; `target` and `threshold` copied
exactly as INPUT states them, in the INPUT unit (no conversion; `threshold: null` when there is none);
other requirements `kind: "qualitative"` with `target_text` and `threshold_text`;
- `weight` = the INPUT priority or weight number (3 as ASSUMPTION when INPUT gives none);
- the derived entries (at least eight, including DR-SAF-*), `source: "derived"`.
Do not write `compliance.csv` (code generates it). The concept agents do not wait for you, but the scorer ranks
their concepts with your JSON afterwards: copy the numbers exactly.
Gate R: `python3 method/gates/check_requirements.py`
files/method/roles/safety.md (7 wierszy)
# Safety and certification stage (full tier): `deliverables/08-safety-certification/` Handbook §11. `hazards.md` (ISO 12100 table, hazards numbered H-01 …, risk before and after, measures, residual risk); `safety-requirements.md` (every DR-SAF id → design measure → evidence path); `regulatory.md` (EU and US map, each item with SOURCE); `conformity-route.md` (route, lab test list, pre-compliance plan, cost, duration). Gate: `python3 method/gates/check_full.py --part safety`
files/method/roles/scouts.md (63 wierszy)
# Scouts (stage A‖B): six agents in parallel, one section each, eight minutes
Common rules: do only your section and write only its files. Every external fact carries its URL and the date you
read it (SOURCE); every number carries a unit and an origin label. A page you cannot read is reported as such and
replaced by another source, never by a guess. Web search and fetch are deferred tools: load them with `ToolSearch`
(query `select:WebSearch,WebFetch`) before the first look-up. Breadth over polish: the requirements stage and the
concept agents read your files as soon as you finish. End each file with a short section "Gaps" (what you could
not find).
## use: customer and use
Read `INPUT/brief.md` and `INPUT/requirements.md` (handbook §1). Write `work/intake/use.md`: buyer and user; where,
how, how often and how long the product is used; environment ranges (temperature, humidity, dust and water,
transport and storage); the job to be done; the main functions as verb–noun pairs; energy, material and signal
flows; the price expectation; three differentiator hypotheses against what exists; candidate derived requirements
(area, statement, metric, target, origin).
## physics: physics map and morphological box
Read `INPUT/*` (handbook §4 and §5). Write `deliverables/02-concept/physics-map.md` as handbook §4 asks: every
domain rated CRITICAL, RELEVANT or MINOR; for each CRITICAL domain the governing relations, the dominant
parameters, order-of-magnitude numbers with a short hand calculation, and the typical failure modes; end with a
section "Seed feasibility" that checks each seed below to an order of magnitude.
Then write `work/morph.json`:
```
{"schema": "flow-morph/1",
"dimensions": [{"function": "<verb-noun>", "options": ["<option>", "<option>"], "decisive": true}],
"seeds": [{"id": "C1", "lens": "<lens text from your prompt>", "choices": {"<function>": "<option>"},
"why": "<one line: why this combination serves the lens>"}]}
```
4–8 functions that decide the architecture (for example energy source and conversion, the core working
principle, actuation, sensing and control, structure and enclosure, user interface), each with 2–5 real options;
mark the function(s) that set the working principle `"decisive": true`. Six seeds C1–C6, one per lens in your
prompt, each choosing one listed option for every function; every two seeds differ in at least two functions;
C6 uses an option of a decisive function that no other seed uses. No seed may fail the feasibility check. A few
web look-ups on how existing products work are allowed. Gate: `python3 method/gates/check_concept.py --stage morph`.
## hazards: hazards and safety requirements
Read `INPUT/*` (handbook §11). Write `work/intake/hazards.md`: an ISO 12100-style list (hazard → hazardous
situation → harm → severity × probability), foreseeable misuse included; for each hazard the candidate measures in
the order inherent design, protective device, information for use, and whether a protection can act
independently of firmware; candidate DR-SAF requirements (statement, metric, limit, the standard it comes from
with SOURCE).
## benchmark: comparable products
Handbook §3. Write `deliverables/02-concept/benchmark.md`: a table of at least five comparable products
(commercial and open source; adjacent categories count), one row per product with name, maker, working
principle, the specifications that matter for the INPUT requirement ids (name the ids), price with date, and the
product URL in the row; then weak points from reviews, teardowns or forums (with links); realistic target values;
the differentiator that is feasible for us.
## parts: key components and cost anchors
Handbook §12. Write `work/research/parts.md`: for every function that needs a bought component, two or three real
candidates (manufacturer, MPN, supplier, supplier part number, URL, price at 1/100/1000 with the date read,
stock) — from the offline parts catalogue first when `INPUT/TOOLCHAIN.md` lists one, then distributor pages
(read-only); cost anchors with SOURCE for enclosure processes (printing, machining, moulding tooling), PCB
fabrication and assembly, certification testing, packaging and freight.
## regulatory: rules that apply
Handbook §11. Write `work/research/regulatory.md`: the EU directives and regulations and the harmonised standards,
the US FCC rules and safety standards that likely apply to this product, each with SOURCE (URL and date) and the
clauses that drive the design (limits, tests, markings); the likely conformity route; typical lab test lists,
costs and durations with SOURCE.
files/method/roles/testplan.md (8 wierszy)
# Test plan stage (full tier): `deliverables/10-prototype-test-plan/` Handbook §13. `test-plan.md`: the build plan of prototype 1; bring-up (current-limited first power-up, what to measure before full power, precautions); one test per INPUT requirement id and per DR-SAF id (method, equipment, conditions, pass criterion, the id it closes); pre-compliance EMC and the safety tests a lab would run; durability or life tests where lifetime is required. Gate: `python3 method/gates/check_full.py --part testplan`
files/method/roles/verification.md (10 wierszy)
# Verification stage (full tier): `work/full/verification.json` Compare every INPUT requirement with what the detailed design now shows (CAD figures, calculation results, ERC and DRC, firmware tests, BOM cost, test plan). Write `work/full/verification.json` (format `method/templates/concept.json`; values from the full-tier homes, `evidence` = deliverables paths, `known_misses`), `work/full/key_figures.json` (dimensions from the CAD bounding box, mass, power, `origins`), `work/full/evidence-verification.json`, and blocked items in `work/blocked/verification.json`. Then run the gate and fix what it reports in your files. Gate: `python3 method/tools/make_package.py --tier full`
files/method/roles/writers.md (52 wierszy)
# Writers (stage J and the report draft): four parts in parallel, all for the chosen concept
Common inputs: `deliverables/02-concept/selection.md`, `work/final/estimates.json` and `work/final/key_figures.json`
(the stable homes of the numbers at the concept tier: quote them, never re-derive them; the grafts are in
`grafted`), `deliverables/02-concept/calc/` (the winner's calculation homes), `work/deep/<winner>/design.md`,
`deliverables/01-requirements/` (code refreshed `compliance.csv` from the final estimates just before the writers
started), `work/intake/*`, `work/research/*`. At the full tier the homes are `work/full/verification.json`,
`work/full/key_figures*.json` and `deliverables/05-calculations/params.json`; the concept-tier files then describe
only the concept. Every number you write carries its origin and comes from a home. If you find a number that looks
wrong, do not change it in your text: append one line to `work/open-issues.md`. Each part also writes
`work/final/evidence-<part>.json`
(`{"<requirement id>": ["deliverables/..."]}` for the ids your files show) and `work/manifest-desc/<part>.json`
(`{"deliverables/<path>": "<one-line description>"}`).
## main: `deliverables/02-concept/concept.md`
Handbook §5 order, with these headings: 1 the product in one paragraph · 2 functions and flows · 3 Alternatives
(all six concepts: principle, key numbers, why each lost, taken from `selection.md`) · 4 Evaluation (knock-outs,
weighted scoring, blind panel, sensitivity; link `selection.md`; why the winner won and what was grafted) · 5
Chosen concept and architecture (Mermaid block diagram, optionally rendered next to it as SVG or PNG; subsystems
with make or buy; critical components with real parts; interfaces) · 6 Key estimates (table: quantity, value, unit,
origin, home path) · 7–10 one paragraph each that links `safety-concept.md`, `compliance-route.md`, `cost.md`,
`risks.md` · 11 Decisions taken without a human (with the rejected alternatives and the POLICY.md defaults used) ·
12 Open issues and what detailed design must do first.
## safety: `deliverables/02-concept/safety-concept.md` and `compliance-route.md`
Handbook §5 items 7–8 and §11 at concept depth. `safety-concept.md`: the hazard table (H-01 …) with severity,
probability, measures in the order inherent design / protective device / information, the protections that act
without firmware, residual risk, and the trace to the DR-SAF ids. `compliance-route.md`: the EU and US rules and
standards with SOURCE, why each applies, the clauses that drive the design, the conformity route, the lab tests,
rough cost and time.
## cost: `deliverables/02-concept/cost.md`, `risks.md`, `deliverables/09-bom-cost/cost.json`, `work/final/key_figures-cost.json`
Handbook §5 items 9–10 and §12. `cost.md`: the major items with real parts and dated prices (ASSUMPTION ranges
where a price could not be read); unit cost at 1/100/1000 with every non-BOM line (PCB, mechanical parts,
assembly and test, packaging, certification and tooling amortised over a stated quantity, freight, contingency);
retail price logic with platform and payment fees, VAT and margin; say plainly if the numbers do not close.
`cost.json` in the RUN CONTRACT format (rough; bom + other = unit_cost). `key_figures-cost.json`:
`{"unit_cost": {"currency", "q1", "q100", "q1000"}, "retail_price": {"currency", "value"}, "origins": {...}}`.
`risks.md`: the top risks with likelihood, impact, mitigation and the analysis or test that retires each.
## writeup: `work/writeup.md` (the report draft; the top session turns it into the report)
Handbook §15 order: the product in one paragraph; then a line containing only `<!-- KEY-FIGURES -->` where the
key-parameter table belongs (code fills it from the homes); unit cost and retail price discussed in words; then a
line containing only `<!-- STATUS-TABLE -->` where the per-requirement status table belongs (code fills it from
`compliance.csv`), followed by a short discussion of every requirement that is `not_met` or `partly` (read
`deliverables/01-requirements/compliance.csv`: it is current; a row whose note says PROVISIONAL means the final
estimates were missing, then use `work/final/estimates.json` or say so); the top risks and
open hazards; the decisions taken without a human; how the design was produced (the pipeline, its gates, and the
PARTIAL stages named in your prompt); what is open, blocked or not run (include `work/open-issues.md`); at the
full tier, the file map of `deliverables/`. Never type the key figures or the status table yourself; never name a
file REPORT*.
files/method/templates/concept.json (23 wierszy)
{
"schema": "flow-concept/1",
"_comment": "Format of work/concepts/C<i>.json, work/deep/C<i>/estimates.json, work/final/estimates.json and work/full/verification.json. Values illustrative. One estimate per INPUT requirement id, in the requirement's own unit. Numeric: value (+ low/high range). Qualitative: level target | threshold | miss | unknown (+ optional value text). origin = SOURCE | CALCULATION | SIMULATION | ASSUMPTION; basis = '<run-relative path to a .results.json>#<key>' for calculated numbers, else a one-line reason. evidence = run-relative paths under deliverables/ (verification files; code adds the calculation homes to the final estimates). grafted (deepened files only) = [{from, feature, why, affects}], at most 3; known_misses (deepened and verification files) = [{id, why}]. safety_ko is a list ([] when every hazard has an adequate protection). Paths are relative to the run directory.",
"id": "C1",
"name": "short descriptive name (no lens words)",
"principle": "one sentence: working principle and architecture",
"morph": {"<function from work/morph.json>": "<chosen option>"},
"estimates": {
"ABC-WGT-01": {"value": 2.6, "unit": "kg", "low": 2.3, "high": 3.1, "origin": "CALCULATION",
"basis": "work/concepts/C1_calc.results.json#mass_total"},
"ABC-FUN-02": {"level": "target", "value": "segment display, 14 mm digits", "origin": "ASSUMPTION",
"basis": "digit height read at 5 m per common legibility rule of thumb; low sensitivity"}
},
"safety_ko": [],
"cost_q1000": 48.0,
"currency": "EUR",
"feasible": true,
"screened": false,
"flags": [],
"grafted": [],
"known_misses": []
}
files/method/templates/requirements.json (22 wierszy)
{
"schema": "flow-req/1",
"_comment": "Format of deliverables/01-requirements/requirements.json. Ids and values are illustrative. One entry per INPUT id (source INPUT): kind numeric when the INPUT target is a number with a direction (copy target and threshold exactly as INPUT states them, in the INPUT unit; threshold null if none), else qualitative with target_text/threshold_text; weight = the INPUT priority or weight (3 as ASSUMPTION when none). Derived entries: DR-<AREA>-<nn>, source derived, parents, statement, verification, origin label.",
"currency": "EUR",
"requirements": [
{"id": "ABC-WGT-01", "source": "INPUT", "title": "Dry mass", "kind": "numeric", "direction": "<=",
"target": 3, "threshold": 4, "unit": "kg", "weight": 3, "verification": "Analysis",
"interpretation": "Mass of the complete product without consumables, from CAD volumes and densities."},
{"id": "ABC-FUN-02", "source": "INPUT", "title": "Status visible on the unit", "kind": "qualitative",
"target_text": "status readable from 5 m", "threshold_text": "status readable from 1 m", "unit": "",
"weight": 2, "verification": "Inspection", "interpretation": "Indicator legible at the stated distance in office light."},
{"id": "DR-SAF-01", "source": "derived", "parents": ["ABC-SAF-01", "regulation"], "title": "Accessible surface temperature",
"statement": "No surface a user can touch in normal use exceeds the limit at the highest ambient temperature.",
"kind": "numeric", "direction": "<=", "target": 48, "threshold": 60, "unit": "degC", "weight": 4,
"verification": "Analysis + Test", "origin": "SOURCE: <standard and clause, URL, date read>"},
{"id": "DR-MFG-01", "source": "derived", "parents": ["manufacturing"], "title": "Assembly without special tools",
"statement": "The product assembles with standard hand tools and at most two fastener types.",
"kind": "qualitative", "target_text": "2 fastener types", "threshold_text": "3 fastener types", "weight": 2,
"verification": "Inspection", "origin": "ASSUMPTION: keeps assembly time low at 1000 units; low sensitivity"}
]
}
files/method/tools/make_package.py (335 wierszy)
#!/usr/bin/env python3
"""make_package.py - write the package files that code can write: compliance.csv, manifest.json, report tables.
Usage (from the run directory):
python3 method/tools/make_package.py --tier concept|full [--report] [--json]
- compliance.csv: one row per INPUT requirement id; status computed from the best available estimates
(work/full/verification.json at the full tier, else work/final/estimates.json, else - marked PROVISIONAL -
the leading concept), with the same comparison rules as the concept scoring, except that `met` also needs an
origin other than ASSUMPTION and evidence of its own (else `partly`); evidence from the estimates and the
evidence maps work/final/evidence-*.json and work/full/evidence-*.json.
- manifest.json: every file under deliverables/ (descriptions from work/manifest-desc/*.json), key figures
merged from work/final|full/key_figures*.json and 09-bom-cost/cost.json, blocked[] from work/blocked/*.json.
- work/final/status-table.md and work/final/keyfig-table.md: tables for the report.
- Housekeeping: removes build caches (__pycache__, .pio) from deliverables/.
- --report (TOP SESSION ONLY - agents never pass it): also write deliverables/REPORT.md from work/writeup.md,
replacing the lines <!-- KEY-FIGURES --> and <!-- STATUS-TABLE --> with the tables; without a draft it writes
a short automatic status report. Standard library only.
"""
import csv
import glob
import io
import os
import shutil
import sys
import time
sys.path.insert(0, os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "gates")))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import flowlib as fl # noqa: E402
import score_concepts as sc # noqa: E402
KIND = {"01-requirements": "requirements", "02-concept": "concept", "03-cad": "cad", "04-drawings": "drawing",
"05-calculations": "calculation", "06-electronics": "electronics", "07-firmware": "firmware",
"08-safety-certification": "safety-certification", "09-bom-cost": "bom-cost",
"10-prototype-test-plan": "test-plan", "11-renders": "render"}
JUNK_DIRS = ("__pycache__", ".pio", ".ipynb_checkpoints")
KEYS = ("dimensions_mm", "mass_kg", "power_w", "unit_cost", "retail_price")
def estimate_source(run, tier):
cands = []
if tier == "full":
cands.append(("work/full/verification.json", False))
cands.append(("work/final/estimates.json", False))
fin, _ = fl.load_json(os.path.join(run, "work", "scoring", "final.json"))
if isinstance(fin, dict) and fin.get("winner"):
cands += [(f"work/deep/{fin['winner']}/estimates.json", True), (f"work/concepts/{fin['winner']}.json", True)]
for name in ("combined", "numeric"):
rec, _ = fl.load_json(os.path.join(run, "work", "scoring", f"{name}.json"))
if isinstance(rec, dict) and rec.get("top"):
cands.append((f"work/concepts/{rec['top'][0]}.json", True))
for path, provisional in cands:
data, err = fl.load_json(os.path.join(run, path))
if not err and isinstance(data, dict) and isinstance(data.get("estimates"), dict):
return path, data, provisional
return None, {}, True
def evidence_maps(run, tier):
pats = ["work/final/evidence-*.json"] + (["work/full/evidence-*.json"] if tier == "full" else [])
merged, warnings = {}, []
for pat in pats:
for p in sorted(glob.glob(os.path.join(run, pat))):
data, err = fl.load_json(p)
if err or not isinstance(data, dict):
warnings.append(f"evidence map {fl.rel(p, run)} ignored: {err or 'not an object'}")
continue
for rid, paths in data.items():
paths = [paths] if isinstance(paths, str) else paths
if isinstance(paths, list):
merged.setdefault(rid, []).extend(str(x) for x in paths)
return merged, warnings
def deliv_rel(path):
path = str(path).split("#", 1)[0].strip().strip("`")
return path[len("deliverables/"):] if path.startswith("deliverables/") else path
def compliance(run, tier, reqs):
src, data, provisional = estimate_source(run, tier)
est = data.get("estimates", {}) if data else {}
emap, warnings = evidence_maps(run, tier)
errors, rows, deliv = [], [], os.path.join(run, "deliverables")
if src is None:
warnings.append("no estimates yet: every requirement is not_addressed")
elif provisional:
warnings.append(f"PROVISIONAL: statuses come from {src} (the pipeline has not finished)")
fallback = "02-concept/selection.md" if os.path.isfile(os.path.join(deliv, "02-concept", "selection.md")) else None
for r in reqs:
e = est.get(r["id"]) if isinstance(est.get(r["id"]), dict) else None
if src and not provisional and e is None:
errors.append(f"{src} has no estimate for {r['id']}")
a = sc.assess(r, e)
refs = []
for x in (e or {}).get("evidence") or []:
refs.append(deliv_rel(x))
refs += [deliv_rel(x) for x in emap.get(r["id"], [])]
basis = str((e or {}).get("basis") or "")
if basis.startswith("deliverables/"):
refs.append(deliv_rel(basis))
good, seen = [], set()
for ref in refs:
if ref and ref not in seen and os.path.exists(os.path.join(deliv, ref)):
good.append(ref)
seen.add(ref)
elif ref and ref not in seen and not provisional:
errors.append(f"{r['id']}: evidence path not found in deliverables/: {ref}")
seen.add(ref)
status = {"target": "met", "threshold": "partly", "miss": "not_met"}.get(a["level"])
if status is None:
status = "partly" if (e and (good or fallback)) else "not_addressed"
assumed = str((e or {}).get("origin") or "").strip().upper().startswith("ASSUMPTION")
demoted = None
if status == "met" and (assumed or not good): # 'met' needs a shown value: an origin and evidence of its own
status = "partly"
demoted = "assumed, not yet quantified" if assumed else "no evidence of its own yet"
if status != "not_addressed" and not good:
if fallback:
good = [fallback]
else:
status = "not_addressed"
v = a["value"]
value = f"{v:g}" if isinstance(v, (int, float)) and not isinstance(v, bool) else ("" if v is None else str(v))
if not value and a["level"] != "unknown":
value = {"target": "meets target", "threshold": "meets threshold", "miss": "misses threshold"}[a["level"]]
prefix = "no estimate yet (pipeline in progress)" if src is None else ("PROVISIONAL; " if provisional else "") + (
"design analysis, not a physical test" if src.startswith("work/full/") else "prediction from concept estimates")
bits = [prefix]
if e:
bits.append(f"{e.get('origin', 'origin missing')}" + (f" ({basis})" if basis else ""))
tgt = r.get("target", r.get("target_text"))
thr = r.get("threshold", r.get("threshold_text"))
if tgt not in (None, ""):
d = sc.direction(r) or ""
bits.append(f"target {d} {tgt} / threshold {d} {thr if thr not in (None, '') else 'none'} {r.get('unit', '')}".replace(" ", " ").strip())
if demoted:
bits.append(f"meets the target on paper but {demoted}")
if a["at_risk"]:
bits.append("pessimistic estimate misses the limit")
if a["level"] == "unknown" and src is not None:
bits.append("not quantified yet")
unit = a["unit"] if value else ""
if status == "not_addressed" and value:
bits.append(f"leading concept predicts {value} {unit}".strip())
value = unit = ""
rows.append([r["id"], status, value, unit, ";".join(good), "; ".join(b for b in bits if b)])
return rows, src, provisional, errors, warnings
def clean(deliv):
removed = []
for root, dirs, _files in os.walk(deliv):
for d in list(dirs):
if d in JUNK_DIRS:
shutil.rmtree(os.path.join(root, d), ignore_errors=True)
removed.append(fl.rel(os.path.join(root, d), deliv))
dirs.remove(d)
return removed
def key_figures(run, tier):
files = ["work/final/key_figures.json", "work/final/key_figures-cost.json"]
if tier == "full":
files += ["work/full/key_figures.json", "work/full/key_figures-cost.json"]
kf, origins, warnings = {k: None for k in KEYS}, {}, []
for f in files:
data, err = fl.load_json(os.path.join(run, f))
if err:
continue
for k in KEYS:
if isinstance(data, dict) and data.get(k) is not None:
kf[k] = data[k]
if isinstance(data, dict) and isinstance(data.get("origins"), dict):
origins.update(data["origins"])
cost, err = fl.load_json(os.path.join(run, "deliverables", "09-bom-cost", "cost.json"))
if not err and isinstance(cost, dict) and isinstance(cost.get("unit_cost"), dict) and (tier == "full" or kf["unit_cost"] is None):
uc = cost["unit_cost"]
kf["unit_cost"] = {"currency": cost.get("currency"), "q1": uc.get("q1"), "q100": uc.get("q100"), "q1000": uc.get("q1000")}
origins.setdefault("unit_cost", "CALCULATION deliverables/09-bom-cost/cost.json")
d = kf["dimensions_mm"]
if d is not None and not (isinstance(d, list) and len(d) == 3 and all(fl.num(x) is not None for x in d)):
warnings.append(f"key figure dimensions_mm {d!r} is not [L, W, H]: set to null")
kf["dimensions_mm"] = None
for k in ("mass_kg", "power_w"):
if kf[k] is not None and fl.num(kf[k]) is None:
warnings.append(f"key figure {k} {kf[k]!r} is not a number: set to null")
kf[k] = None
elif kf[k] is not None:
kf[k] = fl.num(kf[k])
uc = kf["unit_cost"]
if uc is not None and not (isinstance(uc, dict) and all(q in uc for q in ("currency", "q1", "q100", "q1000"))
and all(uc[q] is None or fl.num(uc[q]) is not None for q in ("q1", "q100", "q1000"))):
warnings.append("key figure unit_cost malformed: set to null")
kf["unit_cost"] = None
rp = kf["retail_price"]
if rp is not None and not (isinstance(rp, dict) and "currency" in rp and (rp.get("value") is None or fl.num(rp.get("value")) is not None)):
warnings.append("key figure retail_price malformed: set to null")
kf["retail_price"] = None
return kf, origins, warnings
def manifest(run, tier, kf):
deliv = os.path.join(run, "deliverables")
desc = {}
for p in sorted(glob.glob(os.path.join(run, "work", "manifest-desc", "*.json"))):
data, err = fl.load_json(p)
if not err and isinstance(data, dict):
desc.update({deliv_rel(k): str(v) for k, v in data.items()})
files = []
for root, _dirs, names in os.walk(deliv):
for n in sorted(names):
p = os.path.join(root, n)
r = os.path.relpath(p, deliv)
if r == "manifest.json" or os.path.getsize(p) == 0 or n.endswith(".tmp"):
continue
top = r.split(os.sep)[0]
kind = "report" if r == "REPORT.md" else ("results" if n.endswith(".results.json") else
("script" if n.endswith(".py") else KIND.get(top, "other")))
files.append({"path": r, "kind": kind, "desc": desc.get(r) or f"{KIND.get(top, 'package')} file {n}"})
files.sort(key=lambda f: f["path"])
blocked, seen = [], set()
for p in sorted(glob.glob(os.path.join(run, "work", "blocked", "*.json"))):
data, err = fl.load_json(p)
for b in (data if isinstance(data, list) else []):
if isinstance(b, dict) and b.get("what") and b.get("why") and b["what"] not in seen:
blocked.append({"what": str(b["what"]), "why": str(b["why"])})
seen.add(b["what"])
fl.write_json(os.path.join(deliv, "manifest.json"), {"tier": tier, "files": files, "key_figures": kf, "blocked": blocked})
return len(files), len(blocked)
def fmt(x, unit=""):
if x is None:
return "unknown"
return (f"{x:g}" if isinstance(x, float) else str(x)) + (f" {unit}" if unit else "")
def tables(run, rows, kf, origins, src, provisional):
d = os.path.join(run, "work", "final")
os.makedirs(d, exist_ok=True)
count = {s: sum(1 for r in rows if r[1] == s) for s in ("met", "partly", "not_met", "not_addressed")}
st = [f"Status per requirement (from `01-requirements/compliance.csv`; {count['met']} met, {count['partly']} partly, "
f"{count['not_met']} not met, {count['not_addressed']} not addressed"
+ (f"; PROVISIONAL, from {src}" if provisional and src else "") + "):", "",
"| id | status | value | unit | evidence | note |", "|---|---|---|---|---|---|"]
st += ["| " + " | ".join(str(c).replace("|", "/").replace("\n", " ") for c in r) + " |" for r in rows]
dims = kf.get("dimensions_mm")
uc, rp = kf.get("unit_cost") or {}, kf.get("retail_price") or {}
k = ["| figure | value | origin |", "|---|---|---|",
f"| dimensions L x W x H | {' x '.join(fmt(x) for x in dims) + ' mm' if dims else 'unknown'} | {origins.get('dimensions_mm', '')} |",
f"| mass | {fmt(kf.get('mass_kg'), 'kg')} | {origins.get('mass_kg', '')} |",
f"| power | {fmt(kf.get('power_w'), 'W')} | {origins.get('power_w', '')} |",
f"| unit cost at 1 / 100 / 1000 | {fmt(uc.get('q1'))} / {fmt(uc.get('q100'))} / {fmt(uc.get('q1000'))} {uc.get('currency', '') or ''} | {origins.get('unit_cost', '')} |",
f"| retail price | {fmt(rp.get('value'))} {rp.get('currency', '') or ''} | {origins.get('retail_price', '')} |"]
with open(os.path.join(d, "status-table.md"), "w", encoding="utf-8") as fh:
fh.write("\n".join(st) + "\n")
with open(os.path.join(d, "keyfig-table.md"), "w", encoding="utf-8") as fh:
fh.write("\n".join(k) + "\n")
return "\n".join(st), "\n".join(k)
def report(run, tier, st, kt):
draft = fl.read_text(os.path.join(run, "work", "writeup.md"))
out = os.path.join(run, "deliverables", "REPORT.md")
if draft.strip():
text = draft
for marker, table, title in (("<!-- KEY-FIGURES -->", kt, "Key figures"), ("<!-- STATUS-TABLE -->", st, "Status per requirement")):
text = text.replace(marker, table) if marker in text else text.rstrip() + f"\n\n## {title}\n\n{table}\n"
mode = "draft"
else:
brief = [ln.strip() for ln in fl.read_text(os.path.join(run, "INPUT", "brief.md")).splitlines() if ln.strip()]
title = next((ln.lstrip("# ").strip() for ln in brief if ln.startswith("#")), "the requested product")
para = next((ln for ln in brief if not ln.startswith("#")), "")
gates = []
for p in sorted(glob.glob(os.path.join(run, "work", "gates", "*.json"))):
g, err = fl.load_json(p)
if not err and isinstance(g, dict):
gates.append(f"| {g.get('gate', os.path.basename(p))} | {'PASS' if g.get('pass') else 'FAIL'} | "
f"{str((g.get('errors') or [''])[0])[:160].replace('|', '/')} |")
text = "\n".join([
f"# Design report - {title}", "",
f"Status: the design pipeline has not produced its written report yet (automatic report, "
f"{time.strftime('%Y-%m-%d %H:%M %Z')}). The numbers below are read from their homes in the package; "
"requirement statuses marked PROVISIONAL are predictions from the leading concept.", "",
"## The request", "", para, "",
"## Key figures", "", kt, "",
"## Status per requirement", "", st, "",
"## Pipeline state (gates)", "", "| gate | result | first error |", "|---|---|---|", *gates, "",
"## What remains open", "",
"Every requirement that is not `met` above, and every deliverable of the tier that is missing from "
"`manifest.json`. See `02-concept/` for the concept work that exists.", ""])
mode = "automatic"
with open(out, "w", encoding="utf-8") as fh:
fh.write(text if text.endswith("\n") else text + "\n")
return mode
def main():
ap = fl.parser(__doc__)
ap.add_argument("--tier", choices=["concept", "full"], required=True)
ap.add_argument("--report", action="store_true", help="top session only: also write deliverables/REPORT.md")
a = ap.parse_args()
run = os.path.abspath(a.run_dir)
deliv = os.path.join(run, "deliverables")
os.makedirs(os.path.join(deliv, "01-requirements"), exist_ok=True)
reqs, rw = fl.requirements(run)
rows, src, provisional, errors, warnings = compliance(run, a.tier, reqs)
if rw:
warnings.insert(0, rw)
if not reqs:
errors.append("no requirement ids found in INPUT/requirements.md")
buf = io.StringIO()
w = csv.writer(buf, lineterminator="\n")
w.writerow(["req_id", "status", "value", "unit", "evidence", "note"])
w.writerows(rows)
with open(os.path.join(deliv, "01-requirements", "compliance.csv"), "w", encoding="utf-8") as fh:
fh.write(buf.getvalue())
removed = clean(deliv)
kf, origins, kw = key_figures(run, a.tier)
warnings += kw
st, kt = tables(run, rows, kf, origins, src, provisional)
notes = [f"compliance.csv: {len(rows)} rows from {src or 'no estimates'}"] + [f"removed build cache {r}" for r in removed]
if a.report:
notes.append(f"REPORT.md written ({report(run, a.tier, st, kt)} mode)")
n_files, n_blocked = manifest(run, a.tier, kf)
notes.append(f"manifest.json: {n_files} files, {n_blocked} blocked item(s)")
return fl.emit(f"make-package-{a.tier}", errors, warnings, notes,
{"source": src, "provisional": provisional}, run)
if __name__ == "__main__":
sys.exit(main())
files/method/tools/score_concepts.py (682 wierszy)
#!/usr/bin/env python3
"""score_concepts.py - FLOWCODE concept selection in code: knock-outs, weighted satisfaction, panel, final pick.
Run-directory modes (run from the run directory; the JSON verdict is the last output line):
python3 method/tools/score_concepts.py numeric work/concepts/C*.json -> work/scoring/numeric.json
python3 method/tools/score_concepts.py combine + work/panel/*.json -> work/scoring/combined.json
python3 method/tools/score_concepts.py final --deep C3,C1 [--tier concept|full] deepened estimates ->
work/scoring/final.json, deliverables/02-concept/selection.md + selection.json and a re-runnable copy of
this script with its inputs in deliverables/02-concept/calc/; then it PROMOTES the winner (its
work/deep/<id>/calc/ into deliverables/02-concept/calc/, its estimates and key figures into work/final/,
paths rewritten), runs check_concept.py --stage final and refreshes the package (make_package.py). The
verdict fails when the promoted winner does not pass that final check.
Deliverable mode (no arguments, inside 02-concept/calc/): recompute the selection from selection_inputs.json
and write selection_score.results.json.
Rules (method/POLICY.md): only INPUT requirement ids are scored, with their weights. Numeric requirement:
satisfaction 1 at or beyond the target, 0.5 at the threshold, linear between; a violated threshold is a
knock-out (KO). Qualitative requirement: level target 1, threshold 0.5, miss = KO. Unknown: 0.5, and the
score loses 0.10 x the unknown weight share. A concept is also KO when its screening says infeasible or it
lists a hazard without an independent protection (safety_ko); a fatal flaw flagged by two or more judges is
KO. Knocked-out concepts rank last; among them a missed threshold ranks before a safety, infeasibility or
fatal-flaw KO. Panel: share of pairwise wins (tie = half). Combined = 0.6 x numeric normalised on the scale of
the panel candidates (min-max of the best four, floor 0.10 on the spread; the final pick reuses that scale) +
0.4 x panel share. Ties (< 0.005) go to the lower unit cost at 1000. Standard library only.
"""
import glob
import json
import os
import re
import shutil
import subprocess
import sys
W_NUMERIC, W_PANEL, UNKNOWN_PENALTY, NORM_FLOOR, TIE = 0.6, 0.4, 0.10, 0.10, 0.005
LEVEL_S = {"target": 1.0, "threshold": 0.5, "miss": 0.0}
NONE_WORDS = {"", "none", "no", "n/a", "na", "-", "nil", "null", "[]"}
HARD_KO = ("safety:", "screening:", "panel:")
HERE = os.path.dirname(os.path.abspath(__file__))
# ---------------------------------------------------------------- scoring math (standalone)
def num(x):
if isinstance(x, bool):
return None
if isinstance(x, (int, float)):
return float(x)
if isinstance(x, str):
try:
return float(x.strip().replace(",", "."))
except ValueError:
return None
return None
def direction(req):
return {"<=": "<=", "≤": "<=", ">=": ">=", "≥": ">="}.get(str(req.get("direction") or "").strip())
def assess(req, est):
"""Level (target/threshold/miss/unknown), satisfaction s, knock-out and risk flag of one estimate."""
est = est if isinstance(est, dict) else {}
out = {"level": "unknown", "s": 0.5, "ko": False, "at_risk": False,
"value": est.get("value"), "unit": est.get("unit") or req.get("unit") or ""}
d, t, thr, v = direction(req), num(req.get("target")), num(req.get("threshold")), num(est.get("value"))
if req.get("kind") == "numeric" and d and t is not None and v is not None:
ok = (lambda a, b: a <= b + 1e-12) if d == "<=" else (lambda a, b: a >= b - 1e-12)
if ok(v, t):
out.update(level="target", s=1.0)
elif thr is not None and ok(v, thr):
span = abs(thr - t)
out.update(level="threshold", s=0.5 + 0.5 * (abs(thr - v) / span if span else 1.0))
elif thr is not None:
out.update(level="miss", s=0.0, ko=True)
else:
out.update(level="miss", s=max(0.0, 0.5 * (1 - abs(v - t) / max(abs(t), 1e-9))))
pess = num(est.get("high") if d == "<=" else est.get("low"))
lim = thr if thr is not None else t
if pess is not None and out["level"] != "miss" and not ok(pess, lim):
out["at_risk"] = True
return out
lvl = est.get("level")
if lvl in LEVEL_S:
out.update(level=lvl, s=LEVEL_S[lvl], ko=(lvl == "miss"))
return out
def weight_of(req, weights=None):
w = weights.get(req["id"]) if weights else num(req.get("weight"))
return w if (w is not None and w > 0) else 1.0
def safety_list(x):
"""safety_ko as a list of hazard texts; '', 'none' and the like mean that no hazard is unprotected."""
items = x if isinstance(x, list) else ([] if x is None else [x])
return [str(i).strip() for i in items if str(i).strip().lower() not in NONE_WORDS]
def hard_ko(result):
"""Knocked out by safety, an infeasible screening or a panel fatal flaw - worse than a missed threshold."""
return any(str(x).startswith(HARD_KO) for x in result.get("ko_reasons") or [])
def concept_result(reqs, concept, weights=None):
est = concept.get("estimates") if isinstance(concept.get("estimates"), dict) else {}
tot = acc = unk = 0.0
per, ko, risk = {}, [], []
for r in reqs:
w = weight_of(r, weights)
e = est.get(r["id"])
if unit_clashes([r], {r["id"]: e}):
e = None # a value in another unit is not comparable: scored as unknown (reported by the warnings)
a = assess(r, e)
per[r["id"]] = a
tot += w
acc += w * a["s"]
if a["level"] == "unknown":
unk += w
if a["ko"]:
ko.append(r["id"])
if a["at_risk"]:
risk.append(r["id"])
score = (acc / tot - UNKNOWN_PENALTY * unk / tot) if tot else 0.0
reasons = [f"{i}: threshold missed" for i in ko]
reasons += [f"safety: {x}" for x in safety_list(concept.get("safety_ko"))]
if concept.get("feasible") is False:
reasons.append("screening: infeasible")
return {"score": round(score, 4), "unknown_share": round(unk / tot, 3) if tot else 1.0, "ko": bool(reasons),
"ko_reasons": reasons, "ko_weight": sum(weight_of(r, weights) for r in reqs if r["id"] in ko),
"at_risk": risk, "cost_q1000": num(concept.get("cost_q1000")), "per_req": per}
def rank(results, key="score"):
def cost(i):
c = results[i].get("cost_q1000")
return c if c is not None else float("inf")
def cls(i): # without KO, then missed thresholds only, then safety / infeasible / fatal flaw
return (results[i]["ko"], hard_ko(results[i]))
order = sorted(results, key=lambda i: (*cls(i), -results[i][key], cost(i), i))
for n in range(len(order) - 1): # near-ties go to the lower unit cost at 1000
a, b = order[n], order[n + 1]
if cls(a) == cls(b) and abs(results[a][key] - results[b][key]) < TIE and cost(b) < cost(a):
order[n], order[n + 1] = b, a
return order
def leader(results, key="score"):
order = rank(results, key)
ok = [i for i in order if not results[i]["ko"]]
return ok[0] if ok else (order[0] if order else None)
def extreme(concept, reqs, pessimistic):
est = dict(concept.get("estimates") or {})
for r in reqs:
d, e = direction(r), est.get(r["id"])
if not d or not isinstance(e, dict):
continue
side = ("high" if d == "<=" else "low") if pessimistic else ("low" if d == "<=" else "high")
if num(e.get(side)) is not None:
est[r["id"]] = dict(e, value=num(e.get(side)))
return dict(concept, estimates=est)
def sensitivity(reqs, concepts):
base = {i: concept_result(reqs, c) for i, c in concepts.items()}
lead, holds, tests, flips = leader(base), 0, 0, []
for r in reqs:
for f in (0.8, 1.2):
w = {x["id"]: weight_of(x) * (f if x["id"] == r["id"] else 1.0) for x in reqs}
other = leader({i: concept_result(reqs, c, w) for i, c in concepts.items()})
tests += 1
if other == lead:
holds += 1
else:
flips.append(f"{r['id']} weight x{f}: {other}")
pes = leader({i: concept_result(reqs, extreme(c, reqs, True)) for i, c in concepts.items()})
opt = leader({i: concept_result(reqs, extreme(c, reqs, False)) for i, c in concepts.items()})
return {"leader": lead, "weight_tests": tests, "leader_holds": holds, "flips": flips[:12],
"pessimistic_leader": pes, "optimistic_leader": opt}
def scale_of(scores):
"""Normalisation scale of the numeric score: min-max over the given scores with a floor on the spread."""
lo, hi = (min(scores), max(scores)) if scores else (0.0, 0.0)
return {"lo": round(lo, 6), "span": round(max(hi - lo, NORM_FLOOR), 6)}
def combine(cands, numeric, share, ko_extra, scale=None):
"""Combined score; `scale` fixes the numeric normalisation (the final pick reuses the panel candidates' scale,
so the panel share can still move the pick between two deepened concepts)."""
ok = isinstance(scale, dict) and num(scale.get("lo")) is not None and (num(scale.get("span")) or 0) > 0
scl = {"lo": num(scale["lo"]), "span": num(scale["span"])} if ok else scale_of([numeric[c]["score"] for c in cands])
out = {}
for c in cands:
norm = (numeric[c]["score"] - scl["lo"]) / scl["span"]
reasons = list(numeric[c]["ko_reasons"]) + list(ko_extra.get(c, []))
out[c] = {"numeric": numeric[c]["score"], "numeric_norm": round(norm, 4), "panel_share": round(share.get(c, 0.5), 4),
"combined": round(W_NUMERIC * norm + W_PANEL * share.get(c, 0.5), 4), "ko": bool(reasons),
"ko_reasons": reasons, "ko_weight": numeric[c].get("ko_weight", 0.0),
"cost_q1000": numeric[c].get("cost_q1000")}
return out
def pick(final):
"""Winner and order of the deepened concepts. If all are KO: a concept knocked out only by missed thresholds
beats one with a safety, infeasibility or fatal-flaw KO; then the smallest knocked-out weight wins."""
order = rank(final, "combined")
if order and all(final[i]["ko"] for i in order):
order = sorted(order, key=lambda i: (hard_ko(final[i]), final[i]["ko_weight"], -final[i]["combined"], i))
return order
# ---------------------------------------------------------------- deliverable mode (no arguments)
def recompute_here(quiet=False):
path = "selection_inputs.json"
if not os.path.isfile(path):
print("usage: run from the run directory with a mode (numeric|combine|final), or with no arguments "
"inside 02-concept/calc/ next to selection_inputs.json")
return 2
data = json.load(open(path, encoding="utf-8"))
reqs, concepts = data["requirements"], data["concepts"]
numeric = {i: concept_result(reqs, c) for i, c in concepts.items()}
final_in = {i: data["deepened_concepts"][i] for i in data["deepened"] if i in data.get("deepened_concepts", {})}
deep_num = {i: concept_result(reqs, c) for i, c in final_in.items()}
final = combine(list(deep_num), deep_num, data.get("panel_share", {}), data.get("panel_ko", {}), data.get("scale"))
order = pick(final)
res = {f"numeric_score_{i}": {"value": r["score"], "unit": "1"} for i, r in sorted(numeric.items())}
res.update({f"final_score_{i}": {"value": r["combined"], "unit": "1"} for i, r in sorted(final.items())})
if order:
res["winner"] = {"value": order[0], "unit": "concept id"}
res["winner_index"] = {"value": int(re.sub(r"\D", "", order[0]) or 0), "unit": "1"}
with open("selection_score.results.json", "w", encoding="utf-8") as fh:
json.dump(res, fh, indent=1)
if not quiet:
print(json.dumps(res))
expected = data.get("winner")
if expected and order and order[0] != expected:
print(f"MISMATCH: recomputed winner {order[0]} differs from the recorded winner {expected}")
return 1
return 0
# ---------------------------------------------------------------- run-directory modes
def flowlib():
sys.path.insert(0, os.path.normpath(os.path.join(HERE, "..", "gates")))
import flowlib as fl # noqa: E402
return fl
def validate(cid, data, fl):
if not isinstance(data, dict):
return ["not a JSON object"]
errs = []
est = data.get("estimates")
if not isinstance(est, dict) or not est:
errs.append("estimates{} missing or empty")
else:
for k, e in est.items():
if not isinstance(e, dict):
errs.append(f"estimate {k} is not an object")
elif e.get("level") is not None and e.get("level") not in fl.LEVELS:
errs.append(f"estimate {k}: level '{e.get('level')}' not in {list(fl.LEVELS)}")
if not isinstance(data.get("safety_ko"), (list, type(None))):
errs.append("safety_ko must be a list of unprotected hazards ([] when every hazard has an adequate protection)")
if data.get("id") not in (None, cid):
errs.append(f"id '{data.get('id')}' differs from the file name {cid}")
return errs[:8]
def load_concepts(run, fl, deep=()):
concepts, invalid, errors, warnings, sources = {}, [], [], [], {}
for p in sorted(glob.glob(os.path.join(run, "work", "concepts", "C*.json"))):
cid = os.path.basename(p)[:-5]
if not re.fullmatch(r"C\d+", cid):
continue
path = p
dp = os.path.join(run, "work", "deep", cid, "estimates.json")
if cid in deep:
if os.path.isfile(dp):
path = dp
else:
warnings.append(f"{cid}: deepened estimates {fl.rel(dp, run)} missing; using the screened concept")
data, err = fl.load_json(path)
probs = [err] if err else validate(cid, data, fl)
if probs and path == dp:
warnings.append(f"{cid}: deepened estimates invalid ({probs[0]}); using the screened concept")
path = p
data, err = fl.load_json(p)
probs = [err] if err else validate(cid, data, fl)
if probs:
invalid.append(cid)
errors += [f"{cid}: {x}" for x in probs]
continue
concepts[cid] = data
sources[cid] = fl.rel(path, run)
return concepts, invalid, errors, warnings, sources
def norm_unit(u):
u = str(u or "").lower().replace(" ", "").replace("(", "").replace(")", "").replace("\u00b0", "deg")
return u.replace("\u20ac", "eur").replace("\u00b5", "u").replace("\u03bc", "u").replace("degreesc", "degc")
def unit_clashes(reqs, est):
"""Numeric requirements whose estimate states a different unit (values must be in the requirement's unit)."""
out = []
for r in reqs:
e = est.get(r["id"]) if isinstance(est, dict) else None
if r.get("kind") == "numeric" and isinstance(e, dict) and e.get("unit") and r.get("unit") \
and num(e.get("value")) is not None and norm_unit(e["unit"]) != norm_unit(r["unit"]):
out.append(f"{r['id']}: estimate unit '{e['unit']}' differs from the requirement unit '{r['unit']}'")
return out
def coverage_warnings(reqs, concepts):
out = []
for cid, c in sorted(concepts.items()):
est = c.get("estimates") or {}
miss = [r["id"] for r in reqs if r["id"] not in est]
if miss:
out.append(f"{cid}: no estimate for {len(miss)} requirement id(s) (scored unknown): {', '.join(miss[:6])}")
out += [f"{cid}: {x} (convert the value)" for x in unit_clashes(reqs, est)[:6]]
return out
def lens_names(run, fl):
data, _ = fl.load_json(os.path.join(run, "work", "morph.json"))
out = {}
for s in (data or {}).get("seeds", []) if isinstance(data, dict) else []:
if isinstance(s, dict) and s.get("id"):
out[s["id"]] = s.get("lens", "")
return out
def slim(results):
return {i: {k: v for k, v in r.items() if k != "per_req"} | {"per_req": {q: {"level": a["level"], "s": round(a["s"], 3),
"value": a["value"], "unit": a["unit"], "ko": a["ko"], "at_risk": a["at_risk"]} for q, a in r["per_req"].items()}}
for i, r in results.items()}
def mode_numeric(run, fl):
reqs, rw = fl.requirements(run)
concepts, invalid, errors, warnings, sources = load_concepts(run, fl)
warnings = ([rw] if rw else []) + warnings + coverage_warnings(reqs, concepts)
results = {i: concept_result(reqs, c) for i, c in concepts.items()}
order = rank(results)
ok = [i for i in order if not results[i]["ko"]]
top = ok[:4] if len(ok) >= 2 else order[:4]
if len(ok) < 2 and order:
warnings.append(f"only {len(ok)} concept(s) without knock-out; the panel compares the best by score")
if len(concepts) < 2:
errors.append(f"fewer than two valid concepts ({len(concepts)})")
sens = sensitivity(reqs, concepts) if concepts else {}
fl.write_json(os.path.join(run, "work", "scoring", "numeric.json"),
{"schema": "flow-score/1", "mode": "numeric", "sources": sources, "ranking": order, "top": top,
"concepts": slim(results), "sensitivity": sens})
notes = [f"{i}: score {results[i]['score']:.3f}{' KO ' + '; '.join(results[i]['ko_reasons'][:3]) if results[i]['ko'] else ''}" for i in order]
return fl.emit("score-numeric", errors, warnings, notes, {"ranking": order, "top": top,
"ko": [i for i in order if results[i]["ko"]], "invalid": invalid}, run)
def panel(run, fl, cands):
wins, games, fatal, rows, warnings = {c: 0.0 for c in cands}, {c: 0 for c in cands}, {c: {} for c in cands}, [], []
files = sorted(glob.glob(os.path.join(run, "work", "panel", "*.json")))
for p in files:
data, err = fl.load_json(p)
if err or not isinstance(data, dict) or not isinstance(data.get("pairs"), list):
warnings.append(f"panel file {fl.rel(p, run)} ignored: {err or 'no pairs[]'}")
continue
judge = str(data.get("judge") or os.path.basename(p)[:-5].split("-")[0])
for pr in data["pairs"]:
a, b, w = (pr or {}).get("a"), (pr or {}).get("b"), (pr or {}).get("winner")
if a not in wins or b not in wins or a == b or w not in (a, b, "tie"):
warnings.append(f"{fl.rel(p, run)}: pair {a} vs {b} winner {w!r} ignored")
continue
games[a] += 1
games[b] += 1
if w == "tie":
wins[a] += 0.5
wins[b] += 0.5
else:
wins[w] += 1
rows.append({"judge": judge, "order": data.get("order", ""), "a": a, "b": b, "winner": w,
"confidence": pr.get("confidence"), "why": str(pr.get("why", ""))[:300]})
for f in data.get("fatal") or []:
if isinstance(f, dict) and f.get("id") in fatal:
fatal[f["id"]][judge] = str(f.get("why", ""))[:300]
share = {c: (wins[c] / games[c] if games[c] else 0.5) for c in cands}
ko = {c: [f"panel: fatal flaw ({'; '.join(f'{j}: {w}' for j, w in v.items())})"] for c, v in fatal.items() if len(v) >= 2}
if not files:
warnings.append("no panel verdicts: combined score = numeric only (panel share 0.5 each)")
return {"files": len(files), "pairs": rows, "share": share, "fatal": fatal, "ko": ko}, warnings
def mode_combine(run, fl):
num_rec, err = fl.load_json(os.path.join(run, "work", "scoring", "numeric.json"))
if err:
return fl.emit("score-combine", [err + " (run the numeric mode first)"], extra={"ranking": [], "top": [], "ko": [], "invalid": []}, run=run)
numeric, cands = num_rec["concepts"], list(num_rec.get("top") or [])
pan, warnings = panel(run, fl, cands)
scale = scale_of([numeric[c]["score"] for c in cands])
comb = combine(cands, numeric, pan["share"], pan["ko"], scale)
order = rank(comb, "combined")
ok = [i for i in order if not comb[i]["ko"]]
top = (ok if ok else order)[:2]
rest = [i for i in num_rec["ranking"] if i not in order]
fl.write_json(os.path.join(run, "work", "scoring", "combined.json"),
{"schema": "flow-score/1", "mode": "combine", "candidates": cands, "panel": pan, "scale": scale,
"combined": comb, "ranking": order + rest, "top": top})
notes = [f"{i}: combined {comb[i]['combined']:.3f} (numeric {comb[i]['numeric']:.3f}, panel {comb[i]['panel_share']:.2f})"
+ (" KO" if comb[i]["ko"] else "") for i in order]
return fl.emit("score-combine", [] if cands else ["no candidates in numeric.json"], warnings, notes,
{"ranking": order + rest, "top": top, "ko": [i for i in order if comb[i]["ko"]], "invalid": []}, run)
def fmt_cell(a):
if a["level"] == "unknown" and a["value"] in (None, ""):
return "? (0.50)"
v = a["value"]
val = f"{v:g}" if isinstance(v, (int, float)) and not isinstance(v, bool) else (str(v)[:24] if v not in (None, "") else a["level"])
mark = " **KO**" if a["ko"] else (" (risk)" if a["at_risk"] else "")
return f"{val} ({a['s']:.2f}){mark}".replace("|", "/")
def esc(s, n=160):
return str(s if s is not None else "").replace("|", "/").replace("\n", " ")[:n]
def write_selection(run, fl, reqs, concepts, numeric, sens, comb_rec, deep_ids, deep_concepts, final, order, scale):
"""Selection record + re-runnable copy; returns an error text when the copy recomputes another winner."""
d = os.path.join(run, "deliverables", "02-concept")
calc = os.path.join(d, "calc")
os.makedirs(calc, exist_ok=True)
lenses = lens_names(run, fl)
pan = (comb_rec or {}).get("panel", {"pairs": [], "share": {}, "fatal": {}, "files": 0})
comb = (comb_rec or {}).get("combined", {})
ids = sorted(concepts, key=lambda x: int(re.sub(r"\D", "", x) or 0))
winner = order[0] if order else None
L = ["# Concept selection record", "",
"Generated by code from the pipeline's files (method/tools/score_concepts.py; reproducible with "
"`ddpy selection_score.py` in `calc/`). Concepts were generated from one morphological box under six "
"lenses, each screened by its own script, scored against the INPUT requirements, compared by a blind "
"three-judge panel; the top two were deepened (each could graft at most three features of the other) "
"before the final pick.", "",
"Rules: satisfaction 1 at the target, 0.5 at the threshold, linear between; a missed threshold, an "
"infeasible screening result, a hazard without an independent protection or a fatal flaw named by two "
f"judges is a knock-out (KO); unknown = 0.5 with a penalty of {UNKNOWN_PENALTY} x the unknown weight "
f"share; combined = {W_NUMERIC} x numeric (normalised on the scale of the panel candidates: lowest score "
f"{scale['lo']:.3f}, spread {scale['span']:.3f}, floor {NORM_FLOOR}; the final pick uses the same scale) + "
f"{W_PANEL} x panel share; ties go to the lower unit cost at 1000; if every deepened concept is KO, a "
"concept knocked out only by missed thresholds beats a safety, infeasibility or fatal-flaw KO, then the "
"smallest knocked-out weight wins. Weights are the requirement weights.", "",
"## 1. The concepts", "",
"| id | lens | name | principle | numeric score | unknown share | cost @1000 | knock-out |",
"|---|---|---|---|---|---|---|---|"]
for i in ids:
c, r = concepts[i], numeric[i]
L.append(f"| {i} | {esc(lenses.get(i, c.get('lens', '')), 60)} | {esc(c.get('name'), 50)} | {esc(c.get('principle'))} | "
f"{r['score']:.3f} | {r['unknown_share']:.2f} | {esc(c.get('cost_q1000'))} {esc(c.get('currency', ''))} | "
f"{esc('; '.join(r['ko_reasons']) or 'no', 200)} |")
L += ["", "## 2. Requirement satisfaction (screened estimates; value (satisfaction))", "",
"| requirement | weight | target / threshold | " + " | ".join(ids) + " |",
"|---|---|---|" + "---|" * len(ids)]
for r in reqs:
tt = f"{esc(r.get('direction', ''))} {esc(r.get('target', r.get('target_text', '')), 40)} / {esc(r.get('threshold', r.get('threshold_text', '')), 40)} {esc(r.get('unit', ''))}"
L.append(f"| {r['id']} | {weight_of(r):g} | {tt.strip()} | " + " | ".join(fmt_cell(numeric[i]["per_req"][r["id"]]) for i in ids) + " |")
L += ["", f"## 3. Blind panel ({pan.get('files', 0)} verdict files)", ""]
if pan.get("pairs"):
L += ["| judge | order | pair (A vs B) | winner | confidence | reason |", "|---|---|---|---|---|---|"]
L += [f"| {esc(p['judge'])} | {esc(p['order'])} | {p['a']} vs {p['b']} | {p['winner']} | {esc(p.get('confidence'))} | {esc(p['why'], 240)} |" for p in pan["pairs"]]
fat = [f"{c}: " + "; ".join(f"{j}: {w}" for j, w in v.items()) for c, v in pan.get("fatal", {}).items() if v]
L += ["", "Fatal-flaw flags: " + (" / ".join(esc(x, 300) for x in fat) if fat else "none")]
else:
L.append("No panel verdicts were available; the combination used the numeric score only.")
if comb:
L += ["", "## 4. Combination of the candidates", "", "| id | numeric | normalised | panel share | combined | KO |", "|---|---|---|---|---|---|"]
L += [f"| {i} | {c['numeric']:.3f} | {c['numeric_norm']:.3f} | {c['panel_share']:.2f} | {c['combined']:.3f} | {esc('; '.join(c['ko_reasons']) or 'no', 200)} |"
for i, c in sorted(comb.items(), key=lambda kv: -kv[1]["combined"])]
L += ["", "## 5. Deepened concepts and final pick", "", "| id | numeric (deepened) | panel share | final | KO |", "|---|---|---|---|---|"]
L += [f"| {i} | {final[i]['numeric']:.3f} | {final[i]['panel_share']:.2f} | {final[i]['combined']:.3f} | {esc('; '.join(final[i]['ko_reasons']) or 'no', 200)} |" for i in order]
if winner:
L += ["", f"**Winner: {winner} - {esc(concepts.get(winner, {}).get('name'), 80)}.** "
+ (f"Runner-up (graft source): {order[1]}." if len(order) > 1 else "No runner-up was deepened.")]
if final[winner]["ko"]:
L.append(f"All deepened concepts were knocked out; the winner has the smallest knocked-out weight. Known misses: {esc('; '.join(final[winner]['ko_reasons']), 400)}.")
L += ["", "## 6. Sensitivity (all concepts, screened estimates)", ""]
if sens:
L += [f"- Numeric leader {sens['leader']} holds in {sens['leader_holds']} of {sens['weight_tests']} single-weight changes of +/-20 %.",
f"- Flips: {esc('; '.join(sens['flips']), 600) or 'none'}.",
f"- Leader with every estimate at its pessimistic bound: {sens['pessimistic_leader']}; at its optimistic bound: {sens['optimistic_leader']}."]
with open(os.path.join(d, "selection.md"), "w", encoding="utf-8") as fh:
fh.write("\n".join(L) + "\n")
fl.write_json(os.path.join(d, "selection.json"),
{"schema": "flow-selection/1", "winner": winner, "runner_up": order[1] if len(order) > 1 else None,
"deepened": deep_ids, "lenses": lenses, "numeric": slim(numeric), "sensitivity": sens,
"panel": pan, "combined": comb, "final": final})
slim_c = lambda c: {k: c.get(k) for k in ("estimates", "safety_ko", "feasible", "cost_q1000", "name")}
fl.write_json(os.path.join(calc, "selection_inputs.json"),
{"requirements": [{k: r.get(k) for k in ("id", "kind", "direction", "target", "threshold", "unit", "weight")} for r in reqs],
"concepts": {i: slim_c(c) for i, c in concepts.items()},
"deepened": deep_ids, "deepened_concepts": {i: slim_c(c) for i, c in deep_concepts.items()},
"panel_share": pan.get("share", {}), "panel_ko": pan.get("ko", {}), "scale": scale, "winner": winner})
shutil.copyfile(os.path.abspath(__file__), os.path.join(calc, "selection_score.py"))
fl.write_json(os.path.join(run, "work", "manifest-desc", "selection.json"), {
"deliverables/02-concept/selection.md": "concept selection record: all six concepts, scores, panel verdicts, sensitivity, final pick (generated by code)",
"deliverables/02-concept/selection.json": "the selection record as data",
"deliverables/02-concept/calc/selection_score.py": "re-runs the concept selection from selection_inputs.json",
"deliverables/02-concept/calc/selection_inputs.json": "inputs of the selection (requirements, estimates, panel shares, scale)",
"deliverables/02-concept/calc/selection_score.results.json": "scores and winner recomputed by selection_score.py"})
cwd = os.getcwd()
try:
os.chdir(calc)
if recompute_here(quiet=True):
return "selection_score.py recomputed a different winner than the pick - scoring code bug"
finally:
os.chdir(cwd)
return None
CALC_HOME = "deliverables/02-concept/calc/"
REQ_ID = re.compile(r"\b[A-Z][A-Z0-9]{1,7}-[A-Z0-9]{1,7}-\d{1,4}\b")
def promote(run, fl, winner, source):
"""Make the winner's deepened folder the package's calculation home: copy work/deep/<id>/calc/ into
deliverables/02-concept/calc/ (the previous promotion's files go, the selection files stay), write
work/final/estimates.json and key_figures.json with the paths rewritten, the evidence map of the scripts and
their manifest descriptions. `source` is the estimates file the pick used for the winner. Returns (errors, notes)."""
errors, notes = [], []
calc = os.path.join(run, CALC_HOME)
os.makedirs(calc, exist_ok=True)
for name in os.listdir(calc):
if not name.startswith("selection_"):
p = os.path.join(calc, name)
shutil.rmtree(p, ignore_errors=True) if os.path.isdir(p) else os.remove(p)
deep = f"work/deep/{winner}/"
if source.startswith(deep):
src, kf_path = os.path.join(run, deep, "calc"), os.path.join(run, deep, "key_figures.json")
prefix, target = deep + "calc/", CALC_HOME
files = [os.path.relpath(os.path.join(r, n), src) for r, _d, ns in os.walk(src) for n in ns
if "__pycache__" not in r and not n.endswith(".tmp")] if os.path.isdir(src) else []
else:
errors.append(f"{winner}: {deep}estimates.json missing or invalid - the screened concept was promoted instead; "
f"fix {deep} (estimates.json, calc/, key_figures.json)")
src, kf_path = os.path.join(run, "work", "concepts"), None
prefix, target = f"work/concepts/{winner}_calc", f"{CALC_HOME}{winner}_calc"
files = sorted(n for n in os.listdir(src) if n.startswith(f"{winner}_calc")) if os.path.isdir(src) else []
for f in files:
os.makedirs(os.path.dirname(os.path.join(calc, f)), exist_ok=True)
shutil.copy2(os.path.join(src, f), os.path.join(calc, f))
def rw(x): # run-relative paths of the source folder -> the package's calculation home
if isinstance(x, str):
return x.replace(prefix, target)
if isinstance(x, list):
return [rw(i) for i in x]
return {k: rw(v) for k, v in x.items()} if isinstance(x, dict) else x
data, err = fl.load_json(os.path.join(run, source))
if err or not isinstance(data, dict):
errors.append(f"{winner}: {err or source + ' is not a JSON object'}")
data = {"estimates": {}}
data = rw(data)
for e in (data.get("estimates") or {}).values():
if isinstance(e, dict):
ev = e.get("evidence")
ev = [ev] if isinstance(ev, str) else (list(ev) if isinstance(ev, list) else [])
home = str(e.get("basis") or "").split("#", 1)[0].strip()
if home.startswith(CALC_HOME) and home not in ev:
ev.append(home)
e["evidence"] = ev
data.update(id=winner, promoted_from=source)
fl.write_json(os.path.join(run, "work", "final", "estimates.json"), data)
kf, kerr = fl.load_json(kf_path) if kf_path else (None, "missing")
if kerr or not isinstance(kf, dict):
errors.append(f"{winner}: {deep}key_figures.json {kerr or 'is not a JSON object'} (null key figures written)")
kf = {"dimensions_mm": None, "mass_kg": None, "power_w": None, "origins": {}}
fl.write_json(os.path.join(run, "work", "final", "key_figures.json"), rw(kf))
emap, desc = {}, {}
for r, _d, ns in os.walk(calc):
for n in sorted(ns):
if n.startswith("selection_") or "__pycache__" in r:
continue
rel = os.path.relpath(os.path.join(r, n), run).replace(os.sep, "/")
if n.endswith(".py"):
text = fl.read_text(os.path.join(r, n))[:4000]
m = re.search(r"Requirements:\s*([^\n]+)", text)
res = rel[:-3] + ".results.json"
for rid in (REQ_ID.findall(m.group(1)) if m else []):
if os.path.isfile(os.path.join(run, res)) and res not in emap.get(rid, []):
emap.setdefault(rid, []).append(res)
doc = re.search(r'(?:"""|\'\'\')\s*([^\n]+)', text)
desc[rel] = "concept calculation: " + (doc.group(1).strip()[:140] if doc else n)
elif n.endswith(".results.json"):
desc[rel] = f"results of {n[:-len('.results.json')]}.py"
else:
desc[rel] = f"input data of the concept calculations ({n})"
fl.write_json(os.path.join(run, "work", "final", "evidence-calc.json"), emap)
fl.write_json(os.path.join(run, "work", "manifest-desc", "calc.json"), desc)
notes.append(f"promoted {winner} from {source}: {len(files)} file(s) into {CALC_HOME}, work/final/estimates.json "
f"and key_figures.json")
return errors, notes
def run_tool(path, args, timeout):
"""Run a sibling gate/tool and read its JSON verdict (last output line): (verdict, error text)."""
try:
p = subprocess.run([sys.executable, path] + args, capture_output=True, text=True, timeout=timeout)
return json.loads(p.stdout.strip().splitlines()[-1]), None
except Exception as exc: # noqa: BLE001
return None, f"{os.path.basename(path)} did not run to its verdict: {str(exc)[:160]}"
def mode_final(run, fl, deep_ids, tier):
comb_rec, err = fl.load_json(os.path.join(run, "work", "scoring", "combined.json"))
warnings = []
if err:
warnings.append(f"{err}; panel shares default to 0.5")
comb_rec = None
reqs, rw = fl.requirements(run)
if rw:
warnings.append(rw)
concepts, invalid, errors, w2, _ = load_concepts(run, fl)
deep_all, _, _, w3, sources = load_concepts(run, fl, deep=tuple(deep_ids))
warnings += w2 + w3
deep_ids = [i for i in deep_ids if i in deep_all]
if not deep_ids:
return fl.emit("score-final", errors + ["no valid deepened concept"], warnings,
extra={"ranking": [], "top": [], "ko": [], "invalid": invalid}, run=run)
deep_concepts = {i: deep_all[i] for i in deep_ids}
numeric = {i: concept_result(reqs, c) for i, c in concepts.items()}
deep_num = {i: concept_result(reqs, c) for i, c in deep_concepts.items()}
pan = (comb_rec or {}).get("panel", {})
scale = (comb_rec or {}).get("scale")
if not isinstance(scale, dict):
scale = scale_of([deep_num[i]["score"] for i in deep_ids])
warnings.append("no panel-candidate scale in combined.json: normalised over the deepened concepts")
final = combine(deep_ids, deep_num, pan.get("share", {}), pan.get("ko", {}), scale)
order = pick(final)
rest = [i for i in ((comb_rec or {}).get("ranking") or rank(numeric)) if i not in order]
runner = order[1] if len(order) > 1 else (rest[0] if rest else None)
sens = sensitivity(reqs, concepts) if concepts else {}
fl.write_json(os.path.join(run, "work", "scoring", "final.json"),
{"schema": "flow-score/1", "mode": "final", "deepened": deep_ids, "sources": sources, "scale": scale,
"final": final, "winner": order[0], "runner_up": runner, "ranking": order + rest})
p_err, p_notes = promote(run, fl, order[0], sources.get(order[0], ""))
bug = write_selection(run, fl, reqs, concepts, numeric, sens, comb_rec, deep_ids, deep_concepts, final, order, scale)
notes = [f"{i}: final {final[i]['combined']:.3f}" + (" KO" if final[i]["ko"] else "") for i in order] + p_notes
notes.append("wrote deliverables/02-concept/selection.md, selection.json and calc/selection_score.py + inputs + results")
out_err = (errors if len(concepts) < 2 else []) + p_err + ([bug] if bug else [])
gates = os.path.normpath(os.path.join(HERE, "..", "gates"))
res, terr = run_tool(os.path.join(gates, "check_concept.py"), ["--stage", "final", "--run-dir", run, "--json"], 400)
out_err += [terr] if terr else res.get("errors", [])
warnings += [] if terr else res.get("warnings", [])
res, terr = run_tool(os.path.join(HERE, "make_package.py"), ["--tier", tier, "--run-dir", run, "--json"], 120)
warnings += [terr] if terr else [f"package refreshed before the writers: {e}" for e in res.get("errors", [])[:8]]
top = [order[0]] + ([runner] if runner else [])
return fl.emit("score-final", out_err, warnings + [f"invalid concept {e}" for e in errors],
notes, {"ranking": order + rest, "top": top, "ko": [i for i in order if final[i]["ko"]],
"invalid": invalid}, run)
def main(argv):
if not argv:
return recompute_here()
fl = flowlib()
ap = fl.parser(__doc__)
ap.add_argument("mode", choices=["numeric", "combine", "final"])
ap.add_argument("--deep", default="", help="comma-separated deepened concept ids (final mode)")
ap.add_argument("--tier", choices=["concept", "full"], default="concept", help="final mode: tier of the package refresh")
a = ap.parse_args(argv)
run = os.path.abspath(a.run_dir)
if a.mode == "numeric":
return mode_numeric(run, fl)
if a.mode == "combine":
return mode_combine(run, fl)
return mode_final(run, fl, [x.strip() for x in a.deep.split(",") if x.strip()], a.tier)
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
files/method/workflows/concept.js (262 wierszy)
export const meta = {
name: 'dd-flowcode-concept',
description: 'FLOWCODE concept pipeline: six scouts beside the requirements stage, six lens concepts each written and screened by its own script, code scoring, blind pairwise panel, top two deepened (each may graft up to three features of the other), final pick and promotion in code, writers, package gate',
phases: [
{ title: 'Scout', detail: 'use, physics + morphological box, hazards, benchmark, parts, regulatory - in parallel' },
{ title: 'Requirements', detail: 'requirements.md + requirements.json beside the scouts and the concepts; gate R; retried with the gate errors' },
{ title: 'Concepts', detail: 'six lens concepts, each written and then screened by its own script' },
{ title: 'Score', detail: 'knock-outs and weighted satisfaction in code' },
{ title: 'Panel', detail: 'three blind judges x two presentation orders; code combines 0.6 numeric + 0.4 panel' },
{ title: 'Deepen', detail: 'top two in parallel; each may graft up to three features of the other and writes its final calculation homes' },
{ title: 'Pick', detail: 'final pick in code; the winner promoted into 02-concept; package refreshed; one fix round' },
{ title: 'Write', detail: 'concept, safety + compliance route, cost + risks, report draft - in parallel' },
{ title: 'Package', detail: 'gate C (code refreshes compliance.csv and manifest.json first); one fixer round' },
],
}
// ---------------------------------------------------------------- configuration (args come from the top session)
const A = args || {}
const RUN = String(A.run_dir || '').replace(/\/+$/, '')
if (!RUN.startsWith('/')) throw new Error('args.run_dir must be the absolute run directory')
const TIER = A.tier === 'full' ? 'full' : 'concept'
const T0 = Number(A.start_epoch) || 0
const HARD = Number(A.hard_stop_min) || (TIER === 'full' ? 600 : 120)
let NOW = T0
const partial = []
const elapsed = () => (T0 && NOW > T0 ? Math.round((NOW - T0) / 60) : 0)
const frac = () => elapsed() / HARD
const tick = r => { if (r && typeof r.epoch === 'number' && r.epoch > NOW) NOW = r.epoch; return r }
// Planned start of the stages in minutes after START (MAIN.md section 5); lateness is measured against this plan.
const PLAN = { concepts: 12, score: 22, panel: 23, deepen: 30, pick: 41, write: 42, package: 50 }
const late = stage => elapsed() - PLAN[stage]
const SCOUTS = ['use', 'physics', 'hazards', 'benchmark', 'parts', 'regulatory']
const LENSES = [
{ id: 'C1', lens: 'lowest unit cost at 1000 units that still meets every threshold' },
{ id: 'C2', lens: 'best result on the highest-weighted requirements' },
{ id: 'C3', lens: 'smallest and lightest' },
{ id: 'C4', lens: 'most robust, serviceable and inherently safe' },
{ id: 'C5', lens: 'easiest to certify and to manufacture at 1, 100 and 1000 units' },
{ id: 'C6', lens: 'boldest buildable: a working principle that none of the other seeds uses' },
]
const JUDGES = [
{ id: 'physics', lens: 'physics and requirements - are the numbers physically credible, and which concept really meets the highest-weighted requirements' },
{ id: 'cost', lens: 'cost and manufacturing - real parts, unit cost at 1/100/1000, tooling, assembly effort, supply risk' },
{ id: 'safety', lens: 'safety, compliance and honesty - hazards with independent protections, certification route, claims without evidence' },
]
const CMD = {
morph: 'python3 method/gates/check_concept.py --stage morph --json',
req: 'python3 method/gates/check_requirements.py --json',
concept: id => `python3 method/gates/check_concept.py --stage concept --id ${id} --json`,
numeric: 'python3 method/tools/score_concepts.py numeric --json',
combine: 'python3 method/tools/score_concepts.py combine --json',
deep: id => `python3 method/gates/check_concept.py --stage deep --id ${id} --json`,
final: ids => `python3 method/tools/score_concepts.py final --deep ${ids.join(',')} --tier ${TIER} --json`,
gateC: `python3 method/gates/check_concept.py --stage package --tier ${TIER} --refresh --json`,
}
// ---------------------------------------------------------------- schemas
const S = { type: 'string' }
const B = { type: 'boolean' }
const I = { type: 'integer' }
const L = items => ({ type: 'array', items })
const O = (properties, required) => ({ type: 'object', properties, required: required || Object.keys(properties) })
const DONE = O({ done: B, files: L(S), note: S, epoch: I })
const GATE = O({ pass: B, errors: L(S), warnings: L(S), epoch: I }, ['pass', 'errors', 'epoch'])
const RANK = O({ pass: B, errors: L(S), epoch: I, ranking: L(S), top: L(S), ko: L(S), invalid: L(S) })
// ---------------------------------------------------------------- prompt pieces
const head = (role, box) => [
`Run directory: ${RUN} - cd there first; every path below is relative to it. Tier: ${TIER}.`,
'Standing rules: .claude/agents/flow-stage.md (read it unless it already is your system prompt).',
`Read method/HYGIENE.md, your role brief method/roles/${role}, and sections 2, 3 and 5 of work/run-contract.md.`,
`Time box: ${box} min from your start (check with date).`,
].join('\n')
const RET = '\nReturn {done, files (paths you wrote), note (one line: the result or the problem), epoch (output of date +%s)}.'
const selfCheck = cmd => `\nSELF-CHECK before returning: run \`${cmd}\` and fix every ERROR it prints (fix the content; never weaken it to pass).`
const retryText = (cmd, errs, box) => `\n\nRETRY: the previous attempt's files are on disk. The gate \`${cmd}\` rejected them with:\n- ${errs.slice(0, 25).join('\n- ')}\nFix exactly these problems in place (edit, do not start over) and keep what already passed. Time box for this fix: ${box} min (it replaces the one above).`
// ---------------------------------------------------------------- agent wrappers (control flow never depends on prose)
async function call(prompt, opts) {
try { return await agent(prompt, opts) } catch (e) { log(`${opts.label}: agent error ${String((e && e.message) || e).slice(0, 160)}`); return null }
}
// One stage agent: retried once, then once as general-purpose; an empty result marks the stage PARTIAL
// (inside gated() the gate verdict decides instead).
async function run(label, ph, prompt, x) {
const c = x || {}
const o = { label, phase: ph, schema: c.schema || DONE, agentType: c.type || 'flow-stage', effort: c.effort || 'high' }
let r = await call(prompt, o)
if (!r) r = await call(prompt, { ...o, label: `${label} (retry)` })
if (!r) r = await call(prompt, { ...o, label: `${label} (general-purpose)`, agentType: 'general-purpose' })
if (!r) {
log(`${label}: no result after three attempts`)
if (!c.gated) partial.push({ stage: label, errors: ['no result after three attempts'] })
}
return tick(r)
}
const gatePrompt = cmd => [
`Run directory: ${RUN}.`,
'Run exactly this command with the Bash tool (timeout 600000 ms) and nothing else:',
'',
`cd ${RUN} && ${cmd}`,
'',
'Its last output line is one JSON object. Return its fields exactly as printed (lists verbatim, at most 40 items).',
'If the command does not start, times out or prints no JSON line: pass=false, errors=["<one line: what happened>"], all other lists empty, epoch = output of date +%s.',
'Never edit files, never fix anything, never run another command.',
].join('\n')
async function gate(label, ph, cmd, schema) {
const s = schema || GATE
const p = gatePrompt(cmd)
let r = await call(p, { label, phase: ph, schema: s, agentType: 'flow-gate', model: 'haiku', effort: 'low' })
if (!r) r = await call(p, { label: `${label} (retry)`, phase: ph, schema: s, agentType: 'flow-gate', model: 'sonnet', effort: 'low' })
if (!r) r = await call(p, { label: `${label} (general-purpose)`, phase: ph, schema: s, agentType: 'general-purpose', model: 'sonnet', effort: 'low' })
if (!r) r = { pass: false, errors: [`gate runner returned nothing for: ${cmd}`], epoch: 0 }
for (const k of ['errors', 'warnings', 'ranking', 'top', 'ko', 'invalid']) if (!Array.isArray(r[k])) r[k] = []
return tick(r)
}
// Build -> gate -> retry with the gate's errors; still failing after the retries -> PARTIAL, the line continues.
async function gated(label, ph, prompt, cmd, tries, fixBox) {
let errs = []
let res = null
let g = null
for (let i = 0; i <= tries; i++) {
const p = i === 0 ? prompt + selfCheck(cmd) + RET : prompt + retryText(cmd, errs, fixBox) + selfCheck(cmd) + RET
res = (await run(i === 0 ? label : `${label} (fix ${i})`, ph, p, { gated: true })) || res
g = await gate(`gate: ${label}${i ? ' ' + (i + 1) : ''}`, ph, cmd)
if (g.pass) return { ok: true, res, gate: g }
errs = g.errors.length ? g.errors : ['the gate failed without error text']
const more = i < tries && frac() <= 0.75 // no retries above 75 % of the hard stop
log(`${label}: gate FAIL (${errs.length} errors)${more ? ' - retrying with the errors' : ' - marked PARTIAL'}`)
if (!more) break
}
partial.push({ stage: label, errors: errs.slice(0, 8) })
return { ok: false, res, gate: g }
}
// ---------------------------------------------------------------- stage prompts
const scoutPrompt = s => head('scouts.md', 8) +
`\nYou are scout "${s}": do section "${s}" of the brief and nothing else; five other scouts and the requirements stage run in parallel.` +
(s === 'physics' ? `\nThe six lenses for the morphological-box seeds:\n${LENSES.map(l => `${l.id}: ${l.lens}`).join('\n')}` : '')
const reqPrompt = head('requirements.md', 15) + '\nThe six scouts start with you and the six concept agents follow them; follow the order of work in your brief.'
const conceptPrompt = l => head('concepts.md', 10) +
`\nConcept ${l.id}. Your lens: ${l.lens}. Your seed: entry "${l.id}" of work/morph.json seeds. Write it, then screen it with your own script (both parts of the brief). Five other concept agents work in parallel; write only work/concepts/${l.id}.* and work/concepts/${l.id}_calc.*.` +
selfCheck(CMD.concept(l.id))
const pairsOf = list => {
const out = []
for (let i = 0; i < list.length; i++) for (let j = i + 1; j < list.length; j++) out.push(`${list[i]} vs ${list[j]}`)
return out
}
const judgePrompt = (j, order, list) => [
`Run directory: ${RUN} - cd there first. Read method/HYGIENE.md.`,
'Standing rules and the verdict file format: .claude/agents/flow-judge.md (read it unless it already is your system prompt).',
`You are judge "${j.id}" of a blind panel. Your lens: ${j.lens}.`,
`Candidates: ${list.join(', ')}. For each read work/concepts/<id>.md and work/concepts/<id>_calc.results.json - nothing else under work/.`,
`Compare these pairs, each presented in this order (the first named is A): ${pairsOf(list).join('; ')}.`,
`Write work/panel/${j.id}-${order}.json in the format of the standing rules. Time box: 6 min.`,
].join('\n') + RET
const deepPrompt = (id, src, ids) => head('deepen.md', 11) + `\nDeepen concept ${id}; output folder work/deep/${id}/.` +
(src ? ` Graft source: concept ${src} - read only its screened dossier work/concepts/${src}.md and .json; you may graft at most three of its features.` : ' No graft source: make no graft.') +
(ids.length > 1 ? ` Another agent deepens ${ids.filter(x => x !== id).join(', ')} at the same time; never touch work/deep/ folders other than yours.` : '') +
' Code picks the winner from your estimates and copies the winner\'s folder into the package: write everything as final.' +
selfCheck(CMD.deep(id))
const deepFixPrompt = (id, errs) => head('deepen.md', 6) + `\nFIX ROUND for concept ${id}, the winner of the final pick. Code copied work/deep/${id}/ into the package and the final check rejected it with:\n- ${errs.slice(0, 25).join('\n- ')}\nFix the causes in work/deep/${id}/ only (code copies the folder again; never edit deliverables/02-concept/calc/). Keep the design and what already passed.` +
selfCheck(CMD.deep(id)) + RET
const writePrompt = (part, w) => head('writers.md', 8) + `\nDo part "${part}" for the chosen concept ${w}. Three other writers work in parallel on the other parts.` +
(part === 'writeup' ? `\nStages that ended PARTIAL so far: ${partial.length ? partial.map(p => p.stage).join(', ') : 'none'}.` : '') + RET
const fixPrompt = errs => head('fixer.md', 8) + `\nThe gate \`${CMD.gateC}\` failed with:\n- ${errs.slice(0, 30).join('\n- ')}` + selfCheck(CMD.gateC) + RET
// ---------------------------------------------------------------- A || B, C: scouts, then concepts; requirements beside both
phase('Scout')
await parallel([
() => gated('requirements', 'Requirements', reqPrompt, CMD.req, 2, 6),
async () => {
await parallel(SCOUTS.map(s => () =>
s === 'physics' ? gated('scout physics', 'Scout', scoutPrompt(s), CMD.morph, 1, 5) : run(`scout ${s}`, 'Scout', scoutPrompt(s) + RET)))
log(`Scouts done at about ${elapsed()} min (plan ${PLAN.concepts})`)
// E+F: each concept agent writes its concept and screens it with its own script
await parallel(LENSES.map(l => () => run(`concept ${l.id}`, 'Concepts', conceptPrompt(l) + RET)))
log(`Concepts done at about ${elapsed()} min (plan ${PLAN.score})`)
},
])
// ---------------------------------------------------------------- G: numeric scoring in code
phase('Score')
let sc = await gate('score: numeric', 'Score', CMD.numeric, RANK)
const broken = LENSES.filter(l => sc.invalid.includes(l.id))
if (broken.length && frac() < 0.5) {
log(`Concepts rejected by the scorer: ${broken.map(l => l.id).join(', ')} - regenerating once with the errors`)
await parallel(broken.map(l => () =>
run(`concept ${l.id} (fix)`, 'Concepts', conceptPrompt(l) + retryText(CMD.numeric, sc.errors.filter(e => e.startsWith(`${l.id}:`)), 6) + RET)))
sc = await gate('score: numeric 2', 'Score', CMD.numeric, RANK)
}
if (!sc.ranking.length) {
partial.push({ stage: 'score', errors: sc.errors.slice(0, 8) })
return { status: 'failed', stage: 'Score', errors: sc.errors.slice(0, 10), invalid: sc.invalid, partial, elapsed_min: elapsed() }
}
// ---------------------------------------------------------------- G: blind panel, both presentation orders
phase('Panel')
const cand = sc.top.slice(0, 4)
if (cand.length >= 2) {
const fwd = cand.slice().sort()
const rev = fwd.slice().reverse()
await parallel(JUDGES.flatMap(j => [['forward', fwd], ['reverse', rev]].map(([o, list]) => () =>
run(`judge ${j.id} ${o}`, 'Panel', judgePrompt(j, o, list), { type: 'flow-judge' }))))
} else {
log('Panel skipped: fewer than two candidates')
}
const cb = await gate('score: combine', 'Panel', CMD.combine, RANK)
const ranked = cb.pass && cb.top.length ? cb : sc
// ---------------------------------------------------------------- H: deepen the top two; each may graft from the other
phase('Deepen')
const nDeep = late('deepen') > 12 && frac() > 0.3 ? 1 : 2
const deepIds = ranked.top.slice(0, nDeep)
if (!deepIds.length) deepIds.push(sc.ranking[0])
const graftSource = id => (deepIds.length > 1 ? deepIds.find(x => x !== id) : ranked.ranking.find(x => x !== id)) || null
if (deepIds.length === 1) log(`Behind the plan by ${late('deepen')} min (${elapsed()} of ${HARD} min): deepening only ${deepIds[0]}, graft source ${graftSource(deepIds[0]) || 'none'}`)
await parallel(deepIds.map(id => () => run(`deepen ${id}`, 'Deepen', deepPrompt(id, graftSource(id), deepIds) + RET)))
// ---------------------------------------------------------------- I: final pick + promotion in code, one fix round
phase('Pick')
let fin = await gate('final pick', 'Pick', CMD.final(deepIds), RANK)
if (!fin.pass && fin.top.length && frac() < 0.6) {
log(`final pick: the promoted winner ${fin.top[0]} failed the final check (${fin.errors.length} errors) - one fix round`)
await run(`deepen ${fin.top[0]} (fix)`, 'Pick', deepFixPrompt(fin.top[0], fin.errors))
fin = await gate('final pick 2', 'Pick', CMD.final(deepIds), RANK)
}
if (!fin.pass) partial.push({ stage: 'final pick', errors: fin.errors.slice(0, 8) })
const winner = fin.top[0] || deepIds[0]
const runnerUp = fin.top[1] || ranked.ranking.find(x => x !== winner) || null
log(`Final pick: ${winner} (runner-up ${runnerUp || 'none'}) at about ${elapsed()} min (plan ${PLAN.write})`)
// ---------------------------------------------------------------- J + L: writers in parallel (compliance.csv is current)
phase('Write')
await parallel(['main', 'safety', 'cost', 'writeup'].map(part => () => run(`write ${part}`, 'Write', writePrompt(part, winner))))
// ---------------------------------------------------------------- K: gate C (refreshes the package first), one fixer round
phase('Package')
let gc = await gate('gate C', 'Package', CMD.gateC)
if (!gc.pass && frac() < 0.75) {
await run('fixer', 'Package', fixPrompt(gc.errors))
gc = await gate('gate C 2', 'Package', CMD.gateC)
}
if (!gc.pass) partial.push({ stage: 'gate C', errors: gc.errors.slice(0, 8) })
return {
status: gc.pass && !partial.length ? 'ok' : 'partial',
tier: TIER,
winner,
runner_up: runnerUp,
deepened: deepIds,
ranking: ranked.ranking,
gate_c_pass: !!gc.pass,
gate_c_errors: gc.errors.slice(0, 10),
partial,
elapsed_min: elapsed(),
}
files/method/workflows/full.js (216 wierszy)
export const meta = {
name: 'dd-flowcode-full',
description: 'FLOWCODE full-tier pipeline: architecture freeze, parallel discipline builds behind code gates with retries, integration cross-checks with targeted re-runs, drawings/renders/test plan, verification, package and final gate',
phases: [
{ title: 'Freeze', detail: 'params.json as the single home of numbers, interfaces, plan; gate' },
{ title: 'Build', detail: 'CAD | calculation groups | schematic -> (board | firmware | BOM) | safety; build -> gate -> retry' },
{ title: 'Integrate', detail: 'cross-checks in code and by an integration agent; targeted re-runs of the owning stage' },
{ title: 'Finish', detail: 'drawings, renders, test plan in parallel, each gated' },
{ title: 'Package', detail: 'verification, compliance and manifest by code, write-up, final gate, routed repair round' },
],
}
// ---------------------------------------------------------------- configuration (args come from the top session)
const A = args || {}
const RUN = String(A.run_dir || '').replace(/\/+$/, '')
if (!RUN.startsWith('/')) throw new Error('args.run_dir must be the absolute run directory')
const T0 = Number(A.start_epoch) || 0
const HARD = Number(A.hard_stop_min) || 600
let NOW = T0
const partial = []
const elapsed = () => (T0 && NOW > T0 ? Math.round((NOW - T0) / 60) : 0)
const frac = () => elapsed() / HARD
const T = n => (frac() > 0.75 ? 0 : n)
const tick = r => { if (r && typeof r.epoch === 'number' && r.epoch > NOW) NOW = r.epoch; return r }
const G = part => `python3 method/gates/check_full.py --part ${part} --json`
const CMD = {
params: G('params'), cad: G('cad'), sch: G('sch'), pcb: G('pcb'), firmware: G('firmware'), safety: G('safety'),
bom: G('bom'), drawings: G('drawings'), renders: G('renders'), testplan: G('testplan'), calc: G('calc'), all: G('all'),
pkg: 'python3 method/tools/make_package.py --tier full --json',
}
// ---------------------------------------------------------------- schemas
const S = { type: 'string' }
const B = { type: 'boolean' }
const I = { type: 'integer' }
const L = items => ({ type: 'array', items })
const O = (properties, required) => ({ type: 'object', properties, required: required || Object.keys(properties) })
const DONE = O({ done: B, files: L(S), note: S, epoch: I })
const GATE = O({ pass: B, errors: L(S), warnings: L(S), epoch: I }, ['pass', 'errors', 'epoch'])
const PLAN = O({ done: B, files: L(S), note: S, epoch: I, board: B, mcu_env: S,
calc_groups: L(O({ name: S, questions: L(S) })) })
const ISSUES = O({ issues: L(O({ owner: S, issue: S })), note: S, epoch: I })
// ---------------------------------------------------------------- prompt pieces
const head = (role, box) => [
`Run directory: ${RUN} - cd there first; every path below is relative to it. Tier: full.`,
'Standing rules: .claude/agents/flow-stage.md (read it unless it already is your system prompt).',
`Read method/HYGIENE.md, your role brief method/roles/${role}, and sections 2, 3 and 5 of work/run-contract.md.`,
`Time box: ${box} min from your start (check with date).`,
].join('\n')
const RET = '\nReturn {done, files (paths you wrote), note (one line: the result or the problem), epoch (output of date +%s)}.'
const selfCheck = cmd => `\nSELF-CHECK before returning: run \`${cmd}\` and fix every ERROR it prints (fix the content; never weaken it to pass).`
const retryText = (cmd, errs) => `\n\nRETRY: the previous attempt's files are on disk. The gate \`${cmd}\` rejected them with:\n- ${errs.slice(0, 25).join('\n- ')}\nFix exactly these problems in place (edit, do not start over) and keep what already passed.`
const ROLE = {
cad: ['cad.md'], calc: ['calc.md'], sch: ['electronics.md', 'sch'], pcb: ['electronics.md', 'pcb'], firmware: ['firmware.md'],
safety: ['safety.md'], bom: ['bom.md'], drawings: ['drawings.md'], renders: ['renders.md'], testplan: ['testplan.md'],
verification: ['verification.md'],
}
const brief = (key, box, extra) => head(ROLE[key][0], box) +
'\nAlso read method/roles/full-common.md (rules for every full-tier stage).' +
(ROLE[key][1] ? `\nDo section "${ROLE[key][1]}" of the brief.` : '') + (extra || '')
// ---------------------------------------------------------------- agent wrappers (same contract as concept.js)
async function call(prompt, opts) {
try { return await agent(prompt, opts) } catch (e) { log(`${opts.label}: agent error ${String((e && e.message) || e).slice(0, 160)}`); return null }
}
// One stage agent: retried once, then once as general-purpose; an empty result marks the stage PARTIAL
// (inside gated() the gate verdict decides instead).
async function run(label, ph, prompt, x) {
const c = x || {}
const o = { label, phase: ph, schema: c.schema || DONE, agentType: c.type || 'flow-stage', effort: c.effort || 'high' }
let r = await call(prompt, o)
if (!r) r = await call(prompt, { ...o, label: `${label} (retry)` })
if (!r) r = await call(prompt, { ...o, label: `${label} (general-purpose)`, agentType: 'general-purpose' })
if (!r) {
log(`${label}: no result after three attempts`)
if (!c.gated) partial.push({ stage: label, errors: ['no result after three attempts'] })
}
return tick(r)
}
const gatePrompt = cmd => [
`Run directory: ${RUN}.`,
'Run exactly this command with the Bash tool (timeout 600000 ms) and nothing else:',
'',
`cd ${RUN} && ${cmd}`,
'',
'Its last output line is one JSON object. Return its fields exactly as printed (lists verbatim, at most 40 items).',
'If the command does not start, times out or prints no JSON line: pass=false, errors=["<one line: what happened>"], all other lists empty, epoch = output of date +%s.',
'Never edit files, never fix anything, never run another command.',
].join('\n')
async function gate(label, ph, cmd) {
const p = gatePrompt(cmd)
let r = await call(p, { label, phase: ph, schema: GATE, agentType: 'flow-gate', model: 'haiku', effort: 'low' })
if (!r) r = await call(p, { label: `${label} (retry)`, phase: ph, schema: GATE, agentType: 'flow-gate', model: 'sonnet', effort: 'low' })
if (!r) r = await call(p, { label: `${label} (general-purpose)`, phase: ph, schema: GATE, agentType: 'general-purpose', model: 'sonnet', effort: 'low' })
if (!r) r = { pass: false, errors: [`gate runner returned nothing for: ${cmd}`], epoch: 0 }
for (const k of ['errors', 'warnings']) if (!Array.isArray(r[k])) r[k] = []
return tick(r)
}
async function gated(label, ph, prompt, cmd, tries, schema) {
let errs = []
let res = null
let g = null
for (let i = 0; i <= tries; i++) {
const p = i === 0 ? prompt + selfCheck(cmd) + RET : prompt + retryText(cmd, errs) + selfCheck(cmd) + RET
res = (await run(i === 0 ? label : `${label} (fix ${i})`, ph, p, { schema: schema || DONE, gated: true })) || res
g = await gate(`gate: ${label}${i ? ' ' + (i + 1) : ''}`, ph, cmd)
if (g.pass) return { ok: true, res, gate: g }
errs = g.errors.length ? g.errors : ['the gate failed without error text']
const more = i < tries && frac() <= 0.75 // no retries above 75 % of the hard stop
log(`${label}: gate FAIL (${errs.length} errors)${more ? ' - retrying with the errors' : ' - marked PARTIAL, continuing'}`)
if (!more) break
}
partial.push({ stage: label, errors: errs.slice(0, 8) })
return { ok: false, res, gate: g }
}
// ---------------------------------------------------------------- P: architecture freeze
phase('Freeze')
const fz = await gated('architecture', 'Freeze', head('architecture.md', 35) +
'\nAlso return board (true when the design has a custom PCB), mcu_env (the PlatformIO environment of the target MCU) and calc_groups (1-4 groups: name = short lowercase word, questions = the calculations that group owns).',
CMD.params, T(2), PLAN)
const plan = fz.res && Array.isArray(fz.res.calc_groups) ? fz.res : { calc_groups: [], board: true, mcu_env: '' }
let groups = plan.calc_groups.filter(g => g && /^[a-z][a-z0-9]{1,15}$/.test(g.name || '')).slice(0, 4)
if (!groups.length) groups = [{ name: 'core', questions: ['every key estimate of 02-concept that detailed design must confirm'] }]
const board = plan.board !== false
log(`Plan: ${groups.length} calculation group(s) ${groups.map(g => g.name).join(', ')}; custom board: ${board}; at about ${elapsed()} min`)
// ---------------------------------------------------------------- Q: parallel discipline pipelines, each build -> gate -> retry
phase('Build')
const calcText = g => `\nYour calculation group: "${g.name}" - name every script ${g.name}_<question>.py. Questions you own:\n- ${(g.questions || []).join('\n- ')}`
async function electronics() {
await gated('schematic', 'Build', brief('sch', 60), CMD.sch, T(2))
await parallel([
async () => (board ? gated('board', 'Build', brief('pcb', 75), CMD.pcb, T(2)) : null),
() => gated('firmware', 'Build', brief('firmware', 60, plan.mcu_env ? `\nTarget environment from the plan: ${plan.mcu_env}.` : ''), CMD.firmware, T(2)),
() => gated('bom', 'Build', brief('bom', 60), CMD.bom, T(2)),
])
}
await parallel([
() => gated('cad', 'Build', brief('cad', 90), CMD.cad, T(2)),
() => pipeline(groups, (_, g) => gated(`calc ${g.name}`, 'Build', brief('calc', 45, calcText(g)),
`python3 method/gates/check_full.py --part calc --prefix ${g.name}_ --json`, T(2))),
() => electronics(),
() => gated('safety', 'Build', brief('safety', 60), CMD.safety, T(2)),
])
log(`Build done at about ${elapsed()} min; partial so far: ${partial.map(p => p.stage).join(', ') || 'none'}`)
// ---------------------------------------------------------------- R: integration, targeted re-runs of the owning stage
phase('Integrate')
const OWN = { cad: CMD.cad, sch: CMD.sch, pcb: CMD.pcb, firmware: CMD.firmware, safety: CMD.safety, bom: CMD.bom, calc: CMD.calc }
const rounds = frac() > 0.5 ? 1 : 2
for (let round = 1; round <= rounds; round++) {
const ig = await run(`integration ${round}`, 'Integrate', head('integration.md', 25) +
`\nRound ${round} of ${rounds}. Write work/full/integration-${round}.json and return {issues: [{owner, issue}], note, epoch}.`, { schema: ISSUES })
const issues = ((ig && ig.issues) || []).filter(i => i && OWN[i.owner] && i.issue).slice(0, 12)
if (!issues.length) { log(`integration ${round}: no issues`); break }
const owners = [...new Set(issues.map(i => i.owner))].filter(o => o !== 'pcb' || board)
log(`integration ${round}: ${issues.length} issue(s) for ${owners.join(', ')}`)
await parallel(owners.map(o => () => {
const text = `\nINTEGRATION ISSUES for your section (fix each in your own files; numbers stay in params.json):\n- ${issues.filter(i => i.owner === o).map(i => i.issue).join('\n- ')}`
const extra = o === 'calc' ? '\nYou may edit any script in deliverables/05-calculations/ that the issues name; keep the script names.' + text : text
return gated(`${o} fix ${round}`, 'Integrate', brief(o, 40, extra), OWN[o], T(1))
}))
}
// ---------------------------------------------------------------- S: drawings, renders, test plan
phase('Finish')
await parallel([
() => gated('drawings', 'Finish', brief('drawings', 60), CMD.drawings, T(2)),
() => gated('renders', 'Finish', brief('renders', 45), CMD.renders, T(2)),
() => gated('test plan', 'Finish', brief('testplan', 45), CMD.testplan, T(2)),
])
// ---------------------------------------------------------------- T: verification, package by code, write-up, final gate
phase('Package')
await gated('verification', 'Package', brief('verification', 30), CMD.pkg, T(1))
await run('writeup', 'Package', head('writers.md', 25) +
'\nDo part "writeup" for the full tier: the homes of the numbers are work/full/verification.json, work/full/key_figures*.json and deliverables/05-calculations/params.json (the concept-tier files describe the concept only).' +
`\nStages that ended PARTIAL: ${partial.length ? partial.map(p => p.stage).join(', ') : 'none'}.` + RET)
await gate('package', 'Package', CMD.pkg)
let fg = await gate('final gate', 'Package', CMD.all)
const REPAIR = { ...OWN, drawings: CMD.drawings, renders: CMD.renders, testplan: CMD.testplan }
if (!fg.pass && frac() < 0.85) {
const byPart = {}
for (const e of fg.errors) {
const part = (e.match(/^([a-z]+):/) || [])[1]
const owner = REPAIR[part] ? part : 'fixer'
;(byPart[owner] = byPart[owner] || []).push(e)
}
log(`final gate FAIL: repair round for ${Object.keys(byPart).join(', ')}`)
await parallel(Object.keys(byPart).map(o => () => {
const errs = `\nFINAL-GATE ERRORS for your section:\n- ${byPart[o].slice(0, 20).join('\n- ')}`
if (o === 'fixer') return run('fixer', 'Package', head('fixer.md', 30) + `\nThe gate \`${CMD.all}\` failed with:\n- ${byPart[o].slice(0, 30).join('\n- ')}` + selfCheck(CMD.all) + RET)
return gated(`${o} repair`, 'Package', brief(o, 40, errs), REPAIR[o], 0)
}))
await gate('package (after repair)', 'Package', CMD.pkg)
fg = await gate('final gate 2', 'Package', CMD.all)
}
if (!fg.pass) partial.push({ stage: 'final gate', errors: fg.errors.slice(0, 8) })
return {
status: fg.pass && !partial.length ? 'ok' : 'partial',
tier: 'full',
board,
calc_groups: groups.map(g => g.name),
final_gate_pass: !!fg.pass,
final_gate_errors: fg.errors.slice(0, 12),
partial,
elapsed_min: elapsed(),
}
skills/dd-hw-handbook/SKILL.md (266 wierszy)
---
name: dd-hw-handbook
description: Product-agnostic quality bars and methods for every deliverable of a dd hardware design run (requirements, concept, calculations, CAD, drawings, electronics, firmware, safety and certification, BOM and cost, prototype and test plan, renders, report). Read it before producing any deliverable. It says WHAT GOOD LOOKS LIKE; your main prompt says who does the work and in which order.
---
# dd hardware design handbook
Your main prompt defines the team, the roles, the communication and the order of work. This handbook defines
the quality bar of each deliverable and the methods that reach it. Where your main prompt is more specific,
follow your main prompt. Where the RUN CONTRACT fixes a file name or a format, the RUN CONTRACT wins.
Exact tool commands, versions and paths are in `INPUT/TOOLCHAIN.md` (read the working examples it points to
before writing your own scripts); this handbook names tools only by purpose.
Contents: 1 inputs · 2 requirements · 3 research · 4 physics map · 5 concept · 6 calculations · 7 CAD ·
8 drawings · 9 electronics · 10 firmware · 11 safety and certification · 12 sourcing, BOM and cost ·
13 prototype and test plan · 14 renders · 15 report and manifest · 16 claims · 17 time and context.
---
## 1. Reading the inputs
- `INPUT/brief.md` — one sentence plus customer context. Extract: who buys it, where and how it is used, how
often and how long, the price expectation, the one or two things that must be better than what exists.
- `INPUT/requirements.md` — the acceptance criteria. Each item has an id, a metric, a target and usually a
threshold (worst acceptable), a verification method and a weight. **Every id must be addressed and keep its
exact spelling** wherever you mention it. Weights tell you what matters most when requirements conflict.
- `INPUT/TOOLCHAIN.md` — what is installed and how to call it headless. Use only what it lists. A tool that
misbehaves gets three honest attempts with new information each time, then a labelled fallback (for example
an analytical model instead of a simulation) and a note in `manifest.json → blocked[]` if it mattered.
## 2. Requirements — `deliverables/01-requirements/`
Quality bar: an engineer who has never seen the brief can design and verify the product from this folder
alone, and every number in it has an origin.
Write `requirements.md` (and, if you like, a machine-readable `requirements.yaml`) with three parts:
1. **Given requirements**, one entry per id from `INPUT/requirements.md`: the id, the original statement, your
interpretation (the exact condition under which it will be verified: temperature, voltage, distance,
duty cycle, test method), and the design consequence (what it forces in the design).
2. **Derived requirements**, ids `DR-<AREA>-<nn>`, each with: parent id(s) or "brief"/"regulation"/
"manufacturing", statement, metric, target, threshold, verification method (Analysis, Inspection,
Demonstration, Test), rationale with origin label, priority. Walk at least these areas and write down
either the requirement or why the area does not apply: function and performance under all operating
conditions; user interface and ergonomics; power source and consumption (including standby); mechanical
interfaces and mounting; environment (temperature, humidity, dust and water, altitude, transport and
storage); safety (electric shock, heat and burns, fire, mechanical hazards, chemical, optical and other
radiation, noise, misuse that can be foreseen); EMC and, if any radio, radio; materials and restricted
substances; lifetime, duty cycle and reliability; maintenance, cleaning, consumables and repair;
manufacturing process at 1, 100 and 1000 units; assembly and test in production; packaging, shipping and
drop; labelling, manuals and markings; unit cost and retail price; prototype schedule.
3. **Conflicts and decisions**: requirement pairs that pull against each other, how you resolved each (use the
weights; safety first), and the assumptions register (each ASSUMPTION with reason and sensitivity).
Write `compliance.csv` exactly as the RUN CONTRACT specifies, one row per id of `INPUT/requirements.md`:
- `status`: `met`, `partly`, `not_met` or `not_addressed`. At the concept tier the status is a prediction
from your estimates: say so in `note` (for example "predicted, CALCULATION"). Be honest: a predicted miss
is `not_met` with the reason, not a hidden problem.
- `value` and `unit`: the predicted or designed value in the requirement's own metric and unit.
- `evidence`: path(s) inside `deliverables/` that show it (a calculation result, a section of the concept).
## 3. Research — benchmark, users, prior art
- Find at least five comparable products (commercial and open-source; adjacent categories count). For each:
the specifications that matter for our requirements, price with date and URL, and weak points from
reviews, teardowns, forums or return reasons. Put the table in `02-concept/benchmark.md`.
- From the benchmark derive: realistic target values, the differentiator ("our own, better version" — what we
do better and why it is feasible), and the failure modes users actually complain about. Feed both back into
the derived requirements.
- Note patents only as prior art you noticed; never give legal opinions. If an idea of yours looks genuinely
new, flag it as `novelty` in `REPORT.md` (one line) — do not elaborate on it in public-facing text.
- Read pages, do not guess them. If a site blocks automated reading, use a different source.
## 4. Physics map — which disciplines decide this product
Rate each domain CRITICAL, RELEVANT or MINOR for this product: structure and strength; heat; fluids and
aerosols; electrical power; electronics and control; firmware; mechanisms and motion; materials and
chemistry; optics and light; acoustics; human factors; environment. For each CRITICAL domain write: the
governing relations, the few parameters that dominate the result, the order-of-magnitude numbers, and the
ways it typically fails (from first principles and from what users report). The map decides which analyses
you run and how deep; it lives in `02-concept/physics-map.md`.
## 5. Concept — `deliverables/02-concept/`
Quality bar: the decision is visibly earned. A reader sees real alternatives, numbers for each, a fair
comparison, and a chosen architecture concrete enough to start detailed design tomorrow.
Contents (file names are suggestions; keep them inside `02-concept/`):
- `concept.md` — the main document, in this order:
1. The product in one paragraph: what it is, for whom, what it does better.
2. Functions and flows: the main functions as verb–noun pairs and the energy, material and signal flows.
3. Alternatives: at least three that differ in **working principle or architecture**, not in cosmetics.
For each: how it works, a simple diagram, first-order numbers against the top-weighted requirements
(performance, power, size, mass, unit cost), the main risks and hazards.
4. Evaluation: knock-out by thresholds first; then a weighted comparison whose weights come from the
requirement weights; a sensitivity check (vary each weight ±20 %, and the most uncertain estimates
within their range) showing whether the winner holds. Record why each loser lost.
5. The chosen concept: architecture block diagram (Mermaid in the markdown and a rendered PNG or SVG if
the toolchain has a renderer), the subsystem list with make-or-buy, the critical components with real
candidate parts (manufacturer, part number, supplier link, price with date), the interfaces between
subsystems (mechanical, electrical, thermal, data).
6. Key estimates: every number the decision rests on, each with its origin; the scripts that compute them.
7. Safety concept: the top hazards, the design measures (inherently safe design first, then protective
devices, then information for the user), and which protections act **independently of firmware**.
8. Compliance route: which EU directives and harmonised standards, and which FCC rules, apply and why
(with SOURCE), the likely conformity route, rough cost and time.
9. Cost: a rough BOM of the major items, the unit cost at 1, 100 and 1000 units with the method, and the
retail price logic (section 12). Optionally `09-bom-cost/cost.json` already at this tier.
10. Risks: the top risks with likelihood, impact, mitigation and the analysis or test that would retire
each one.
11. Decisions taken without a human (taste, priorities, trade-offs) with the rejected alternatives.
12. Open issues and what detailed design must do first.
- `calc/` — the scripts behind the key estimates (same rules as section 6), with their results files.
- `benchmark.md`, `physics-map.md`, and anything your method produces (scores, votes, issue logs, models).
Concept anti-patterns: two strawmen next to the favourite; alternatives that differ only in colour or
material; scores without numbers behind them; a decision matrix whose weights were tuned until the favourite
won; a concept that silently misses a top-weighted requirement; prose without a single calculated number.
## 6. Calculations — `05-calculations/` (full tier) and `02-concept/calc/` (concept tier)
- One Python script per question, run with the toolchain's Python (`ddpy`, see `INPUT/TOOLCHAIN.md`; the
system `python3` has no scientific libraries). Physical quantities carry units (the `pint` library) and every
input is labelled at the top of the script with its origin.
- Scripts run with `ddpy <script>` from their own folder, offline, in under five minutes, and write
`<stem>.results.json` (`{"<name>": {"value": x, "unit": "..."}}`). The RUN CONTRACT requires this for
`05-calculations/`; use the same convention in `02-concept/calc/`.
- Order of work: hand estimate → script → (only if it changes a decision) simulation. For any simulation:
first a known-answer case, then the real case on at least three mesh or step sizes with the trend shown,
sanity checks (balances of mass, energy or force), and a sensitivity run on the least certain inputs.
- Static strength is not drop, impact or fatigue; steady thermal is not transient; say which remain open.
- Margins: state each result as value, requirement, margin, and status.
## 7. CAD — `03-cad/` (full tier)
- Parametric code-CAD (CadQuery or build123d, whichever `INPUT/TOOLCHAIN.md` lists as working), one script
per designed part, parameters at the top, an assembly script placing parts with names. Re-running the
scripts reproduces every STEP. FreeCAD files are acceptable if that is your route.
- Deliver `assembly.step` and one STEP per designed part, plus the source. Purchased parts appear as
simplified bodies with their real envelope dimensions from the datasheet (label them).
- Check every part: valid solid, volume and mass from density, bounding box against size requirements,
pairwise interference zero (or an intended fit), clearances for assembly tools and fingers, wall thickness
and draft suited to the chosen process, standard fasteners.
- Choose a manufacturing process per part for the prototype (3D printing, CNC, laser-cut sheet) and for
1000 units (injection moulding, die cutting, sheet-metal bending), and note what changes between them.
- Export a GLB of the assembly for the renders and the website (see `INPUT/TOOLCHAIN.md` for the converter).
## 8. Drawings — `04-drawings/` (full tier)
- A drawing for every manufactured part: views, overall and functional dimensions, tolerances (a general
tolerance class plus explicit tolerances on functional features), material, finish, quantity, part number
and revision matching the BOM. Flat patterns for sheet metal.
- An assembly drawing with item balloons that match `bom.csv` item numbers. PDF or SVG; DXF optional.
- Open each output after you write it (render a PNG preview and look at it, or parse it) — a drawing file
that does not open does not exist.
## 9. Electronics — `06-electronics/` (full tier)
- Start from a block diagram and a power budget over the full supply range and all operating modes.
- KiCad 9 project with schematic and, where custom electronics is needed, a routed board. Real parts with
matching symbols and footprints; every part has a manufacturer part number that exists at a distributor or in
the offline LCSC/JLC catalogue described in `INPUT/TOOLCHAIN.md`.
- Protection where it applies: input fuse or current limit, reverse polarity, transients and ESD on external
connectors, inrush, thermal shutdown of hot parts in hardware, safe state on power loss.
- Calculations: currents, trace widths, dissipation and temperature rise of the hot parts, derating, and for
any mains part the creepage and clearance with the standard you took the values from.
- Run ERC and DRC from the command line; deliver the reports; fix every error or justify each waiver.
Fabrication outputs: Gerbers, drill, pick-and-place and the board BOM. Passing DRC is not a hardware test.
- Simulate uncertain circuit behaviour with ngspice where it changes a decision (start-up, protection trip,
filter response) after one known-answer test of the simulator.
## 10. Firmware — `07-firmware/` (full tier)
- PlatformIO project at the folder root that builds offline for the real target MCU of the schematic; use
exactly the platform versions `INPUT/TOOLCHAIN.md` pins (anything else downloads packages and blocks
other runs).
- A state machine with explicit fault states and a safe default; a watchdog; inputs debounced; every
safety-relevant reaction has a hardware backstop and the firmware is never the only protection.
- Host tests (`native` environment) for the logic and the interlocks: write a test that fails first, then
passes. Build with warnings as errors where the framework allows. Never flash or power anything.
- Pin map agreed with the schematic; configuration constants generated from or checked against the design
parameters.
## 11. Safety and certification — `08-safety-certification/` (both tiers: summary in 02; full tier: complete)
- Hazard analysis in the ISO 12100 manner: hazard → hazardous situation → harm → risk estimate (severity ×
probability) → reduction in the order inherently safe design, protective measures, information for use →
residual risk. Derive safety requirements with ids and trace them to the design.
- Regulatory map with SOURCE for each item. For the EU think through at least: the Low Voltage Directive
(2014/35/EU) when any voltage is in its range, the EMC Directive (2014/30/EU), the Radio Equipment Directive
(2014/53/EU) when there is any radio, RoHS (2011/65/EU), WEEE (2012/19/EU), the batteries rules when there
is a battery, the General Product Safety Regulation (EU) 2023/988, REACH, ecodesign and standby rules.
For the US: FCC Part 15 (Subpart B for unintentional radiators, Subpart C for intentional ones) and the
voluntary safety listings retailers expect. Then find the harmonised product-family standards that
apply to THIS product by research, and quote which clauses drive the design.
- Prefer architectures that make conformity easy: low voltage inside the product through a certified external
supply when the power level allows it, pre-certified radio modules, protections that do not depend on
firmware, materials with suitable flammability ratings near heat and live parts.
- State the conformity route (self-declaration, notified body or not), the tests an accredited lab would run,
a pre-compliance plan, and a rough cost and duration. You identify; you do not certify.
## 12. Sourcing, BOM and cost — `09-bom-cost/`
- Real, purchasable parts only. For each line: manufacturer, part number, supplier, supplier part number, URL,
unit price at 1, 100 and 1000 (price breaks as shown by the supplier, currency stated), the date you read it
in `notes`. A price you could not read is an ASSUMPTION with a range, never a guess dressed up as SOURCE.
Prefer parts in stock at two or more suppliers, active (not end-of-life), and note a second source for
critical parts. Use the offline catalogue from `INPUT/TOOLCHAIN.md` first, then distributor pages
read-only.
- `bom.csv` with exactly the header the RUN CONTRACT gives; `cost.json` and `cost.md` as it specifies.
- The unit cost at each quantity is more than the BOM: PCB fabrication and assembly, mechanical parts (at 1
and 100 units printed or machined; at 1000 units moulded with the tooling cost amortised and stated
separately), assembly and test labour, packaging, certification and other one-off costs amortised over
the quantity (state the assumption), freight to the fulfilment point, and a contingency. Show each line.
- Retail price logic: start from benchmark prices and a target gross margin; check that the price leaves room
for crowdfunding platform and payment fees, shipping to backers, VAT where it applies, returns and warranty.
An underpriced product that cannot be delivered is the classic crowdfunding failure; say plainly if the
numbers do not close.
## 13. Prototype and test plan — `10-prototype-test-plan/` (full tier)
- Build plan for prototype 1: how each part is made, the order of assembly, tools, and the checks after each
step.
- Bring-up for a person later: first power-up with a current-limited or isolated supply, what to measure
before applying full power, safety precautions.
- One test per requirement (and per safety requirement): method, equipment, conditions, pass criterion, the
requirement id it closes. Include pre-compliance EMC and the safety tests a lab would run, and durability
or life tests where lifetime is a requirement.
## 14. Renders — `11-renders/` (full tier)
- Renders come from the real CAD geometry (the assembly GLB) through Blender in background mode: a product
view, an exploded view, and a view that shows the user interface. Check the scale against a known
dimension. Label them as renders of a digital design.
- Never use generative image models to depict the product: they invent geometry that does not exist.
- You cannot judge images reliably by looking. Check what can be measured (file exists, resolution, the
object fills the frame, parts are present by name) instead of trusting a glance.
## 15. Report and manifest
`REPORT.md` (both tiers) in this order: the product in one paragraph; a key-parameter table (value, unit,
origin label); unit cost at 1/100/1000 and the retail price; status per requirement (from `compliance.csv`);
the top risks and open hazards; decisions taken without a human with their alternatives; what is open,
blocked or not run; for the full tier, the file map. Every number in it must match its home.
`manifest.json` exactly as the RUN CONTRACT specifies, including `key_figures` (use `null` for an unknown
figure, never a guess) and `blocked[]`. List every file under `deliverables/`.
Before you finish: re-read `REPORT.md` and `02-concept/concept.md` against the numbers' homes, check that
`compliance.csv` has one row per requirement id, that every evidence path exists, and that `manifest.json`
parses and lists the files that exist.
## 16. Claims
The project designs real products for honest crowdfunding. In every document: no feature that the design
does not implement, no performance number that is not calculated or sourced, no "certified", "safe",
"waterproof", "AI" or similar word unless the evidence for it is in the deliverables. Under-promise.
## 17. Time and context
- The RUN CONTRACT gives the time target and the hard stop. Plan time boxes per phase at the start, check
`date` at each boundary, and cut scope (depth, not honesty) when behind.
- Long runs lose context. Keep a short state file (decisions, key numbers' homes, open issues, next step)
current, so you or a fresh agent can resume from files alone.
- Write deliverables as soon as each part exists and keep them current.