nonbinary.computer
OR-1 dataflow CPU sketch
1"""Tests for monitor/graph_json.py.
2
3Verifies:
4- or1-monitor.AC3.1 (partial): Graph structure for rendering
5- or1-monitor.AC3.7: PE state serialization
6- or1-monitor.AC3.8: SM state serialization
7"""
8
9from cm_inst import ArithOp, MemOp, Port, FrameDest, Instruction, OutputStyle
10from asm.ir import IRGraph, IRNode, IREdge, SourceLoc, SystemConfig
11from emu.events import TokenReceived, Matched, Executed, Emitted
12from monitor.graph_json import graph_to_monitor_json, graph_loaded_json
13from monitor.snapshot import StateSnapshot, PESnapshot, SMSnapshot, SMCellSnapshot
14from sm_mod import Presence
15from tokens import DyadToken
16
17
18class TestGraphJsonStructure:
19 """Test basic JSON structure output."""
20
21 def test_graph_loaded_json_structure(self):
22 """Test graph_loaded_json returns correct structure."""
23 # Create minimal IR graph
24 node = IRNode(
25 name="&test",
26 opcode=ArithOp.ADD,
27 pe=0,
28 iram_offset=0,
29 act_id=0,
30 loc=SourceLoc(1, 1),
31 )
32 ir_graph = IRGraph(
33 nodes={"&test": node},
34 edges=[],
35 system=SystemConfig(pe_count=1, sm_count=0),
36 )
37
38 # Create Instruction for IRAM
39 inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
40
41 # Create minimal snapshot
42 pe_snap = PESnapshot(
43 pe_id=0,
44 iram={0: inst},
45 frames=(),
46 tag_store={},
47 presence=(),
48 port_store=(),
49 match_data=(),
50 free_frames=(),
51 lane_count=4,
52 input_queue=(),
53 output_log=(),
54 )
55 snapshot = StateSnapshot(
56 sim_time=0.0,
57 next_time=1.0,
58 pes={0: pe_snap},
59 sms={},
60 )
61
62 result = graph_loaded_json(ir_graph, snapshot)
63
64 # Check structure
65 assert result["type"] == "graph_loaded"
66 assert "graph" in result
67 assert "nodes" in result["graph"]
68 assert "edges" in result["graph"]
69 assert "state" in result
70 assert "pes" in result["state"]
71 assert "sms" in result["state"]
72 assert result["sim_time"] == 0.0
73 assert result["finished"] is False
74
75 def test_graph_loaded_json_node_fields(self):
76 """Test node fields in graph_loaded_json."""
77 node = IRNode(
78 name="&add",
79 opcode=ArithOp.ADD,
80 pe=0,
81 iram_offset=5,
82 act_id=1,
83 const=42,
84 loc=SourceLoc(1, 1),
85 )
86 ir_graph = IRGraph(
87 nodes={"&add": node},
88 edges=[],
89 system=SystemConfig(pe_count=1, sm_count=0),
90 )
91
92 inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
93 pe_snap = PESnapshot(
94 pe_id=0,
95 iram={5: inst},
96 frames=(),
97 tag_store={},
98 presence=(),
99 port_store=(),
100 match_data=(),
101 free_frames=(),
102 lane_count=4,
103 input_queue=(),
104 output_log=(),
105 )
106 snapshot = StateSnapshot(
107 sim_time=0.0,
108 next_time=1.0,
109 pes={0: pe_snap},
110 sms={},
111 )
112
113 result = graph_loaded_json(ir_graph, snapshot)
114 nodes = result["graph"]["nodes"]
115
116 assert len(nodes) == 1
117 node_json = nodes[0]
118 assert node_json["id"] == "&add"
119 assert node_json["opcode"] == "add"
120 assert node_json["pe"] == 0
121 assert node_json["iram_offset"] == 5
122 assert node_json["act_id"] == 1
123 assert node_json["const"] == 42
124 # Check execution overlay fields are present and False
125 assert node_json["active"] is False
126 assert node_json["matched"] is False
127 assert node_json["executed"] is False
128
129 def test_graph_loaded_json_pe_state(self):
130 """Test PE state serialization."""
131 node = IRNode(
132 name="&add",
133 opcode=ArithOp.ADD,
134 pe=0,
135 iram_offset=5,
136 act_id=1,
137 )
138 ir_graph = IRGraph(nodes={"&add": node})
139
140 inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
141 pe_snap = PESnapshot(
142 pe_id=0,
143 iram={5: inst},
144 frames=(),
145 tag_store={},
146 presence=(),
147 port_store=(),
148 match_data=(),
149 free_frames=(),
150 lane_count=4,
151 input_queue=(),
152 output_log=(),
153 )
154 snapshot = StateSnapshot(
155 sim_time=0.0,
156 next_time=1.0,
157 pes={0: pe_snap},
158 sms={},
159 )
160
161 result = graph_loaded_json(ir_graph, snapshot)
162 pe_state = result["state"]["pes"]["0"]
163
164 assert "iram" in pe_state
165 assert "frames" in pe_state
166 assert "tag_store" in pe_state
167 assert "free_frames" in pe_state
168 assert pe_state["input_queue_size"] == 0
169
170 def test_graph_loaded_json_sm_state(self):
171 """Test SM state serialization."""
172 ir_graph = IRGraph()
173
174 cell_snap = SMCellSnapshot(
175 pres=Presence.FULL,
176 data_l=42,
177 data_r=None,
178 )
179 sm_snap = SMSnapshot(
180 sm_id=0,
181 cells={10: cell_snap},
182 deferred_read=None,
183 t0_store=(),
184 input_queue=(),
185 )
186 snapshot = StateSnapshot(
187 sim_time=0.0,
188 next_time=1.0,
189 pes={},
190 sms={0: sm_snap},
191 )
192
193 result = graph_loaded_json(ir_graph, snapshot)
194 sm_state = result["state"]["sms"]["0"]
195
196 assert "cells" in sm_state
197 assert "10" in sm_state["cells"]
198 assert sm_state["cells"]["10"]["presence"] == "FULL"
199 assert sm_state["cells"]["10"]["data_l"] == 42
200 assert sm_state["deferred_read"] is None
201 assert sm_state["t0_store_size"] == 0
202
203
204class TestGraphJsonWithEvents:
205 """Test execution overlay with events."""
206
207 def test_monitor_json_with_token_received(self):
208 """Test TokenReceived sets active flag."""
209 node = IRNode(
210 name="&add",
211 opcode=ArithOp.ADD,
212 pe=0,
213 iram_offset=0,
214 act_id=0,
215 )
216 ir_graph = IRGraph(nodes={"&add": node})
217
218 token = DyadToken(target=0, offset=0, act_id=0, data=42, port=Port.L)
219 event = TokenReceived(time=1.0, component="pe:0", token=token)
220
221 inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
222 pe_snap = PESnapshot(
223 pe_id=0,
224 iram={0: inst},
225 frames=(),
226 tag_store={},
227 presence=(),
228 port_store=(),
229 match_data=(),
230 free_frames=(),
231 lane_count=4,
232 input_queue=(),
233 output_log=(),
234 )
235 snapshot = StateSnapshot(
236 sim_time=1.0,
237 next_time=2.0,
238 pes={0: pe_snap},
239 sms={},
240 )
241
242 result = graph_to_monitor_json(ir_graph, snapshot, [event])
243
244 assert result["type"] == "monitor_update"
245 # Find the node in the result
246 nodes = result["graph"]["nodes"]
247 node_json = next((n for n in nodes if n["id"] == "&add"), None)
248 assert node_json is not None
249 assert node_json["active"] is True
250
251 def test_monitor_json_with_matched(self):
252 """Test Matched event sets matched flag."""
253 node = IRNode(
254 name="&add",
255 opcode=ArithOp.ADD,
256 pe=0,
257 iram_offset=5,
258 act_id=0,
259 )
260 ir_graph = IRGraph(nodes={"&add": node})
261
262 event = Matched(time=1.0, component="pe:0", left=42, right=7, act_id=0, offset=5, frame_id=0)
263
264 inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
265 pe_snap = PESnapshot(
266 pe_id=0,
267 iram={5: inst},
268 frames=(),
269 tag_store={},
270 presence=(),
271 port_store=(),
272 match_data=(),
273 free_frames=(),
274 lane_count=4,
275 input_queue=(),
276 output_log=(),
277 )
278 snapshot = StateSnapshot(
279 sim_time=1.0,
280 next_time=2.0,
281 pes={0: pe_snap},
282 sms={},
283 )
284
285 result = graph_to_monitor_json(ir_graph, snapshot, [event])
286
287 nodes = result["graph"]["nodes"]
288 node_json = next((n for n in nodes if n["id"] == "&add"), None)
289 assert node_json is not None
290 assert node_json["matched"] is True
291
292 def test_monitor_json_with_executed(self):
293 """Test Executed event sets executed flag."""
294 node = IRNode(
295 name="&add",
296 opcode=ArithOp.ADD,
297 pe=0,
298 iram_offset=0,
299 act_id=0,
300 )
301 ir_graph = IRGraph(nodes={"&add": node})
302
303 event = Executed(time=1.0, component="pe:0", op=ArithOp.ADD, result=49, bool_out=False)
304
305 inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
306 pe_snap = PESnapshot(
307 pe_id=0,
308 iram={0: inst},
309 frames=(),
310 tag_store={},
311 presence=(),
312 port_store=(),
313 match_data=(),
314 free_frames=(),
315 lane_count=4,
316 input_queue=(),
317 output_log=(),
318 )
319 snapshot = StateSnapshot(
320 sim_time=1.0,
321 next_time=2.0,
322 pes={0: pe_snap},
323 sms={},
324 )
325
326 result = graph_to_monitor_json(ir_graph, snapshot, [event])
327
328 nodes = result["graph"]["nodes"]
329 node_json = next((n for n in nodes if n["id"] == "&add"), None)
330 assert node_json is not None
331 assert node_json["executed"] is True
332
333 def test_monitor_json_events_serialization(self):
334 """Test events are properly serialized."""
335 ir_graph = IRGraph()
336 token = DyadToken(target=0, offset=0, act_id=0, data=42, port=Port.L)
337 events = [
338 TokenReceived(time=1.0, component="pe:0", token=token),
339 Executed(time=1.1, component="pe:0", op=ArithOp.ADD, result=49, bool_out=False),
340 ]
341
342 snapshot = StateSnapshot(
343 sim_time=1.1,
344 next_time=2.0,
345 pes={},
346 sms={},
347 )
348
349 result = graph_to_monitor_json(ir_graph, snapshot, events)
350
351 assert "events" in result
352 assert len(result["events"]) == 2
353
354 event_json_0 = result["events"][0]
355 assert event_json_0["type"] == "TokenReceived"
356 assert event_json_0["time"] == 1.0
357 assert event_json_0["component"] == "pe:0"
358 assert "details" in event_json_0
359
360 event_json_1 = result["events"][1]
361 assert event_json_1["type"] == "Executed"
362 assert event_json_1["time"] == 1.1
363 assert event_json_1["component"] == "pe:0"
364
365
366class TestGraphJsonEdges:
367 """Test edge serialization."""
368
369 def test_edge_serialization(self):
370 """Test edges are properly serialized."""
371 source = IRNode(
372 name="&a",
373 opcode=ArithOp.ADD,
374 pe=0,
375 iram_offset=0,
376 act_id=0,
377 )
378 dest = IRNode(
379 name="&b",
380 opcode=ArithOp.SUB,
381 pe=0,
382 iram_offset=1,
383 act_id=0,
384 )
385 edge = IREdge(
386 source="&a",
387 dest="&b",
388 port=Port.L,
389 source_port=Port.L,
390 )
391 ir_graph = IRGraph(
392 nodes={"&a": source, "&b": dest},
393 edges=[edge],
394 )
395
396 snapshot = StateSnapshot(
397 sim_time=0.0,
398 next_time=1.0,
399 pes={},
400 sms={},
401 )
402
403 result = graph_loaded_json(ir_graph, snapshot)
404 edges = result["graph"]["edges"]
405
406 assert len(edges) == 1
407 edge_json = edges[0]
408 assert edge_json["source"] == "&a"
409 assert edge_json["target"] == "&b"
410 assert edge_json["port"] == "L"
411 # source_port can be None or "L" depending on how it's set
412 assert edge_json["source_port"] in [None, "L"]
413 assert edge_json["token_flow"] is False
414
415 def test_edge_token_flow(self):
416 """Test token_flow flag is set on emitted edges."""
417 source = IRNode(
418 name="&a",
419 opcode=ArithOp.ADD,
420 pe=0,
421 iram_offset=0,
422 act_id=0,
423 )
424 dest = IRNode(
425 name="&b",
426 opcode=ArithOp.SUB,
427 pe=1, # Destination on a different PE
428 iram_offset=1,
429 act_id=0,
430 )
431 edge = IREdge(source="&a", dest="&b", port=Port.L)
432 ir_graph = IRGraph(
433 nodes={"&a": source, "&b": dest},
434 edges=[edge],
435 )
436
437 # Emitted event with token targeting PE 1 should mark the edge
438 token = DyadToken(target=1, offset=1, act_id=0, data=42, port=Port.L)
439 event = Emitted(time=1.0, component="pe:0", token=token)
440
441 snapshot = StateSnapshot(
442 sim_time=1.0,
443 next_time=2.0,
444 pes={},
445 sms={},
446 )
447
448 result = graph_to_monitor_json(ir_graph, snapshot, [event])
449 edges = result["graph"]["edges"]
450
451 assert len(edges) == 1
452 edge_json = edges[0]
453 # Edge token_flow should be True when an Emitted event targets the destination PE
454 assert edge_json["token_flow"] is True
455
456
457class TestGraphJsonFinished:
458 """Test finished flag."""
459
460 def test_finished_false_when_next_time_not_inf(self):
461 """Test finished=False when next_time is finite."""
462 ir_graph = IRGraph()
463 snapshot = StateSnapshot(
464 sim_time=0.0,
465 next_time=1.0,
466 pes={},
467 sms={},
468 )
469
470 result = graph_loaded_json(ir_graph, snapshot)
471 assert result["finished"] is False
472
473 def test_finished_true_when_next_time_inf(self):
474 """Test finished=True when next_time is infinity."""
475 ir_graph = IRGraph()
476 snapshot = StateSnapshot(
477 sim_time=10.0,
478 next_time=float('inf'),
479 pes={},
480 sms={},
481 )
482
483 result = graph_loaded_json(ir_graph, snapshot)
484 assert result["finished"] is True
485
486
487class TestTagStoreSerialisation:
488 """Test non-empty tag_store JSON serialization in PE state."""
489
490 def test_non_empty_tag_store_serialization(self):
491 """Test that tag_store with non-empty mapping serializes correctly."""
492 node = IRNode(
493 name="&add",
494 opcode=ArithOp.ADD,
495 pe=0,
496 iram_offset=0,
497 act_id=0,
498 )
499 ir_graph = IRGraph(nodes={"&add": node})
500
501 inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
502
503 # Create PESnapshot with non-empty tag_store: act_id 0 maps to frame 2, lane 1
504 pe_snap = PESnapshot(
505 pe_id=0,
506 iram={0: inst},
507 frames=(),
508 tag_store={0: (2, 1)}, # act_id=0 -> (frame_id=2, lane=1)
509 presence=(),
510 port_store=(),
511 match_data=(),
512 free_frames=(),
513 lane_count=4,
514 input_queue=(),
515 output_log=(),
516 )
517 snapshot = StateSnapshot(
518 sim_time=0.0,
519 next_time=1.0,
520 pes={0: pe_snap},
521 sms={},
522 )
523
524 result = graph_loaded_json(ir_graph, snapshot)
525 pe_state = result["state"]["pes"]["0"]
526
527 # Verify tag_store is correctly serialized
528 assert "tag_store" in pe_state
529 assert "0" in pe_state["tag_store"] # act_id 0 should be a string key "0"
530 assert pe_state["tag_store"]["0"]["frame_id"] == 2
531 assert pe_state["tag_store"]["0"]["lane"] == 1
532 assert pe_state["lane_count"] == 4
533
534 def test_multiple_entries_tag_store_serialization(self):
535 """Test tag_store with multiple act_id entries serializes correctly."""
536 ir_graph = IRGraph()
537
538 inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
539
540 # Create PESnapshot with multiple tag_store entries
541 pe_snap = PESnapshot(
542 pe_id=0,
543 iram={0: inst},
544 frames=(),
545 tag_store={
546 0: (2, 1),
547 1: (3, 0),
548 5: (7, 2),
549 },
550 presence=(),
551 port_store=(),
552 match_data=(),
553 free_frames=(),
554 lane_count=4,
555 input_queue=(),
556 output_log=(),
557 )
558 snapshot = StateSnapshot(
559 sim_time=0.0,
560 next_time=1.0,
561 pes={0: pe_snap},
562 sms={},
563 )
564
565 result = graph_loaded_json(ir_graph, snapshot)
566 pe_state = result["state"]["pes"]["0"]
567
568 # Verify all entries are correctly serialized
569 assert "tag_store" in pe_state
570 assert pe_state["tag_store"]["0"]["frame_id"] == 2
571 assert pe_state["tag_store"]["0"]["lane"] == 1
572 assert pe_state["tag_store"]["1"]["frame_id"] == 3
573 assert pe_state["tag_store"]["1"]["lane"] == 0
574 assert pe_state["tag_store"]["5"]["frame_id"] == 7
575 assert pe_state["tag_store"]["5"]["lane"] == 2
576 assert pe_state["lane_count"] == 4
577
578
579class TestAC72_NewEventTypesSerialization:
580 """AC7.2: Verify new frame-based event types serialize correctly to JSON."""
581
582 def test_frame_allocated_event_serialization(self):
583 """FrameAllocated event should serialize with correct type and fields."""
584 from emu.events import FrameAllocated
585
586 event = FrameAllocated(
587 time=5.0,
588 component="pe:0",
589 act_id=1,
590 frame_id=2,
591 lane=0,
592 )
593
594 ir_graph = IRGraph()
595 snapshot = StateSnapshot(
596 sim_time=5.0,
597 next_time=6.0,
598 pes={},
599 sms={},
600 )
601
602 result = graph_to_monitor_json(ir_graph, snapshot, [event])
603
604 # Extract the events from result
605 assert "events" in result
606 events_list = result["events"]
607 assert len(events_list) == 1
608
609 event_json = events_list[0]
610 assert event_json["type"] == "FrameAllocated"
611 assert event_json["time"] == 5.0
612 assert event_json["component"] == "pe:0"
613 assert "details" in event_json
614 assert event_json["details"]["act_id"] == 1
615 assert event_json["details"]["frame_id"] == 2
616 assert event_json["details"]["lane"] == 0
617
618 def test_frame_freed_event_serialization(self):
619 """FrameFreed event should serialize with correct type and fields."""
620 from emu.events import FrameFreed
621
622 event = FrameFreed(
623 time=10.0,
624 component="pe:1",
625 act_id=3,
626 frame_id=4,
627 lane=0,
628 frame_freed=True,
629 )
630
631 ir_graph = IRGraph()
632 snapshot = StateSnapshot(
633 sim_time=10.0,
634 next_time=11.0,
635 pes={},
636 sms={},
637 )
638
639 result = graph_to_monitor_json(ir_graph, snapshot, [event])
640
641 events_list = result["events"]
642 assert len(events_list) == 1
643
644 event_json = events_list[0]
645 assert event_json["type"] == "FrameFreed"
646 assert event_json["time"] == 10.0
647 assert event_json["component"] == "pe:1"
648 assert "details" in event_json
649 assert event_json["details"]["act_id"] == 3
650 assert event_json["details"]["frame_id"] == 4
651 assert event_json["details"]["lane"] == 0
652 assert event_json["details"]["frame_freed"] == True
653
654 def test_frame_slot_written_event_serialization(self):
655 """FrameSlotWritten event should serialize with correct type and fields."""
656 from emu.events import FrameSlotWritten
657
658 event = FrameSlotWritten(
659 time=7.0,
660 component="pe:0",
661 frame_id=1,
662 slot=5,
663 value=0x1234,
664 )
665
666 ir_graph = IRGraph()
667 snapshot = StateSnapshot(
668 sim_time=7.0,
669 next_time=8.0,
670 pes={},
671 sms={},
672 )
673
674 result = graph_to_monitor_json(ir_graph, snapshot, [event])
675
676 events_list = result["events"]
677 assert len(events_list) == 1
678
679 event_json = events_list[0]
680 assert event_json["type"] == "FrameSlotWritten"
681 assert event_json["time"] == 7.0
682 assert event_json["component"] == "pe:0"
683 assert "details" in event_json
684 assert event_json["details"]["frame_id"] == 1
685 assert event_json["details"]["slot"] == 5
686 assert event_json["details"]["value"] == 0x1234
687
688 def test_token_rejected_event_serialization(self):
689 """TokenRejected event should serialize with correct type and fields."""
690 from emu.events import TokenRejected
691 from tokens import DyadToken
692
693 token = DyadToken(
694 target=0,
695 offset=1,
696 act_id=99,
697 data=0x5678,
698 port=Port.L,
699 )
700
701 event = TokenRejected(
702 time=3.0,
703 component="pe:0",
704 token=token,
705 reason="invalid act_id",
706 )
707
708 ir_graph = IRGraph()
709 snapshot = StateSnapshot(
710 sim_time=3.0,
711 next_time=4.0,
712 pes={},
713 sms={},
714 )
715
716 result = graph_to_monitor_json(ir_graph, snapshot, [event])
717
718 events_list = result["events"]
719 assert len(events_list) == 1
720
721 event_json = events_list[0]
722 assert event_json["type"] == "TokenRejected"
723 assert event_json["time"] == 3.0
724 assert event_json["component"] == "pe:0"
725 assert "details" in event_json
726 assert event_json["details"]["reason"] == "invalid act_id"