🚀 Rocket Lab Technical Screen Reference

📍 CHRONOLOGICAL CALL NAVIGATION & VISUAL TRIGGER MAP

Click any blue button to jump to that section. Every section has a [ ↩ Back ] button to return here.

1. The Ground Station Mindset: Plain-English Primer

• Conveyor Belt Analogy: Satellite passes transmit high-volume sensor packets over radio (downlink). Noise and packet drops happen constantly. Defensive code inspects every item, gently sifts out corrupted ones, and keeps the pipeline rolling.
• dict.get() vs dict[]: packet["voltage"] crashes with KeyError if key is absent. packet.get("voltage") returns None safely.
• The None Math Trap: You cannot do > or + on None. Always check if val is not None: before any math.
• The Empty Stream Trap: Dividing by a count of zero crashes. Always guard: if count > 0:.
• Batch O(N) vs Streaming O(1): Storing 50M raw packets in a list exhausts RAM. Streaming means maintaining only a running sum + count + max — three variables per subsystem, forever.

2. The 3 Clarifying Questions to Ask Before Coding

1. "Can packets arrive as None, or with missing keys like subsystem or voltage?"
2. "Can timestamps arrive out of order, or can the input list be completely empty?"
3. "Is this a discrete in-memory batch, or should I structure it for an unbounded stream?"

3. The 4 Ground Software Code Archetypes

Archetype 1 · Telemetry Aggregation & Anomaly Filter Scenario 1 process_bus_telemetry

Keywords: voltage, temperature, status, corrupt, missing, average, max, threshold, flag

The 5-Step Mental Model:

1. Guard empty input → return {}
2. Loop with .get() on every key
3. Skip: status != "OK" or voltage is None or not name
4. Accumulate: running sum, count, max per subsystem
5. Build output with avg and flagged = avg > 32.0

def process_bus_telemetry(packets):
    # SAY: "First thing — guard the empty case up front"
    if not packets:
        return {}

    subsystems = {}  # SAY: "Dict to accumulate per-subsystem running stats"

    for packet in packets:
        if not isinstance(packet, dict):    # SAY: "Skip corrupt non-dict entries from the downlink"
            continue

        # SAY: "Using .get() — never bracket access on untrusted data"
        name    = packet.get("subsystem")
        voltage = packet.get("voltage")
        status  = packet.get("status")

        # SAY: "Triple filter — skip ERROR status, None voltages, missing names"
        if status != "OK" or voltage is None or not name:
            continue

        # SAY: "Initialize new subsystem with first voltage as seed max"
        if name not in subsystems:
            subsystems[name] = {"sum": 0, "count": 0, "max": voltage}

        # SAY: "Running accumulation — O(1) per packet"
        subsystems[name]["sum"]   += voltage
        subsystems[name]["count"] += 1
        if voltage > subsystems[name]["max"]:
            subsystems[name]["max"] = voltage

    # SAY: "Build output dict — count > 0 is guaranteed since we only stored valid packets"
    result = {}
    for name, data in subsystems.items():
        avg = data["sum"] / data["count"]
        result[name] = {
            "max"    : data["max"],
            "avg"    : avg,
            "count"  : data["count"],
            "flagged": avg > 32.0  # SAY: "Flag subsystems above 32V threshold"
        }
    return result

Archetype 2 · Ground Station Contact Window Merging Scenario 2 merge_passes

Keywords: passes, schedules, contact windows, overlap, intervals, merge, antenna, LOS

The 3-Step Mental Model:

1. Guard empty → return []
2. Sort by start time
3. Walk through: if current_start <= prev_end → merge (extend end); else → append new block

def merge_passes(intervals):
    # Guard: empty input
    if not intervals:
        return []

    # Step 1: Sort chronologically by start time
    intervals.sort(key=lambda x: x[0])

    merged = [intervals[0][:]  ]    # copy of first window

    for start, end in intervals[1:]:
        prev_start, prev_end = merged[-1]

        if start <= prev_end:          # overlapping — extend
            merged[-1][1] = max(prev_end, end)
        else:                           # gap — new block
            merged.append([start, end])

    return merged

Key insight: Adjacent intervals like [1,5] and [6,10] are not overlapping — overlap requires start <= prev_end (strictly less than or equal).

Archetype 3 · Uplink Command Rate Limiter (Sliding Window) Scenario 3 CommandLimiter

Keywords: throttle, max commands, rate limit, sliding window, transceiver, uplink

The 3-Step Mental Model:

1. __init__: store max_cmds, window_sec, and a deque of timestamps
2. allow_command(t): purge timestamps older than t - window_sec
3. If len(deque) < max_cmds: append t → return True. Else False.

from collections import deque

