feat: add lick latency plot to qc

src/dynamic_foraging_processing/qc/_core/result.py

@@ -24,8 +24,10 @@ class QCResult:
         Metric name.
     value : Any
         The computed value.
-    passed : bool
-        Whether the check passed.
+    passed : bool or None
+        Whether the check passed. ``None`` for a metric with no automated
+        pass/fail (its value is reported but the status is left ``PENDING`` for
+        manual review).
     description : str, optional
         Human-readable description.
     reference : str, optional
@@ -36,7 +38,7 @@ class QCResult:
 
     name: str
     value: t.Any
-    passed: bool
+    passed: t.Optional[bool]
     description: t.Optional[str] = None
     reference: t.Optional[str] = None
     tags: t.Dict[str, str] = dataclasses.field(default_factory=dict)

src/dynamic_foraging_processing/qc/_core/schema.py

@@ -51,23 +51,30 @@ def now_utc() -> datetime.datetime:
     return datetime.datetime.now(datetime.timezone.utc)
 
 
-def bool_to_status(passed: bool, timestamp: t.Optional[datetime.datetime] = None) -> QCStatus:
-    """Convert a boolean pass/fail into an automated ``QCStatus``.
+def bool_to_status(
+    passed: t.Optional[bool], timestamp: t.Optional[datetime.datetime] = None
+) -> QCStatus:
+    """Convert a boolean pass/fail (or ``None``) into an automated ``QCStatus``.
 
     Parameters
     ----------
-    passed : bool
-        ``True`` for a passing metric, ``False`` for a failing one.
+    passed : bool or None
+        ``True`` for a passing metric, ``False`` for a failing one, and ``None``
+        for a metric with no automated pass/fail — the value is reported but the
+        judgment is deferred, so the status is ``PENDING`` (needs manual review).
     timestamp : datetime.datetime, optional
         Timezone-aware evaluation time. Defaults to the current Seattle time.
 
     Returns
     -------
     QCStatus
