fix: incorrect lickspout position

docs/trials_table_mapping.md

@@ -123,11 +123,11 @@ seconds, far too coarse for the ~tens-of-ms valve pulse.
 | --- | --- |
 | `goCue_start_time` | `PlaySoundOrFrequency` `WRITE` message. |
 
-### From `InitialManipulatorPosition` (software event)
+### From `HarpManipulator` `AccumulatedSteps` (+ `InputSchemas.Rig`)
 
 | Trials column | Mapping |
 | --- | --- |
-| `lickspout_positions` | `data` field. |
+| `lickspout_position_x` / `y1` / `y2` / `z` | Per-motor cumulative microstep count from the `AccumulatedSteps` stream, converted to millimetres via the rig manipulator calibration (`full_step_to_mm / microstep_resolution`) and re-referenced to the session-start position (displacement **relative to session start**, mm). The manipulator is a continuously-sampled hardware value, so — like the go cue — each trial takes the sample within its `[start_time, stop_time)` window nearest the start. `Motor{i}` drives `Axis(i + 1)` (X, Y1, Y2, Z). `None` when no sample falls in the trial window. The rig and `AccumulatedSteps` streams are required inputs (`build` raises if either is missing with trials present). |
 
 ### From `trainer_state.json` and `acquisition.json` (autoTrain — can be disregarded)
 
@@ -155,3 +155,4 @@ These were mapped during exploration but are no longer in scope:
 | 2026-06-17 | `auto_waterL` / `auto_waterR` now encode no auto-response (`is_auto_reward_right` is `None`) and missing trials as `0` instead of `NULL`. The columns are non-nullable (`int`, default `0`). |
 | 2026-06-20 | Added `reward_size_left` / `reward_size_right` (reward volume in uL) from `task_parameters.reward_size`, and `side_bias` from the per-trial `TrialMetrics` event (`bias` field). |
 | 2026-06-20 | `reward_probabilityL` / `reward_probabilityR` now read the block probability from `trial.metadata.p_reward_left` / `p_reward_right` instead of the top-level per-trial `trial.p_reward_left` / `p_reward_right`. |
+| 2026-07-22 | `lickspout_position_x` / `y1` / `y2` / `z` now derive from the `HarpManipulator` `AccumulatedSteps` stream (microsteps → mm via the `InputSchemas.Rig` manipulator calibration, `full_step_to_mm / microstep_resolution`), sampled per trial via the closest sample in the `[start_time, stop_time)` window and re-referenced to the session-start position (displacement relative to session start, mm), replacing the static `InitialManipulatorPosition` software event. `Motor{i}` maps to `Axis(i + 1)` (X, Y1, Y2, Z). The rig and `AccumulatedSteps` streams are required when there are trials (`build` raises if either is missing). Column descriptions corrected from `um` to `mm`. |

src/dynamic_foraging_processing/processing/_trial_table.py

@@ -10,6 +10,7 @@
 
 import numpy as np
 import pandas as pd
