Merge pull request #394 from ttngu207/datajoint_pipeline

Pull from `main`, minor updates to automated worker

aeon/dj_pipeline/create_experiments/create_socialexperiment.py

@@ -0,0 +1,58 @@
+from pathlib import Path
+from datetime import datetime
+from aeon.dj_pipeline import acquisition
+from aeon.dj_pipeline.utils.paths import get_repository_path
+
+
+# ---- Programmatic creation of a new social experiment ----
+# Infer experiment metadata from the experiment name
+# User-specified "experiment_name" (everything else should be automatically inferred)
+experiment_name = "social0.4-aeon3"
+
+
+# Find paths
+ceph_dir = get_repository_path("ceph_aeon")
+ceph_data_dir = ceph_dir / "aeon" / "data"
+
+
+def create_new_social_experiment(experiment_name):
+    exp_name, machine_name = experiment_name.split("-")
+    raw_dir = ceph_data_dir / "raw" / machine_name.upper() / exp_name
+    if not raw_dir.exists():
+        raise FileNotFoundError(f"Raw directory not found: {raw_dir}")
+
+    start_time = sorted([f.name for f in raw_dir.glob("*") if f.is_dir()])[0]
+
+    new_experiment_entry = {
+        "experiment_name": experiment_name,
+        "experiment_start_time": datetime.strptime(start_time, "%Y-%m-%dT%H-%M-%S"),
+        "experiment_description": f"{exp_name.capitalize()} experiment on {machine_name.upper()} machine",
+        "arena_name": "circle-2m",
+        "lab": "SWC",
+        "location": machine_name.upper(),
+        "experiment_type": "social",
+    }
+    experiment_directories = [
+        {
+            "experiment_name": experiment_name,
+            "repository_name": "ceph_aeon",
+            "directory_type": dir_type,
+            "directory_path": (ceph_data_dir / dir_type / machine_name.upper() / exp_name)
+            .relative_to(ceph_dir)
+            .as_posix(),
+            "load_order": load_order,
+        }
+        for load_order, dir_type in enumerate(["processed", "raw"])
+    ]
+
+    with acquisition.Experiment.connection.transaction:
+        acquisition.Experiment.insert1(
+            new_experiment_entry,
+            skip_duplicates=True,
+        )
+        acquisition.Experiment.Directory.insert(experiment_directories, skip_duplicates=True)
+        acquisition.Experiment.DevicesSchema.insert1(
+            {"experiment_name": experiment_name, "devices_schema_name": exp_name.replace(".", "")},
+            skip_duplicates=True,
+        )
+

aeon/dj_pipeline/populate/worker.py

@@ -1,5 +1,5 @@
 import datajoint as dj
-from datajoint_utilities.dj_worker import DataJointWorker, ErrorLog, WorkerLog
+from datajoint_utilities.dj_worker import DataJointWorker, ErrorLog, WorkerLog, RegisteredWorker
 from datajoint_utilities.dj_worker.worker_schema import is_djtable
 
 from aeon.dj_pipeline import db_prefix

aeon/schema/foraging.py

@@ -83,7 +83,7 @@ class DeliverPellet(Stream):
     """Pellet delivery commands."""
 
     def __init__(self, pattern):
-        super().__init__(_reader.BitmaskEvent(f"{pattern}_35_*", 0x80, "TriggerPellet"))
+        super().__init__(_reader.BitmaskEvent(f"{pattern}_35_*", 0x01, "TriggerPellet"))
 
 
 class Patch(StreamGroup):

docker/cron_script.bash

@@ -48,4 +48,8 @@ if [ $? -eq 0 ]; then
     print_verbose "Workers restarted successfully."
 else
     print_verbose "Failed to restart workers."
-fi
+fi
+
+# Clean up old logs (older than 30 days)
+print_verbose "Cleaning up old logs..."
+find "${ROOT_LOG_DIR}" -mtime +30 -delete;

pyproject.toml

@@ -75,8 +75,8 @@ Repository = "https://github.com/sainsburyWellcomeCentre/aeon_mecha"
 Documentation = "https://sainsburywellcomecentre.github.io/aeon_docs/"
 DataJoint = "https://docs.datajoint.org/"
 
-[tool.setuptools]
-packages = ["aeon"]
+[tool.setuptools.packages.find]
+include = ["aeon*"]
 
 [tool.black]
 line-length = 108