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.

1. Clarify sensors/events, robot command API, priorities, cancellation, and fault behavior.
2. Model explicit states: IDLE, PICKING, PLACING, FAULTED.
3. Put incoming work into per-conveyor queues; choose round-robin or oldest-first to prevent starvation.
4. Issue one idempotent command and transition only on matching completion events.
5. On timeout/fault: stop conveyors if required, record context, and require an explicit recovery policy.

from collections import deque
from dataclasses import dataclass

@dataclass(frozen=True)
class Pick:
    conveyor: int
    item_id: str

class CellScheduler:
    def __init__(self, conveyor_count: int) -> None:
        self.queues = [deque() for _ in range(conveyor_count)]
        self.next_conveyor = 0
        self.active: Pick | None = None

    def submit(self, pick: Pick) -> None:
        self.queues[pick.conveyor].append(pick)

    def dispatch(self) -> Pick | None:
        if self.active is not None:
            return None
        for offset in range(len(self.queues)):
            index = (self.next_conveyor + offset) % len(self.queues)
            if self.queues[index]:
                self.active = self.queues[index].popleft()
                self.next_conveyor = (index + 1) % len(self.queues)
                return self.active
        return None

    def complete(self, item_id: str) -> None:
        if self.active is None or self.active.item_id != item_id:
            raise ValueError("completion does not match active pick")
        self.active = None

s = CellScheduler(2)
s.submit(Pick(0, "a")); s.submit(Pick(1, "b"))
assert s.dispatch() == Pick(0, "a")
s.complete("a")
assert s.dispatch() == Pick(1, "b")

import math
from dataclasses import dataclass
from typing import Any

@dataclass(frozen=True)
class JointSample:
    timestamp_s: float
    positions_rad: tuple[float, ...]

def parse_sample(data: dict[str, Any], joint_count: int) -> JointSample:
    required = {"timestamp_s", "positions", "unit"}
    missing = required - data.keys()
    if missing:
        raise ValueError(f"missing fields: {sorted(missing)}")
    if data["unit"] != "rad":
        raise ValueError("unit must be 'rad'")
    try:
        timestamp = float(data["timestamp_s"])
        positions = tuple(float(x) for x in data["positions"])
    except (TypeError, ValueError) as exc:
        raise ValueError("timestamp and positions must be numeric") from exc
    if len(positions) != joint_count:
        raise ValueError(f"expected {joint_count} joints, got {len(positions)}")
    if not all(math.isfinite(x) for x in (timestamp, *positions)):
        raise ValueError("all numeric values must be finite")
    return JointSample(timestamp, positions)

x = parse_sample({"timestamp_s": 1, "positions": [0, 0.5], "unit": "rad"}, 2)
assert x.positions_rad == (0.0, 0.5)

Validate at system boundaries, return a typed internal representation, and preserve context in errors. Do not accept ambiguous units and “fix them later.”

def transpose_in_place(matrix: list[list[int]]) -> None:
    n = len(matrix)
    if any(len(row) != n for row in matrix):
        raise ValueError("in-place transpose requires a square matrix")
    for row in range(n):
        for col in range(row + 1, n):
            matrix[row][col], matrix[col][row] = matrix[col][row], matrix[row][col]

m = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
transpose_in_place(m)
assert m == [[1, 4, 7], [2, 5, 8], [3, 6, 9]]

O(n²) time and O(1) auxiliary space. Explain why only the upper triangle is visited; swapping every pair twice would undo the work.

1. Project input (B,T,D) into Q, K, V.
2. Reshape to (B,H,T,D/H) and transpose the head axis forward.
3. Compute QKᵀ / sqrt(head_dim); apply mask before softmax.
4. Multiply attention weights by V.
5. Transpose/reshape heads back to (B,T,D); apply output projection.

Key checks: D % H == 0; mask broadcasting; softmax across key positions; avoid accidentally mixing batch/head dimensions. Study conceptually only unless later rounds explicitly mention ML coding.

1. Requirements: objects, poses, throughput, accuracy, payload, variability, humans, failure budget.
2. Hardware: arm/reach/payload, gripper, conveyor/fixture, lighting, safety-rated devices.
3. Perception: fixed and/or wrist RGB-D, object pose/confidence, occlusion strategy.
4. Frames/calibration: camera intrinsics, camera-to-base extrinsics, tool center point, conveyor frame, time synchronization.
5. Runtime: detect → select grasp → plan → execute → verify; explicit state machine and timeouts.
6. Recovery: retry with new view/grasp, reject bin, human escalation, safe stop.
7. Validation: unit/integration/simulation/HIL; success rate stratified by object/pose; cycle time; collision and near-miss metrics.

