Get the servo velocity right.

Velocity is displacement divided by elapsed time. Signed displacement preserves direction.

Validate finite numeric inputs and require t1 > t0. Do not return infinity for a duplicate timestamp.

1. Define the units and signed-output contract.
2. Reject non-finite inputs so NaN cannot silently propagate.
3. Compute dt and reject zero/negative intervals.
4. Return Δposition / Δtime.

import math

def servo_velocity(p0: float, t0: float, p1: float, t1: float) -> float:
    values = (p0, t0, p1, t1)
    if not all(math.isfinite(x) for x in values):
        raise ValueError("positions and timestamps must be finite")
    dt = t1 - t0
    if dt <= 0:
        raise ValueError("timestamps must be strictly increasing")
    return (p1 - p0) / dt

assert servo_velocity(1.0, 10.0, 1.5, 10.25) == 2.0
assert servo_velocity(1.0, 0.0, 0.5, 0.25) == -2.0
assert servo_velocity(0.7, 3.0, 0.7, 4.0) == 0.0

Complexity: O(1) time and memory. Key interview statement: this is average velocity over the interval; with sufficiently close samples it approximates instantaneous velocity.

Keep only the previous valid sample. A generator makes the function work with lists, files, and live streams.

Decide whether a bad sample should replace the previous valid state. A safe default is to raise before updating it.

import math
from collections.abc import Iterable, Iterator

def velocities(
    samples: Iterable[tuple[float, float]],
) -> Iterator[tuple[float, float]]:
    previous: tuple[float, float] | None = None
    for timestamp, position in samples:
        if previous is None:
            previous = (timestamp, position)
            continue
        prev_t, prev_p = previous
        dt = timestamp - prev_t
        if dt <= 0:
            raise ValueError(f"non-increasing timestamp: {timestamp}")
        yield timestamp, (position - prev_p) / dt
        previous = (timestamp, position)

data = [(0.0, 1.0), (0.1, 1.2), (0.4, 1.8)]
result = list(velocities(data))
assert [t for t, _ in result] == [0.1, 0.4]
assert all(math.isclose(v, 2.0) for _, v in result)

Complexity: O(n) total time, O(1) auxiliary memory. Subtlety: the assignment to previous occurs after validation and calculation, so a rejected sample cannot corrupt valid state.

Use a monotonic clock for intervals. Separate one loop iteration from the infinite loop so it remains testable.

Repeatedly sleeping 20 ms after work makes the period “work time + 20 ms.” Schedule against an advancing deadline instead.

import time
from collections.abc import Callable

def monitor_servo(
    read_position_rad: Callable[[], float],
    rate_hz: float = 50.0,
) -> None:
    if rate_hz <= 0:
        raise ValueError("rate_hz must be positive")
    period = 1.0 / rate_hz
    previous: tuple[float, float] | None = None
    deadline = time.monotonic()

    try:
        while True:
            # Best case: hardware returns position and capture time atomically.
            position = read_position_rad()
            now = time.monotonic()

            if previous is not None:
                prev_t, prev_p = previous
                dt = now - prev_t
                if dt > 0:
                    print(f"{(position - prev_p) / dt:.4f} rad/s")
            previous = (now, position)

            deadline += period
            delay = deadline - time.monotonic()
            if delay > 0:
                time.sleep(delay)
            else:
                # We missed one or more periods; don't burst to catch up.
                deadline = time.monotonic()
    except KeyboardInterrupt:
        return

Tradeoff to discuss: ordinary Python cannot promise hard real-time behavior. This is adequate for a practical measurement/telemetry loop, but a control/safety loop may need hardware timestamps, a real-time process, bounds, and explicit missed-deadline metrics.

Convert the positions to an immutable tuple at the boundary so a caller cannot mutate stored state.

class JointVelocityEstimator:
    def __init__(self, joint_count: int) -> None:
        if joint_count <= 0:
            raise ValueError("joint_count must be positive")
        self.joint_count = joint_count
        self.previous: tuple[float, tuple[float, ...]] | None = None

    def update(self, timestamp: float, positions: list[float]) -> tuple[float, ...] | None:
        current = tuple(positions)
        if len(current) != self.joint_count:
            raise ValueError(
                f"expected {self.joint_count} joints, got {len(current)}"
            )
        if self.previous is None:
            self.previous = (timestamp, current)
            return None
        prev_t, prev_q = self.previous
        dt = timestamp - prev_t
        if dt <= 0:
            raise ValueError("timestamps must increase")
        result = tuple((q - old_q) / dt for q, old_q in zip(current, prev_q))
        self.previous = (timestamp, current)
        return result

est = JointVelocityEstimator(3)
assert est.update(0.0, [1.0, 2.0, 3.0]) is None
assert est.update(0.5, [2.0, 2.0, 2.0]) == (2.0, 0.0, -2.0)

Clarify whether the interviewer wants per-joint velocities, speed magnitudes, or end-effector velocity. Joint velocities do not directly equal Cartesian end-effector velocity; that conversion involves the manipulator Jacobian and is a different problem.

