Add np_sync module for VR-to-NP barcode alignment and tests

.github/workflows/cron-scenario.yml

@@ -167,6 +167,7 @@ jobs:
             interpolated_trajectories
             session_metrics
             latency_tests
+            np_sync
             summary_emails
             inputs_videos
             decision
@@ -275,6 +276,10 @@ jobs:
           dataframe_count="$(docker exec vr4mice-mysql-restore mysql -N -uroot -psimple -e 'SELECT COUNT(*) FROM base_analysis.__data_frame;')"
           summary_plot_count="$(docker exec vr4mice-mysql-restore mysql -N -uroot -psimple -e 'SELECT COUNT(*) FROM base_analysis.__summary_plots;')"
           summary_email_count="$(docker exec vr4mice-mysql-restore mysql -N -uroot -psimple -e 'SELECT COUNT(*) FROM summary_emails.summary_plot_email WHERE send_error IS NULL;')"
+          np_sync_barcode_sync_count=""
+          if grep -qx 'np_sync' "$RUNNER_TEMP/mysql-dump/schemas.txt"; then
+            np_sync_barcode_sync_count="$(docker exec vr4mice-mysql-restore mysql -N -uroot -psimple -e 'SELECT COUNT(*) FROM np_sync.__barcode_sync;')"
+          fi
           if [ "${summary_email_count}" -lt 1 ] 2>/dev/null; then
             echo "summary_emails tables:" >&2
             docker exec vr4mice-mysql-restore mysql -uroot -psimple -e 'SHOW TABLES FROM summary_emails;' >&2 || true
@@ -290,6 +295,11 @@ jobs:
           echo "base_analysis.__data_frame rows: ${dataframe_count}"
           echo "base_analysis.__summary_plots rows: ${summary_plot_count}"
           echo "summary_emails.summary_plot_email successful rows: ${summary_email_count}"
+          if [ -n "${np_sync_barcode_sync_count}" ]; then
+            echo "np_sync.__barcode_sync rows: ${np_sync_barcode_sync_count}"
+          else
+            echo "np_sync schema not present in dump; skipping np_sync.__barcode_sync row-count check"
+          fi
 
           test "${dataset_count}" -ge 1
           test "${batch_count}" -ge 2
@@ -298,6 +308,9 @@ jobs:
           test "${state_count}" -ge 1
           test "${dataframe_count}" -ge 1
           test "${summary_plot_count}" -ge 1
+          if [ -n "${np_sync_barcode_sync_count}" ]; then
+            test "${np_sync_barcode_sync_count}" -ge 1
+          fi
 
       - name: Verify decision tables (--aws populated)
         run: |

dj_pipeline/.env.compose.example

@@ -11,6 +11,9 @@ DB_DATA_PATH=/mnt/database/vr4mice/vr4mice_database/database/
 SHARED_PATH=/mnt/database/shared
 DATA_PATH=/mnt/database/vr4mice/vr4mice_database/data
 SCREEN_RECORDINGS_PATH=/mnt/neuropixel_data/vr4mice/raw_screen_recordings
+# Optional: point to a real FreelyMoving_Mouse_NP checkout to enable np_sync.
+# If omitted, docker-compose defaults to ./np_pipeline_stub and np_sync is skipped.
+NP_PIPELINE_SRC_PATH=./np_pipeline_stub
 JUPYTER_PORT=8887
 CLIENT_IMAGE=mmathislab/vr4mice_app:0.1.0
 CLIENT_CONTAINER_NAME=vr4mice_${USER}

dj_pipeline/.env.example

@@ -1,5 +1,6 @@
 # Pipeline / DataJoint credentials (loaded into the client container via docker-compose env_file).
 # DJ_HOST port must match DB_PORT in .env.compose when using the local Docker database.
+# For the "np_sync" mode: DJ_HOST must match np_pipeline's DJ_HOST (same MySQL server).
 DJ_HOST=your-db-hostname-or-ip:3306
 DJ_USER=your-db-user
 DJ_PWD=your-db-password

dj_pipeline/Makefile

@@ -37,11 +37,16 @@ BUILD_ARGS := \
 .PHONY: build_all git_info git-whoami git-user-env git-user-local up_all base_install down_all client_build client_up \
 	client_stop client_rm client_down ipython bash mysql aws notebook sync_build \
 	sync_up add-cron cron-local cron-aws cron-now cron-aws-only cron-reboot import-dumps export-dumps \
-	check-compose-project
+	check-compose-project format
 
 check-compose-project:
 	@bash docker/check_compose_conflict.sh "$(COMPOSE_PROJECT)"
 