Tie back to velocity: joint-velocity quality is useful for execution monitoring, stall detection, and verifying that commanded motion matches observed motion.

TCP: ordered, reliable byte stream with retransmission and congestion control; good for configuration, firmware, logs, and commands that must arrive exactly and in order. A lost packet can delay later bytes (head-of-line blocking).

UDP: message/datagram based, no built-in delivery/order guarantee; good when freshness is more valuable than retransmitting stale telemetry. Add sequence number, source timestamp, schema/version, integrity check, and loss/reorder metrics at the application layer.

Safety caveat: neither transport alone makes commands safe. Define authorization, idempotency, acknowledgment, timeouts, stale-command rejection, watchdogs, and a safe state.

1. Do not enable full-power motion immediately. Confirm part, gearing, encoder type/resolution, wiring, firmware, current/thermal limits, brakes, and mechanical installation.
2. Establish encoder zero/homing reference, direction sign, counts-per-revolution/gear ratio, joint offset, and hard/soft limits.
3. Verify position at known poses, then command very small low-current moves and confirm direction.
4. Measure velocity independently where possible; compare commanded vs observed sign and magnitude.
5. Tune or restore controller parameters only after kinematics/calibration are valid.
6. Run staged acceptance tests: no-load, limited workspace, increasing speed/load, thermal soak; preserve rollback and calibration provenance.

High-signal failure modes: reversed sign causing positive feedback, wrong gear ratio scaling velocity, stale zero offset, phase/commutation mismatch, and a limit configuration copied from the wrong joint.

from collections import deque

def shortest_path(grid: list[list[int]], start: tuple[int, int], goal: tuple[int, int]) -> int | None:
    rows, cols = len(grid), len(grid[0])
    queue = deque([(start, 0)])
    visited = {start}
    while queue:
        (row, col), distance = queue.popleft()
        if (row, col) == goal:
            return distance
        for dr, dc in ((1, 0), (-1, 0), (0, 1), (0, -1)):
            nxt = (row + dr, col + dc)
            r, c = nxt
            if 0 <= r < rows and 0 <= c < cols and grid[r][c] == 0 and nxt not in visited:
                visited.add(nxt)
                queue.append((nxt, distance + 1))
    return None

g = [[0, 1, 0], [0, 0, 0], [1, 0, 0]]
assert shortest_path(g, (0, 0), (2, 2)) == 4

BFS is optimal for equal-cost edges. Mark visited when enqueuing, not when dequeuing, to avoid duplicates. Real motion planning needs geometry, dynamics, collision margins, and often weighted/continuous methods.

import math

def tick_velocity(tick0: int, time0: float, tick1: int, time1: float,
                  counts_per_rev: int = 4096, gear_ratio: float = 5.0) -> float:
    dt = time1 - time0
    if dt <= 0 or counts_per_rev <= 0 or gear_ratio <= 0:
        raise ValueError("invalid time or encoder configuration")
    raw = tick1 - tick0
    half = counts_per_rev // 2
    delta_ticks = (raw + half) % counts_per_rev - half
    motor_radians = delta_ticks * (2 * math.pi / counts_per_rev)
    return motor_radians / gear_ratio / dt

# 100 motor ticks in 0.1 s, reduced by 5:1 at the output.
expected = 100 * 2 * math.pi / 4096 / 5 / 0.1
assert math.isclose(tick_velocity(4000, 0.0, 4, 0.1), expected)

Ask whether the counter can make more than half a revolution between samples; if so, two wrapped samples are insufficient to infer direction/count without a rate bound or unwrapped counter.

def pair_nearest(
    positions: list[tuple[float, float]],
    currents: list[tuple[float, float]],
    tolerance_s: float,
) -> list[tuple[tuple[float, float], tuple[float, float]]]:
    result = []
    j = 0
    for position in positions:
        t = position[0]
        while j + 1 < len(currents) and abs(currents[j + 1][0] - t) <= abs(currents[j][0] - t):
            j += 1
        if j < len(currents) and abs(currents[j][0] - t) <= tolerance_s:
            result.append((position, currents[j]))
            j += 1
        if j >= len(currents):
            break
    return result

p = [(0.010, 1.0), (0.030, 2.0)]
c = [(0.012, 4.0), (0.027, 5.0)]
assert len(pair_nearest(p, c, 0.005)) == 2

O(P+C) for sorted streams. Clarify whether reuse is allowed and whether timestamps share a clock. “Nearest” pairing can create causal problems online because the future sample may not have arrived yet.