Normalize the raw difference into [−period/2, period/2).

import math

def wrapped_delta(current: float, previous: float, period: float) -> float:
    if period <= 0:
        raise ValueError("period must be positive")
    return (current - previous + period / 2) % period - period / 2

def wrapped_velocity(p0: float, t0: float, p1: float, t1: float) -> float:
    dt = t1 - t0
    if dt <= 0:
        raise ValueError("timestamps must increase")
    return wrapped_delta(p1, p0, 2 * math.pi) / dt

eps = 1e-12
v = wrapped_velocity(math.radians(179), 0, math.radians(-179), 1)
assert abs(v - math.radians(2)) < eps
v = wrapped_velocity(math.radians(-179), 0, math.radians(179), 1)
assert abs(v - math.radians(-2)) < eps

Ambiguous boundary: exactly half a period has two equivalent directions; this formula chooses the negative direction. Document it if that case matters. Also ask whether the hardware already provides unwrapped angles.

Stream rows with csv.DictReader. Convert nanoseconds to seconds and degrees to radians before differentiation.

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 = None

    for line_number, row in enumerate(reader, start=2):
        try:
            timestamp_ns = int(row["timestamp_ns"])
            position_rad = math.radians(float(row["position_deg"]))
        except (TypeError, ValueError) as exc:
            raise ValueError(f"invalid row at line {line_number}") from exc

        if previous is not None:
            prev_ns, prev_rad = previous
            dt_s = (timestamp_ns - prev_ns) / 1_000_000_000
            if dt_s <= 0:
                raise ValueError(f"time did not increase at line {line_number}")
            writer.writerow({
                "timestamp_ns": timestamp_ns,
                "velocity_rad_s": (position_rad - prev_rad) / dt_s,
            })
        previous = (timestamp_ns, position_rad)

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.

For a line q = a + vt, compute v = Σ(t−t̄)(q−q̄) / Σ(t−t̄)².

Center timestamps before multiplication. Use a bounded deque. Need at least two distinct timestamps.

from collections import deque

class WindowSlope:
    def __init__(self, window_size: int) -> None:
        if window_size < 2:
            raise ValueError("window_size must be at least 2")
        self.samples: deque[tuple[float, float]] = deque(maxlen=window_size)

    def update(self, timestamp: float, position: float) -> float | None:
        if self.samples and timestamp <= self.samples[-1][0]:
            raise ValueError("timestamps must increase")
        self.samples.append((timestamp, position))
        if len(self.samples) < 2:
            return None

        mean_t = sum(t for t, _ in self.samples) / len(self.samples)
        mean_q = sum(q for _, q in self.samples) / len(self.samples)
        numerator = sum(
            (t - mean_t) * (q - mean_q) for t, q in self.samples
        )
        denominator = sum((t - mean_t) ** 2 for t, _ in self.samples)
        if denominator == 0:
            return None
        return numerator / denominator

est = WindowSlope(4)
assert est.update(10.0, 21.0) is None
assert abs(est.update(10.1, 21.2) - 2.0) < 1e-12
assert abs(est.update(10.4, 21.8) - 2.0) < 1e-12

Tradeoff: a larger window reduces variance but increases lag when velocity changes. This simple implementation is O(N) per sample; for a bounded small window that may be the right clarity/performance tradeoff. Running sums can make it O(1), but are easier to get wrong.

Use alpha = 1 - exp(-dt / tau). This adapts the weight to actual elapsed time.

import math

class SmoothedVelocity:
    def __init__(self, time_constant_s: float) -> None:
        if time_constant_s <= 0:
            raise ValueError("time constant must be positive")
        self.tau = time_constant_s
        self.previous_sample: tuple[float, float] | None = None
        self.filtered: float | None = None

    def update(self, timestamp: float, position: float) -> float | None:
        if self.previous_sample is None:
            self.previous_sample = (timestamp, position)
            return None
        prev_t, prev_q = self.previous_sample
        dt = timestamp - prev_t
        if dt <= 0:
            raise ValueError("timestamps must increase")
        raw = (position - prev_q) / dt
        alpha = 1.0 - math.exp(-dt / self.tau)
        self.filtered = (
            raw if self.filtered is None
            else self.filtered + alpha * (raw - self.filtered)
        )
        self.previous_sample = (timestamp, position)
        return self.filtered

Interpretation: after one time constant, the filter moves about 63% toward a step input. Smaller τ responds faster but passes more noise. Filtering velocity after differentiation is simple; filtering position before differentiating has different phase/noise behavior.

Track when the threshold was first crossed. Use different entry and exit thresholds to avoid chatter.

class OverspeedDetector:
    def __init__(self, enter: float, exit: float, dwell_s: float) -> None:
        if not (0 < exit < enter and dwell_s >= 0):
            raise ValueError("require 0 < exit < enter and nonnegative dwell")
        self.enter = enter
        self.exit = exit
        self.dwell_s = dwell_s
        self.above_since: float | None = None
        self.active = False

    def update(self, timestamp: float, velocity: float) -> bool:
        speed = abs(velocity)
        if self.active:
            if speed < self.exit:
                self.active = False
                self.above_since = None
            return self.active

        if speed > self.enter:
            if self.above_since is None:
                self.above_since = timestamp
            if timestamp - self.above_since >= self.dwell_s:
                self.active = True
        else:
            self.above_since = None
        return self.active

