Q5 - Fleet Telemetry: Normalize, Then Group Into Trips

Scenario: Millions of Tesla vehicles stream telemetry to the fleet backend. Different firmware versions emit the same signals under different short names and in different units, packed into compact frames. Before anything downstream can use the data, it has to be cleaned up and then organized.

Part 1 (normalize): Each raw frame is a string "<vin>@<epoch>|<sig>=<value><unit>|<sig>=<value><unit>|...". Turn a list of frames into clean readings. For each signal in each frame: map its name to the canonical signal, convert its value into the canonical unit, and drop any signal you do not recognize. Return the readings as [vin, epoch, signal, value].

canonical signal   aliases              units -> canonical
range               rng, range, r         km -> meters (x1000),  m -> meters (x1)
energy              nrg, energy, e         kwh -> watt-hours (x1000),  wh -> watt-hours (x1)
soc                 soc, batt, charge      pct -> percent (x1)

any other signal name (for example gps) is unknown and is dropped.

Input:  frames = [
          "KX9@160|range=405000m|nrg=2kwh|gps=abc",
          "TZ3@120|r=300km|e=1kwh"
        ]
Output: [
          ["KX9", 160, "range",  405000],   // 405000 m stays in meters
          ["KX9", 160, "energy", 2000],     // 2 kwh -> 2000 wh
          ["TZ3", 120, "range",  300000],   // r is range; 300 km -> 300000 m
          ["TZ3", 120, "energy", 1000]      // e is energy; 1 kwh -> 1000 wh
        ]
Why: gps is not a known signal, so it is dropped. Every value ends up in one unit.

Part 1: Clean the Feed

The parsing is mechanical once you see the shape of a frame: a header, then pipe-separated signal=valueunit fields. The only judgment calls are canonicalizing names, scaling units, and skipping anything unrecognized.

Step 1: Split a frame into its pieces

Split the frame on |. The first piece is the header vin@epoch; the rest are the signal fields. Split each field on = into a raw name and a raw value, then peel the numeric prefix off the value to separate the number from its unit.

"KX9@160|range=405000m|nrg=2kwh"
  header  -> "KX9@160"   ->  vin = "KX9",  epoch = 160
  field 1 -> "range=405000m"  ->  raw name "range", raw value "405000m"  ->  405000 + "m"
  field 2 -> "nrg=2kwh"       ->  raw name "nrg",   raw value "2kwh"     ->  2 + "kwh"

Step 2: Canonicalize the name, scale the unit

Look the raw name up in the alias table to get the canonical signal. If it is not there, the signal is unknown, so skip it. Then look the (signal, unit) pair up in the scale table; multiply the value by that factor to land in the canonical unit. An unrecognized unit for a known signal is also skipped.

"range=405000m":  alias(range) = range,  scale(range, m)  = 1     ->  405000
"nrg=2kwh":       alias(nrg)   = energy, scale(energy, kwh) = 1000 ->  2000
"gps=abc":        alias(gps)   = unknown                          ->  dropped

signal = ALIASES.get(raw_signal)
if signal is None:
    continue                            # unknown signal: drop it
scale = SCALE.get((signal, unit))
if scale is None:
    continue                            # unrecognized unit for this signal: drop it
readings.append(Reading(vin, epoch, signal, value * scale))

# O(total characters) time | O(readings) space
ALIASES = {"rng": "range", "range": "range", "r": "range",
           "nrg": "energy", "energy": "energy", "e": "energy",
           "soc": "soc", "batt": "soc", "charge": "soc"}
SCALE = {("range", "km"): 1000, ("range", "m"): 1,
         ("energy", "kwh"): 1000, ("energy", "wh"): 1,
         ("soc", "pct"): 1}

class Reading:
    def __init__(self, vin, epoch, signal, value):
        self.vin, self.epoch, self.signal, self.value = vin, epoch, signal, value

def normalize(frames):
    readings = []
    for frame in frames:
        parts = frame.split("|")                       # header, then signal fields
        vin, epoch_str = parts[0].split("@")
        epoch = int(epoch_str)
        for field in parts[1:]:
            raw_signal, raw_value = field.split("=")    # "range", "405000m"
            signal = ALIASES.get(raw_signal)
            if signal is None:
                continue                                # unknown signal: drop it
            digits = 0
            while digits < len(raw_value) and raw_value[digits].isdigit():
                digits += 1
            value, unit = int(raw_value[:digits]), raw_value[digits:]   # 405000, "m"
            scale = SCALE.get((signal, unit))
            if scale is None:
                continue                                # unrecognized unit: drop it
            readings.append(Reading(vin, epoch, signal, value * scale))
    return readings

Part 2: Group the Stream Into Trips