-        An ``"Automated"`` status with ``PASS`` or ``FAIL``.
+        An ``"Automated"`` status with ``PASS``, ``FAIL``, or ``PENDING``.
     """
     timestamp = timestamp if timestamp is not None else now_seattle()
-    status = Status.PASS if passed else Status.FAIL
+    if passed is None:
+        status = Status.PENDING
+    else:
+        status = Status.PASS if passed else Status.FAIL
     return QCStatus(evaluator="Automated", status=status, timestamp=timestamp)
 
 

src/dynamic_foraging_processing/qc/processed/__init__.py

@@ -3,10 +3,12 @@
 from dynamic_foraging_processing.qc.processed.behavior import (
     calculate_lick_intervals,
     lick_interval_results,
+    lick_latency_result,
     side_bias_result,
 )
 from dynamic_foraging_processing.qc.processed.plots import (
     plot_lick_intervals,
+    plot_lick_latency,
     plot_side_bias,
 )
 from dynamic_foraging_processing.qc.processed.results import behavior_qc_results
@@ -17,7 +19,9 @@
     "behavior_qc_results",
     "calculate_lick_intervals",
     "lick_interval_results",
+    "lick_latency_result",
     "plot_lick_intervals",
+    "plot_lick_latency",
     "plot_side_bias",
     "side_bias_result",
 ]

src/dynamic_foraging_processing/qc/processed/behavior.py

@@ -17,6 +17,7 @@
 #: Reference plot assets shared by the behavior metrics.
 SIDE_BIAS_PLOT = "side_bias.png"
 LICK_INTERVALS_PLOT = "lick_intervals.png"
+LICK_LATENCY_PLOT = "lick_latency.png"
 
 
 def _plot_reference(plot_name: str, results_folder: t.Optional[str]) -> str:
@@ -165,6 +166,105 @@ def side_bias_result(side_bias: np.ndarray, results_folder: t.Optional[str] = No
     )
 
 
+def _first_lick_latency(go_cue: float, licks: np.ndarray) -> float:
+    """Return the latency (s) from ``go_cue`` to the first lick after it.
+
+    Parameters
+    ----------
+    go_cue : float
+        The trial's go-cue time (s). ``nan`` yields ``nan`` (no lick compares
+        greater than ``nan``).
+    licks : numpy.ndarray
+        Ascending lick timestamps (s).
+
+    Returns
+    -------
+    float
+        The first-lick latency, or ``nan`` when no lick follows the go cue.
+    """
+    after = licks[licks > go_cue]
+    if after.size:
+        return float(after[0] - go_cue)
+    return float("nan")
+
+
+def lick_latency_by_side(
+    go_cue_times: t.Optional[np.ndarray],
+    animal_response: t.Optional[np.ndarray],
+    left_lick_times: np.ndarray,
+    right_lick_times: np.ndarray,
+) -> t.Tuple[np.ndarray, np.ndarray]:
+    """Return per-trial first-lick latency (s) after the go cue, split by chosen side.
+
+    For each trial the latency is the time from the go cue to the first lick on
+    the *chosen* side — left when ``animal_response == 0``, right when ``== 1``.
+    Trials with no response (``2``), and any go cue after which the chosen side
+    never licks, are ``nan``. Slow or one-sided licking is the diagnostic signal
+    (e.g. deafness, or a non-functional lickport on one side).
+
+    Parameters
+    ----------
+    go_cue_times : numpy.ndarray or None
+        Per-trial go-cue times (s). ``None`` (column absent) is treated as no
+        trials.
+    animal_response : numpy.ndarray or None
+        Per-trial choice codes (``0`` left, ``1`` right, ``2`` ignore). ``None``
+        is treated as no trials.
+    left_lick_times, right_lick_times : numpy.ndarray
+        Timestamps (s) of left/right-port licks (need not be sorted).
+
+    Returns
+    -------
+    tuple of numpy.ndarray
+        The ``(left_latency, right_latency)`` per-trial arrays; each trial has a
+        latency on at most its chosen side, ``nan`` elsewhere.
+    """
+    if go_cue_times is None or animal_response is None:
+        return np.empty(0), np.empty(0)
+    go_cue = np.asarray(go_cue_times, dtype=float)
+    response = np.asarray(animal_response)
+    left = np.sort(np.asarray(left_lick_times, dtype=float))
+    right = np.sort(np.asarray(right_lick_times, dtype=float))
+    left_latency = np.full(go_cue.shape, np.nan)
+    right_latency = np.full(go_cue.shape, np.nan)
+    for i, cue in enumerate(go_cue):
+        if response[i] == 0:
+            left_latency[i] = _first_lick_latency(cue, left)
+        elif response[i] == 1:
+            right_latency[i] = _first_lick_latency(cue, right)
+    return left_latency, right_latency
+
+
+def lick_latency_result(results_folder: t.Optional[str] = None) -> QCResult:
+    """Build the review-only first-lick-latency ``QCResult``.
+
+    A single review-only metric surfacing the lick-latency plot (per-side
+    first-lick latency after the go cue): there is no computed value
+    (``value=None``) and no automated pass/fail (``passed=None`` -> ``PENDING``).
+    Tagged ``type="Lick_Interval"`` so it groups with the lick-interval metrics.
+
+    Parameters
+    ----------
+    results_folder : str, optional
+        Directory the lick-latency plot is written to; used to build the
+        result's reference. When ``None``, the reference is the bare plot name.
+
+    Returns
+    -------
+    QCResult
+        The lick-latency result (``PENDING``, no value or auto pass/fail)
+        referencing the lick-latency plot.
+    """
+    return QCResult(
+        name="Lick_Latency",
+        value=None,
+        passed=None,  # no automated pass/fail -> PENDING for manual review
+        description="First-lick latency (s) after the go cue, by side (review-only).",
+        reference=_plot_reference(LICK_LATENCY_PLOT, results_folder),
+        tags={"metric": "Lick_Latency", "type": "Lick_Interval"},
+    )
+
+
 def lick_interval_results(
     left_lick_times: np.ndarray,
     right_lick_times: np.ndarray,

src/dynamic_foraging_processing/qc/processed/plots.py

@@ -17,7 +17,9 @@
 
 from dynamic_foraging_processing.qc.processed.behavior import (
     LICK_INTERVALS_PLOT,
+    LICK_LATENCY_PLOT,
     SIDE_BIAS_PLOT,
+    lick_latency_by_side,
 )
 
 
@@ -85,6 +87,57 @@ def plot_lick_intervals(
     return LICK_INTERVALS_PLOT
 
 
+def plot_lick_latency(
+    go_cue_times: t.Optional[np.ndarray],
+    animal_response: t.Optional[np.ndarray],
+    left_lick_times: np.ndarray,
+    right_lick_times: np.ndarray,
+    results_folder: str,
+) -> str:
+    """Save the per-side first-lick-latency histogram (response to the go cue).
+
+    One overlaid histogram of the time from the go cue to the first lick on the
+    chosen side (right and left, density-normalized). It shows how quickly the
+    animal licks each side after the go cue; a shifted or absent distribution on
+    one side is the diagnostic signal (e.g. deafness or a dead lickport).
+
+    Parameters
+    ----------
+    go_cue_times, animal_response : numpy.ndarray or None
+        Per-trial go-cue times and choice codes (see ``lick_latency_by_side``).
+    left_lick_times, right_lick_times : numpy.ndarray
+        Timestamps (s) of left/right-port licks.
+    results_folder : str
+        Directory to write ``lick_latency.png`` into.
+
+    Returns
+    -------
+    str
+        The plot filename (``lick_latency.png``), for use as a metric
+        ``reference``.
+    """
+    left_latency, right_latency = lick_latency_by_side(
+        go_cue_times, animal_response, left_lick_times, right_lick_times
+    )
+
+    fig, ax = plt.subplots(figsize=(5, 4))
+    ax.spines["top"].set_visible(False)
+    ax.spines["right"].set_visible(False)
+    bins = np.arange(0, 1, 0.05)
+    ax.hist(right_latency[~np.isnan(right_latency)], bins=bins, alpha=0.5, label="R", density=True)
+    ax.hist(left_latency[~np.isnan(left_latency)], bins=bins, alpha=0.5, label="L", density=True)
+    ax.legend()
+    ax.set_title("lick latency by lick side")
+    ax.set_xlabel("Time from go cue (s)")
+    ax.set_ylabel("density %")
+    ax.set_xlim(left=0)
+
+    fig.tight_layout()
+    fig.savefig(Path(results_folder) / LICK_LATENCY_PLOT, dpi=300, bbox_inches="tight")
+    plt.close(fig)
+    return LICK_LATENCY_PLOT
+
+
 def _add_bias_plot(ax: plt.Axes, side_bias: np.ndarray) -> None:
     """Draw the per-trial side-bias trace from the trial-table column."""
     ax.set_xlabel("Trial #")

src/dynamic_foraging_processing/qc/processed/results.py

@@ -17,9 +17,14 @@
 from dynamic_foraging_processing.qc._core.result import QCResult
 from dynamic_foraging_processing.qc.processed.behavior import (
     lick_interval_results,
+    lick_latency_result,
     side_bias_result,
 )
-from dynamic_foraging_processing.qc.processed.plots import plot_lick_intervals, plot_side_bias
+from dynamic_foraging_processing.qc.processed.plots import (
+    plot_lick_intervals,
+    plot_lick_latency,
+    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
@@ -64,9 +69,9 @@ def behavior_qc_results(
 ) -> t.List[QCResult]:
     """Build the behavior QC results (side bias + lick intervals).
 
