"""Behaviour of every component in the IMS co-verification twin. One module so the seven behaviour files stay thin, and so the whole system's logic can be read in one place during a demo. TWO RULES, and they are what make this a co-verification demo rather than an animation: 1. **Every timing number comes from the Cameo model.** Nothing here invents a latency, a frame size, a parse cost or a deadline. They arrive in ims.params.json, generated from the model's value properties. If a number looks wrong, the model is where it gets fixed. 2. **The software takes clock cycles.** MissionProcessor runs the model's `Fuse And Transmit` activity as a timed interpreter: one action at a time, each occupying its bound duration in simulated microseconds, advancing only as fabric steps advance. Nothing completes atomically. Chris, 2026-09-09. Scenario knobs (targets, collection window, Vendor B mode, telemetry, and whether the planted defect is present) live in ims.scenario.json so a campaign can sweep them without a rebuild. """ import json import os _HERE = os.path.dirname(os.path.abspath(__file__)) def _find(name): for d in (_HERE, os.path.join(_HERE, ".."), os.path.join(_HERE, "..", "..")): p = os.path.join(d, name) if os.path.isfile(p): return p raise RuntimeError("cannot find %s near %s" % (name, _HERE)) def _load(name, default=None): try: with open(_find(name), encoding="utf-8") as fh: return json.load(fh) except RuntimeError: if default is None: raise return default PARAMS = _load("ims.params.json") SCENARIO = _load("ims.scenario.json", {}) BLOCKS = PARAMS["blocks"] STEP_US = PARAMS["step_us"] ETHERTYPE = PARAMS["ethertypes"] ENDPOINT = PARAMS["ethernet"]["endpoints"] OVERHEAD = PARAMS["ethernet"]["serialization_overhead_bytes"] REPORT = ETHERTYPE["ReceiverReportMsg"] COMMAND = ETHERTYPE["TransmitCommandMsg"] TELEMETRY = ETHERTYPE["TelemetryMsg"] def opt(key, default): v = SCENARIO.get(key, default) return default if v is None else v # Scenario, with the model's own value properties as the defaults. TARGETS = int(opt("targets", BLOCKS["TargetScenario"]["targets"])) WINDOW_US = int(opt("window_us", BLOCKS["MissionProcessor"]["window_us"])) EARLY_PARTIAL = int(opt("early_partial", BLOCKS["ReceiverVendorB"]["early_partial"])) TELEMETRY_ON = int(opt("telemetry", 1)) # The planted freshness defect. 1 reproduces the shipped software, 0 is fix 1. STALE_DEFECT = int(opt("stale_defect", 1)) FRAME_PERIOD_US = int(BLOCKS["MissionProcessor"]["frame_period_us"]) def truth_hash(frame, targets): """Ground truth for a frame: what a correct combined message must carry. Deterministic and frame-dependent, which is the whole point: a report from the previous frame carries a different value, so a stale fusion is provable from the traces rather than a matter of opinion.""" return ((frame * 37 + targets * 11) % 251) + 1 def serialization_us(nbytes): """Wire time for one frame, from the model's link rate.""" mbps = int(BLOCKS["RuggedizedSwitch"]["link_mbps"]) return int(round(((nbytes + OVERHEAD) * 8.0) / mbps)) class Eth(object): """One Ethernet frame as it rides the signal set.""" FIELDS = ("valid", "src", "dst", "ethertype", "bytes", "frame", "hash", "frag_index", "frag_total") def __init__(self, src=0, dst=0, ethertype=0, nbytes=0, frame=0, hsh=0, frag_index=0, frag_total=1): self.valid = 1 self.src = src self.dst = dst self.ethertype = ethertype self.bytes = nbytes self.frame = frame self.hash = hsh self.frag_index = frag_index self.frag_total = frag_total @classmethod def read(cls, inputs, prefix): if not int(inputs.get(prefix + "valid", 0) or 0): return None f = cls() for name in cls.FIELDS: setattr(f, name, int(inputs.get(prefix + name, 0) or 0)) return f def write(self, outs, prefix): for name in self.FIELDS: outs[prefix + name] = int(getattr(self, name)) @staticmethod def blank(outs, prefix): for name in Eth.FIELDS: outs[prefix + name] = 0 class Component(object): name = "?" def __init__(self): self.log = [] def step(self, step, time_ns, inputs): now_us = time_ns // 1000 outs = {} self.tick(step, now_us, inputs, outs) return outs def tick(self, step, now_us, inputs, outs): raise NotImplementedError def say(self, now_us, msg): print("[%s %8d us] %s" % (self.name, now_us, msg)) # --------------------------------------------------------------- the fixture class TargetScenario(Component): """Injects detections past the receivers' RF front end, and is the ground truth the checker compares the transmitted message against. Not a system element: DetectionEventIF is marked modelled=0 in the model.""" name = "TargetScenario" def tick(self, step, now_us, inputs, outs): frame = now_us // FRAME_PERIOD_US if now_us % FRAME_PERIOD_US == 0: outs["detOut_det_valid"] = 1 outs["detOut_det_frame"] = frame outs["detOut_det_count"] = TARGETS outs["detOut_det_t0_us"] = now_us outs["detOut_det_hash"] = truth_hash(frame, TARGETS) self.say(now_us, "frame %d: %d target(s), truth hash %d" % (frame, TARGETS, truth_hash(frame, TARGETS))) else: for f in ("valid", "frame", "count", "t0_us", "hash"): outs["detOut_det_" + f] = 0 # ------------------------------------------------------------- vendor blocks class Receiver(Component): """A vendor receiver, as a TIMING ENVELOPE and nothing else. No functional detail whatsoever, because a vendor does not give you any. Latency, jitter, fragmentation and payload size, straight off the model's value properties, and that is genuinely enough to find the defect.""" block = None endpoint = None def __init__(self): Component.__init__(self) self.v = BLOCKS[self.block] self.pending = [] # [(emit_at_us, Eth)] def latency_for(self, targets): base = int(self.v["latency_typ_us"]) per = int(self.v.get("latency_per_target_us", 0) or 0) return base + per * max(0, targets - 1) def jitter_for(self, frame): j = int(self.v.get("jitter_us", 0) or 0) if j <= 0: return 0 # Deterministic, so a run is reproducible and a campaign is comparable. return (frame * 7919) % (2 * j) - j def on_detection(self, now_us, frame, targets, hsh): raise NotImplementedError def tick(self, step, now_us, inputs, outs): det = int(inputs.get("rfIn_det_valid", 0) or 0) if det: self.on_detection(now_us, int(inputs.get("rfIn_det_frame", 0) or 0), int(inputs.get("rfIn_det_count", 0) or 0), int(inputs.get("rfIn_det_hash", 0) or 0)) due = [p for p in self.pending if p[0] <= now_us] if due: due.sort() emit_at, frame_obj = due[0] self.pending.remove((emit_at, frame_obj)) frame_obj.write(outs, "eth_tx_") self.say(now_us, "report frame %d fragment %d of %d, %d bytes" % (frame_obj.frame, frame_obj.frag_index + 1, frame_obj.frag_total, frame_obj.bytes)) else: Eth.blank(outs, "eth_tx_") class ReceiverVendorA(Receiver): name = "ReceiverVendorA" block = "ReceiverVendorA" def on_detection(self, now_us, frame, targets, hsh): at = now_us + self.latency_for(targets) + self.jitter_for(frame) self.pending.append((at, Eth(src=ENDPOINT["ReceiverVendorA"], dst=ENDPOINT["MissionProcessor"], ethertype=REPORT, nbytes=int(self.v["frame_bytes"]), frame=frame, hsh=hsh, frag_index=0, frag_total=1))) class ReceiverVendorB(Receiver): """Vendor B, and the datasheet footnote that breaks the system. The headline latency is latency_typ_us. What a software team is rarely handed is latency_per_target_us, and that it fragments: one 1500 byte fragment per target, up to burst_fragments. early_partial is the vendor OPTION: fragments emitted as they are produced rather than batched at the end. It is off by default, matching what ships.""" name = "ReceiverVendorB" block = "ReceiverVendorB" def on_detection(self, now_us, frame, targets, hsh): frags = max(1, min(targets, int(self.v["burst_fragments"]))) per = int(self.v.get("latency_per_target_us", 0) or 0) base = int(self.v["latency_typ_us"]) jit = self.jitter_for(frame) for k in range(frags): if EARLY_PARTIAL: at = now_us + base + per * k + jit else: at = now_us + self.latency_for(targets) + jit + k * STEP_US self.pending.append((at, Eth(src=ENDPOINT["ReceiverVendorB"], dst=ENDPOINT["MissionProcessor"], ethertype=REPORT, nbytes=int(self.v["frame_bytes"]), frame=frame, hsh=hsh, frag_index=k, frag_total=frags))) # ------------------------------------------------------------- the network class RuggedizedSwitch(Component): """Store and forward with a per-port output queue. This is where the Ethernet actually lives. VSI's tcpIp gateway is hub and spoke with two ports per socket and cannot carry a five-endpoint switched network, so the wire is modelled here and carried over GenericPayload signals (see the assumptions file). The queueing physics is real and every parameter comes from the model.""" name = "RuggedizedSwitch" PORTS = {"portRxA": 1, "portRxB": 2, "portMp": 3, "portTx": 4} def __init__(self): Component.__init__(self) v = BLOCKS["RuggedizedSwitch"] self.fabric_us = int(v["fabric_latency_us"]) self.free_at = dict((p, 0) for p in self.PORTS) # egress busy until self.queue = dict((p, []) for p in self.PORTS) # [(ready_us, Eth)] self.by_endpoint = dict((v2, p) for p, v2 in self.PORTS.items()) self.dropped = 0 def tick(self, step, now_us, inputs, outs): # ingress: a frame is fully received only after its serialization time for port in self.PORTS: f = Eth.read(inputs, port + "_tx_") if f is None: continue egress = self.by_endpoint.get(f.dst) if egress is None: self.dropped += 1 self.say(now_us, "DROP: frame for unknown endpoint %d" % f.dst) continue ready = now_us + serialization_us(f.bytes) + self.fabric_us self.queue[egress].append((ready, f)) # egress: one frame at a time per port, serialized for port in self.PORTS: Eth.blank(outs, port + "_rx_") q = self.queue[port] if not q: continue q.sort(key=lambda e: e[0]) ready, f = q[0] if ready > now_us or self.free_at[port] > now_us: continue q.pop(0) self.free_at[port] = now_us + serialization_us(f.bytes) f.write(outs, port + "_rx_") # ------------------------------------------------- processor plus software class Activity(object): """The model's `Fuse And Transmit` activity, executed as a timed machine. This is the part that must not be atomic. One action is current at a time and it occupies its bound duration in simulated microseconds; the machine only advances when the fabric clock has advanced past that. A frame's software work therefore spans tens of fabric steps, exactly as the hardware does, and it can miss a deadline.""" def __init__(self, params): sw = params["software"] self.duration = sw["durations"] self.per_fragment = set(sw["per_fragment"]) self.next_by_guard = {} for e in sw["edges"]: self.next_by_guard.setdefault(e["source"], []).append( (e["guard"], e["target"])) self.node = None self.busy_until = 0 def enter(self, node, now_us, fragments=1): self.node = node d = int(self.duration.get(node, 0)) if node in self.per_fragment: d = d * max(1, fragments) self.busy_until = now_us + d return d def is_busy(self, now_us): return now_us < self.busy_until def successors(self, node): return self.next_by_guard.get(node, []) class MissionProcessor(Component): """The processor, and the mission software that runs on it. There is no interface between them: the model says the software is an activity owned by this block, so the twin runs it inside this component's own on_step and both share one clock. Hardware time (dma_irq_us) and software time (the activity durations) are the same clock.""" name = "MissionProcessor" def __init__(self): Component.__init__(self) self.v = BLOCKS["MissionProcessor"] self.dma_us = int(self.v["dma_irq_us"]) self.act = Activity(PARAMS) self.inbox = [] # [(deliver_at_us, Eth)] self.reset_frame(0, 0) # The report buffer deliberately SURVIVES a frame boundary. That is the # planted defect: nothing clears it and nothing timestamps what it holds. self.buf_a = {"hash": 0, "frame": -1, "at": -1} self.buf_b = {"hash": 0, "frame": -1, "at": -1, "frags": 0} self.pending_tx = None self.drops = 0 def reset_frame(self, frame, now_us): self.frame = frame self.t0 = now_us self.got_a = False self.got_b = False self.b_frags = 0 self.act.node = None self.act.busy_until = 0 def tick(self, step, now_us, inputs, outs): Eth.blank(outs, "eth_tx_") # --- NIC: a received frame costs DMA and interrupt time ------------- f = Eth.read(inputs, "eth_rx_") if f is not None: self.inbox.append((now_us + self.dma_us, f)) due = [e for e in self.inbox if e[0] <= now_us] for e in due: self.inbox.remove(e) self.deliver(e[1], now_us) # --- frame boundary: the software starts a new cycle ---------------- if now_us % FRAME_PERIOD_US == 0: self.reset_frame(now_us // FRAME_PERIOD_US, now_us) self.act.enter("start", now_us) self.run_activity(now_us) if self.pending_tx is not None: self.pending_tx.write(outs, "eth_tx_") self.pending_tx = None def deliver(self, f, now_us): if f.ethertype != REPORT: return if f.src == ENDPOINT["ReceiverVendorA"]: self.buf_a = {"hash": f.hash, "frame": f.frame, "at": now_us} self.got_a = True elif f.src == ENDPOINT["ReceiverVendorB"]: self.b_frags += 1 self.buf_b = {"hash": f.hash, "frame": f.frame, "at": now_us, "frags": f.frag_total} if f.frag_index + 1 >= f.frag_total: self.got_b = True def run_activity(self, now_us): if self.act.node is None or self.act.is_busy(now_us): return node = self.act.node if node == "Await Receiver Reports": # A wait, not work: hold here until both reports have arrived or the # collection window closes. This is the only node whose duration is # data dependent, which is why it is bound to zero and justified. if self.got_a and self.got_b: self.act.enter("Window Expired", now_us) elif now_us - self.t0 >= WINDOW_US: self.act.enter("Window Expired", now_us) return if node == "Window Expired": complete = self.got_a and self.got_b guard = "both reports present" if complete else "either report missing" for (g, target) in self.act.successors(node): if g == guard: self.act.enter(target, now_us) if not complete: self.say(now_us, "frame %d: window closed at %d us with " "A=%s B=%s" % (self.frame, WINDOW_US, self.got_a, self.got_b)) return return if node in ("Issue Transmit Command", "Issue Degraded Command"): self.emit_command(node, now_us) if node == "done": self.act.node = None return for (g, target) in self.act.successors(node): if g: continue frags = self.b_frags if target == "Parse Receiver B Report" else 1 self.act.enter(target, now_us, fragments=max(1, frags)) return self.act.node = None def emit_command(self, node, now_us): a, b = self.buf_a, self.buf_b degraded = 0 hash_b = b["hash"] if node == "Issue Degraded Command": if STALE_DEFECT: # THE PLANTED DEFECT. The buffer is read without checking when # what is in it arrived, so a Vendor B report from the PREVIOUS # frame is fused with this frame's Vendor A report and the # message goes out marked complete. Nothing throws. No counter # moves. A bench test never sees it, because a stub receiver # always answers inside the window. degraded = 0 if b["frame"] >= 0 else 1 else: # Fix 1: reject anything not from this frame. if b["frame"] != self.frame: degraded = 1 hash_b = 0 self.drops += 1 self.pending_tx = Eth(src=ENDPOINT["MissionProcessor"], dst=ENDPOINT["Transmitter"], ethertype=COMMAND, nbytes=512, frame=self.frame, hsh=a["hash"], frag_index=hash_b, frag_total=degraded) self.say(now_us, "frame %d: command hash_a=%d hash_b=%d degraded=%d " "(b from frame %d)" % (self.frame, a["hash"], hash_b, degraded, b["frame"])) # ------------------------------------------------------------- the RF chain class Transmitter(Component): """Holds the command until its RF slot, and emits to the array. Also the source of the background telemetry that shares the switch with everything else, which is what makes run-to-run timing vary.""" name = "Transmitter" def __init__(self): Component.__init__(self) self.v = BLOCKS["Transmitter"] self.tlm_period = int(self.v["telemetry_period_us"]) self.tlm_bytes = int(self.v["telemetry_bytes"]) self.seq = 0 self.held = None self.ready_us = -1 def tick(self, step, now_us, inputs, outs): Eth.blank(outs, "eth_tx_") for f in ("valid", "frame", "hash_a", "hash_b", "emit_us", "degraded"): outs["rfOut_rf_" + f] = 0 f = Eth.read(inputs, "eth_rx_") if f is not None and f.ethertype == COMMAND: self.held = f self.ready_us = now_us self.say(now_us, "command for frame %d held for the slot" % f.frame) # The RF slot is the frame boundary: everything for frame N must be in # hand before it. if now_us % FRAME_PERIOD_US == 0 and self.held is not None: h = self.held outs["rfOut_rf_valid"] = 1 outs["rfOut_rf_frame"] = h.frame outs["rfOut_rf_hash_a"] = h.hash outs["rfOut_rf_hash_b"] = h.frag_index outs["rfOut_rf_emit_us"] = self.ready_us outs["rfOut_rf_degraded"] = h.frag_total self.held = None if TELEMETRY_ON and now_us > 0 and now_us % self.tlm_period == 0: self.seq = (self.seq + 1) % 251 Eth(src=ENDPOINT["Transmitter"], dst=ENDPOINT["MissionProcessor"], ethertype=TELEMETRY, nbytes=self.tlm_bytes, frame=now_us // FRAME_PERIOD_US, hsh=self.seq).write(outs, "eth_tx_") class CommunicationArray(Component): """The end of the chain. PR-01 is measured here, because this is where the message actually reaches the world. Receives only: RfTransmitIF's flow properties are all out, and this port is conjugated, so the component has no outputs at all.""" name = "CommunicationArray" def tick(self, step, now_us, inputs, outs): if int(inputs.get("rfIn_rf_valid", 0) or 0): self.say(now_us, "emission frame %d hash_a=%s hash_b=%s ready_at=%s " "degraded=%s" % (inputs.get("rfIn_rf_frame"), inputs.get("rfIn_rf_hash_a"), inputs.get("rfIn_rf_hash_b"), inputs.get("rfIn_rf_emit_us"), inputs.get("rfIn_rf_degraded"))) _IMPL = {c.name: c for c in (TargetScenario, ReceiverVendorA, ReceiverVendorB, RuggedizedSwitch, MissionProcessor, Transmitter, CommunicationArray)} def make(name): if name not in _IMPL: raise RuntimeError("no behaviour for %r" % name) return _IMPL[name]()