Measure velocity from a large CSV log
Read timestamp_ns,position_deg from a large CSV and write timestamp_ns,velocity_rad_s . Include useful errors for malformed rows.
Make the mechanism intuitive.
Stream rows with csv.DictReader . Convert nanoseconds to seconds and degrees to radians before differentiation.
For an interview, state the choice; do not silently discard data.
One decision at a time.
Policy question: fail fast versus skip-and-count malformed rows depends on the use case.
For an interview, state the choice; do not silently discard data.
Streaming gives O(1) auxiliary memory.
Make the contract visible in code.
import csv
import math
from typing import TextIO
def convert_log(source: TextIO, destination: TextIO) -> None:
reader = csv.DictReader(source)
required = {"timestamp_ns", "position_deg"}
if not required.issubset(reader.fieldnames or []):
raise ValueError(f"required columns: {sorted(required)}")
writer = csv.DictWriter(
destination, fieldnames=["timestamp_ns", "velocity_rad_s"]
)
writer.writeheader()
previous: tuple[int, float] | None = NoneFor an interview, state the choice; do not silently discard data.
Finish like an engineer.
Policy question: fail fast versus skip-and-count malformed rows depends on the use case. For an interview, state the choice; do not silently discard data. Streaming gives O(1) auxiliary memory.
invariant · policy · complexity · next step