-    When ``results_folder`` is provided, the supporting ``side_bias.png`` and
-    ``lick_intervals.png`` plots are written there so the result references
-    resolve. Convert the returned results to schema metrics with
+    When ``results_folder`` is provided, the supporting ``side_bias.png``,
+    ``lick_intervals.png``, and ``lick_latency.png`` plots are written there so
+    the result references resolve. Convert the returned results to schema metrics with
     ``to_metrics`` / ``QCResult.to_metric`` when assembling a ``QualityControl``.
 
     Parameters
@@ -89,12 +94,16 @@ def behavior_qc_results(
     Returns
     -------
     list of QCResult
-        The average-side-bias result followed by the four lick-interval results.
+        The average-side-bias result, the four lick-interval results, and the
+        review-only lick-latency result.
     """
     side_bias = _column(trials, "side_bias")
+    go_cue_times = _column(trials, "go_cue_times")
+    animal_response = _column(trials, "animal_response")
     results = [
         side_bias_result(side_bias, results_folder),
         *lick_interval_results(left_lick_times, right_lick_times, results_folder),
+        lick_latency_result(results_folder),
     ]
     if results_folder is not None:
         plot_side_bias(
@@ -116,4 +125,7 @@ def behavior_qc_results(
             manual_right_times=manual_right_times,
         )
         plot_lick_intervals(left_lick_times, right_lick_times, results_folder)
+        plot_lick_latency(
+            go_cue_times, animal_response, left_lick_times, right_lick_times, results_folder
+        )
     return results

src/dynamic_foraging_processing/qc/processed/stage.py

@@ -51,7 +51,8 @@ def run(
         Returns
         -------
         list of QCMetric
-            The side-bias metric followed by the four lick-interval metrics.
+            The side-bias metric, the four lick-interval metrics, and the
+            review-only lick-latency metric.
         """
         results = behavior_qc_results(
             trials,

tests/test_qc/test_behavior.py

@@ -78,6 +78,50 @@ def test_lick_interval_results_names_and_count():
     assert all(r.tags == {"metric": r.name, "type": "Lick_Interval"} for r in results)
 
 
+def test_first_lick_latency_after_go_cue_and_none():
+    """The first lick after the go cue gives the latency; no later lick -> nan."""
+    licks = np.array([0.5, 1.2, 2.0])
+    assert _behavior._first_lick_latency(1.0, licks) == pytest.approx(0.2)
+    # No lick after the cue -> nan.
+    assert np.isnan(_behavior._first_lick_latency(2.5, licks))
+    # A nan go cue has no lick strictly greater than it -> nan.
+    assert np.isnan(_behavior._first_lick_latency(float("nan"), licks))
+
+
+def test_lick_latency_by_side_splits_on_choice():
+    """Latency is measured on the chosen side; other side / ignore trials are nan."""
+    go_cue = np.array([0.0, 1.0, 2.0, 3.0])
+    response = np.array([0, 1, 2, 1])  # left, right, ignore, right
+    left_licks = np.array([0.3])  # after the trial-0 cue
+    right_licks = np.array([1.4, 3.2])  # after the trial-1 and trial-3 cues
+    left_latency, right_latency = _behavior.lick_latency_by_side(
+        go_cue, response, left_licks, right_licks
+    )
+    assert left_latency[0] == pytest.approx(0.3)
+    assert np.isnan(left_latency[1])  # right-choice trial has no left latency
+    assert right_latency[1] == pytest.approx(0.4)
+    assert right_latency[3] == pytest.approx(0.2)
+    assert np.isnan(right_latency[2])  # ignore trial
+
+
+def test_lick_latency_by_side_none_inputs_return_empty():
+    """Absent go-cue / response columns yield empty latency arrays."""
+    left, right = _behavior.lick_latency_by_side(None, None, np.array([1.0]), np.array([2.0]))
+    assert left.size == 0 and right.size == 0
+
+
+def test_lick_latency_result_is_pending_review_only():
+    """The single latency result is review-only: no value, PENDING, plot ref."""
+    result = _behavior.lick_latency_result("/data/my_results")
+    assert result.name == "Lick_Latency"
+    # No computed value yet, and no automated pass/fail (renders as PENDING).
+    assert result.value is None
+    assert result.passed is None
+    assert result.reference == f"my_results/{_behavior.LICK_LATENCY_PLOT}"
+    # Tagged Lick_Interval so it groups with the lick-interval metrics.
+    assert result.tags == {"metric": "Lick_Latency", "type": "Lick_Interval"}
+
+
 def test_reference_includes_results_folder_name():
     """With a results_folder, references are '<folder-name>/<plot>'."""
     side_bias = _behavior.side_bias_result(np.array([0.1]), "/data/my_results")