def interpolate(t0: float, q0: float, t1: float, q1: float, t: float) -> float:
    if t1 <= t0:
        raise ValueError("bracket times must increase")
    if not t0 <= t <= t1:
        raise ValueError("requested time is outside the bracket")
    fraction = (t - t0) / (t1 - t0)
    return q0 + fraction * (q1 - q0)

assert interpolate(1.0, 10.0, 3.0, 20.0, 2.0) == 15.0

Linear interpolation assumes motion is approximately linear inside the interval. Angular wrap requires interpolation along the chosen shortest/continuous path, not raw values.

1. Catch only known transient exceptions, not every exception.
2. Use a small bounded retry count/deadline; record attempts and final failure.
3. Apply backoff only if it fits the measurement latency budget.
4. Return/raise an explicit unavailable result; never retimestamp cached position.
5. After repeated failures, trip a health state/watchdog and let a higher layer decide safe behavior.

import time
from collections.abc import Callable
from typing import TypeVar

T = TypeVar("T")

def read_with_retries(read: Callable[[], T], attempts: int = 3) -> T:
    if attempts < 1:
        raise ValueError("attempts must be positive")
    for attempt in range(attempts):
        try:
            return read()
        except TimeoutError:
            if attempt == attempts - 1:
                raise
            time.sleep(0.002 * (2 ** attempt))
    raise AssertionError("unreachable")

from collections import deque
from threading import Lock
from typing import Generic, TypeVar

T = TypeVar("T")

class LatestValue(Generic[T]):
    def __init__(self) -> None:
        self._items: deque[T] = deque(maxlen=1)
        self._lock = Lock()

    def publish(self, value: T) -> None:
        with self._lock:
            self._items.append(value)

    def read(self) -> T | None:
        with self._lock:
            return self._items[-1] if self._items else None

b = LatestValue[int]()
b.publish(1); b.publish(2)
assert b.read() == 2

This is correct for a display where freshness beats completeness. It is wrong for an audit log or estimator that requires every sample. State the data-loss policy explicitly and count overwritten samples if observability matters.

import math
from dataclasses import dataclass

@dataclass(frozen=True)
class EncoderCalibration:
    zero_count: int
    direction: int
    counts_per_rev: int
    gear_ratio: float

    def __post_init__(self) -> None:
        if self.direction not in (-1, 1):
            raise ValueError("direction must be -1 or 1")
        if self.counts_per_rev <= 0 or self.gear_ratio <= 0:
            raise ValueError("scale values must be positive")

    def radians(self, raw_count: int) -> float:
        motor_revs = self.direction * (raw_count - self.zero_count) / self.counts_per_rev
        return motor_revs * 2 * math.pi / self.gear_ratio

c = EncoderCalibration(100, -1, 4096, 4.0)
assert c.radians(100) == 0.0
assert math.isclose(c.radians(100 + 4096), -math.pi / 2)

Calibration metadata needs versioning and association with the exact joint/actuator. Mixing calibrated and raw values should be difficult at the API boundary.

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

    def update(self, t: float, q: float) -> float | None:
        if self.previous is None:
            self.previous = (t, q)
            return None
        old_t, old_q = self.previous
        dt = t - old_t
        if dt <= 0:
            raise ValueError("timestamps must increase")
        velocity = (q - old_q) / dt
        if abs(velocity) > self.max_speed:
            raise ValueError("sample implies physically impossible speed")
        self.previous = (t, q)
        return velocity

g = VelocityGate(4.0)
g.update(0.0, 0.0)
try: g.update(0.1, 100.0)
except ValueError: pass
assert g.update(0.5, 1.0) == 2.0

A gate can also reject legitimate fast motion if configuration is wrong. Report rejected samples and avoid treating the threshold as a proof of sensor failure.

def acceleration(v0: float, t0: float, v1: float, t1: float) -> float:
    dt = t1 - t0
    if dt <= 0:
        raise ValueError("timestamps must increase")
    return (v1 - v0) / dt

assert acceleration(1.0, 0.0, 3.0, 0.5) == 4.0

A second derivative amplifies noise even more strongly than velocity. For robust acceleration, consider fitting a quadratic to a window of positions and differentiating the fitted curve, or filter with a latency budget. Do not blindly difference already noisy two-point velocities.