class CommandLimiter:
    def __init__(self, max_cmds: int, window_sec: int):
        self.max_cmds   = max_cmds
        self.window_sec = window_sec
        self.timestamps = deque()          # O(1) popleft

    def allow_command(self, timestamp: float) -> bool:
        cutoff = timestamp - self.window_sec

        # Step 1: Purge expired timestamps from the front
        while self.timestamps and self.timestamps[0] <= cutoff:
            self.timestamps.popleft()

        # Step 2: Check capacity
        if len(self.timestamps) < self.max_cmds:
            self.timestamps.append(timestamp)
            return True

        return False

Why deque? popleft() is O(1). A list's pop(0) is O(N). At 100k commands/sec that matters.

Archetype 4 · Flight State Machine Sequence Validator Scenario 4 validate_sequence

Keywords: flight modes, stage separation, transitions, ABORT, illegal jump, state, sequence

The 4-Step Mental Model:

1. Guard: empty list → {"valid": False, "error": "No events"}
2. Guard: first event must be "PRE_LAUNCH"
3. Dict maps each state to allowed next states
4. Walk pairs: if next not in allowed → return error with index

VALID_TRANSITIONS = {
    "PRE_LAUNCH"      : ["BOOST"],
    "BOOST"           : ["STAGE_SEP"],
    "STAGE_SEP"       : ["COAST"],
    "COAST"           : ["PAYLOAD_DEPLOY"],
    "PAYLOAD_DEPLOY"  : ["MISSION_COMPLETE"],
    "MISSION_COMPLETE": []
}

def validate_sequence(events):
    if not events:
        return {"valid": False, "error": "No events provided"}

    if events[0] != "PRE_LAUNCH":
        return {"valid": False,
                "error": f"Must start with PRE_LAUNCH, got {events[0]}"}

    for i in range(1, len(events)):
        current = events[i - 1]
        nxt     = events[i]
        allowed = VALID_TRANSITIONS.get(current, [])  # safe: unknown state → []

        if nxt not in allowed:
            return {"valid": False,
                    "error": f"Invalid transition from {current} to {nxt} at index {i}"}

    return {"valid": True, "final_state": events[-1]}

Archetype 5 · Telemetry Packet Loss & Sequence Continuity Scenario 5 detect_packet_drops

Keywords: seq_num, sequence counter, monotonic, dropped frames, packet loss, RF drop, gap, loss_rate_pct, frame_gap

The 4-Step Mental Model:

1. Guard empty → return {received:0, dropped:0, loss_rate_pct:0.0, missing_seqs:[]}
2. Extract valid seq_nums with .get() + isinstance check, sort them
3. Span = (last_seq - first_seq) + 1 — this is expected count, not received
4. Convert to set(), loop range(first, last+1) — anything missing goes in missing[]

def detect_packet_drops(packets):
    # SAY: "Guard empty input first — can't index an empty list"
    if not packets:
        return {"received": 0, "dropped": 0, "loss_rate_pct": 0.0, "missing_seqs": []}

    # SAY: "Filter with .get() — skip None seq_nums and non-dict packets"
    valid_seqs = sorted([
        p.get("seq_num")
        for p in packets
        if isinstance(p, dict) and p.get("seq_num") is not None
    ])

    if not valid_seqs:
        return {"received": 0, "dropped": 0, "loss_rate_pct": 0.0, "missing_seqs": []}

    first_seq = valid_seqs[0]
    last_seq  = valid_seqs[-1]
    expected_count = (last_seq - first_seq) + 1   # SAY: "Span of expected IDs"
    received_set   = set(valid_seqs)               # SAY: "O(1) membership check"

    missing = []
    for num in range(first_seq, last_seq + 1):
        if num not in received_set:
            missing.append(num)

    dropped_count = len(missing)
    # SAY: "Loss rate over EXPECTED span — not over received"
    loss_pct = round((dropped_count / expected_count) * 100.0, 2)

    return {
        "received"     : len(valid_seqs),
        "dropped"      : dropped_count,
        "loss_rate_pct": loss_pct,
        "missing_seqs" : missing
    }

Key trap: Loss rate denominator = expected span (last − first + 1), not received count. 4 received out of span 5 → 20% loss, not 25%.

Curveball: "The seq_num wraps at 65535 (16-bit). How does your detector handle [65534, 65535, 0, 1]?" — Answer: naively it would report 65534 false drops. Fix: detect rollover with if last_seq < first_seq: last_seq += 65536.

Archetype 6 · Out-of-Limits (OOL) Telemetry Alarm Monitor Scenario 6 evaluate_telemetry_limits

Keywords: OOL, Out-of-Limits, yellow_low/high, red_low/high, WARNING, CRITICAL, alarm_severity, abort threshold, limit band

The 4-Step Mental Model:

