feat: ingest new manual water software events

docs/trials_table_mapping.md

@@ -38,7 +38,12 @@ Each reward-delivery timestamp carries a label in the series' `data` field:
 | --- | --- |
 | `earned` | Water the animal worked for: the matched trial has no free water (`is_auto_reward_right` is `None`). |
 | `auto` | Free water: the matched trial has `is_auto_reward_right` set. Scheduled autowater and the anti-bias intervention share that channel and are **not** split here — `auto_waterL` / `auto_waterR` and `anti_bias_left_water` / `anti_bias_right_water` record the mechanism per trial. |
-| `manual` | The delivery is the closest valve opening to a `GiveManualWater` software event for this port. Takes precedence over the other labels, since manual water is not aligned to a go cue. |
+| `manual_go_cue_aligned` | The delivery is the closest valve opening to a `LeftManualAutoReward` / `RightManualAutoReward` software event for this port: water the *experimenter* triggered to land on the go cue. It fires at the go cue like autowater, but the task did not schedule it, so it is neither `auto` nor `earned`. Takes precedence over both trial-derived labels. |
+| `manual` | The delivery is the closest valve opening to a `LeftManualWater` / `RightManualWater` software event for this port: experimenter water given at an arbitrary moment, not tied to a go cue. Highest precedence of all four. |
+
+The side of an experimenter-water event comes from the **stream name**, not from
+an event payload. Each of the four streams exists only when the experimenter gave
+water of that kind, so a session with none of them is normal.
 
 Two properties of this series are worth stating explicitly, because both differ
 from "every time the valve opened":
@@ -230,3 +235,4 @@ These were mapped during exploration but are no longer in scope:
 | 2026-08-20 | `block_max` is now one below `block_length`'s configured maximum, which accounts for the floor applied upstream: a block is a whole number of trials, so the configured bound is never itself reachable. `block_min`, `block_beta`, and the `ITI_*` / `delay_*` bounds are unchanged — those durations are continuous and take no such adjustment. |
 | 2026-08-20 | `ITI_min` now reports `inter_trial_interval_duration`'s scaling `offset` instead of its truncation minimum: the sampled ITI is shifted by the offset, so the offset is the shortest ITI the generator can produce. Falls back to the truncation minimum when no scaling parameters are configured. |
 | 2026-08-20 | `bait_left` / `bait_right` now read `trial.metadata.extra.is_left_baited` / `is_right_baited` from the acquisition software instead of being re-derived from `p_reward_left` / `p_reward_right` and the `is_auto_reward_right` channel. The software is the authority on bait state, so the two can disagree — notably a port with `p_reward == 1` is no longer assumed baited. `False` when the trial carries no extra metadata. |
+| 2026-09-15 | **Breaking:** experimenter water is now read from the four side-specific software-event streams the acquisition software emits (`LeftManualWater` / `RightManualWater`, not aligned to a go cue, and `LeftManualAutoReward` / `RightManualAutoReward`, aligned to it) instead of the single `GiveManualWaterRight` stream whose `data` payload selected the side. The `GiveManualWaterRight` path is removed, not deprecated. A fourth reward-delivery label, `manual_go_cue_aligned`, joins `earned` / `auto` / `manual`. **This fixes a mislabel:** manual auto-rewards fire at the go cue but leave `is_auto_reward_right` unset, so they previously fell through to `earned` — water the animal never worked for, counted as earned. The QC `side_bias.png` behavior raster gains an `L` / `R Manual Water (go cue)` row per side (now 11 rows, y-limits `[-0.8, 1.8]`), drawn dotted where unaligned manual water is dashed. |

pyproject.toml

@@ -17,7 +17,7 @@ readme = "README.md"
 version = "0.3.8"
 
 dependencies = [
-    "aind-behavior-dynamic-foraging[data] @ git+https://github.com/AllenNeuralDynamics/Aind.Behavior.DynamicForaging.git@933a7aee2627de1979a982847f3553cf7435cbd2",
+    "aind-behavior-dynamic-foraging[data] @ git+https://github.com/AllenNeuralDynamics/Aind.Behavior.DynamicForaging.git@4e4faa4941c8d9cc8024e760fdaec30339b626dc",
     "ipykernel",
 ]
 

src/dynamic_foraging_processing/nwb/acquisition/acquisition_builder.py

@@ -11,7 +11,7 @@
 )
 from dynamic_foraging_processing.nwb.utils import clean_for_nwb
 from dynamic_foraging_processing.raw_data_loader import RawDataLoader