class PID:
    def __init__(self, kp: float, ki: float, kd: float, output_limit: float) -> None:
        if output_limit <= 0:
            raise ValueError("output_limit must be positive")
        self.kp, self.ki, self.kd = kp, ki, kd
        self.limit = output_limit
        self.integral = 0.0
        self.previous_error: float | None = None

    def update(self, error: float, dt: float) -> float:
        if dt <= 0:
            raise ValueError("dt must be positive")
        derivative = 0.0 if self.previous_error is None else (error - self.previous_error) / dt
        candidate_integral = self.integral + error * dt
        unclamped = self.kp * error + self.ki * candidate_integral + self.kd * derivative
        output = max(-self.limit, min(self.limit, unclamped))
        # Conditional integration: accept it if not saturated, or if error drives out of saturation.
        if output == unclamped or output * error < 0:
            self.integral = candidate_integral
        self.previous_error = error
        return output

p = PID(2.0, 0.0, 0.0, 5.0)
assert p.update(10.0, 0.1) == 5.0

Real control demands more: derivative filtering, bumpless reset, actuator dynamics, saturation/rate limits, stability analysis, safety supervision, and deterministic timing. Do not claim this snippet is production-ready control.

1. Reference: compare against a calibrated external encoder/tachometer or high-rate position capture with quantified uncertainty.
2. Static: stationary output near zero across orientations/temperatures; characterize bias/noise.
3. Known profiles: constant positive/negative speeds, steps, ramps, and sinusoidal motion within safe limits.
4. Semantics: sign, rad/s vs deg/s, output-shaft vs motor-shaft, gear ratio, timestamp point.
5. Timing: latency, update rate, jitter, dropped/stale outputs, behavior during clock or communication faults.
6. Boundaries: wrap crossing, near-zero speed, maximum rated speed, startup, stop, reconnect, calibration reload.
7. Statistics: bias, MAE/RMSE, 95th/99th percentile error and latency; define pass thresholds before observing results.
8. Reproducibility: recorded inputs, calibration version, hardware serial, firmware, temperature/load, and automated regression report.

Separate the sensor/API’s measurement error from the external reference’s uncertainty and time-alignment error. A beautiful error plot is meaningless if the signals are shifted in time.

class VelocityCrossCheck:
    def __init__(self, tolerance: float, consecutive: int = 3) -> None:
        if tolerance < 0 or consecutive < 1:
            raise ValueError("invalid cross-check configuration")
        self.tolerance = tolerance
        self.consecutive = consecutive
        self.mismatch_count = 0

    def update(self, supplied: float, derived: float) -> bool:
        if abs(supplied - derived) > self.tolerance:
            self.mismatch_count += 1
        else:
            self.mismatch_count = 0
        return self.mismatch_count >= self.consecutive

c = VelocityCrossCheck(0.2, 2)
assert not c.update(1.5, 1.0)
assert c.update(1.5, 1.0)

Prefer the firmware value if its definition, units, timestamps, filtering, and latency satisfy the contract; it may use higher-rate encoder data. Deriving your own is valuable for validation and when only position is exposed. Comparing them requires time alignment—filter delay alone can create apparent disagreement during acceleration.

def elapsed_u32_us(previous: int, current: int) -> float:
    modulus = 1 << 32
    if not (0 <= previous < modulus and 0 <= current < modulus):
        raise ValueError("timestamps must be uint32 values")
    delta_us = (current - previous) % modulus
    if delta_us == 0:
        raise ValueError("zero elapsed device time")
    return delta_us / 1_000_000

assert elapsed_u32_us((1 << 32) - 100, 100) == 0.0002

This assumes fewer than one full counter period elapsed between samples. Without that bound, rollover count is unknowable from two values. For a 32-bit microsecond timer the full period is about 71.6 minutes.

import math

def unwrap_angles(angles: list[float], period: float = 2 * math.pi) -> list[float]:
    if not angles:
        return []
    result = [angles[0]]
    previous_wrapped = angles[0]
    for current in angles[1:]:
        delta = (current - previous_wrapped + period / 2) % period - period / 2
        result.append(result[-1] + delta)
        previous_wrapped = current
    return result

a = [math.radians(170), math.radians(-170), math.radians(-150)]
u = unwrap_angles(a)
assert all(math.isclose(math.degrees(x), y) for x, y in zip(u, [170, 190, 210]))

Unwrapping assumes true motion between samples is less than half a period. If the servo can move faster, sample faster or use a multi-turn encoder/turn counter.

1. Quantify one-tick angle: 2π / (counts_per_rev × gear_ratio).
2. Use a longer time/window slope so several samples or ticks contribute; report latency.
3. For very low speed, measure time between encoder edges (reciprocal method) instead of ticks per fixed interval.
4. Apply a zero-speed timeout when no edge arrives; otherwise the last reciprocal estimate can remain stale forever.
5. Return a quality indicator based on tick count/window duration.