Important: this reports a condition; it does not command or stop hardware. Measurement, alerting, and safety actuation should remain explicit boundaries unless the prompt asks you to integrate them.

Inject time and position sources. Do not call time.sleep inside the pure estimator test.

from collections.abc import Callable

def take_measurement(
    read_time: Callable[[], float],
    read_position: Callable[[], float],
    previous: tuple[float, float] | None,
) -> tuple[tuple[float, float], float | None]:
    timestamp = read_time()
    position = read_position()
    current = (timestamp, position)
    if previous is None:
        return current, None
    prev_t, prev_q = previous
    dt = timestamp - prev_t
    if dt <= 0:
        raise ValueError("timestamps must increase")
    return current, (position - prev_q) / dt

class FakeMotion:
    def __init__(self, velocity: float) -> None:
        self.t = 0.0
        self.velocity = velocity
    def time(self) -> float:
        return self.t
    def position(self) -> float:
        return self.velocity * self.t

fake = FakeMotion(3.0)
state, velocity = take_measurement(fake.time, fake.position, None)
assert velocity is None
fake.t = 0.2
state, velocity = take_measurement(fake.time, fake.position, state)
assert abs(velocity - 3.0) < 1e-12

In production, prefer an atomic read_sample() returning both capture time and position. The separate callbacks here illustrate dependency injection, but also expose possible timestamp/position skew.

For a simpler interview version, ask whether (q[i+1]-q[i-1])/(t[i+1]-t[i-1]) is acceptable. It is a secant across neighbors, but is not the exact three-point nonuniform quadratic formula.

Let h0=t[i]-t[i-1], h1=t[i+1]-t[i]. The derivative weights are -h1/(h0*(h0+h1)), (h1-h0)/(h0*h1), and h0/(h1*(h0+h1)).

def offline_velocity(times: list[float], positions: list[float]) -> list[float]:
    if len(times) != len(positions) or len(times) < 2:
        raise ValueError("need equal arrays with at least two samples")
    if any(b <= a for a, b in zip(times, times[1:])):
        raise ValueError("times must be strictly increasing")

    n = len(times)
    result = [0.0] * n
    result[0] = (positions[1] - positions[0]) / (times[1] - times[0])
    result[-1] = (positions[-1] - positions[-2]) / (times[-1] - times[-2])

    for i in range(1, n - 1):
        h0 = times[i] - times[i - 1]
        h1 = times[i + 1] - times[i]
        w_prev = -h1 / (h0 * (h0 + h1))
        w_here = (h1 - h0) / (h0 * h1)
        w_next = h0 / (h1 * (h0 + h1))
        result[i] = (
            w_prev * positions[i - 1]
            + w_here * positions[i]
            + w_next * positions[i + 1]
        )
    return result

# q=t^2; exact derivative is 2t at the interior even with uneven samples.
t = [0.0, 1.0, 3.0]
q = [x * x for x in t]
v = offline_velocity(t, q)
assert abs(v[1] - 2.0) < 1e-12

Interview judgment: do not volunteer this formula unless the prompt needs it. For a live servo, it adds future-sample latency. The two-point estimator is usually the correct starting point.

Compute consecutive intervals. Jitter is dt - target_period. Estimate missed periods from how many target periods fit in a gap.

import math
from dataclasses import dataclass

@dataclass(frozen=True)
class TimingHealth:
    mean_rate_hz: float
    max_interval_s: float
    rms_jitter_s: float
    estimated_missed: int

def timing_health(times: list[float], target_hz: float) -> TimingHealth:
    if len(times) < 2 or target_hz <= 0:
        raise ValueError("need two timestamps and a positive target rate")
    intervals = [b - a for a, b in zip(times, times[1:])]
    if any(dt <= 0 for dt in intervals):
        raise ValueError("timestamps must increase")
    target = 1.0 / target_hz
    mean_dt = sum(intervals) / len(intervals)
    rms_jitter = math.sqrt(
        sum((dt - target) ** 2 for dt in intervals) / len(intervals)
    )
    # A 2.1-period gap implies one likely missing sample.
    missed = sum(max(0, round(dt / target) - 1) for dt in intervals)
    return TimingHealth(1.0 / mean_dt, max(intervals), rms_jitter, missed)

h = timing_health([0.00, 0.02, 0.04, 0.08, 0.10], 50.0)
assert h.estimated_missed == 1
assert abs(h.mean_rate_hz - 40.0) < 1e-12

Nuance: “missed” is an estimate because timestamps alone do not prove whether the producer dropped a sample or intentionally paused. Production observability should distinguish source sequence numbers, acquisition failures, and consumer backpressure when possible.