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, Addr, ALUInst
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 ctx=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 ALUInst for IRAM
39 inst = ALUInst(op=ArithOp.ADD, dest_l=None, dest_r=None, const=None)
40
41 # Create minimal snapshot
42 pe_snap = PESnapshot(
43 pe_id=0,
44 iram={0: inst},
45 matching_store=(),
46 gen_counters=(),
47 input_queue=(),
48 output_log=(),
49 )
50 snapshot = StateSnapshot(
51 sim_time=0.0,
52 next_time=1.0,
53 pes={0: pe_snap},
54 sms={},
55 )
56
57 result = graph_loaded_json(ir_graph, snapshot)
58
59 # Check structure
60 assert result["type"] == "graph_loaded"
61 assert "graph" in result
62 assert "nodes" in result["graph"]
63 assert "edges" in result["graph"]
64 assert "state" in result
65 assert "pes" in result["state"]
66 assert "sms" in result["state"]
67 assert result["sim_time"] == 0.0
68 assert result["finished"] is False
69
70 def test_graph_loaded_json_node_fields(self):
71 """Test node fields in graph_loaded_json."""
72 node = IRNode(
73 name="&add",
74 opcode=ArithOp.ADD,
75 pe=0,
76 iram_offset=5,
77 ctx=1,
78 const=42,
79 loc=SourceLoc(1, 1),
80 )
81 ir_graph = IRGraph(
82 nodes={"&add": node},
83 edges=[],
84 system=SystemConfig(pe_count=1, sm_count=0),
85 )
86
87 inst = ALUInst(op=ArithOp.ADD, dest_l=None, dest_r=None, const=None)
88 pe_snap = PESnapshot(
89 pe_id=0,
90 iram={5: inst},
91 matching_store=(),
92 gen_counters=(),
93 input_queue=(),
94 output_log=(),
95 )
96 snapshot = StateSnapshot(
97 sim_time=0.0,
98 next_time=1.0,
99 pes={0: pe_snap},
100 sms={},
101 )
102
103 result = graph_loaded_json(ir_graph, snapshot)
104 nodes = result["graph"]["nodes"]
105
106 assert len(nodes) == 1
107 node_json = nodes[0]
108 assert node_json["id"] == "&add"
109 assert node_json["opcode"] == "add"
110 assert node_json["pe"] == 0
111 assert node_json["iram_offset"] == 5
112 assert node_json["ctx"] == 1
113 assert node_json["const"] == 42
114 # Check execution overlay fields are present and False
115 assert node_json["active"] is False
116 assert node_json["matched"] is False
117 assert node_json["executed"] is False
118
119 def test_graph_loaded_json_pe_state(self):
120 """Test PE state serialization."""
121 node = IRNode(
122 name="&add",
123 opcode=ArithOp.ADD,
124 pe=0,
125 iram_offset=5,
126 ctx=1,
127 )
128 ir_graph = IRGraph(nodes={"&add": node})
129
130 inst = ALUInst(op=ArithOp.ADD, dest_l=None, dest_r=None, const=None)
131 pe_snap = PESnapshot(
132 pe_id=0,
133 iram={5: inst},
134 matching_store=(),
135 gen_counters=(0, 1, 2),
136 input_queue=(),
137 output_log=(),
138 )
139 snapshot = StateSnapshot(
140 sim_time=0.0,
141 next_time=1.0,
142 pes={0: pe_snap},
143 sms={},
144 )
145
146 result = graph_loaded_json(ir_graph, snapshot)
147 pe_state = result["state"]["pes"]["0"]
148
149 assert "iram" in pe_state
150 assert "matching_store" in pe_state
151 assert "gen_counters" in pe_state
152 assert pe_state["gen_counters"] == [0, 1, 2]
153 assert pe_state["input_queue_depth"] == 0
154 assert pe_state["output_count"] == 0
155
156 def test_graph_loaded_json_sm_state(self):
157 """Test SM state serialization."""
158 ir_graph = IRGraph()
159
160 cell_snap = SMCellSnapshot(
161 pres=Presence.FULL,
162 data_l=42,
163 data_r=None,
164 )
165 sm_snap = SMSnapshot(
166 sm_id=0,
167 cells={10: cell_snap},
168 deferred_read=None,
169 t0_store=(),
170 input_queue=(),
171 )
172 snapshot = StateSnapshot(
173 sim_time=0.0,
174 next_time=1.0,
175 pes={},
176 sms={0: sm_snap},
177 )
178
179 result = graph_loaded_json(ir_graph, snapshot)
180 sm_state = result["state"]["sms"]["0"]
181
182 assert "cells" in sm_state
183 assert "10" in sm_state["cells"]
184 assert sm_state["cells"]["10"]["presence"] == "FULL"
185 assert sm_state["cells"]["10"]["data_l"] == 42
186 assert sm_state["deferred_read"] is None
187 assert sm_state["t0_store_size"] == 0
188
189
190class TestGraphJsonWithEvents:
191 """Test execution overlay with events."""
192
193 def test_monitor_json_with_token_received(self):
194 """Test TokenReceived sets active flag."""
195 node = IRNode(
196 name="&add",
197 opcode=ArithOp.ADD,
198 pe=0,
199 iram_offset=0,
200 ctx=0,
201 )
202 ir_graph = IRGraph(nodes={"&add": node})
203
204 token = DyadToken(target=0, offset=0, ctx=0, data=42, port=Port.L, gen=0, wide=False)
205 event = TokenReceived(time=1.0, component="pe:0", token=token)
206
207 inst = ALUInst(op=ArithOp.ADD, dest_l=None, dest_r=None, const=None)
208 pe_snap = PESnapshot(
209 pe_id=0,
210 iram={0: inst},
211 matching_store=(),
212 gen_counters=(),
213 input_queue=(),
214 output_log=(),
215 )
216 snapshot = StateSnapshot(
217 sim_time=1.0,
218 next_time=2.0,
219 pes={0: pe_snap},
220 sms={},
221 )
222
223 result = graph_to_monitor_json(ir_graph, snapshot, [event])
224
225 assert result["type"] == "monitor_update"
226 # Find the node in the result
227 nodes = result["graph"]["nodes"]
228 node_json = next((n for n in nodes if n["id"] == "&add"), None)
229 assert node_json is not None
230 assert node_json["active"] is True
231
232 def test_monitor_json_with_matched(self):
233 """Test Matched event sets matched flag."""
234 node = IRNode(
235 name="&add",
236 opcode=ArithOp.ADD,
237 pe=0,
238 iram_offset=5,
239 ctx=0,
240 )
241 ir_graph = IRGraph(nodes={"&add": node})
242
243 event = Matched(time=1.0, component="pe:0", left=42, right=7, ctx=0, offset=5)
244
245 inst = ALUInst(op=ArithOp.ADD, dest_l=None, dest_r=None, const=None)
246 pe_snap = PESnapshot(
247 pe_id=0,
248 iram={5: inst},
249 matching_store=(),
250 gen_counters=(),
251 input_queue=(),
252 output_log=(),
253 )
254 snapshot = StateSnapshot(
255 sim_time=1.0,
256 next_time=2.0,
257 pes={0: pe_snap},
258 sms={},
259 )
260
261 result = graph_to_monitor_json(ir_graph, snapshot, [event])
262
263 nodes = result["graph"]["nodes"]
264 node_json = next((n for n in nodes if n["id"] == "&add"), None)
265 assert node_json is not None
266 assert node_json["matched"] is True
267
268 def test_monitor_json_with_executed(self):
269 """Test Executed event sets executed flag."""
270 node = IRNode(
271 name="&add",
272 opcode=ArithOp.ADD,
273 pe=0,
274 iram_offset=0,
275 ctx=0,
276 )
277 ir_graph = IRGraph(nodes={"&add": node})
278
279 event = Executed(time=1.0, component="pe:0", op=ArithOp.ADD, result=49, bool_out=False)
280
281 inst = ALUInst(op=ArithOp.ADD, dest_l=None, dest_r=None, const=None)
282 pe_snap = PESnapshot(
283 pe_id=0,
284 iram={0: inst},
285 matching_store=(),
286 gen_counters=(),
287 input_queue=(),
288 output_log=(),
289 )
290 snapshot = StateSnapshot(
291 sim_time=1.0,
292 next_time=2.0,
293 pes={0: pe_snap},
294 sms={},
295 )
296
297 result = graph_to_monitor_json(ir_graph, snapshot, [event])
298
299 nodes = result["graph"]["nodes"]
300 node_json = next((n for n in nodes if n["id"] == "&add"), None)
301 assert node_json is not None
302 assert node_json["executed"] is True
303
304 def test_monitor_json_events_serialization(self):
305 """Test events are properly serialized."""
306 ir_graph = IRGraph()
307 token = DyadToken(target=0, offset=0, ctx=0, data=42, port=Port.L, gen=0, wide=False)
308 events = [
309 TokenReceived(time=1.0, component="pe:0", token=token),
310 Executed(time=1.1, component="pe:0", op=ArithOp.ADD, result=49, bool_out=False),
311 ]
312
313 snapshot = StateSnapshot(
314 sim_time=1.1,
315 next_time=2.0,
316 pes={},
317 sms={},
318 )
319
320 result = graph_to_monitor_json(ir_graph, snapshot, events)
321
322 assert "events" in result
323 assert len(result["events"]) == 2
324
325 event_json_0 = result["events"][0]
326 assert event_json_0["type"] == "TokenReceived"
327 assert event_json_0["time"] == 1.0
328 assert event_json_0["component"] == "pe:0"
329 assert "details" in event_json_0
330
331 event_json_1 = result["events"][1]
332 assert event_json_1["type"] == "Executed"
333 assert event_json_1["time"] == 1.1
334 assert event_json_1["component"] == "pe:0"
335
336
337class TestGraphJsonEdges:
338 """Test edge serialization."""
339
340 def test_edge_serialization(self):
341 """Test edges are properly serialized."""
342 source = IRNode(
343 name="&a",
344 opcode=ArithOp.ADD,
345 pe=0,
346 iram_offset=0,
347 ctx=0,
348 )
349 dest = IRNode(
350 name="&b",
351 opcode=ArithOp.SUB,
352 pe=0,
353 iram_offset=1,
354 ctx=0,
355 )
356 edge = IREdge(
357 source="&a",
358 dest="&b",
359 port=Port.L,
360 source_port=Port.L,
361 )
362 ir_graph = IRGraph(
363 nodes={"&a": source, "&b": dest},
364 edges=[edge],
365 )
366
367 snapshot = StateSnapshot(
368 sim_time=0.0,
369 next_time=1.0,
370 pes={},
371 sms={},
372 )
373
374 result = graph_loaded_json(ir_graph, snapshot)
375 edges = result["graph"]["edges"]
376
377 assert len(edges) == 1
378 edge_json = edges[0]
379 assert edge_json["source"] == "&a"
380 assert edge_json["target"] == "&b"
381 assert edge_json["port"] == "L"
382 # source_port can be None or "L" depending on how it's set
383 assert edge_json["source_port"] in [None, "L"]
384 assert edge_json["token_flow"] is False
385
386 def test_edge_token_flow(self):
387 """Test token_flow flag is set on emitted edges."""
388 source = IRNode(
389 name="&a",
390 opcode=ArithOp.ADD,
391 pe=0,
392 iram_offset=0,
393 ctx=0,
394 )
395 dest = IRNode(
396 name="&b",
397 opcode=ArithOp.SUB,
398 pe=1, # Destination on a different PE
399 iram_offset=1,
400 ctx=0,
401 )
402 edge = IREdge(source="&a", dest="&b", port=Port.L)
403 ir_graph = IRGraph(
404 nodes={"&a": source, "&b": dest},
405 edges=[edge],
406 )
407
408 # Emitted event with token targeting PE 1 should mark the edge
409 token = DyadToken(target=1, offset=1, ctx=0, data=42, port=Port.L, gen=0, wide=False)
410 event = Emitted(time=1.0, component="pe:0", token=token)
411
412 snapshot = StateSnapshot(
413 sim_time=1.0,
414 next_time=2.0,
415 pes={},
416 sms={},
417 )
418
419 result = graph_to_monitor_json(ir_graph, snapshot, [event])
420 edges = result["graph"]["edges"]
421
422 assert len(edges) == 1
423 edge_json = edges[0]
424 # Edge token_flow should be True when an Emitted event targets the destination PE
425 assert edge_json["token_flow"] is True
426
427
428class TestGraphJsonFinished:
429 """Test finished flag."""
430
431 def test_finished_false_when_next_time_not_inf(self):
432 """Test finished=False when next_time is finite."""
433 ir_graph = IRGraph()
434 snapshot = StateSnapshot(
435 sim_time=0.0,
436 next_time=1.0,
437 pes={},
438 sms={},
439 )
440
441 result = graph_loaded_json(ir_graph, snapshot)
442 assert result["finished"] is False
443
444 def test_finished_true_when_next_time_inf(self):
445 """Test finished=True when next_time is infinity."""
446 ir_graph = IRGraph()
447 snapshot = StateSnapshot(
448 sim_time=10.0,
449 next_time=float('inf'),
450 pes={},
451 sms={},
452 )
453
454 result = graph_loaded_json(ir_graph, snapshot)
455 assert result["finished"] is True