Part 2 (sessionize): The normalized readings arrive interleaved across the fleet and out of order. For each vehicle, split its readings into trips: a run of readings where each consecutive pair is at most maxGap apart in time. A gap larger than maxGap starts a new trip. Report each trip as [vin, start, end, count] (first time, last time, number of readings), sorted by vin then start.

Input:  readings for KX9 at times 160, 560, 100, 500, 220 (plus TZ3 at 120),
        maxGap = 100
Output: [
          ["KX9", 100, 220, 5],   // sorted times 100,160,220 form one trip (5 readings)
          ["KX9", 500, 560, 3],   // gap 220 -> 500 is 280 > 100, so a new trip
          ["TZ3", 120, 120, 1 or more]
        ]
Why: per vehicle, sort by time and cut wherever consecutive readings are more
        than maxGap apart. The count is how many readings fall inside each trip.

The mechanics: bucket the readings by vehicle, sort each bucket by time (this is what tames the out-of-order arrival), then sweep once, comparing each reading's time to the previous one. If the gap exceeds maxGap, close the current trip and open a new one; otherwise the trip grows.

KX9 sorted times: 100, 160, 220, 500, 560   maxGap = 100
  100 -> 160:  gap 60,  within limit   trip grows
  160 -> 220:  gap 60,  within limit   trip grows
  220 -> 500:  gap 280, over limit     close [100, 220], start new at 500
  500 -> 560:  gap 60,  within limit   trip grows
result: [100, 220] and [500, 560]

for epoch in epochs[1:]:
    if epoch - prev > max_gap:          # a gap over the limit ends the current trip
        result.append(Trip(vin, start, prev, count))
        start, count = epoch, 1
    else:
        count += 1
    prev = epoch

The slideshow traces one vehicle end to end. Watch the sorted readings on the time axis, and how the one oversized gap cuts the stream into two trips:

1 / 6

# O(n log n) time | O(n) space  (n readings; the sort dominates)
class Trip:
    def __init__(self, vin, start, end, count):
        self.vin, self.start, self.end, self.count = vin, start, end, count

def trips(readings, max_gap):
    by_vin = {}
    for r in readings:
        by_vin.setdefault(r.vin, []).append(r.epoch)
    result = []
    for vin in by_vin:
        epochs = sorted(by_vin[vin])                  # tame the out-of-order arrival
        start = prev = epochs[0]
        count = 1
        for epoch in epochs[1:]:
            if epoch - prev > max_gap:                # a gap over the limit ends the trip
                result.append(Trip(vin, start, prev, count))
                start, count = epoch, 1
            else:
                count += 1
            prev = epoch
        result.append(Trip(vin, start, prev, count))
    result.sort(key=lambda t: (t.vin, t.start))
    return result

Common Pitfalls

• Assuming the feed is clean. Aliases, mixed units, and unknown signals are the whole point of Part 1. Hard-code one signal name or one unit and half the fleet's data silently goes wrong.
• Floating-point unit conversion. Every conversion here is an exact integer scale, so keep it in integers. Reaching for floats invites rounding differences you then have to defend.
• Forgetting to sort before sessionizing. Readings arrive out of order, so a raw sweep would cut trips at the wrong places. Sort each vehicle's readings by time first; that one step is what makes the gap logic correct.
• Off-by-one on the gap test. maxGap is the largest gap allowed inside a trip, so a new trip starts only when the gap is strictly greater than maxGap. Using >= splits trips that should stay together.

Complexity Analysis

Part 1: O(C) time, where C is the total number of characters across all frames: each frame is scanned once and every lookup is O(1). Space is O(R) for the R readings produced.

Part 2: O(n log n) time for n readings, dominated by sorting each vehicle's readings by time; the sweep itself is linear. Space is O(n) for the per-vehicle buckets.

Final Notes

1. Two lookup tables do the cleaning. Aliases to a canonical name, (signal, unit) to an integer factor. Saying that structure out loud turns a messy prompt into two tiny maps and a scan.
2. Keep conversions in integers. Choosing unit scales that are exact integers (km -> m is x1000) sidesteps floating point entirely. Calling that out shows you are thinking about correctness, not just parsing.
3. Sorting is what tames out-of-order data. The sessionizing sweep is only correct on a time-sorted stream, so the sort per vehicle is the load-bearing step, and it is why Part 2 is O(n log n), not linear.
4. A gap threshold turns a stream into groups. "Cut wherever the gap exceeds the limit" is the whole sessionization idea, and getting the strict > versus >= right is the detail interviewers watch for.
5. This is a pipeline, and pipelines compose. Normalize, then group; the same two-stage shape (clean the feed, then aggregate) shows up across ingestion, logging, and analytics. Naming it signals you have built this before. 🚀