1. Guard: if not telemetry_points or not limit_table: return []
2. Per point: isinstance check, .get() for sensor/value/ts, skip if unknown sensor or val is None
3. Red (CRITICAL) check FIRST — then elif Yellow (WARNING). Never check yellow before red.
4. Append alarm dict only if severity is not None

def evaluate_telemetry_limits(telemetry_points, limit_table):
    if not telemetry_points or not limit_table:
        return []

    alarms = []
    for point in telemetry_points:
        if not isinstance(point, dict):
            continue
        sensor = point.get("sensor")
        val    = point.get("value")
        ts     = point.get("timestamp")

        if sensor not in limit_table or val is None:
            continue

        limits   = limit_table[sensor]
        severity = None

        # SAY: "Red check FIRST — critical violations take precedence"
        if val < limits.get("red_low",  float('-inf')) or \
           val > limits.get("red_high", float('inf')):
            severity = "CRITICAL"
        elif val < limits.get("yellow_low",  float('-inf')) or \
             val > limits.get("yellow_high", float('inf')):
            severity = "WARNING"

        if severity:
            alarms.append({
                "sensor"  : sensor,
                "value"   : val,
                "severity": severity,
                "timestamp": ts
            })
    return alarms

Key trap: Use float('-inf') / float('inf') as .get() defaults so missing limit keys never raise a TypeError.

Curveball: "Transient spikes — alert only after 3 consecutive OOL readings." — Answer: add a consecutive = {} dict tracking per-sensor streak; only emit alarm when consecutive[sensor] >= 3.

Archetype 7 · Uplink Command ACK & Timeout Tracker Scenario 7 verify_command_acks

Keywords: ACK, command acknowledgment, uplink, telemetry_ack, round_trip_latency, command_timeout, CONFIRMED, REJECTED, TIMEOUT

The 3-Step Mental Model:

1. Guard: if not commands: return {}. Build ACK index: ack_map = {a["cmd_id"]: a for a in (acks or [])}
2. Loop commands with .get() — skip missing cmd_id or sent_time
3. If cmd_id not in ack_map → "TIMEOUT". Else check status == "REJECTED" first, then timing: ack_time - sent_time <= timeout_sec

def verify_command_acks(commands, acks):
    if not commands:
        return {}

    # SAY: "Index ACKs by cmd_id for O(1) lookup — never loop inside a loop"
    ack_map = {}
    for a in (acks or []):    # (acks or []) handles None acks safely
        if isinstance(a, dict) and "cmd_id" in a:
            ack_map[a["cmd_id"]] = a

    results = {}
    for cmd in commands:
        if not isinstance(cmd, dict):
            continue
        cid     = cmd.get("cmd_id")
        sent    = cmd.get("sent_time")
        timeout = cmd.get("timeout_sec", 5.0)

        if not cid or sent is None:
            continue

        if cid not in ack_map:
            results[cid] = "TIMEOUT"       # No ACK received at all
            continue

        ack        = ack_map[cid]
        ack_time   = ack.get("ack_time", float('inf'))
        ack_status = ack.get("status")

        # SAY: "REJECTED check before timing — status takes priority"
        if ack_status == "REJECTED":
            results[cid] = "REJECTED"
        elif ack_time - sent <= timeout:
            results[cid] = "CONFIRMED"
        else:
            results[cid] = "TIMEOUT"       # ACK arrived, but too late

    return results

Key trap: Use (acks or []) — if the caller passes acks=None, iterating over None raises a TypeError. The or [] idiom handles it silently.

Curveball: "Spacecraft sends two ACKs per command: Accepted + Executed. A REJECTED on Executed means it accepted but failed mid-execution." — Answer: store a list of ACKs per cmd_id; take the last ACK as the definitive status.

4. The 4 Fatal Bugs & Exact Fixes

5. Verbal Scripts — What to Say Out Loud While Coding

• Opening: "Before I type anything, let me trace the expected input shape and confirm how we handle corrupt or missing fields."
• Typing .get(): "Using .get() so missing telemetry keys don't throw an unhandled KeyError."
• Adding None guard: "Explicit None check here — missing sensor readings shouldn't break our numerical calculations."
• Before division: "Checking count is greater than zero before dividing to prevent a ZeroDivisionError on an empty pass."
• After passing tests: "All tests green. The main thing I'd add next is a staleness flag for sensors that go silent mid-pass."

Scale / O(1) Memory Script

6. High-Impact Questions to Ask Anh (Pick 2)

1. "How is the Ground Software team currently architecting the boundary between real-time binary ingestion from the ground stations and the WebSocket feeds pushing to mission control displays?"
2. "What does the handoff and telemetry schema synchronization look like between the Flight Software team on the vehicle and your Ground Software team?"
3. "How does the team handle packet ordering and replay when the ground station loses contact mid-pass and then reconnects?"