+# Matches CI's formatter.yml: black==22.6 against dj_pipeline.
+format:
+	pip install --quiet black==22.6
+	black .
+
 # Fail with logs if the client service is not running (common after compose project / name mismatch).
 define verify_client_running
 	@if [ "$$($(DOCKER_COMPOSE) ps --status running -q client 2>/dev/null | wc -l | tr -d ' ')" = "0" ]; then \

dj_pipeline/README.md

@@ -298,6 +298,7 @@ When using the local Docker database, `DJ_HOST` port must match `DB_PORT` in `.e
 - `COMPOSE_PROJECT` (default `vr4mice`) — must match between `make`, cron scripts, and manual `docker compose -p …` calls
 - `DB_BIND_IP`, `DB_PORT`, `MYSQL_ROOT_PASSWORD`
 - `DB_DATA_PATH`, `SHARED_PATH`, `DATA_PATH`, `SCREEN_RECORDINGS_PATH`
+- `NP_PIPELINE_SRC_PATH` (optional): path to a real `FreelyMoving_Mouse_NP` checkout. If unset, compose mounts `./np_pipeline_stub`, and only `run.py np_sync` is skipped while the rest of the pipeline remains functional.
 - `CLIENT_IMAGE`, `CLIENT_CONTAINER_NAME`, `DB_CONTAINER_NAME`, `CLIENT_NETWORK_MODE`, `JUPYTER_PORT`
 
 Remote/AWS DB credentials for scheduled AWS runs live in `.env-aws` (copy from `.env-aws.example`); this file is **not** committed.

dj_pipeline/cron_scenario.py

@@ -295,6 +295,33 @@ def import_decision_schema():
 
         return decision
 
+    def import_np_sync_schema():
+        from vr4mice.schema import np_sync
+
+        return np_sync
+
+    # np_sync depends on the separate np_pipeline package/repo and should not
+    # fail the whole cron scenario when unavailable in a deployment/CI setup.
+    try:
+        np_sync = import_np_sync_schema()
+    except Exception as err:
+        logger.warning(
+            "Skipping np_sync: np_pipeline is not available (%s). "
+            "Behavioral analysis is unaffected.",
+            err,
+        )
+    else:
+        from vr4mice.utils.populate_helpers import populate_pending
+
+        run_step(
+            "np_sync.BarcodeSync.populate",
+            lambda: populate_pending(
+                np_sync.BarcodeSync,
+                np_sync.BarcodeSync().key_source,
+                logger=logger,
+            ),
+        )
+
     decision = run_import("import decision schema", import_decision_schema)
     if decision:
         run_step(

dj_pipeline/docker-compose.yml

@@ -1,3 +1,5 @@
+name: ${COMPOSE_PROJECT:-vr4mice}
+
 services:
   db:
     image: ${DB_IMAGE:-mysql:8.0}
@@ -23,6 +25,7 @@ services:
       - MPLCONFIGDIR=/app/.cache/matplotlib
       - XDG_CONFIG_HOME=/app/.cache
       - PATH=/app/.local/bin:/usr/local/bin:/usr/bin:/bin
+      - PYTHONPATH=/app/np_pipeline/src:/app/np_pipeline:${PYTHONPATH}
       - TAG=${TAG}
     network_mode: ${CLIENT_NETWORK_MODE:-host}
     # Run via bash so the bind-mounted script does not need +x on shared/NFS checkouts.
@@ -36,7 +39,8 @@ services:
       - "${SCREEN_RECORDINGS_PATH:-/mnt/neuropixel_data/vr4mice/raw_screen_recordings}:/vr4mice_screen_recordings"
       - "./base/base_min_schemas:/base_schemas"
       - "./base/base_actions:/base_actions"
-    ports: 
+      - "${NP_PIPELINE_SRC_PATH:-./np_pipeline_stub}:/app/np_pipeline"
+    ports:
       - "${JUPYTER_PORT:-8887}:8888"
     working_dir: /app
     env_file:

dj_pipeline/run.py

@@ -25,6 +25,9 @@
     "latency": compute latencies based on photodiode signals
     "inputs_videos": process input videos and extract frames
     "decision": analyze decision-making metrics
+    "np_sync": align VR time to Neuropixels time via barcodes (requires the
+        np_pipeline package/repo; skips gracefully, without affecting any other
+        mode, if it isn't installed or no dataset has a linked NP recording yet)
     "maintenance": rebuild DataJoint lineage tables (one-time setup)
 """
 
@@ -77,9 +80,10 @@ def check_folder_existence(folder_path):
             "sync_days",
             "inputs_videos",
             "decision",
+            "np_sync",
             "maintenance",
         ],
-        help="Mode to execute: 'connect', 'populate', 'summary', 'dlc', 'fetch', 'sync_days', 'analysis', 'inputs_videos', 'decision', 'maintenance'",
+        help="Mode to execute: 'connect', 'populate', 'summary', 'dlc', 'fetch', 'sync_days', 'analysis', 'inputs_videos', 'decision', 'np_sync', 'maintenance'",
     )
 
     args = parser.parse_args()
@@ -281,6 +285,29 @@ def check_folder_existence(folder_path):
         decision.PredictionModel10Windows().populate()
         decision.DecisionPoints10Windows().populate()
 
+    elif args.mode == "np_sync":
+        # np_sync depends on the separate np_pipeline package/repo, which may be
+        # absent (ModuleNotFoundError) or unreachable (e.g. its own DB connection
+        # fails while its schema modules activate). Either way, isolate the
+        # failure to this mode so every other (purely behavioral) mode is
+        # unaffected.
+        try:
+            from vr4mice.schema import np_sync
+        except Exception as err:
+            logger.warning(
+                "Skipping np_sync: np_pipeline is not available (%s). "
+                "Behavioral analysis is unaffected.",
+                err,
+            )
+        else:
+            from vr4mice.utils.populate_helpers import populate_pending
+
+            populate_pending(
+                np_sync.BarcodeSync,
+                np_sync.BarcodeSync().key_source,
+                logger=logger,
+            )
+
     elif args.mode == "fetch":
         from vr4mice.actions.fetch_data import fetch_data
 

dj_pipeline/vr4mice/analysis/np_sync.py

@@ -0,0 +1,78 @@
+"""Fit a VR-time-to-NP-time alignment from barcode values shared by both streams."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import numpy as np
+import scipy.interpolate
+import scipy.stats
+
+# Number of leading (chronologically earliest) VR barcode events excluded from the
+# regression fit by default. DLC-live starts receiving data slightly after the
+# Unity/game stream does, which is why downstream analysis already drops
+# `trial == 1` as a DLC-live initialization trial (see vr4mice/analysis/analysis.py).
+# Barcodes in that same early window can carry an unreliable onset_time_unity,
+# biasing the fit if used as tie points, so we drop the earliest few by default.
+DEFAULT_SKIP_FIRST_N_BARCODES = 10
+
+
+@dataclass(frozen=True)
+class BarcodeAlignmentFit:
+    """Linear fit + interpolator mapping VR time to NP time."""
+
+    slope: float
+    intercept: float
+    r2: float
+    interpol_func: scipy.interpolate.interp1d
+    shared_barcodes: np.ndarray
+
+
+def align_barcodes(
+    vr_times: np.ndarray,
+    vr_values: np.ndarray,
+    np_times: np.ndarray,
+    np_values: np.ndarray,
+    skip_first_n_barcodes: int = 0,
+) -> BarcodeAlignmentFit:
+    """Fit VR time -> NP time from barcode values shared between both streams.
+
+    Alignment approach (intersect1d + linregress + interp1d) ported from
+    cross_analysis_schemas/schemas/vr_np_sync.py::BarcodeSync.align_barcodes in
+    https://github.com/AdaptiveMotorControlLab/auxPipelines-DataJoint_Mathis,
+    adapted for this repo's VR (vr4mice) / NP (np_pipeline) schemas.
+
+    Args:
+        vr_times: VR-side barcode onset times, ordered by event index (chronological).
+        vr_values: VR-side barcode integer payloads, same order as `vr_times`.
+        np_times: NP-side barcode onset times, ordered by event index.
+        np_values: NP-side barcode integer payloads, same order as `np_times`.
+        skip_first_n_barcodes: number of leading (earliest) VR events to exclude
+            before matching, to avoid the DLC-live startup-lag window.
+    """
+    if skip_first_n_barcodes:
+        vr_times = vr_times[skip_first_n_barcodes:]
+        vr_values = vr_values[skip_first_n_barcodes:]
+
+    shared_barcodes, vr_index, np_index = np.intersect1d(
+        vr_values, np_values, return_indices=True
+    )
+
+    vr_shared_times = np.asarray(vr_times)[vr_index]
+    np_shared_times = np.asarray(np_times)[np_index]
+
+    linreg = scipy.stats.linregress(vr_shared_times, np_shared_times)
+    interpol_func = scipy.interpolate.interp1d(
+        vr_shared_times,
+        np_shared_times,
+        bounds_error=False,
+        fill_value="extrapolate",
+    )
+
+    return BarcodeAlignmentFit(
+        slope=linreg.slope,
+        intercept=linreg.intercept,
+        r2=linreg.rvalue**2,
+        interpol_func=interpol_func,
+        shared_barcodes=shared_barcodes,
+    )