diff --git a/plans/rover learning service.md b/plans/rover learning service.md index d8104f55..be615461 100644 --- a/plans/rover learning service.md +++ b/plans/rover learning service.md @@ -2,6 +2,19 @@ ## Resume here +Latest repair: full learning-path audit. Fixed missing command labels between +video samples with a shared causal 100ms grid; added real-data score/baseline/ +training-probe diagnostics and normalized saved control profiles for retained +recordings. Continues policy 5 / lineage v6; this audit does not reset progress. +See the final audit results section for measured synthetic evidence and limits. + +Latest scope (authorized IPI): only drive, camera tilt, headlight and the four +operating-mode/dock commands (128/131/132/143). Exclude horn, song, laser and all +brush/vacuum motor commands from both learning and execution, including raw OI +bypasses. Raw recordings stay intact. Policy format 5, training lineage v6; +startup removes earlier models and lineages through v5. Earlier scope below is +historical. No configuration change required. + Latest change (2026-09-29 IPI): format 4 seeds known user-control command shapes before training. Generic peripherals are excluded; sensorStream is ignored without invalidating footage. Drive includes zero-speed drive, with no separate stop type. @@ -962,3 +975,103 @@ and browser appearance remain unverified. No new tests or background processes. Deploying starts a new learner using existing recordings; enabled-service startup removes older policy models and optimizer/scheduler lineages through v4. Current lineage is v5. No settings changes or prescribed human driving are required. + +### Simplified controls, policy 5 + +User authorized removal of horn/song/laser/brush motors. Retained 13 seeded shapes: +four drive zero/nonzero combinations, two camera angle shapes, three headlight +actions, four raw operating-mode/dock constants. Generic peripherals remain excluded. +Removed excluded command history/targets via policy allowlist and execution via +Node allowlist; raw payloads must exactly match one of the four mode commands at +both boundaries. Controller shutdown now sends only zero-speed drive, leaving +manual brush/horn state alone. Raw recording and sensor telemetry unchanged. +Old policies are deleted on enabled startup; resume-v6.pt/schedule-v6.json begin a +fresh learner using retained recordings. Existing evaluation behavior unchanged. +Verification passed: production train.py --check-dependencies (policy 5, 13 seeded +shapes), infer.py --help, changed Node module syntax, focused card ESLint, and diff +whitespace checks. No live hardware/data available locally; actual learning and +driving quality remain unverified. No tests added or background processes left. + +### Active full learning-path audit (authorized IPI) + +User explicitly permits local training/inference with controlled synthetic input; +there are NO real recordings on this machine. Existing user-control simplification +edits were present at start and must be preserved. No new schema reset is planned. + +Reproduced: production reader with a generated lossless 25 FPS video and 31 valid +human drive events labeled ZERO commands. Old sampler used 120ms frame intervals +with only 100ms target coverage. Fixed causal 100ms grid shared by dataset/inference +uses latest past image; no future pixels. The repaired fixture preserves all 31 +commands. Sensor cutoff is strictly before decision in both paths. + +Original objective experiment: 12 easy colored-frame samples, 120 steps, seed 7, +4 threads: loss 0.8206 -> 0.1996, mean true-command score 0.5106 -> 0.7268, recall +at 0.7 reached 2/3; at 80 steps recall was still zero. This disproves a wholly +broken gradient path; it does not diagnose real-data score suppression. Keep +existing loss/threshold until real-data diagnostics justify changing them. + +New learning.py shares the actual objective for ongoing training and disposable +probe. Per-job before/after evaluation on <=96 unchanged training examples, full +held-out score distributions, threshold sweep, training-only frequency and no-op +baselines, decoded pre-server-limit parameter error, gradient norms and sampled +versus natural frequencies. Bounded <=20s/80-step memorization probe on <=12 +training examples at first job/every 1000 steps; copy only, never published or +used for generalization gate. Continues current checkpoint/optimizer lineage. + +Audit results (synthetic only): +- Round-tripped eight representative retained command payloads, including signed + drive/servo values, zero-speed drive, headlight and all four mode commands. +- Lossless 25 FPS recording with 31 command events: old sampler labeled 0; new + sampler labeled 31/31, including interleaved ignored sensorStream events. +- Full production train.py decoded five generated recordings, trained/exported + checkpoints, evaluated held-out sessions, and resumed optimizer/weights for a + second job. 120 -> 240 cumulative steps improved recall at unchanged 0.7 from + 0 to 0.2553, with precision 1.0 at step 240. Mean true-event score rose 0.5346 + -> 0.6303. Gate correctly remained failed. A lower 0.5 threshold had precision + 0.6629/recall 0.8369 and 120 false-positive actions, so threshold was NOT lowered. +- Disposable diverse probe covered zero-speed drive, moving drive and servo; + reached recall 0.75 after 39 steps under its 20s budget. Sampling was corrected + to avoid probing only the first event category. Probe uses a copy and higher + learning rate for memorization; never adopts weights or changes main optimizer. +- Exported trained checkpoint reload, forward inference and decoding passed. +- Retained recordings had extra removed actuator fields, causing false model + incompatibility. Normalize saved and current profiles through controlProfile. + Direct Node exercise accepts removed laser/peripheral differences and still + rejects a changed retained camera configuration. + +Diagnostics now report accepted vs represented command counts, request/dispatch +delay, maximum sensor/image age; sampled vs natural command frequencies; gradient +norms; same training-subset before/after scores; per-command score distributions; +threshold sweep and no-op/training-frequency baselines; pre-server-limit decoded +parameter errors; and bounded diverse memorization probe. Evaluation-step identity +separates last evaluation from current-job progress. Compact history avoids +repeating entire command/score tables for every job. Four concise numeric rows +were added to expanded details; full evidence is in Copy diagnostics. + +The original loss and configured 0.7 threshold remain unchanged because experiments +did not establish that a different objective/threshold fixes real recordings. +Promotion keeps existing checks and additionally must beat the simple training- +frequency baseline F1. One-time retry of old reader-specific no-aligned-window +rejections preserves optimizer state and unrelated errors. + +Limits: actual human footage and rover/RTSP transport unavailable. Isolated held-out +windows cannot measure sequential latch behavior or physical driving; decoded +parameter errors are explicitly before Node's rover limits/permission/safety +handling. Main model learned some synthetic cues but has not demonstrated useful +autonomous driving. Camera command encoding still uses whole-degree predictions. +The user must not be asked to drive prescribed routes or perform a manual checklist. + +Verification: full synthetic worker jobs and checkpoint inference above passed; +focused Node syntax/card ESLint/diff checks passed before final report additions. +One experimental seeded run failed because its runpy launcher omitted worker +import path; corrected launcher succeeded. No production import failure occurred. +Synthetic inference-worker media exercise passed: eight prediction batches, +threshold update to 0.3, retained command types and discrete latching exercised. +This check used a generated checkpoint and a FIFO media adapter preserving +synthetic PTS (RTSP wall-clock override disabled in the experiment only); it does +not validate real network capture timing. Trained checkpoint forward/decode was +verified separately. Temporary media launchers needed fixes for file lifetime +and command-string escaping; these were experiment issues, not production fixes. +No repository tests added. Experiments finished and no background worker remains. +Final focused Node syntax and diff whitespace checks passed. Card lint passed. +No additional schema/optimizer reset introduced by this audit. diff --git a/server/src/services/roverLearningService/capabilities.js b/server/src/services/roverLearningService/capabilities.js index a3c2b0f1..136816c2 100644 --- a/server/src/services/roverLearningService/capabilities.js +++ b/server/src/services/roverLearningService/capabilities.js @@ -1,16 +1,18 @@ // Match actuator configuration, not a rover's display name or changing telemetry. const { isDeepStrictEqual } = require('node:util'); -const POLICY_VERSION = 4; +const POLICY_VERSION = 5; function controlProfile(meta = {}) { - return Object.fromEntries(['cameraServo', 'headlight', 'laser', 'roomba', 'platform'] + return Object.fromEntries(['cameraServo', 'headlight', 'roomba', 'platform'] .map((key) => [key, meta?.[key] ?? null])); } function compatible(model, record) { if (model?.specification?.version !== POLICY_VERSION || !record) return false; const profile = controlProfile(record.meta); - return Array.isArray(model.controlProfiles) && model.controlProfiles.some((saved) => isDeepStrictEqual(saved, profile)); + // Retained recordings may include removed actuator fields. Compare only + // controls this policy actually uses, on both sides of the boundary. + return Array.isArray(model.controlProfiles) && model.controlProfiles.some((saved) => isDeepStrictEqual(controlProfile(saved), profile)); } module.exports = { POLICY_VERSION, controlProfile, compatible }; diff --git a/server/src/services/roverLearningService/diagnostics.js b/server/src/services/roverLearningService/diagnostics.js index 9c9703e0..780be8ce 100644 --- a/server/src/services/roverLearningService/diagnostics.js +++ b/server/src/services/roverLearningService/diagnostics.js @@ -7,7 +7,7 @@ const path = require('node:path'); const { createHash } = require('node:crypto'); const applicationVersion = require('../../../package.json').version; const hash = createHash('sha256'); -for (const file of ['workers/policy.py', 'workers/dataset.py', 'workers/train.py', 'workers/infer.py', +for (const file of ['workers/policy.py', 'workers/dataset.py', 'workers/train.py', 'workers/infer.py', 'workers/observations.py', 'workers/learning.py', 'capabilities.js', 'errors.js', 'storage.js', 'runtime.js', 'models.js', 'driving.js', 'recording.js', 'training.js', 'socketGateway.js', 'diagnostics.js']) { hash.update(fs.readFileSync(path.join(__dirname, file))); } @@ -41,11 +41,12 @@ function report(state, controller) { sessions: state.recording.sessions.map((item) => ({ startedAtUnixMs: item.startedAt, status: item.video, droppedEvents: item.droppedEvents, error: redact(item.error), failureDetails: item.failureDetails || null, stderr: redact(item.stderr), errorCategory: errorCategory(item.error) })) }, training: { status: training.status, steps: training.steps, examples: training.examples, - selection: training.selection || null, + selection: training.selection || null, evaluatedSteps: training.evaluatedSteps ?? null, loss: training.loss, losses: training.losses || null, publication: training.publication || null, + trainingAudit: metrics.trainingAudit || null, evaluation: training.evaluation || null, dataset: metrics.dataset || training.dataset || null, resources: { decodeSeconds: metrics.decodeSeconds ?? null, trainingSeconds: metrics.trainingSeconds ?? null, - evaluationSeconds: metrics.evaluationSeconds ?? null, checkpointSeconds: metrics.checkpointSeconds ?? null, cpuSeconds: metrics.cpuSeconds ?? null, + evaluationSeconds: metrics.evaluationSeconds ?? null, probeSeconds: metrics.probeSeconds ?? null, beforeAuditSeconds: metrics.beforeAuditSeconds ?? null, checkpointSeconds: metrics.checkpointSeconds ?? null, cpuSeconds: metrics.cpuSeconds ?? null, peakRssMiB: metrics.peakRssMiB ?? null }, history: training.history || [], rejectionReasons: (training.rejectionReasons || []).map(redact), rejectionCategories: (training.rejectionReasons || []).map(errorCategory), error: redact(training.error), errorCategory: errorCategory(training.error), nextAttemptAtUnixMs: training.nextAttemptAt }, diff --git a/server/src/services/roverLearningService/driving.js b/server/src/services/roverLearningService/driving.js index e4ffded3..37be153f 100644 --- a/server/src/services/roverLearningService/driving.js +++ b/server/src/services/roverLearningService/driving.js @@ -9,7 +9,8 @@ const { recordCommandRequest } = require('../commandService/recording'); const { acquireModel } = require('./models'); const { compatible } = require('./capabilities'); -const OPERATING = new Set(['drive', 'motors', 'servo', 'headlight', 'laser', 'horn', 'song', 'raw']); +const OPERATING = new Set(['drive', 'servo', 'headlight', 'raw']); +const MODE_COMMANDS = new Set([128, 131, 132, 143].map((opcode) => Buffer.from([opcode]).toString('base64'))); const clamp = (value, min, max) => Math.max(min, Math.min(max, Number(value) || 0)); function canControl(socket, roverId) { @@ -34,14 +35,12 @@ function createDriving({ root, config, onChange, activity }) { if (session.status === 'stopping') return; if (!canControl(session.socket, session.roverId)) { stop(session.roverId, 'Control permission lost'); return; } if (!OPERATING.has(command?.type)) throw new Error('Unsupported model command'); + if (command.type === 'raw' && !MODE_COMMANDS.has(command.raw)) throw new Error('Unsupported model operating-mode command'); const record = roverManager.rovers.get(session.roverId); if (command.type === 'drive') { const limit = Math.min(500, Number(record.meta?.maxWheelSpeed) || 500); command.driveDirect = { left: Math.round(clamp(command.driveDirect?.left, -limit, limit)), right: Math.round(clamp(command.driveDirect?.right, -limit, limit)) }; } - if (command.type === 'motors') { - command.motorPwm = { main: clamp(command.motorPwm?.main, -127, 127), side: clamp(command.motorPwm?.side, -127, 127), vacuum: clamp(command.motorPwm?.vacuum, 0, 127) }; - } if (command.type === 'servo') { const servo = record.meta?.cameraServo; if (!servo?.enabled) { session.submissions.push({ command, result: 'Servo unavailable' }); return; } @@ -64,8 +63,6 @@ function createDriving({ root, config, onChange, activity }) { // A stop must still reach the rover after turn permission is revoked. for (const command of [ { type: 'drive', driveDirect: { left: 0, right: 0 } }, - { type: 'motors', motorPwm: { main: 0, side: 0, vacuum: 0 } }, - { type: 'horn', horn: { action: 'stop' } }, ]) { try { issueCommand(roverId, command); } catch { /* Rover may already be offline. */ } } session.worker.kill('SIGTERM'); const timer = setTimeout(() => session.worker.kill('SIGKILL'), 2000); diff --git a/server/src/services/roverLearningService/models.js b/server/src/services/roverLearningService/models.js index ef891c28..8e37e0df 100644 --- a/server/src/services/roverLearningService/models.js +++ b/server/src/services/roverLearningService/models.js @@ -142,8 +142,8 @@ function removeObsoleteModels(root) { else { await fs.rm(target, { recursive: true, force: true }); retired.delete(target); } } // These filenames belong to superseded optimizer/scheduler lineages, not - // policy versions. The current command policy uses lineage v5. - for (const suffix of ['', '-v1', '-v2', '-v3', '-v4']) { + // policy versions. The current command policy uses lineage v6. + for (const suffix of ['', '-v1', '-v2', '-v3', '-v4', '-v5']) { for (const filename of [`resume${suffix}.pt`, `resume${suffix}.tmp`, `schedule${suffix}.json`, `schedule${suffix}.json.tmp`]) { await fs.rm(path.join(root, 'training', filename), { force: true }); diff --git a/server/src/services/roverLearningService/training.js b/server/src/services/roverLearningService/training.js index bb412a4c..b539ccc5 100644 --- a/server/src/services/roverLearningService/training.js +++ b/server/src/services/roverLearningService/training.js @@ -33,10 +33,17 @@ function createTrainer({ root, config, actionThreshold, logger }) { } } try { - const saved = JSON.parse(await fs.readFile(path.join(trainingRoot, 'schedule-v5.json'), 'utf8')); + const saved = JSON.parse(await fs.readFile(path.join(trainingRoot, 'schedule-v6.json'), 'utf8')); ledger = saved.sessions; state = { ...saved.progress, status: 'waiting' }; if (!ledger || typeof ledger !== 'object') throw new Error('Invalid training schedule'); + // Retry this reader-specific rejection once after the fixed-grid repair. + // Keep optimizer progress and unrelated recording failures intact. + if (saved.samplingVersion !== 2) { + for (const [id, entry] of Object.entries(ledger)) { + if (entry.rejected === 'No complete, aligned observation/action windows') delete ledger[id]; + } + } } catch (error) { if (error.code !== 'ENOENT') throw error; } } catch (error) { startupError = true; @@ -46,8 +53,8 @@ function createTrainer({ root, config, actionThreshold, logger }) { } async function saveLedger() { - const filename = path.join(trainingRoot, 'schedule-v5.json'); - await fs.writeFile(`${filename}.tmp`, JSON.stringify({ sessions: ledger, progress: state })); + const filename = path.join(trainingRoot, 'schedule-v6.json'); + await fs.writeFile(`${filename}.tmp`, JSON.stringify({ samplingVersion: 2, sessions: ledger, progress: state })); await fs.rename(`${filename}.tmp`, filename); } @@ -159,6 +166,7 @@ function createTrainer({ root, config, actionThreshold, logger }) { state = { ...state, status: stopped ? 'stopped' : 'waiting', steps: result.steps ?? state.steps, completedAt: Date.now(), elapsedSeconds: result.elapsedSeconds ?? null, error: null, examples: result.examples ?? state.examples, loss: result.loss ?? null, + evaluatedSteps: result.trained ? result.steps : state.evaluatedSteps, diagnostics: result.diagnostics ?? state.diagnostics, publication: result.publication ?? state.publication, evaluation: result.evaluation ?? state.evaluation, reason: result.reason ?? 'Waiting for the next scheduled training job' }; await saveLedger(); @@ -169,7 +177,9 @@ function createTrainer({ root, config, actionThreshold, logger }) { logger.error('Training result persistence failed', { error: error.message, failure: errorDetails(error, 'training result persistence') }); } finally { history.push({ at: Date.now(), status: state.status, steps: state.steps, loss: state.loss, - losses: state.losses || null, evaluation: state.evaluation || null, + losses: state.losses || null, evaluation: state.evaluation ? Object.fromEntries( + ['status', 'windows', 'precision', 'recall', 'f1', 'predictedActions', 'positiveActions', 'numericMse', 'missedStops', 'gateReasons'] + .map((key) => [key, state.evaluation[key]])) : null, elapsedSeconds: state.elapsedSeconds || null }); if (history.length > 20) history.shift(); leases.clear(); diff --git a/server/src/services/roverLearningService/workers/dataset.py b/server/src/services/roverLearningService/workers/dataset.py index b43ddf1b..3495157f 100644 --- a/server/src/services/roverLearningService/workers/dataset.py +++ b/server/src/services/roverLearningService/workers/dataset.py @@ -8,6 +8,7 @@ from pathlib import Path import av import numpy as np +from observations import decision_frames, history_frames from policy import SLOTS, VALUES, HEIGHT, WIDTH, SENSORS, command_family, is_stop, encode_command, sensor_vector @@ -74,6 +75,7 @@ def prepare(directory, schema, stats, extend): candidate = copy.deepcopy(schema) history, targets, unsupported_times = [], [], [] unsupported = {} + dispatch_delays = [] stats.update({'sensorFrames': len(sensors), 'dispatches': len(dispatches), 'failedCommands': len(failed), 'sensorPresentFraction': np.mean([row[1][len(SENSORS):] for row in sensors], axis=0).tolist()}) for data in dispatches: @@ -103,11 +105,17 @@ def prepare(directory, schema, stats, extend): target_time = request['ts'] if not 0 <= stamp - target_time <= 1000: raise ValueError('Invalid request/dispatch timing') + dispatch_delays.append(stamp - target_time) targets.append((target_time, *encoded)) else: # Server safety overrides belong in history, never demonstration # targets. Exclude windows around them rather than teaching inaction. boundaries.append(stamp) + stats['requestToDispatchMs'] = {'mean': sum(dispatch_delays) / max(1, len(dispatch_delays)), + 'max': max(dispatch_delays, default=0)} + stats['acceptedCommandEvents'] = len(targets) + stats['representedCommandEvents'] = 0 + stats['commandEventCounts'] = [0] * SLOTS stats['unsupportedCommands'] = list(unsupported.values()) stats['unsupportedCommandEvents'] = len(unsupported_times) // 2 if not targets and not unsupported_times: @@ -121,7 +129,7 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True) stats = stats if stats is not None else {} metadata, sensors, history, targets, boundaries, unsupported_times = prepare(directory, schema, stats, extend) stats.update({'usableWindows': 0, 'commandWindows': 0, 'stopWindows': 0, 'noCommandWindows': 0, - 'staleSensorWindows': 0, 'boundaryWindows': 0, + 'staleSensorWindows': 0, 'boundaryWindows': 0, 'maxSensorAgeMs': 0, 'maxFrameAgeMs': 0, 'unsupportedWindows': 0, 'ambiguousWindows': 0, 'videoFrames': 0, 'maxVideoGapMs': 0}) sensor_times = [row[0] for row in sensors] target_times = [row[0] for row in targets] @@ -131,8 +139,9 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True) command_times = np.full(SLOTS, -np.inf) cursor = 0 recent = collections.deque(maxlen=32) - last_sample, last_pts = -math.inf, -math.inf + last_pts = -math.inf def frames(): + nonlocal last_pts files = metadata['video'].get('files', ['video.mkv']) for name in files: if Path(name).name != name: @@ -140,27 +149,24 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True) with av.open(str(directory / name)) as container: stream = container.streams.video[0] stream.codec_context.thread_count = 1 - yield from container.decode(stream) + for frame in container.decode(stream): + if frame.pts is None: + raise ValueError('Video frame has no timestamp') + stamp = float(frame.pts * frame.time_base) * 1000 + stats['videoFrames'] += 1 + if math.isfinite(last_pts): + stats['maxVideoGapMs'] = max(stats['maxVideoGapMs'], stamp - last_pts) + last_pts = stamp + if not metadata['startedAt'] - 2000 <= stamp <= metadata['endedAt'] + 2000: + raise ValueError('Video PTS is not aligned to session Unix time') + yield frame - for frame in frames(): + represented = set() + for stamp, frame, captured_at in decision_frames(frames()): if cancelled(): return - if frame.pts is None: - raise ValueError('Video frame has no timestamp') - stamp = float(frame.pts * frame.time_base) * 1000 - if not math.isfinite(stamp) or stamp < last_pts: - raise ValueError('Invalid/nonmonotonic video timestamps') - stats['videoFrames'] += 1 - if math.isfinite(last_pts): - stats['maxVideoGapMs'] = max(stats['maxVideoGapMs'], stamp - last_pts) - last_pts = stamp - if not metadata['startedAt'] - 2000 <= stamp <= metadata['endedAt'] + 2000: - raise ValueError('Video PTS is not aligned to session Unix time') if stamp < metadata.get('trainingStartedAt', metadata['startedAt']): continue - if stamp - last_sample < 99: - continue - last_sample = stamp while cursor < len(history) and history[cursor][0] < stamp: when, slot, values = history[cursor] # Do not let drive predictions simply copy earlier human/model drive commands. @@ -168,22 +174,19 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True) command_values[slot] = values command_times[slot] = when cursor += 1 - sensor_index = bisect.bisect_right(sensor_times, stamp) - 1 + sensor_index = bisect.bisect_left(sensor_times, stamp) - 1 if sensor_index < 0 or stamp - sensor_times[sensor_index] > 1000: stats['staleSensorWindows'] += 1 recent.clear() continue + stats['maxSensorAgeMs'] = max(stats['maxSensorAgeMs'], stamp - sensor_times[sensor_index]) + stats['maxFrameAgeMs'] = max(stats['maxFrameAgeMs'], stamp - captured_at) seen = np.isfinite(command_times).astype(np.float32) ages = np.minimum((stamp - command_times) / 10000, 1).astype(np.float32) state = np.concatenate((sensors[sensor_index][1], command_values.flatten(), ages, seen)) pixels = frame.to_ndarray(width=WIDTH, height=HEIGHT, format='rgb24').transpose(2, 0, 1).copy() recent.append((stamp, pixels, state)) - chosen = [] - for offset in (900, 600, 300, 0): - options = [row for row in recent if row[0] <= stamp - offset + 0.01] - if not options or stamp - offset - options[-1][0] > 200: - break - chosen.append(options[-1]) + chosen = history_frames(recent, stamp) if len(chosen) != 4 or stamp + 100 > metadata['endedAt']: continue if bisect.bisect_right(unsupported_times, stamp + 100) > bisect.bisect_left(unsupported_times, chosen[0][0]): @@ -201,7 +204,7 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True) for _, slot, vector in targets[start:end]: template = schema[slot]['template'] family = command_family(template) - if ((actions[slot] and (template['type'] in ('headlight', 'laser', 'horn', 'song', 'raw') or not np.array_equal(values[slot], vector))) + if ((actions[slot] and (template['type'] in ('headlight', 'raw') or not np.array_equal(values[slot], vector))) or (not actions[slot] and family in families)): ambiguous = True break @@ -210,6 +213,10 @@ def windows(directory, schema, cancelled=lambda: False, stats=None, extend=True) if ambiguous: stats['ambiguousWindows'] += 1 continue + represented.update(range(start, end)) + stats['representedCommandEvents'] = len(represented) + for slot in np.flatnonzero(actions): + stats['commandEventCounts'][slot] += 1 stats['usableWindows'] += 1 stats['commandWindows' if actions.any() else 'noCommandWindows'] += 1 stats['stopWindows'] += int(any(actions[slot] and is_stop(descriptor) for slot, descriptor in enumerate(schema))) diff --git a/server/src/services/roverLearningService/workers/infer.py b/server/src/services/roverLearningService/workers/infer.py index 0f2ad74f..8b0f06d5 100644 --- a/server/src/services/roverLearningService/workers/infer.py +++ b/server/src/services/roverLearningService/workers/infer.py @@ -13,6 +13,7 @@ import time import av import numpy as np import torch +from observations import decision_frames, history_frames from policy import Policy, VERSION, SLOTS, VALUES, HEIGHT, WIDTH, SENSORS, select_slots, decode_command, encode_command, sensor_vector @@ -54,23 +55,16 @@ def main(): sensors, sensor_time = {}, 0 history = collections.deque(maxlen=32) latched = set() - sampled = 0 with av.open(args.url, options={'rtsp_transport': 'tcp', 'timeout': '3000000', 'use_wallclock_as_timestamps': '1'}) as video: stream = video.streams.video[0] stream.codec_context.thread_count = 1 - for frame in video.decode(stream): + for stamp, frame, captured_at in decision_frames(video.decode(stream)): if closed.is_set(): return - if frame.pts is None: - continue - stamp = float(frame.pts * frame.time_base) * 1000 now = time.time() * 1000 - if not math.isfinite(stamp) or abs(now - stamp) > 2000: + if abs(now - captured_at) > 2000: raise ValueError('Video time is stale or not aligned to server time') - if stamp - sampled < 99: - continue - sampled = stamp with mutex: while observations and observations[0]['ts'] < stamp: event = observations.popleft() @@ -94,12 +88,7 @@ def main(): np.isfinite(last_commands).astype(np.float32))) pixels = frame.to_ndarray(width=WIDTH, height=HEIGHT, format='rgb24').transpose(2, 0, 1).copy() history.append((stamp, pixels, state)) - selected = [] - for offset in (900, 600, 300, 0): - candidates = [entry for entry in history if entry[0] <= stamp - offset + .01] - if not candidates or stamp - offset - candidates[-1][0] > 200: - break - selected.append(candidates[-1]) + selected = history_frames(history, stamp) if len(selected) != 4: continue before = time.perf_counter() @@ -116,7 +105,7 @@ def main(): proposals, commands = [], [] for slot, descriptor in enumerate(schema): command = decode_command(descriptor, predictions[slot]) - discrete = command['type'] in ('headlight', 'laser', 'horn', 'song', 'raw') + discrete = command['type'] in ('headlight', 'raw') reason = ('Below threshold' if slot not in active else 'Competing action' if slot not in winners else 'Held discrete action' if discrete and slot in latched else 'Proposed') @@ -125,7 +114,7 @@ def main(): commands.append(command) if slot in winners and discrete: latched.add(slot) - print(json.dumps({'kind': 'prediction', 'frameAt': stamp, 'sensorAt': sensor_time, + print(json.dumps({'kind': 'prediction', 'frameAt': captured_at, 'decisionAt': stamp, 'sensorAt': sensor_time, 'latencyMs': (time.perf_counter() - before) * 1000, 'sensorPresent': sensor_vector(sensors)[len(SENSORS):].tolist(), 'commands': commands, 'proposals': proposals, 'threshold': args.threshold}, allow_nan=False), flush=True) diff --git a/server/src/services/roverLearningService/workers/learning.py b/server/src/services/roverLearningService/workers/learning.py new file mode 100644 index 00000000..58d6b66c --- /dev/null +++ b/server/src/services/roverLearningService/workers/learning.py @@ -0,0 +1,181 @@ +"""Training objective and bounded, observational learning diagnostics.""" +import copy +import time + +import numpy as np +import torch +from torch.nn import functional as F + +from policy import SLOTS, VALUES, select_slots, decode_command, is_stop + + +def loss_setup(samples, schema): + numeric_mask = torch.zeros(SLOTS, VALUES) + for slot, descriptor in enumerate(schema): + numeric_mask[slot, :len(descriptor['fields'])] = 1 + positives = np.sum([sample[2] for sample in samples], axis=0) + opportunities = np.sum([sample[4] for sample in samples], axis=0) + weight = torch.from_numpy(np.clip((opportunities - positives) / np.maximum(positives, 1), 1, 20)).float() + return numeric_mask, weight + + +def tensors(samples): + images, states, targets, values, available = [torch.from_numpy(np.stack(items)) for items in zip(*samples)] + return images.float().div_(255), states, targets, values, available + + +def losses(logits, predictions, targets, values, available, numeric_mask, positive_weight): + events = F.binary_cross_entropy_with_logits(logits, targets, pos_weight=positive_weight, reduction='none') + events = (events * available).sum() / available.sum().clamp_min(1) + mask = numeric_mask.unsqueeze(0) * targets.unsqueeze(-1) + parameters = ((predictions - values).square() * mask).sum() / mask.sum().clamp_min(1) + return events, parameters + + +def distribution(values): + values = np.asarray(values, dtype=np.float64) + if not values.size: + return {'count': 0, 'mean': None, 'p50': None, 'p95': None, 'max': None} + return {'count': int(values.size), 'mean': float(values.mean()), + 'p50': float(np.quantile(values, .5)), 'p95': float(np.quantile(values, .95)), 'max': float(values.max())} + + +def decisions(scores, schema, threshold): + selected = np.zeros_like(scores, dtype=bool) + for row, vector in enumerate(scores): + for slot in select_slots(vector, schema, threshold): + selected[row, slot] = True + return selected + + +def counts(predicted, actual): + matched = int((predicted & actual).sum()) + proposed, wanted = int(predicted.sum()), int(actual.sum()) + empty = ~actual.any(axis=1) + return {'predictedActions': proposed, 'positiveActions': wanted, 'matchedActions': matched, + 'precision': matched / max(1, proposed), 'recall': matched / max(1, wanted), + 'f1': 2 * matched / max(1, proposed + wanted), + 'falsePositiveActions': proposed - matched, + 'noCommandWindows': int(empty.sum()), 'unwantedCommandWindows': int((empty & predicted.any(axis=1)).sum())} + + +def summarize(scores, actual, schema, threshold, frequencies): + active_scores = scores[:, :len(schema)] + labels = actual[:, :len(schema)] + thresholds = sorted(set([.1, .3, .5, .7, .9, threshold])) + sweep = [{'threshold': value, **counts(decisions(scores, schema, value), actual)} for value in thresholds] + frequent = np.broadcast_to(frequencies, scores.shape) + baseline = [{'threshold': value, **counts(decisions(frequent, schema, value), actual)} for value in thresholds] + return {'scoreOnHumanCommands': distribution(active_scores[labels]), + 'scoreOnAbsentCommands': distribution(active_scores[~labels]), + 'maxScoreOnNoCommandWindows': distribution(active_scores[~actual.any(axis=1)].max(axis=1)) if (~actual.any(axis=1)).any() else distribution([]), + 'thresholds': sweep, 'baselines': {'sendNothing': counts(np.zeros_like(actual), actual), + 'trainingFrequency': baseline, 'frequencySource': 'Training windows only; no held-out fitting'}, + 'perCommandScores': [{'slot': slot, 'onHumanCommand': distribution(scores[actual[:, slot], slot]), + 'otherwise': distribution(scores[~actual[:, slot], slot])} for slot in range(len(schema))]} + + +def evaluate(model, samples, schema, threshold, frequencies, batch_size, cancelled): + """Same score selection as runtime; isolated windows cannot measure latching.""" + model.eval() + score_rows, target_rows = [], [] + value_error = value_count = 0 + decoded = {} + with torch.inference_mode(): + for offset in range(0, len(samples), batch_size): + if cancelled(): + break + images, states, targets, values, _ = tensors(samples[offset:offset + batch_size]) + logits, numeric = model(images, states) + if not torch.isfinite(logits).all() or not torch.isfinite(numeric).all(): + raise ValueError('Nonfinite evaluation output') + score_rows.append(logits.sigmoid().numpy()) + target_rows.append(targets.numpy().astype(bool)) + for row, slot in targets.nonzero().tolist(): + descriptor = schema[slot] + n = len(descriptor['fields']) + value_error += (numeric[row, slot, :n] - values[row, slot, :n]).square().sum().item() + value_count += n + prediction = decode_command(descriptor, numeric[row, slot]) + expected = decode_command(descriptor, values[row, slot]) + for field in descriptor['fields']: + a, b = prediction, expected + for key in field['path']: + a, b = a[key], b[key] + key = descriptor['template']['type'] + '.' + '.'.join(map(str, field['path'])) + total, count = decoded.get(key, (0, 0)) + decoded[key] = (total + abs(a - b), count + 1) + if not score_rows: + return {'windows': 0, 'complete': False} + scores, actual = np.concatenate(score_rows), np.concatenate(target_rows) + predicted = decisions(scores, schema, threshold) + stop_slots = [i for i, d in enumerate(schema) if is_stop(d)] + move_slots = [i for i, d in enumerate(schema) if d['template']['type'] == 'drive' and not is_stop(d)] + stop_count = int(actual[:, stop_slots].sum()) + stop_match = int((actual[:, stop_slots] & predicted[:, stop_slots]).sum()) + move_count = int(actual[:, move_slots].sum()) + move_match = int((actual[:, move_slots] & predicted[:, move_slots]).sum()) + return {'windows': len(scores), 'complete': len(scores) == len(samples), + 'stopEvents': stop_count, 'missedStops': stop_count - stop_match, + 'stopRecall': stop_match / max(1, stop_count), + 'falseStops': int((predicted[:, stop_slots] & ~actual[:, stop_slots]).sum()), + 'movementEvents': move_count, 'movementRecall': move_match / max(1, move_count), + 'commands': [{'slot': i, 'command': d['template'], 'actual': int(actual[:, i].sum()), + 'predicted': int(predicted[:, i].sum()), + 'matched': int((actual[:, i] & predicted[:, i]).sum())} for i, d in enumerate(schema)], + 'numericMse': value_error / max(1, value_count), + 'decodedParameterMaeBeforeServerLimits': {key: total / count for key, (total, count) in decoded.items()}, + **counts(decisions(scores, schema, threshold), actual), + **summarize(scores, actual, schema, threshold, frequencies)} + + +def probe(model, samples, schema, threshold, frequencies, cancelled, max_seconds=20): + """Try fitting a small real training subset on a disposable copy, never publish it.""" + started = time.monotonic() + # Include both event and quiet windows; source is training only. No random + # state/optimizer changes escape into the continuing learner. + quiet = [row for row in samples if not row[2].any()][:4] + # A reservoir is grouped by category: taking its first events could probe + # only zero-speed drive and falsely suggest all controls were learnable. + by_slot = [[i for i, row in enumerate(samples) if row[2][slot]] for slot in range(len(schema))] + chosen = [] + for rank in range(8): + for indices in by_slot: + if rank < len(indices) and indices[rank] not in chosen and len(chosen) < 8: + chosen.append(indices[rank]) + subset = [samples[i] for i in chosen] + quiet + if not chosen: + return {'status': 'No command examples', 'steps': 0} + with torch.random.fork_rng(devices=[]): + candidate = copy.deepcopy(model) + optimizer = torch.optim.AdamW(candidate.parameters(), lr=.001) + numeric_mask, weights = loss_setup(subset, schema) + images, states, targets, values, available = tensors(subset) + initial_loss = final_loss = gradient_norm = None + completed = 0 + expired = lambda: cancelled() or time.monotonic() - started >= max_seconds + for _ in range(80): + if expired(): + break + candidate.train() + optimizer.zero_grad(set_to_none=True) + logits, predictions = candidate(images, states) + event_loss, value_loss = losses(logits, predictions, targets, values, available, numeric_mask, weights) + loss = event_loss + value_loss + if not torch.isfinite(loss): + raise ValueError('Nonfinite learning probe loss') + if initial_loss is None: + initial_loss = loss.item() + loss.backward() + norm = torch.nn.utils.clip_grad_norm_(candidate.parameters(), 1) + if not torch.isfinite(norm): + raise ValueError('Nonfinite learning probe gradient') + gradient_norm = float(norm) + optimizer.step() + final_loss = loss.item() + completed += 1 + result = evaluate(candidate, subset, schema, threshold, frequencies, 16, cancelled) + return {'status': 'completed' if completed == 80 else 'budget limited', 'steps': completed, + 'windows': len(subset), 'initialLoss': initial_loss, 'lastStepLoss': final_loss, + 'lastGradientNormBeforeClipping': gradient_norm, 'elapsedSeconds': time.monotonic() - started, + 'evaluation': result, 'meaning': 'Memorization on a disposable copy of training data, not generalization'} diff --git a/server/src/services/roverLearningService/workers/observations.py b/server/src/services/roverLearningService/workers/observations.py new file mode 100644 index 00000000..1afd682d --- /dev/null +++ b/server/src/services/roverLearningService/workers/observations.py @@ -0,0 +1,45 @@ +"""Shared causal 10 Hz sampling for recorded and live video.""" +import math + + +def decision_frames(frames): + """Yield decision time, latest past image, and its actual capture timestamp. + + Sampling every third/fourth decoded frame leaves holes at rates such as 25 + FPS. A fixed grid covers each 100ms command interval exactly once. A frame + after the decision can release the pending tick, but is never its input. + """ + previous = None + previous_time = None + tick = None + for frame in frames: + if frame.pts is None: + raise ValueError('Video frame has no timestamp') + stamp = float(frame.pts * frame.time_base) * 1000 + if not math.isfinite(stamp) or (previous_time is not None and stamp < previous_time): + raise ValueError('Invalid/nonmonotonic video timestamps') + if tick is None: + tick = stamp + while tick < stamp - .01: + # Do not synthesize observations over missing/stalled video. History + # selection will reject a temporal gap until fresh history exists. + if previous is not None and tick - previous_time <= 200: + yield tick, previous, previous_time + elif stamp - tick > 200: + tick += max(1, math.floor((stamp - tick) / 100)) * 100 + continue + tick += 100 + if abs(stamp - tick) <= .01: + yield tick, frame, stamp + tick += 100 + previous, previous_time = frame, stamp + + +def history_frames(history, stamp): + selected = [] + for offset in (900, 600, 300, 0): + candidates = [row for row in history if row[0] <= stamp - offset + .01] + if not candidates or stamp - offset - candidates[-1][0] > 200: + return [] + selected.append(candidates[-1]) + return selected diff --git a/server/src/services/roverLearningService/workers/policy.py b/server/src/services/roverLearningService/workers/policy.py index 543c0d83..70efb310 100644 --- a/server/src/services/roverLearningService/workers/policy.py +++ b/server/src/services/roverLearningService/workers/policy.py @@ -14,12 +14,13 @@ import numpy as np import torch from torch import nn -VERSION = 4 +VERSION = 5 SLOTS = 64 VALUES = 8 FRAMES = 4 HEIGHT, WIDTH = 120, 160 -OPERATING_TYPES = {"drive", "motors", "servo", "headlight", "laser", "horn", "song", "raw"} +OPERATING_TYPES = {"drive", "servo", "headlight", "raw"} +MODE_COMMANDS = tuple(base64.b64encode(bytes([opcode])).decode() for opcode in (128, 131, 132, 143)) SENSORS = [ ("bumpsAndWheelDrops.bumpLeft", 1), ("bumpsAndWheelDrops.bumpRight", 1), ("bumpsAndWheelDrops.wheelDropLeft", 1), ("bumpsAndWheelDrops.wheelDropRight", 1), @@ -56,29 +57,28 @@ def sensor_vector(sensors): def describe_command(command): if command.get('type') not in OPERATING_TYPES: return None + # Raw is restricted to the existing mode/dock buttons, not arbitrary OI + # packets that could reintroduce excluded actuators or songs. + if command['type'] == 'raw' and command.get('raw') not in MODE_COMMANDS: + return None if len(json.dumps(command)) > 4096: raise ValueError('Operating command exceeds 4096-byte schema limit') values, fields = [], [] def visit(value, path=()): - if isinstance(value, list) and command['type'] == 'horn' and path == ('horn', 'freqs'): - return [visit(item, (*path, index)) for index, item in enumerate(value)] if isinstance(value, dict): return {key: visit(item, (*path, key)) for key, item in sorted(value.items())} - # Exact actuator zeros preserve demonstrated releases. Horn frequency - # zeros are parameters so changing a chord does not require a new slot. - if value == 0 and command['type'] != 'horn': + # Exact actuator zeros preserve demonstrated releases. + if value == 0: return copy.deepcopy(value) if isinstance(value, (int, float)) and not isinstance(value, bool): if not math.isfinite(value) or len(values) >= VALUES: raise ValueError('Nonfinite command or too many numeric control fields') - scale = {'drive': 500, 'motors': 255, 'servo': 180, - 'horn': 1000, 'song': 127}.get(command['type'], 1) - fields.append({'path': list(path), 'scale': scale, 'integer': command['type'] in ('drive', 'motors', 'servo', 'horn') or isinstance(value, int)}) + scale = {'drive': 500, 'servo': 180}.get(command['type'], 1) + fields.append({'path': list(path), 'scale': scale, 'integer': command['type'] in ('drive', 'servo') or isinstance(value, int)}) values.append(normalize(value, scale)) return {'$number': len(values) - 1} - # Arrays (e.g. songs) are discrete exact demonstrations, not variable - # length regression outputs. Never reinterpret raw OI bytes as numbers. + # Mode bytes and named actions remain exact constants. return copy.deepcopy(value) template = visit(command) @@ -105,24 +105,17 @@ def encode_command(command, schema, extend=False): def initial_schema(): - # Match commandPipeline.js envelopes, OI_COMMANDS and the four-frequency - # horn control. Values are examples describing fields, not training labels. + # Match the retained commandPipeline.js controls. Values describe fields, + # not artificial demonstrations; zero-speed drive remains a drive command. commands = [] for left, right in itertools.product((0, 1), repeat=2): commands.append({'type': 'drive', 'driveDirect': {'left': left, 'right': right}}) - for main, side, vacuum in itertools.product((0, 1), repeat=3): - commands.append({'type': 'motors', 'motorPwm': {'main': main, 'side': side, 'vacuum': vacuum}}) for angle in (0, 1): commands.append({'type': 'servo', 'servo': {'angle': angle}}) - for kind in ('headlight', 'laser'): - for action in ('on', 'off', 'toggle'): - commands.append({'type': kind, kind: {'action': action}}) - for waveform in ('sine', 'saw'): - commands.append({'type': 'horn', 'horn': {'action': 'start', 'waveform': waveform, 'freqs': [1, 1, 1, 1]}}) - commands.append({'type': 'horn', 'horn': {'action': 'stop'}}) - for opcode in (128, 131, 132, 143): - commands.append({'type': 'raw', 'raw': base64.b64encode(bytes([opcode])).decode()}) - commands.append({'type': 'song', 'song': {'notes': [{'note': 60, 'duration': 8}]}}) + for action in ('on', 'off', 'toggle'): + commands.append({'type': 'headlight', 'headlight': {'action': action}}) + for raw in MODE_COMMANDS: + commands.append({'type': 'raw', 'raw': raw}) schema = [] for command in commands: encode_command(command, schema, extend=True) @@ -188,7 +181,7 @@ class Policy(nn.Module): def specification(schema): - return {'version': VERSION, 'architecture': 'spatial-cnn-gru-events-v4', + return {'version': VERSION, 'architecture': 'spatial-cnn-gru-events-v5', 'frames': FRAMES, 'height': HEIGHT, 'width': WIDTH, 'historyStrideMs': 300, 'decisionIntervalMs': 100, 'stateSize': STATE_SIZE, 'sensorFields': SENSORS, diff --git a/server/src/services/roverLearningService/workers/train.py b/server/src/services/roverLearningService/workers/train.py index df533108..2bdef5e8 100644 --- a/server/src/services/roverLearningService/workers/train.py +++ b/server/src/services/roverLearningService/workers/train.py @@ -39,7 +39,7 @@ def main(): import numpy as np import av import torch - from torch.nn import functional as functional + from learning import loss_setup, tensors, losses as policy_losses, evaluate, probe from dataset import windows from policy import Policy, VERSION, SLOTS, VALUES, specification, is_stop, select_slots, initial_schema @@ -72,12 +72,13 @@ def main(): optimizer = torch.optim.AdamW(model.parameters(), lr=0.0001) # The command-event policy has different inputs, targets and weights. Keep its # optimizer lineage separate; reuse recordings with their unchanged partition. - resume_path = root / 'training' / 'resume-v5.pt' + resume_path = root / 'training' / 'resume-v6.pt' schema, steps, examples, last_published = initial_schema(), 0, 0, 0 trained_rovers = set() control_profiles = [] distinct_windows = {} published_windows = 0 + last_probe_steps, learning_probe = 0, None if resume_path.exists(): saved = torch.load(resume_path, map_location='cpu', weights_only=True) if saved['version'] != VERSION: @@ -91,6 +92,8 @@ def main(): control_profiles = saved.get('controlProfiles', []) distinct_windows = saved.get('distinctWindows', {}) published_windows = saved.get('publishedWindows', 0) + last_probe_steps = saved.get('lastProbeSteps', 0) + learning_probe = saved.get('learningProbe') cpu_started = time.process_time() decode_started = time.monotonic() dataset_stats = [] @@ -171,15 +174,16 @@ def main(): return decode_seconds = time.monotonic() - decode_started + audit_started = time.monotonic() + numeric_mask, positive_weight = loss_setup(pool, schema) + training_rows = [row for row in dataset_stats if row.get('partition') == 'training'] + natural_count = sum(row.get('usableWindows', 0) for row in training_rows) + frequencies = np.sum([row.get('commandEventCounts', [0] * SLOTS) for row in training_rows], axis=0) / max(1, natural_count) + audit_subset = pool[::max(1, len(pool) // 96)][:96] + before_training = evaluate(model, audit_subset, schema, job['actionThreshold'], frequencies, config['batchSize'], cancelled) + before_audit_seconds = time.monotonic() - audit_started train_started = time.monotonic() - numeric_mask = torch.zeros(SLOTS, VALUES) - for slot, descriptor in enumerate(schema): - numeric_mask[slot, :len(descriptor['fields'])] = 1 - # Balance sparse events without assuming that every time step should fire - # every control. This is training loss, not a driving-quality evaluation. - positives = np.sum([sample[2] for sample in pool], axis=0) - opportunities = np.sum([sample[4] for sample in pool], axis=0) - positive_weight = torch.from_numpy(np.clip((opportunities - positives) / np.maximum(positives, 1), 1, 20)).float() + gradient_norms = [] losses = [] model.train() for _ in range(config['passesPerJob']): @@ -188,20 +192,18 @@ def main(): if cancelled() or steps - initial_steps >= config['maxStepsPerJob']: break batch = pool[offset:offset + config['batchSize']] - images, states, targets, values, available = [torch.from_numpy(np.stack(items)) for items in zip(*batch)] - images = images.float().div_(255) + images, states, targets, values, available = tensors(batch) optimizer.zero_grad(set_to_none=True) action_logits, predictions = model(images, states) - event_loss = functional.binary_cross_entropy_with_logits(action_logits, targets, - pos_weight=positive_weight, reduction='none') - event_loss = (event_loss * available).sum() / available.sum().clamp_min(1) - mask = numeric_mask.unsqueeze(0) * targets.unsqueeze(-1) - value_loss = ((predictions - values).square() * mask).sum() / mask.sum().clamp_min(1) + event_loss, value_loss = policy_losses(action_logits, predictions, targets, values, available, numeric_mask, positive_weight) loss = event_loss + value_loss if not torch.isfinite(loss): raise ValueError('Nonfinite training loss; checkpoint not saved') loss.backward() - torch.nn.utils.clip_grad_norm_(model.parameters(), 1) + gradient_norm = torch.nn.utils.clip_grad_norm_(model.parameters(), 1) + if not torch.isfinite(gradient_norm): + raise ValueError('Nonfinite training gradient; checkpoint not saved') + gradient_norms.append(float(gradient_norm)) optimizer.step() steps += 1 examples += len(batch) @@ -226,58 +228,16 @@ def main(): 'windows': len(validation), 'unsupportedCommandEvents': unsupported_events, 'unsupportedWindows': unsupported_windows} if validation and not cancelled(): - model.eval() - true_positive = predicted_positive = actual_positive = 0 - value_error = value_count = evaluated = 0 - stop_actual = stop_detected = false_stops = 0 - drive_actual = drive_detected = 0 - empty_windows = unwanted_windows = 0 - stop_slots = [slot for slot, descriptor in enumerate(schema) if is_stop(descriptor)] - drive_slots = [slot for slot, descriptor in enumerate(schema) - if descriptor['template']['type'] == 'drive' and not is_stop(descriptor)] - per_command = [{'slot': slot, 'command': descriptor['template'], 'actual': 0, - 'predicted': 0, 'matched': 0} for slot, descriptor in enumerate(schema)] - with torch.inference_mode(): - for offset in range(0, len(validation), config['batchSize']): - if cancelled(): - break - batch = validation[offset:offset + config['batchSize']] - images, states, targets, values, available = [torch.from_numpy(np.stack(items)) for items in zip(*batch)] - logits, predictions = model(images.float() / 255, states) - predicted = torch.zeros_like(targets, dtype=torch.bool) - # Use the same threshold and competing-command selection as live - # inference. Reservoir samples cannot assess temporal latching or - # closed-loop motion; report this explicitly rather than implying it. - for row, scores in enumerate(logits.sigmoid().tolist()): - for slot in select_slots(scores, schema, job['actionThreshold']): - predicted[row, slot] = True - actual = targets.bool() - true_positive += (predicted & actual).sum().item() - predicted_positive += predicted.sum().item() - actual_positive += actual.sum().item() - mask = numeric_mask.unsqueeze(0) * targets.unsqueeze(-1) - value_error += ((predictions - values).square() * mask).sum().item() - value_count += mask.sum().item() - empty = ~actual.any(dim=1) - empty_windows += empty.sum().item() - unwanted_windows += (empty & predicted.any(dim=1)).sum().item() - if stop_slots: - stop_actual += actual[:, stop_slots].sum().item() - stop_detected += (actual[:, stop_slots] & predicted[:, stop_slots]).sum().item() - false_stops += (predicted[:, stop_slots] & ~actual[:, stop_slots]).sum().item() - if drive_slots: - drive_actual += actual[:, drive_slots].sum().item() - drive_detected += (actual[:, drive_slots] & predicted[:, drive_slots]).sum().item() - for item in per_command: - slot = item['slot'] - item['actual'] += actual[:, slot].sum().item() - item['predicted'] += predicted[:, slot].sum().item() - item['matched'] += (actual[:, slot] & predicted[:, slot]).sum().item() - evaluated += len(batch) - precision = true_positive / max(1, predicted_positive) - recall = true_positive / max(1, actual_positive) - mse = value_error / max(1, value_count) - stop_recall = stop_detected / max(1, stop_actual) + measured = evaluate(model, validation, schema, job['actionThreshold'], frequencies, config['batchSize'], cancelled) + evaluated = measured['windows'] + precision, recall = measured.get('precision', 0), measured.get('recall', 0) + actual_positive = measured.get('positiveActions', 0) + mse = measured.get('numericMse', 0) + stop_actual = measured.get('stopEvents', 0) + stop_recall = measured.get('stopRecall', 0) + drive_actual = measured.get('movementEvents', 0) + drive_recall = measured.get('movementRecall', 0) + frequency_f1 = max((row['f1'] for row in measured.get('baselines', {}).get('trainingFrequency', [])), default=0) held_out_sessions = sum(row.get('partition') == 'heldOut' and not row.get('skipReason') for row in dataset_stats) gate_reasons = [] for condition, reason in [ @@ -288,27 +248,38 @@ def main(): (actual_positive >= 32, 'Need 32 held-out command events'), (stop_actual >= 8, 'Need eight held-out stop events'), (drive_actual >= 16, 'Need 16 held-out movement commands'), - (drive_detected / max(1, drive_actual) >= .5, 'Movement command recall below 0.5'), + (drive_recall >= .5, 'Movement command recall below 0.5'), (precision >= .5, 'Command precision below 0.5'), (recall >= .5, 'Command recall below 0.5'), (stop_recall >= .5, 'Stop recall below 0.5'), - (value_count == 0 or mse <= .1, 'Command parameter error above 0.1'), + (measured.get('f1', 0) > frequency_f1, 'Command F1 does not beat the training-frequency baseline'), + (mse <= .1, 'Command parameter error above 0.1'), ]: if not condition: gate_reasons.append(reason) - evaluation = {'status': 'imitation gate not met' if gate_reasons else 'passed imitation gate', + evaluation = {**measured, 'status': 'imitation gate not met' if gate_reasons else 'passed imitation gate', 'unsupportedCommandEvents': unsupported_events, 'unsupportedWindows': unsupported_windows, - 'method': 'Held-out 100ms command events; not a closed-loop rollout', + 'method': 'Isolated held-out 100ms events; live selection, no temporal latch/physical rollout', 'threshold': job['actionThreshold'], 'heldOutSessions': held_out_sessions, - 'windows': evaluated, 'precision': precision, 'recall': recall, - 'numericMse': mse, 'positiveActions': actual_positive, - 'predictedActions': predicted_positive, 'falsePositiveActions': predicted_positive - true_positive, - 'stopEvents': stop_actual, 'missedStops': stop_actual - stop_detected, - 'stopRecall': stop_recall, 'falseStops': false_stops, - 'movementEvents': drive_actual, 'movementRecall': drive_detected / max(1, drive_actual), - 'noCommandWindows': empty_windows, 'unwantedCommandWindows': unwanted_windows, - 'commands': per_command, 'gateReasons': gate_reasons} - evaluation_seconds = time.monotonic() - evaluation_started + 'gateReasons': gate_reasons} + after_training = evaluate(model, audit_subset, schema, job['actionThreshold'], frequencies, config['batchSize'], cancelled) + probe_seconds = 0 + if not cancelled() and (learning_probe is None or steps - last_probe_steps >= 1000): + remaining = config['maxJobSeconds'] - (time.monotonic() - started) + if remaining > 10: + probe_started = time.monotonic() + learning_probe = probe(model, audit_subset, schema, job['actionThreshold'], frequencies, + cancelled, max_seconds=min(20, remaining / 2)) + learning_probe['atTrainingStep'] = steps + last_probe_steps = steps + probe_seconds = time.monotonic() - probe_started + training_audit = {'before': before_training, 'after': after_training, + 'gradientNormBeforeClipping': {'min': min(gradient_norms), 'max': max(gradient_norms)}, + 'positiveWeights': positive_weight[:len(schema)].tolist(), + 'naturalTrainingCommandFrequency': frequencies[:len(schema)].tolist(), + 'sampledCommandFrequency': np.mean([row[2] for row in pool], axis=0)[:len(schema)].tolist(), + 'learningProbe': learning_probe} + evaluation_seconds = time.monotonic() - evaluation_started - probe_seconds publication = {'distinctMinutes': unique_count / 600, 'newMinutes': (unique_count - published_windows) / 600, 'minimumMinutes': config['minimumDrivingMinutes'], @@ -361,9 +332,9 @@ def main(): 'schema': schema, 'steps': steps, 'examples': examples, 'lastPublished': last_published, 'trainedRovers': sorted(trained_rovers), 'controlProfiles': control_profiles, 'distinctWindows': distinct_windows, - 'publishedWindows': published_windows, + 'publishedWindows': published_windows, 'lastProbeSteps': last_probe_steps, 'learningProbe': learning_probe, }, resume_path) - diagnostics = {'dataset': dataset_stats, 'decodeSeconds': decode_seconds, + diagnostics = {'dataset': dataset_stats, 'trainingAudit': training_audit, 'probeSeconds': probe_seconds, 'beforeAuditSeconds': before_audit_seconds, 'decodeSeconds': decode_seconds, 'trainingSeconds': training_seconds, 'evaluationSeconds': evaluation_seconds, 'checkpointSeconds': time.monotonic() - save_started, 'cpuSeconds': time.process_time() - cpu_started, 'peakRssMiB': resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024, diff --git a/webui/src/components/vip/VipRoverLearningCard.jsx b/webui/src/components/vip/VipRoverLearningCard.jsx index 935767bc..b49d4e13 100644 --- a/webui/src/components/vip/VipRoverLearningCard.jsx +++ b/webui/src/components/vip/VipRoverLearningCard.jsx @@ -133,7 +133,7 @@ function LearningCard({ roverId }) {