1"""The DAQ box in the hookup, and the stand's own state table.
3A hookup with ``channels`` is *wired*: each connector is one cable from one
4board to one symbol, under the name the DAQ gives it, and the state table opens
5names. So a row drives the valve on the connector of its name and nothing else
6-- never a valve the twin's name matching would have picked behind the
7person's back. A hookup without channels is the old kind and must read, write
8and bind exactly as it did.
10The state table can be edited (``StateMachine.to_dict`` / ``machine_from_dict``)
11and has to come back as the CSV machine it was: the same states, rows, moves,
12panel and aborts, Idle still held shut.
14Each check is against what the old behaviour would have done instead, so it
15fails if the new behaviour were not there.
18from __future__
import annotations
22from dataclasses
import replace
23from pathlib
import Path
40HERE = Path(__file__).resolve().parent
42STAND = STAR /
"feed-twin" /
"backend" /
"diagrams" /
"ethalox_stand.json"
43TABLES = STAR /
"feed-twin" /
"backend" /
"statemachines"
44LE4 = HERE /
"fixtures" /
"le4_rocket_and_gse.json"
46needs_stand = pytest.mark.skipif(
47 not (STAND.exists()
and TABLES.is_dir()), reason=
"stand drawing or tables absent"
49needs_tables = pytest.mark.skipif(
not TABLES.is_dir(), reason=
"no state tables")
51ABORTS = frozenset({
"Engine Abort",
"GSE Abort",
"Emergency Abort"})
55 return assemble_model(
56 read_diagram(json.loads(STAND.read_text()), name=
"s"), diagram_id=
"s"
61 return load_machine(tables=TABLES)
65 """What :func:`binding` matches names against on a one-page drawing."""
69 for n
in model.diagram.nodes
70 if n.id
in built.actuators
71 and not built.actuators[n.id].endswith(
".dome")
72 and n.id
not in built.hand_valves
81 """One valve connector, one row cabled to a transducer, one plain PT. Matched
82 by name the stand binds eight rows; wired it binds the one."""
84 by_name = binding(model, machine,
None).to_symbol
85 assert by_name[
"LOX Main"] ==
"MVO" and by_name[
"Fuel Main"] ==
"MVF"
86 assert len(by_name) > 2,
"the name match binds more than the box does"
90 Channel(
"sol12", 1,
"LOX Main",
"MVO"),
92 Channel(
"pt_low", 1,
"Fuel Main",
"PT_FUU"),
93 Channel(
"pt_high", 1,
"Bottle",
"PT_HI"),
97 b = binding(model, machine, wired)
98 assert dict(b.to_symbol) == {
"LOX Main":
"MVO"}
99 assert b.unmatched == tuple(a
for a
in machine.actuators
if a !=
"LOX Main")
100 assert "Fuel Main" in b.unmatched,
"a PT on the row's connector drives nothing"
101 assert b.by_role == ()
103 none = binding(model, machine,
Hookup(channels=()))
104 assert dict(none.to_symbol) == {}
105 assert none.unmatched == machine.actuators
106 assert sorted(none.uncommanded) == sorted(
_valve_labels(model))
111 """Fire opens both mains and both press valves by name; wired with only
112 "LOX Main" on a connector it opens the LOX main and nothing else."""
114 hookup =
Hookup(channels=(
Channel(
"sol12", 1,
"LOX Main",
"MVO"),))
116 def opened_in_fire(h: Hookup |
None) -> set[str]:
118 session =
Session(model, machine, binding(model, machine, h), hookup=h)
119 assert machine.can_go(
"Ready",
"Fire")
120 session.state =
"Ready"
121 session.command_state(
"Fire")
122 signals = session.signals()
126 for sid, signal
in built.actuators.items()
127 if not signal.endswith(
".dome")
and signals[signal] > 0.5
130 session_binding = binding(
_stand(), machine, hookup)
131 assert session_binding.positions_for(machine,
"Fire") == {
"MVO": 1.0}
132 by_name = opened_in_fire(
None)
133 assert {
"MVO",
"MVF"} <= by_name,
"the name match opens both mains"
134 assert opened_in_fire(hookup) == {
"MVO"}
145 "valves": {
"Fuel Vent":
"SV_GN2_VENT",
"LOX Fill":
""},
150 "regulators": [
"PR1"],
156 "aliases": {
"PT_HI":
"Bottle"},
158 hookup = Hookup.from_dict(raw)
159 assert not hookup.wired
and hookup.channels
is None and hookup.machine
is None
160 assert hookup.to_dict() == raw,
"exactly the schema-1 dict, nothing added"
161 assert list(hookup.to_dict()) == [
"schema",
"valves",
"knobs",
"aliases"]
162 assert Hookup(knobs=(
Knob(
"dome",
"Dome"),)).to_dict()[
"schema"] == 1
164 b = binding(model, machine, hookup)
168 roles=model.built.valve_roles,
169 overrides=raw[
"valves"],
171 assert b == before,
"the pins and the name match, as the old hookup bound"
172 assert b.to_symbol[
"Fuel Vent"] ==
"SV_GN2_VENT" and "LOX Fill" in b.unmatched
173 assert b.to_symbol[
"LOX Main"] ==
"MVO",
"everything else matched by name"
175 with pytest.raises(ValueError, match=
"schema"):
176 Hookup.from_dict({**raw,
"schema": 3})
183 table[
"states"].append({
"name":
"Purge",
"row": 6,
"col": 0,
"abort":
False})
184 table[
"open"][
"Purge"] = [
"Fuel Vent"]
185 table[
"allowed"][
"Idle"].append(
"Purge")
186 table[
"allowed"][
"Purge"] = [
"Idle"]
187 machine = machine_from_dict(table)
189 knobs=(
Knob(
"dome",
"Dome", (), 420.0),),
190 aliases={
"engine.pc":
"Chamber pressure"},
192 Channel(
"sol12", 1,
"LOX Main",
"MVO"),
193 Channel(
"sol12", 2,
"Fuel Vent",
"SV_FUEL_VENT"),
194 Channel(
"pt_low", 3,
"Fuel tank",
"PT_FUU"),
196 rows={
"sol12": 3,
"pt_low": 2},
199 stored = hookup.to_dict()
200 assert stored[
"schema"] == 2
201 back = Hookup.from_dict(json.loads(json.dumps(stored)))
202 assert back == hookup
203 assert back.to_dict() == stored
204 assert back.machine
is not None and "Purge" in back.machine.states
205 assert binding(model, back.machine, back) == binding(model, machine, hookup)
206 assert dict(binding(model, machine, back).to_symbol) == {
208 "Fuel Vent":
"SV_FUEL_VENT",
212@pytest.mark.parametrize(
216 [(
"sol12", 1,
"LOX Main",
"MVO"), (
"sol12", 1,
"Fuel Main",
"MVF")],
217 "two cables in one connector",
220 [(
"sol12", 1,
"LOX Main",
"MVO"), (
"sol24", 1,
"Spare",
"MVO")],
221 "one symbol on two connectors",
224 [(
"sol12", 1,
"LOX Main",
"MVO"), (
"sol12", 2,
"lox main",
"MVF")],
225 "two connectors are named",
227 ([(
"sol12", 1,
" ",
"MVO")],
"no name"),
228 ([(
"sol12", 0,
"LOX Main",
"MVO")],
"numbered from 1"),
232 channels: list[tuple[str, int, str, str]], said: str
236 {
"board": b,
"slot": s,
"name": n,
"symbol": sym}
237 for b, s, n, sym
in channels
240 with pytest.raises(ValueError, match=said):
241 Hookup.from_dict(raw)
245 """The refusals above are of the fault, not of the box."""
246 ok = Hookup.from_dict(
249 {
"board":
"sol12",
"slot": 1,
"name":
"LOX Main",
"symbol":
"MVO"},
250 {
"board":
"sol12",
"slot": 2,
"name":
"Fuel Main",
"symbol":
"MVF"},
251 {
"board":
"sol24",
"slot": 1,
"name":
"Spare",
"symbol":
"SV"},
255 assert ok.wired
and len(ok.channels
or ()) == 3
264 assert csv.table
is not None
265 edited = machine_from_dict(csv.to_dict())
266 assert edited.states == csv.states
267 assert edited.actuators == csv.actuators
268 assert edited.positions == csv.positions
269 assert edited.table == csv.table
270 assert edited.allowed == csv.allowed
271 assert edited.layout == csv.layout
272 assert edited.aborts == csv.aborts
273 for here
in csv.states:
274 assert edited.targets(here) == csv.targets(here), here
277 written = csv.to_dict()[
"open"][
"Idle"]
278 assert set(written) == {
"Fuel Main",
"LOX Press"}
279 assert csv.open_actuators(
"Idle") == frozenset()
280 assert edited.open_actuators(
"Idle") == frozenset()
281 assert edited.to_dict()[
"open"][
"Idle"] == written,
"the cells survive the hold"
282 assert any(
"Idle commands" in w
for w
in edited.warnings)
286@pytest.mark.parametrize(
289 (
lambda t: t[
"states"].append(dict(t[
"states"][0])),
"two states"),
290 (
lambda t: t[
"open"].__setitem__(
"Nowhere", []),
"not a state"),
291 (
lambda t: t[
"allowed"].__setitem__(
"Nowhere", [
"Idle"]),
"not a state"),
292 (
lambda t: t[
"allowed"][
"Idle"].append(
"Nowhere"),
"not states"),
293 (
lambda t: t[
"open"][
"Fire"].append(
"Spare Valve"),
"no row"),
298 machine_from_dict(table)
300 with pytest.raises(ValueError, match=said):
301 machine_from_dict(table)
307 table[
"actuators"].append(
"Purge Valve")
309 {
"name":
"Purge",
"row":
None,
"col":
None,
"abort":
True},
310 {
"name":
"Fake Abort",
"row":
None,
"col":
None,
"abort":
False},
312 table[
"open"][
"Purge"] = [
"Purge Valve"]
313 edited = machine_from_dict(table)
314 assert edited.aborts == ABORTS | {
"Purge"}
315 for here
in (
"Idle",
"Ready",
"Fire",
"Engine Abort"):
316 assert edited.can_go(here,
"Purge"), here
317 assert not edited.can_go(here,
"Fake Abort"), here
318 assert "Fake Abort" not in edited.targets(
"Idle")
320 table[
"allowed"][
"Idle"].append(
"Fake Abort")
321 allowed = machine_from_dict(table)
322 assert allowed.can_go(
"Idle",
"Fake Abort")
323 assert not allowed.can_go(
"Ready",
"Fake Abort")
328 """No ``<name>_states.csv``: the twin's old rule, by name."""
329 for table
in (
"actuators",
"transitions"):
330 shutil.copy(TABLES / f
"diablo_{table}.csv", tmp_path / f
"diablo_{table}.csv")
331 bare = load_machine(tables=tmp_path)
332 assert bare.aborts
is None and bare.layout == {}
333 assert {s
for s
in bare.states
if bare.is_abort(s)} == ABORTS
335 assert "Engine Abort" not in bare.allowed[
"Idle"]
336 assert bare.can_go(
"Idle",
"Engine Abort")
340 states=(*bare.states,
"Fake Abort"),
341 positions={**bare.positions,
"Fake Abort": {}},
343 assert named.can_go(
"Idle",
"Fake Abort"),
"by name, with no flags"
348 machine = load_machine(
"diablo", tables=TABLES)
349 assert machine.layout[
"Idle"] == (0, 0)
350 assert machine.layout[
"Press Standby"] == (2, 0)
351 assert machine.layout[
"Fire"] == (5, 0)
352 assert machine.aborts == ABORTS
353 assert {s
for s
in machine.states
if "abort" in s.lower()} == ABORTS
354 assert not ABORTS & set(machine.layout),
"the aborts are the console's to place"
355 assert set(machine.layout) == set(machine.states) - ABORTS
362 return assemble_model(
363 read_diagram(json.loads(LE4.read_text()), name=
"LE4 (6)"),
370@pytest.mark.skipif(not LE4.exists(), reason=
"LE4 (6) fixture absent")
372 """The box cables Fuel Vent to the cart's FV-SOL and GSE High Press Control
373 to HPC_SOL. The rocket alone keeps the cart's vent line (it is plugged in
374 until launch), so Fuel Vent keeps its connector; HPC_SOL is cut, and its
375 row goes back to being matched. LOX Main keeps its connector; LOX Vent,
376 which the name match would bind to the LOX tank's unpaired disconnect,
377 has no connector and stays unbound."""
378 ids = {n[
"data"][
"label"]: n[
"id"]
for n
in json.loads(LE4.read_text())[
"nodes"]}
379 whole, cut =
_le4(
False),
_le4(
True)
383 Channel(
"sol12", 1,
"LOX Main", ids[
"OM-R"]),
384 Channel(
"sol12", 2,
"Fuel Vent", ids[
"FV-SOL"]),
385 Channel(
"sol24", 3,
"GSE High Press Control", ids[
"HPC_SOL"]),
386 Channel(
"pt_low", 1,
"Fuel tank", ids[
"FU-PT-R"]),
390 assert dict(binding(whole, machine, saved).to_symbol) == {
391 "LOX Main": ids[
"OM-R"],
392 "Fuel Vent": ids[
"FV-SOL"],
393 "GSE High Press Control": ids[
"HPC_SOL"],
396 rocket = on_vehicle(saved, cut)
397 assert rocket.channels == (
398 Channel(
"sol12", 1,
"LOX Main", ids[
"OM-R"]),
399 Channel(
"sol12", 2,
"Fuel Vent", ids[
"FV-SOL"]),
400 Channel(
"pt_low", 1,
"Fuel tank", ids[
"FU-PT-R"]),
402 assert rocket.auto == {
"GSE High Press Control"}
403 assert rocket.rows == saved.rows
and rocket.wired
405 b = binding(cut, machine, rocket)
406 assert dict(b.to_symbol) == {
407 "LOX Main": ids[
"OM-R"],
408 "Fuel Vent": ids[
"FV-SOL"],
410 assert "LOX Vent" in b.unmatched
and "GSE High Press Control" in b.unmatched
411 by_name = binding(cut, machine, on_vehicle(suggest(whole), cut)).to_symbol
412 assert by_name[
"LOX Vent"] == ids[
"QD-OV-B"],
"the name match would bind it"
413 assert by_name[
"Fuel Vent"] == ids[
"FV-SOL"],
"the cart's vent valve, still there"
418 """ "Lox Main" on the box is the table's "LOX Main": typed the way a
419 person types it, the main valve must still open in Fire. Read exactly,
420 it bound nothing and the main was never commanded."""
422 wired =
Hookup(channels=(
Channel(
"sol12", 1,
"Lox Main",
"MVO"),))
423 assert dict(binding(model, machine, wired).to_symbol) == {
"LOX Main":
"MVO"}
427@pytest.mark.parametrize("written", ["0", "false", "no", "", 0, None, False])
429 """An abort is reachable from anywhere, so a flag read loosely is an
430 unguarded path: "0" read as a yes made Fire an abort, and Idle -> Fire
433 for state
in raw[
"states"]:
434 if state[
"name"] ==
"Fire":
435 state[
"abort"] = written
436 table = machine_from_dict(raw)
437 assert not table.is_abort(
"Fire")
438 assert not table.can_go(
"Idle",
"Fire")
444 raw[
"actuators"] = [*raw[
"actuators"],
"lox main"]
445 with pytest.raises(ValueError):
446 machine_from_dict(raw)
450@pytest.mark.skipif(not LE4.exists(), reason=
"LE4 (6) fixture absent")
452 """The connector on the cart's FV-SOL typed "fuel vent": the whole stand
453 joins it to Fuel Vent ignoring case, and so must the rocket alone, whose
454 vent line is still plugged in, or the fuel tank is never vented."""
455 ids = {n[
"data"][
"label"]: n[
"id"]
for n
in json.loads(LE4.read_text())[
"nodes"]}
456 whole, cut =
_le4(
False),
_le4(
True)
458 saved =
Hookup(channels=(
Channel(
"sol24", 2,
"fuel vent", ids[
"FV-SOL"]),))
459 assert binding(whole, machine, saved).to_symbol[
"Fuel Vent"] == ids[
"FV-SOL"]
460 rocket = binding(cut, machine, on_vehicle(saved, cut)).to_symbol
461 assert rocket.get(
"Fuel Vent") == ids[
"FV-SOL"]
465@pytest.mark.skipif(not LE4.exists(), reason=
"LE4 (6) fixture absent")
467 """A cart drawn with no fuel vent solenoid: the whole drawing binds no
468 Fuel Vent, so the box has no connector for it, and on the whole drawing
469 the tank-top disconnect is mated, not a valve the box could take. The
470 rocket alone reads that disconnect as the GSE vent and the old matching
471 vented through it; a saved box must too, or neither Vent nor the burnout
472 vent can empty the fuel tank."""
473 raw = json.loads(LE4.read_text())
475 n[
"id"]
for n
in raw[
"nodes"]
if n[
"data"][
"label"]
in (
"FV-SOL",
"FF-SOL-Vent")
477 raw[
"nodes"] = [n
for n
in raw[
"nodes"]
if n[
"id"]
not in gone]
479 e
for e
in raw[
"edges"]
if e[
"source"]
not in gone
and e[
"target"]
not in gone
481 ids = {n[
"data"][
"label"]: n[
"id"]
for n
in raw[
"nodes"]}
483 def build(cut: bool) -> Any:
484 return assemble_model(
485 read_diagram(raw, name=
"LE4 no vent"), diagram_id=
"le4nv", vehicle_only=cut
488 whole, cut = build(
False), build(
True)
490 legacy = binding(cut, machine, on_vehicle(suggest(whole), cut)).to_symbol
491 assert legacy[
"Fuel Vent"] == ids[
"FV-QD-B"],
"the old matching vents through it"
492 assert "Fuel Vent" not in binding(whole, machine, suggest(whole)).to_symbol
494 channels=(
Channel(
"sol12", 1,
"LOX Main", ids[
"OM-R"]),),
496 rocket = binding(cut, machine, on_vehicle(box, cut)).to_symbol
497 assert rocket.get(
"Fuel Vent") == ids[
"FV-QD-B"]
499 assert set(rocket.values()) <= {ids[
"OM-R"], ids[
"FV-QD-B"]}
506 """The drawing redrawn and the main valve given a new id: the box's LOX
507 Main still names the old one. Pinned to nothing, the main would never
508 open in Fire; matched by name, as a pin to a lost valve always was, it
509 does -- and the connector is listed so the stand can say so."""
513 box =
Hookup(channels=(
Channel(
"sol12", 1,
"LOX Main",
"MVO-old"),))
514 assert binding(model, machine, box).to_symbol.get(
"LOX Main") ==
"MVO"
515 assert [c.symbol
for c
in lost_connectors(box, model)] == [
"MVO-old"]
518 on_pt =
Hookup(channels=(
Channel(
"pt_low", 1,
"LOX Main",
"PT_FUU"),))
519 assert "LOX Main" not in binding(model, machine, on_pt).to_symbol
520 assert lost_connectors(on_pt, model) == []
527@pytest.mark.skipif(not LE4.exists(), reason=
"LE4 (6) fixture absent")
528@pytest.mark.parametrize("cut", [False, True], ids=["whole", "rocket-only"])
530 """LE4 (6) draws the cart charging the COPV, so the built-in charge stands
531 aside and an unwired GSE High Press Control charges nothing: it is not
532 built-in there. The rocket alone has no cart, and the built-in charge is
533 what fills the COPV again. The transfer tank's press is the whole
534 drawing's alone. Held to the stand itself: in GN2 High Press the COPV
535 fills by the built-in, and in Fuel Fill the transfer tank is pressed,
536 exactly when the row is listed."""
540 ids = {n[
"data"][
"label"]: n[
"id"]
for n
in json.loads(LE4.read_text())[
"nodes"]}
542 listed = builtin_rows(machine, model)
543 assert (
"GSE High Press Control" in listed)
is cut
544 assert (
"GSE High Press Vent" in listed)
is cut
545 assert (
"Fuel Fill Press" in listed)
is not cut
547 hookup = suggest(
_le4(
False))
548 session = open_session(
549 model, machine, hookup=on_vehicle(hookup, model)
if cut
else hookup
551 assert session.builtin == listed
552 session.state =
"GN2 High Press"
554 assert session.bottles[ids[
"COPV"]].filling
is cut
555 session.state =
"Fuel Fill"
557 sim.id
for sim
in session.tanks.values()
if session._supply_press(sim)[0] > 0.0
559 assert bool(pressed)
is (
"Fuel Fill Press" in listed)
562def _renamed(table: dict[str, Any], old: str, new: str) -> dict[str, Any]:
563 """``table`` with state ``old`` called ``new`` everywhere it appears."""
564 out: dict[str, Any] = json.loads(json.dumps(table))
565 for row
in out[
"states"]:
566 if row[
"name"] == old:
568 out[
"open"] = {new
if k == old
else k: v
for k, v
in out[
"open"].items()}
570 new
if k == old
else k: [new
if t == old
else t
for t
in v]
571 for k, v
in out[
"allowed"].items()
578 """A state the twin keys on, gone, is warned with its own reason and no
579 other's; a table in which no state loads one side's tank is warned for
580 that side. The DAQ's table, read back, is warned for neither."""
583 def said(table: dict[str, Any], words: str) -> list[str]:
584 return [w
for w
in machine_from_dict(table).warnings
if words
in w]
586 assert said(shipped,
"keys on") == []
and said(shipped,
"No state loads") == []
588 (vent,) = said(
_renamed(shipped,
"Vent",
"Blowdown"),
"keys on")
589 assert "no Vent" in vent
and "a dry tank vents to Vent" in vent
590 assert "Idle" not in vent
and "Ready" not in vent
and "ENG ABORT" not in vent
592 (lox,) = said(
_renamed(shipped,
"Ox Fill",
"Oxidiser Load"),
"No state loads")
593 assert "LOX tank" in lox
and "Ox Fill" in lox
594 assert "fuel" not in lox
and "Fuel Fill" not in lox
596 assert said(
_renamed(shipped,
"Ox Fill",
"LOX Fill"),
"No state loads") == []
None test_a_box_that_cannot_be_is_refused(list[tuple[str, int, str, str]] channels, str said)
None test_a_session_on_a_wired_hookup_opens_only_the_wired_valve()
None test_two_rows_one_name_apart_from_case_are_refused()
None test_a_connector_joins_its_row_as_a_person_reads_the_name()
None test_a_hookup_without_a_box_is_written_and_bound_as_before()
None test_the_same_box_without_the_fault_is_accepted()
None test_without_flags_an_abort_is_whatever_says_abort(Path tmp_path)
None test_a_wired_hookup_with_its_own_table_round_trips()
None test_a_wired_hookup_binds_exactly_its_connectors()
( None) test_a_connector_whose_symbol_left_the_drawing_is_matched_by_name_and_said()
dict[str, str] _valve_labels(Any model)
None test_a_table_that_contradicts_itself_is_refused(Any spoil, str said)
None test_a_table_is_warned_for_only_what_it_lacks()
None test_an_abort_flag_written_as_no_is_no(Any written)
None test_a_state_flagged_abort_is_reachable_and_the_name_no_longer_decides()
None test_an_edited_table_reads_back_as_the_csv_machine()
None test_the_shipped_panel_places_the_states_and_flags_the_aborts()
dict[str, Any] _renamed(dict[str, Any] table, str old, str new)
None test_on_the_vehicle_an_unwired_row_still_drives_the_tank_top_stand_in()
None test_on_the_vehicle_the_box_keeps_the_rocket_and_frees_the_cut_rows()
None test_on_the_vehicle_a_cut_row_is_matched_whatever_case_its_connector_has()
None test_a_row_is_built_in_only_where_the_stand_acts_on_it(bool cut)