Do not “fix” quantization by inventing sub-tick motion unless the estimator’s model and uncertainty are explicit. Larger windows reduce quantization variance at the price of slower response.

from collections import deque
from statistics import median

class RollingMedian:
    def __init__(self, window: int) -> None:
        if window < 1 or window % 2 == 0:
            raise ValueError("window must be a positive odd number")
        self.values: deque[float] = deque(maxlen=window)

    def update(self, value: float) -> float:
        self.values.append(value)
        return float(median(self.values))

m = RollingMedian(3)
assert [m.update(x) for x in (1.0, 100.0, 1.2, 1.1)][-1] == 1.2

A median is robust to isolated impulses but nonlinear and lagging. It can suppress a legitimate abrupt velocity change; use it when spike behavior matches the sensor fault model.

from collections import deque

class TimedWindowSlope:
    def __init__(self, duration_s: float) -> None:
        if duration_s <= 0:
            raise ValueError("duration must be positive")
        self.duration = duration_s
        self.samples: deque[tuple[float, float]] = deque()

    def update(self, t: float, q: float) -> float | None:
        if self.samples and t <= self.samples[-1][0]:
            raise ValueError("timestamps must increase")
        self.samples.append((t, q))
        cutoff = t - self.duration
        while len(self.samples) > 1 and self.samples[1][0] <= cutoff:
            self.samples.popleft()
        if len(self.samples) < 2:
            return None
        mean_t = sum(x for x, _ in self.samples) / len(self.samples)
        mean_q = sum(x for _, x in self.samples) / len(self.samples)
        denominator = sum((x - mean_t) ** 2 for x, _ in self.samples)
        return sum((x - mean_t) * (y - mean_q) for x, y in self.samples) / denominator

w = TimedWindowSlope(0.2)
assert w.update(0.0, 0.0) is None
assert w.update(0.1, 0.2) == 2.0

A time window provides consistent temporal smoothing under rate jitter. State the boundary policy: this version retains one sample at/before the cutoff when needed to maintain a useful span.

import asyncio
from collections.abc import AsyncIterator

async def async_velocities(
    samples: AsyncIterator[tuple[float, float]], timeout_s: float = 0.1
) -> AsyncIterator[float]:
    iterator = aiter(samples)
    previous: tuple[float, float] | None = None
    while True:
        try:
            current = await asyncio.wait_for(anext(iterator), timeout_s)
        except StopAsyncIteration:
            return
        if previous is not None:
            dt = current[0] - previous[0]
            if dt <= 0:
                raise ValueError("timestamps must increase")
            yield (current[1] - previous[1]) / dt
        previous = current

Let cancellation propagate; do not catch CancelledError unless cleanup requires it. Decide whether a timeout should end the stream, emit a health event, or reconnect—do not silently block forever.

1. Buffer arbitrary byte chunks; extract only complete frames using length/delimiter plus escaping rules.
2. Verify checksum/CRC before parsing fields or updating estimator state.
3. Validate schema/version, ranges, finite/unit conversions, and device timestamp progression.
4. Use sequence numbers to count loss, duplicates, and reordering; define wrap behavior.
5. Keep corrupted packets out of valid state. Report counters with last error context.
6. Convert ticks/time with calibration, then feed the pure estimator.

The interviewer may provide a simplified comma-separated line. Even then, handle partial reads at the adapter layer and keep packet validation separate from velocity math.

import math

def motor_rpm_to_output_rad_s(motor_rpm: float, reduction: float) -> float:
    if reduction <= 0:
        raise ValueError("reduction must be positive")
    motor_rad_s = motor_rpm * 2 * math.pi / 60
    return motor_rad_s / reduction

assert math.isclose(motor_rpm_to_output_rad_s(3000, 50), 2 * math.pi)

Write units through the calculation: rev/min × 2π rad/rev × 1 min/60 s ÷ gear ratio. Ask which direction the ratio is defined and whether sign comes from motor orientation.

from dataclasses import dataclass
from enum import Enum, auto

class Quality(Enum):
    STARTUP = auto()
    VALID = auto()
    STALE = auto()
    INVALID = auto()

@dataclass(frozen=True)
class VelocityReading:
    velocity_rad_s: float | None
    timestamp_s: float
    interval_s: float | None
    quality: Quality
    reason: str = ""

startup = VelocityReading(None, 1.0, None, Quality.STARTUP, "need a second sample")
assert startup.velocity_rad_s is None

A numeric zero must mean measured stationary—not “no estimate.” Quality metadata prevents downstream code from confusing missing/stale data with real zero velocity. Define staleness using source timestamp and the consumer’s clock relationship.