+from aind_behavior_dynamic_foraging.rig import AindDynamicForagingRig
 from aind_behavior_dynamic_foraging.task_logic import AindDynamicForagingTaskLogic
 from aind_behavior_dynamic_foraging.task_logic.trial_generators import TrialGeneratorSpec
 from aind_behavior_dynamic_foraging.task_logic.trial_models import (
@@ -24,6 +25,12 @@
 
 logger = logging.getLogger(__name__)
 
+# The four manipulator axes, ordered so that index ``i`` is the axis driven
+# by ``Motor{i}`` in the ``AccumulatedSteps`` stream: ``Motor{i}`` maps to
+# ``Axis(i + 1)`` (Axis.X=1, Y1=2, Y2=3, Z=4), whose names match the
+# ``lickspout_position_*`` suffixes and the ``full_step_to_mm`` attributes.
+_MANIPULATOR_AXES = ("x", "y1", "y2", "z")
+
 
 class TrialTableBuilder:
     """Builds the NWB ``trials`` table from a dynamic foraging ``Dataset``.
@@ -524,41 +531,140 @@ def _session_columns(self, task_logic: AindDynamicForagingTaskLogic) -> t.Dict[s
             columns["min_reward_each_block"] = generator.min_block_reward
         return columns
 
-    def _lickspout_columns(
-        self, manipulator: t.Optional[pd.DataFrame]
+    def _manipulator_mm_per_step(self, rig: AindDynamicForagingRig) -> t.Dict[str, float]:
+        """Return millimetres travelled per accumulated step, keyed by axis.
+
+        ``AccumulatedSteps`` counts *microsteps*. The physical distance per
+        microstep is the full-step-to-mm calibration divided by the microstep
+        resolution (e.g. ``MICROSTEP8`` → 8 microsteps per full step), both read
+        from the rig's manipulator calibration.
+
+        Parameters
+        ----------
+        rig : AindDynamicForagingRig
+            The ``InputSchemas.Rig`` stream data. The rig is a required input
+            schema, so ``build`` guarantees it is present before this is called.
+
+        Returns
+        -------
+        dict of str to float
+            ``{axis: mm_per_step}`` for ``x`` / ``y1`` / ``y2`` / ``z``.
+
+        Raises
+        ------
+        ValueError
+            If the manipulator calibration is missing one of the four axes.
+        """
+        calibration = rig.manipulator.calibration
+        resolution_by_axis = {
+            config.axis.name.lower(): config.microstep_resolution
+            for config in calibration.axis_configuration
+        }
+        mm_per_step: t.Dict[str, float] = {}
+        for axis in _MANIPULATOR_AXES:
+            resolution = resolution_by_axis.get(axis)
+            if resolution is None:
+                raise ValueError(
+                    f"Manipulator calibration is missing axis {axis!r}; "
+                    "cannot derive lickspout position."
+                )
+            # Resolution enum names encode the divisor (MICROSTEP8 -> 8); the
+            # enum *value* (0..3) does not, so parse from the name.
+            microsteps = int(resolution.name.replace("MICROSTEP", ""))
+            mm_per_step[axis] = getattr(calibration.full_step_to_mm, axis) / microsteps
+        return mm_per_step
+
+    def _manipulator_positions(
+        self,
+        accumulated_steps: pd.DataFrame,
+        rig: AindDynamicForagingRig,
+    ) -> pd.DataFrame:
+        """Return a time-indexed frame of lickspout position (mm) over the session.
+
+        Built from the ``HarpManipulator`` ``AccumulatedSteps`` stream: each
+        ``EVENT`` row's per-motor microstep count is converted to millimetres via
+        the rig calibration. ``AccumulatedSteps`` is absolute (zero-referenced at
+        the homing position), so the count maps directly to position with no
+        offset. The frame is then re-referenced to the session start (the first
+        sample subtracted from every row), so the stored values are displacement
+        *relative to session start* — the units the QC plot expects.
+
+        Parameters
+        ----------
+        accumulated_steps : pandas.DataFrame
+            The ``AccumulatedSteps`` stream data (``Motor0``..``Motor3`` columns).
+            Required; guaranteed present by ``build``.
+        rig : AindDynamicForagingRig
+            The ``InputSchemas.Rig`` stream data (required; guaranteed present by
+            ``build``).
+
+        Returns
+        -------
+        pandas.DataFrame
+            Columns ``lickspout_position_x`` / ``y1`` / ``y2`` / ``z`` indexed by
+            time (sorted ascending), relative to the session-start position.
+
+        Raises
+        ------
+        ValueError
+            If the stream has no ``EVENT`` rows, is missing a motor column, or the
+            rig calibration is missing an axis.
+        """
+        mm_per_step = self._manipulator_mm_per_step(rig)
+        events = accumulated_steps
+        if "MessageType" in events.columns:
+            events = events[events["MessageType"] == "EVENT"]
+        if len(events) == 0:
+            raise ValueError(
+                "AccumulatedSteps stream has no EVENT rows; cannot derive lickspout position."
+            )
+        events = events.sort_index()
+        columns: t.Dict[str, np.ndarray] = {}
+        for index, axis in enumerate(_MANIPULATOR_AXES):
+            motor = f"Motor{index}"
+            if motor not in events.columns:
+                raise ValueError(
+                    f"AccumulatedSteps stream is missing column {motor}; "
+                    "cannot derive lickspout position."
+                )
+            columns[f"lickspout_position_{axis}"] = (
+                events[motor].to_numpy(dtype=float) * mm_per_step[axis]
+            )
+        positions = pd.DataFrame(columns, index=events.index)
+        # Re-reference to session start so the stored values are displacement
+        # from the first sample (previously done in the plot as ``values[0]``).
+        return positions - positions.iloc[0]
+
+    def _sample_lickspout(
+        self, positions: pd.DataFrame, start: float, stop: float
     ) -> t.Dict[str, t.Optional[float]]:
-        """Return lickspout position columns from ``InitialManipulatorPosition``.
+        """Return the lickspout position sampled within a trial's window.
 
-        The event's ``data`` payload carries the x / y1 / y2 / z positions. These
-        are the *initial* manipulator coordinates recorded once at experiment
-        start, so the columns are currently constant across trials.
+        The manipulator position is a continuously-sampled hardware value, so —
+        like the go cue — each trial takes the sample within its ``[start, stop)``
+        window nearest the start (see :meth:`_closest_time_in_window`).
 
-        Known limitation: the manipulator can move within a session (e.g. the
-        anti-bias intervention shifts the lickspouts horizontally), which this
-        static initial position does not capture.
+        Parameters
+        ----------
+        positions : pandas.DataFrame
+            The time-indexed position frame from :meth:`_manipulator_positions`.
+        start, stop : float
+            The trial window bounds (seconds); ``start`` inclusive, ``stop``
+            exclusive. May be ``NaN`` when unaligned.
 
-        TODO: think about how to represent
-        this if it becomes relevant. The ideal solution would be to use the harp files
-        to track the manipulator position over time
+        Returns
+        -------
+        dict of str to (float or None)
+            The four ``lickspout_position_*`` values, all ``None`` for a trial
+            whose window contains no manipulator sample.
         """
-        keys = (
-            "lickspout_position_x",
-            "lickspout_position_y1",
-            "lickspout_position_y2",
-            "lickspout_position_z",
-        )
-        empty = {key: None for key in keys}
-        payloads = self._event_payloads(manipulator)
-        if not payloads:
-            return empty
-        payload = payloads[0]
-        components = ("x", "y1", "y2", "z")
-        if isinstance(payload, dict):
-            return {key: payload.get(comp) for key, comp in zip(keys, components)}
-        if isinstance(payload, (list, tuple)) and len(payload) == len(keys):
-            return {key: payload[idx] for idx, key in enumerate(keys)}
-        logger.warning("Unrecognized InitialManipulatorPosition payload; leaving lickspout null.")
-        return empty
+        keys = tuple(f"lickspout_position_{axis}" for axis in _MANIPULATOR_AXES)
+        times = positions.index.to_numpy(dtype=float)
+        sample_time = self._closest_time_in_window(times, start, stop)
+        if sample_time is None:
+            return {key: None for key in keys}
+        row = positions.loc[sample_time]
+        return {key: (None if pd.isna(row[key]) else float(row[key])) for key in keys}
 
     # ------------------------------------------------------------------ #
     # Per-trial assembly
@@ -683,7 +789,8 @@ def build(self) -> pd.DataFrame:
         pulse_supply_left = self._load("Behavior", "HarpBehavior", "PulseSupplyPort0")
         pulse_supply_right = self._load("Behavior", "HarpBehavior", "PulseSupplyPort1")
         go_cue = self._load("Behavior", "HarpSoundCard", "PlaySoundOrFrequency")
-        manipulator = self._load("Behavior", "SoftwareEvents", "InitialManipulatorPosition")
+        accumulated_steps = self._load("Behavior", "HarpManipulator", "AccumulatedSteps")
+        rig = self._load("Behavior", "InputSchemas", "Rig")
         task_logic = self._load("Behavior", "InputSchemas", "TaskLogic")
 
         # Per-trial streams: one payload/timestamp per trial, aligned by index.
@@ -706,6 +813,20 @@ def build(self) -> pd.DataFrame:
                 f"(reward sizes are sourced from it) but it failed to load for {n_trials} trials."
             )
 