-from dynamic_foraging_processing.utils.rewards import get_reward_deliveries
+from dynamic_foraging_processing.utils.rewards import ManualWaterTimes, get_reward_deliveries
 
 
 class LickSource(t.NamedTuple):
@@ -96,28 +96,54 @@ def get_response_times(self) -> np.ndarray:
         )
         return responses.index.to_numpy()
 
-    def get_manual_water_times(self) -> pd.DataFrame:
-        """Get the manual-water software-event stream.
+    def _software_event_times(self, stream_name: str) -> np.ndarray:
+        """Get one ``Behavior/SoftwareEvents`` stream's event timestamps.
+
+        Parameters
+        ----------
+        stream_name : str
+            The software-event stream to read (e.g. ``"LeftManualWater"``).
 
         Returns
         -------
-        pandas.DataFrame
-            The ``GiveManualWaterRight`` stream under ``Behavior/SoftwareEvents``,
-            indexed by event timestamp with a ``data`` column that is ``True``
-            for right-port manual water and ``False`` for left-port manual water.
-            An empty frame (with a ``data`` column) is returned when the stream
-            is absent.
+        numpy.ndarray
+            The stream's event timestamps, or an empty array when the stream is
+            absent. Only the timestamps are used; these events carry no payload
+            this pipeline reads.
         """
         try:
-            return (
-                self.loader.dataset.at("Behavior")
-                .at("SoftwareEvents")
-                .at("GiveManualWaterRight")
-                .load()
-                .data
+            data = (
+                self.loader.dataset.at("Behavior").at("SoftwareEvents").at(stream_name).load().data
             )
         except (KeyError, FileNotFoundError):
-            return pd.DataFrame({"data": []})
+            return np.array([])
+        return data.index.to_numpy()
+
+    def get_manual_water_times(self, *, is_right: bool) -> ManualWaterTimes:
+        """Get one lick port's experimenter-triggered water times.
+
+        The acquisition software emits four side-specific streams:
+        ``{Left,Right}ManualWater`` for water given at an arbitrary moment and
+        ``{Left,Right}ManualAutoReward`` for water triggered to land on the go
+        cue. The side comes from the stream name, so no payload inspection is
+        needed. Each stream is optional -- a session where the experimenter gave
+        no water of that kind has no file -- and reads as an empty array.
+
+        Parameters
+        ----------
+        is_right : bool
+            ``True`` for the right lick port, ``False`` for the left.
+
+        Returns
+        -------
+        ManualWaterTimes
+            This port's ``unaligned`` and ``go_cue_aligned`` event timestamps.
+        """
+        side = "Right" if is_right else "Left"
+        return ManualWaterTimes(
+            unaligned=self._software_event_times(f"{side}ManualWater"),
+            go_cue_aligned=self._software_event_times(f"{side}ManualAutoReward"),
+        )
 
     def get_lick_times(self, device: str, stream_name: str, port: str) -> np.ndarray:
         """Get the lick times for one lick port from a Harp digital-input stream.
@@ -191,38 +217,36 @@ def _reward_delivery_series(
         self,
         writes: pd.DataFrame,
         trial_outcomes: pd.DataFrame,
-        manual_water: pd.DataFrame,
+        manual_water: ManualWaterTimes,
         response_times: np.ndarray,
         *,
         port_column: str,
-        is_right: bool,
         name: str,
         side_label: str,
     ) -> AcquisitionSeries:
         """Build one lick port's reward-delivery series with reward annotations.
 
         Only valve-open events (``port_column`` is truthy) are reward
-        deliveries; the ``data`` field annotates each as earned, manual, or auto
-        via :func:`get_reward_deliveries`. Every valve opening is reported, so
-        the series is a complete record of the water delivered at this port.
+        deliveries; the ``data`` field annotates each as earned, auto, manual, or
+        manual-go-cue-aligned via :func:`get_reward_deliveries`. Every valve
+        opening is reported, so the series is a complete record of the water
+        delivered at this port.
 
         Parameters
         ----------
         writes : pandas.DataFrame
             ``OutputSet`` ``WRITE`` messages indexed by timestamp.
         trial_outcomes : pandas.DataFrame
             The ``TrialOutcome`` stream, indexed by trial timestamp.
-        manual_water : pandas.DataFrame
-            The ``GiveManualWaterRight`` stream; the ``data`` column selects the
-            side (``True`` right, ``False`` left).
+        manual_water : ManualWaterTimes
+            This port's experimenter-triggered water times, already side-specific
+            (see :meth:`get_manual_water_times`).
         response_times : numpy.ndarray
             ``Response`` event timestamps, one per trial, used to match each
             delivery to its trial.
         port_column : str
             Supply-port column for this side (``"SupplyPort0"`` left,
             ``"SupplyPort1"`` right).
-        is_right : bool
-            ``True`` for the right lick port, ``False`` for the left.
         name : str
             Acquisition series name.
         side_label : str
@@ -235,11 +259,10 @@ def _reward_delivery_series(
         """
         open_writes = writes[writes[port_column].fillna(False).astype(bool)]
         delivery_times = open_writes.index.to_numpy()