+        # The rig and AccumulatedSteps streams are the required sources for the
+        # lickspout positions, so a missing one cannot yield valid position
+        # columns when there are trials to build.
+        if n_trials > 0 and rig is None:
+            raise ValueError(
+                "Rig stream is required to build the trials table "
+                f"(lickspout positions are sourced from it) but it failed to load for {n_trials} trials."
+            )
+        if n_trials > 0 and accumulated_steps is None:
+            raise ValueError(
+                "AccumulatedSteps stream is required to build the trials table "
+                f"(lickspout positions are sourced from it) but it failed to load for {n_trials} trials."
+            )
+
         warnings = self._check_aligned(
             n_trials,
             {
@@ -732,7 +853,11 @@ def build(self) -> pd.DataFrame:
         go_cue_times = self._write_times(go_cue)
 
         session = self._session_columns(task_logic)
-        lickspout = self._lickspout_columns(manipulator)
+        # When there are trials, the guards above guarantee both sources are
+        # present; with no trials the frame goes unused (the loop does not run).
+        lickspout_positions = (
+            self._manipulator_positions(accumulated_steps, rig) if n_trials else None
+        )
 
         rows: t.List[TrialConfig] = []
         for i, outcome_payload in enumerate(outcome_payloads):
@@ -743,6 +868,7 @@ def build(self) -> pd.DataFrame:
             stop = float(stop_times[i]) if i < stop_times.size else np.nan
             response = response_payloads[i] if i < len(response_payloads) else None
             side_bias = self._side_bias(metric_payloads[i] if i < len(metric_payloads) else None)
+            lickspout = self._sample_lickspout(lickspout_positions, start, stop)
 
             rows.append(
                 self._build_row(

src/dynamic_foraging_processing/processing/models/trial_config.py

@@ -142,18 +142,20 @@ class TrialConfig(BaseModel):
 
     # --- lickspout_position (mapping's `lickspout_positions` -> these four components) ---
     lickspout_position_x: Optional[float] = Field(
-        default=None, description="x position (um) of the lickspout position (left-right)"
+        default=None,
+        description="x lickspout position (mm), relative to session start (left-right)",
     )
     lickspout_position_y1: Optional[float] = Field(
         default=None,
-        description="y1 position (um) of the left lickspout position (forward-backward)",
+        description="y1 left lickspout position (mm), relative to session start (forward-backward)",
     )
     lickspout_position_y2: Optional[float] = Field(
         default=None,
-        description="y2 position (um) of the right lickspout position (forward-backward)",
+        description="y2 right lickspout position (mm), relative to session start (forward-backward)",
     )
     lickspout_position_z: Optional[float] = Field(
-        default=None, description="z position (um) of the lickspout position (up-down)"
+        default=None,
+        description="z lickspout position (mm), relative to session start (up-down)",
     )
 
     @classmethod

src/dynamic_foraging_processing/qc/processed/plots.py

@@ -122,7 +122,9 @@ def _add_lickspout_position_plot(
         if position is None or len(position) == 0:
             continue
         values = np.asarray(position, dtype=float)
-        ax.plot(values - values[0], color, label=label)
+        # Values already come in relative to session start (normalized by the
+        # trial-table builder), so plot them directly.
+        ax.plot(values, color, label=label)
         plotted = True
     if plotted:
         ax.legend()
@@ -303,6 +305,14 @@ def plot_side_bias(
     )
     _add_reward_probabilities(ax[3], reward_probability_left, reward_probability_right)
 
+    # Align the x-axis across every panel so trials line up vertically. The
+    # panels are all indexed by trial, but some auto-scale (adding margins) while
+    # others set [0, N]; pin them all to a common [0, n_trials].
+    n_trials = max(len(np.asarray(side_bias)), len(np.asarray(animal_response)))
+    if n_trials:
+        for axis in ax:
+            axis.set_xlim([0, n_trials])
+
     fig.savefig(Path(results_folder) / SIDE_BIAS_PLOT, dpi=300, bbox_inches="tight")
     plt.close(fig)
     return SIDE_BIAS_PLOT

src/dynamic_foraging_processing/qc/processed/results.py

@@ -22,15 +22,15 @@
 from dynamic_foraging_processing.qc.processed.plots import plot_lick_intervals, plot_side_bias
 
 # Logical input -> trials-table column name. Centralized so the mapping is easy
-# to correct against the trial-table builder; ``side_bias`` and the
-# ``lickspout_*`` arrays are not yet pinned down in trials_table_mapping.md.
+# to correct against the trial-table builder. The lickspout columns are the
+# ``lickspout_position_*`` fields documented in trials_table_mapping.md.
 _COLUMNS = {
     "animal_response": "animal_response",
     "side_bias": "side_bias",
-    "lickspout_x": "lickspout_x",
-    "lickspout_y1": "lickspout_y1",
-    "lickspout_y2": "lickspout_y2",
-    "lickspout_z": "lickspout_z",
+    "lickspout_x": "lickspout_position_x",
+    "lickspout_y1": "lickspout_position_y1",
+    "lickspout_y2": "lickspout_position_y2",
+    "lickspout_z": "lickspout_position_z",
     "rewarded_left": "rewarded_historyL",
     "rewarded_right": "rewarded_historyR",
     "reward_probability_left": "reward_probabilityL",