-        manual_water_times = manual_water.index[manual_water["data"] == is_right].to_numpy()
         annotations = get_reward_deliveries(
             delivery_times,
             trial_outcomes,
-            manual_water_times,
+            manual_water,
             response_times,
         )
         return AcquisitionSeries(
@@ -249,7 +272,9 @@ def _reward_delivery_series(
             unit="second",
             description=(
                 f"The reward delivery time of the {side_label} lick port. The data field "
-                "annotates whether the reward was earned, manual, or auto"
+                "annotates whether the reward was earned, auto (task-triggered free water), "
+                "manual (experimenter water not aligned to a go cue), or "
+                "manual_go_cue_aligned (experimenter water delivered at the go cue)"
             ),
         )
 
@@ -276,7 +301,8 @@ def build_acquisition(
         """
         rewards = self.get_valve_writes()
         trial_outcomes = self.get_trial_outcomes()
-        manual_water = self.get_manual_water_times()
+        left_manual_water = self.get_manual_water_times(is_right=False)
+        right_manual_water = self.get_manual_water_times(is_right=True)
         response_times = self.get_response_times()
 
         acquisition_streams = self.loader.get_all_raw_data()
@@ -300,10 +326,9 @@ def build_acquisition(
             self._reward_delivery_series(
                 rewards,
                 trial_outcomes,
-                manual_water,
+                left_manual_water,
                 response_times,
                 port_column="SupplyPort0",
-                is_right=False,
                 name="left_reward_delivery_time",
                 side_label="left",
             )
@@ -312,10 +337,9 @@ def build_acquisition(
             self._reward_delivery_series(
                 rewards,
                 trial_outcomes,
-                manual_water,
+                right_manual_water,
                 response_times,
                 port_column="SupplyPort1",
-                is_right=True,
                 name="right_reward_delivery_time",
                 side_label="right",
             )

src/dynamic_foraging_processing/pipeline/_pipeline.py

@@ -48,6 +48,11 @@
 from dynamic_foraging_processing.processing import TrialConfig, TrialTableBuilder
 from dynamic_foraging_processing.qc import ProcessedQC, RawQC, build_quality_control
 from dynamic_foraging_processing.raw_data_loader import RawDataLoader
+from dynamic_foraging_processing.utils.rewards import (
+    MANUAL,
+    MANUAL_GO_CUE_ALIGNED,
+    ManualWaterTimes,
+)
 
 #: Default lick-port sources on the standard behavior board.
 _DEFAULT_LEFT_LICK = LickSource("HarpBehavior", "DigitalInputState", "DIPort0")
@@ -84,9 +89,6 @@
 _LEFT_REWARD_SERIES = "left_reward_delivery_time"
 _RIGHT_REWARD_SERIES = "right_reward_delivery_time"
 
-#: Reward-delivery annotation marking a manual-water event.
-_MANUAL_ANNOTATION = "manual"
-
 
 class Pipeline:
     """Package a raw dynamic foraging acquisition to NWB and run QC.
@@ -345,7 +347,7 @@ def _write_processing(
     # ------------------------------------------------------------------ #
     def _read_processed_inputs(
         self, nwb_file: pynwb.NWBFile
-    ) -> t.Tuple[pd.DataFrame, np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
+    ) -> t.Tuple[pd.DataFrame, np.ndarray, np.ndarray, ManualWaterTimes, ManualWaterTimes]:
         """Read the processed-QC inputs from an NWB file.
 
         Parameters
@@ -356,25 +358,38 @@ def _read_processed_inputs(
         Returns
         -------
         tuple
-            ``(trials, left_lick_times, right_lick_times, manual_left_times,
-            manual_right_times)``.
+            ``(trials, left_lick_times, right_lick_times, manual_left,
+            manual_right)``, where the last two are :class:`ManualWaterTimes`.
         """
         trials = nwb_file.trials.to_dataframe()
         left_lick_times = np.asarray(nwb_file.acquisition[_LEFT_LICK_SERIES].timestamps)
         right_lick_times = np.asarray(nwb_file.acquisition[_RIGHT_LICK_SERIES].timestamps)
-        manual_left_times, manual_right_times = self._manual_water_times(nwb_file)
-        return trials, left_lick_times, right_lick_times, manual_left_times, manual_right_times
+        manual_left, manual_right = self._manual_water_times(nwb_file)
+        return trials, left_lick_times, right_lick_times, manual_left, manual_right
 
     @staticmethod
-    def _manual_water_times(nwb_file: pynwb.NWBFile) -> t.Tuple[np.ndarray, np.ndarray]:
-        """Return the ``(left, right)`` manual-water delivery times from the NWB.
+    def _manual_water_times(
+        nwb_file: pynwb.NWBFile,
+    ) -> t.Tuple[ManualWaterTimes, ManualWaterTimes]:
+        """Return the ``(left, right)`` experimenter-water times from the NWB.
 
-        Manual-water deliveries are the reward-delivery events annotated
-        ``"manual"`` on each side's ``*_reward_delivery_time`` acquisition series.
+        Each side's times are the reward-delivery events annotated ``"manual"``
+        (not aligned to a go cue) and ``"manual_go_cue_aligned"`` on that side's
+        ``*_reward_delivery_time`` acquisition series. The two are read back
+        separately so the QC figure can keep them on their own rows.
         """
-        left = Pipeline._annotated_times(nwb_file, _LEFT_REWARD_SERIES, _MANUAL_ANNOTATION)
-        right = Pipeline._annotated_times(nwb_file, _RIGHT_REWARD_SERIES, _MANUAL_ANNOTATION)
-        return left, right
+        return (
+            Pipeline._side_manual_water_times(nwb_file, _LEFT_REWARD_SERIES),
+            Pipeline._side_manual_water_times(nwb_file, _RIGHT_REWARD_SERIES),
+        )
+
+    @staticmethod
+    def _side_manual_water_times(nwb_file: pynwb.NWBFile, series_name: str) -> ManualWaterTimes:
+        """Split one reward-delivery series' experimenter-water times by alignment."""
+        return ManualWaterTimes(
+            unaligned=Pipeline._annotated_times(nwb_file, series_name, MANUAL),
+            go_cue_aligned=Pipeline._annotated_times(nwb_file, series_name, MANUAL_GO_CUE_ALIGNED),
+        )
 
     @staticmethod
     def _annotated_times(nwb_file: pynwb.NWBFile, series_name: str, annotation: str) -> np.ndarray:
@@ -389,8 +404,8 @@ def _assemble_quality_control(
         trials: pd.DataFrame,
         left_lick_times: np.ndarray,
         right_lick_times: np.ndarray,
-        manual_left_times: np.ndarray,
-        manual_right_times: np.ndarray,
+        manual_left: ManualWaterTimes,
+        manual_right: ManualWaterTimes,
         results_folder: t.Optional[str] = None,
     ) -> QualityControl:
         """Run the raw and processed QC stages and assemble one ``QualityControl``.
@@ -401,9 +416,9 @@ def _assemble_quality_control(
             The trials table, consumed by the processed (behavior) QC stage.
         left_lick_times, right_lick_times : numpy.ndarray
             Left/right-port lick times for the processed QC stage.
-        manual_left_times, manual_right_times : numpy.ndarray
-            Left/right manual-water delivery times passed through to the side-bias
-            figure.
+        manual_left, manual_right : ManualWaterTimes
+            Left/right experimenter-water delivery times, split into unaligned
+            and go-cue-aligned, passed through to the side-bias figure.
         results_folder : str, optional
             Directory to write figure assets into so the metric references
             resolve. If ``None``, assets are skipped.
@@ -420,8 +435,8 @@ def _assemble_quality_control(
             left_lick_times,
             right_lick_times,
             results_folder,
-            manual_left_times=manual_left_times,
-            manual_right_times=manual_right_times,
+            manual_left=manual_left,
+            manual_right=manual_right,
         )
         return build_quality_control([*raw_metrics, *processed_metrics])
 
@@ -486,7 +501,7 @@ def run_qc(
             ``output_path`` is given, the QC JSON and figure assets are written to
             disk as a side effect.
         """
-        trials, left_lick_times, right_lick_times, manual_left_times, manual_right_times = (
+        trials, left_lick_times, right_lick_times, manual_left, manual_right = (
             self._read_processed_inputs(nwb_file)
         )
 
@@ -500,8 +515,8 @@ def run_qc(
             trials,
             left_lick_times,
             right_lick_times,
-            manual_left_times,
-            manual_right_times,
+            manual_left,
+            manual_right,
             results_folder,
         )
         if output_path is not None: