automate video pipeline #19, #18, AllenNeuralDynamics/dynamic-routing#11

src/npc_lims/metadata/codeocean.py

@@ -51,9 +51,9 @@
 MODEL_CAPSULE_MAPPING: dict[str, str] = {
     "dlc_eye": "4cf0be83-2245-4bb1-a55c-a78201b14bfe",
     "dlc_side": "facff99f-d3aa-4ecd-8ef8-a343c38197aa",
-    "dlc_front": "a561aa4c-2066-4ff2-a916-0db86b918cdf",
+    "dlc_face": "a561aa4c-2066-4ff2-a916-0db86b918cdf",
     "facemap": "670de0b3-f73d-4d22-afe6-6449c45fada4",
-    "video_pipeline": "edbc4721-d251-4cca-ba39-db8f167c3468",
+    "video_pipeline": "edbc4721-d251-4cca-ba39-db8f167c3468"
 }
 
 
@@ -112,7 +112,7 @@ def get_session_data_assets(
         asset
         for asset in assets
         if re.match(
-            f"ecephys_{session.subject}_{session.date}_{npc_session.PARSE_TIME}",
+            f"ecephys_{session.subject}_{session.date}_{npc_session.PARSE_TIME}(_[a-z]*_[a-z]*)*",
             asset["name"],
         )
     )
@@ -469,6 +469,26 @@ def get_session_computation_id_and_data_asset_name(
     return session_comp_id_data_asset_name
 
 
+def get_model_data_asset(session: str | npc_session.SessionRecord, model_name: str) -> DataAssetAPI:
+    """
+    Returns the data asset for a given model
+    >>> model_asset = get_model_data_asset('676909_2023-12-13', 'dlc_eye')
+    >>> model_asset['name']
+    'ecephys_676909_2023-12-13_13-43-40_dlc_eye'
+    """
+    session = npc_session.SessionRecord(session)
+    if model_name not in MODEL_CAPSULE_MAPPING:
+        raise ModelCapsuleMappingError(
+            f"No capsule associated with {model_name}. Check codeocean"
+        )
+
+    session_data_assets = get_session_data_assets(session)
+    session_model_asset = tuple(asset for asset in session_data_assets if model_name in asset['name'])
+    if not session_model_asset:
+        raise FileNotFoundError(f'{session} has no {model_name} results')
+
+    return session_model_asset[0]
+
 def create_session_data_asset(
     session: str | npc_session.SessionRecord, model_name: str
 ) -> None:
@@ -494,9 +514,10 @@ def create_session_data_asset(
     source = aind_codeocean_requests.Source(
         computation=aind_codeocean_requests.Sources.Computation(id=computation_id)
     )
+    custom_metadata = {"subject id": str(session.subject)}
     tags = [model_name, "results"]
     create_data_asset_request = aind_codeocean_requests.CreateDataAssetRequest(
-        name=data_asset_name, mount=data_asset_name, tags=tags, source=source
+        name=data_asset_name, mount=data_asset_name, tags=tags, source=source, custom_metadata=custom_metadata
     )
 
     get_codeocean_client().create_data_asset(

src/npc_lims/paths/s3.py

@@ -93,6 +93,53 @@ def get_sorted_data_paths_from_s3(
         raise ValueError("Must provide either session or sorted_data_asset_id")
     return tuple(get_data_asset_s3_path(sorted_data_asset).iterdir())
 
+@functools.cache
+def get_dlc_eye_s3_paths(session: str | npc_session.SessionRecord) -> tuple[upath.UPath, ...]:
+    """
+    >>> paths = get_dlc_eye_s3_paths('676909_2023-12-13')
+    >>> len(paths)
+    7
+    """
+    session = npc_session.SessionRecord(session)
+    dlc_eye_data_asset = codeocean.get_model_data_asset(session, 'dlc_eye')
+
+    return tuple(get_data_asset_s3_path(dlc_eye_data_asset).iterdir())
+
+@functools.cache
+def get_dlc_side_s3_paths(session: str | npc_session.SessionRecord) -> tuple[upath.UPath, ...]:
+    """
+    >>> paths = get_dlc_side_s3_paths('676909_2023-12-13')
+    >>> len(paths)
+    5
+    """
+    session = npc_session.SessionRecord(session)
+    dlc_eye_data_asset = codeocean.get_model_data_asset(session, 'dlc_side')
+
+    return tuple(get_data_asset_s3_path(dlc_eye_data_asset).iterdir())
+
+@functools.cache
+def get_dlc_face_s3_paths(session: str | npc_session.SessionRecord) -> tuple[upath.UPath, ...]:
+    """
+    >>> paths = get_dlc_face_s3_paths('676909_2023-12-13')
+    >>> len(paths)
+    5
+    """
+    session = npc_session.SessionRecord(session)
+    dlc_eye_data_asset = codeocean.get_model_data_asset(session, 'dlc_face')
+
+    return tuple(get_data_asset_s3_path(dlc_eye_data_asset).iterdir())
+
+@functools.cache
+def get_facemap_s3_paths(session: str | npc_session.SessionRecord) -> tuple[upath.UPath, ...]:
+    """
+    >>> paths = get_facemap_s3_paths('676909_2023-12-13')
+    >>> len(paths)
+    4
+    """
+    session = npc_session.SessionRecord(session)
+    dlc_eye_data_asset = codeocean.get_model_data_asset(session, 'facemap')
+
+    return tuple(get_data_asset_s3_path(dlc_eye_data_asset).iterdir())
 
 @functools.cache
 def get_settings_xml_path_from_s3(

src/npc_lims/scripts/generate_session_video_data_assets.py

@@ -0,0 +1,22 @@
+import npc_lims.metadata.codeocean as codeocean
+import npc_lims.status as status
+
+def generate_helper(session_info: status.SessionInfo, model_name: str) -> None:
+      if not getattr(session_info, model_name):
+        try:
+            codeocean.create_session_data_asset(session_info.id, model_name)
+        except UnboundLocalError:
+            pass
+
+def main() -> None:
+    for session_info in status.get_session_info():
+        if not session_info.is_uploaded:
+            continue
+
+        generate_helper(session_info, 'dlc_eye')
+        generate_helper(session_info, 'dlc_face')
+        generate_helper(session_info, 'dlc_side')
+        generate_helper(session_info, 'facemap')
+
+if __name__ == "__main__":
+    main()

src/npc_lims/scripts/run_video_processing.py

@@ -1,20 +1,26 @@
 import npc_lims.metadata.codeocean as codeocean
 import npc_lims.status as status
 
+def run_helper(session_info: status.SessionInfo, model_name: str, num_jobs: int) -> int:
+    if not getattr(session_info, model_name):
+        codeocean.run_capsule(session_info.id, model_name)
+        num_jobs += 1
+
+    return num_jobs
 
 def main() -> None:
+    num_jobs = 0
     for session_info in status.get_session_info():
         if not session_info.is_uploaded:
             continue
+
+        num_jobs = run_helper(session_info, 'dlc_eye', num_jobs)
+        num_jobs = run_helper(session_info, 'dlc_side', num_jobs)
+        num_jobs = run_helper(session_info, 'dlc_face', num_jobs)
+        num_jobs = run_helper(session_info, 'facemap', num_jobs)
 
-        """
-        codeocean.run_capsule(session_info.id, "dlc_eye")
-        codeocean.run_capsule(session_info.id, "dlc_front")
-        codeocean.run_capsule(session_info.id, "dlc_side")
-        codeocean.run_capsule(session_info.id, "facemap")
-        """
-        codeocean.run_capsule(session_info.id, "video_pipeline")
-
+        if num_jobs == 12:
+            break
 
 if __name__ == "__main__":
     main()

src/npc_lims/scripts/update_session_video_data_asset_permissions.py

@@ -0,0 +1,23 @@
+import npc_lims.metadata.codeocean as codeocean
+import npc_lims.status as status
+
+def update_helper(session_info: status.SessionInfo, model_name: str) -> None:
+    if getattr(session_info, model_name):
+        try:
+            model_session_data = codeocean.get_model_data_asset(session_info.id, model_name)
+            codeocean.update_permissions_for_data_asset(model_session_data)
+        except (ValueError, FileNotFoundError):
+            pass
+
+def main() -> None:
+    for session_info in status.get_session_info():
+        if not session_info.is_uploaded:
+            continue
+
+        update_helper(session_info, 'dlc_eye')
+        update_helper(session_info, 'dlc_side')
+        update_helper(session_info, 'dlc_face')
+        update_helper(session_info, 'facemap')
+
+if __name__ == '__main__':
+    main()

src/npc_lims/status/tracked_sessions.py

@@ -117,6 +117,70 @@ def is_surface_channels(self) -> bool:
             return False
         return True
 
+    @functools.cached_property
+    def is_dlc_eye(self) -> bool:
+        """
+        The dlc eye capsule has yield a result for this session
+        >>> get_session_info("676909_2023-12-13").is_dlc_eye
+        True
+        """
+
+        if not self.is_ephys:
+            return False
+
+        try:
+            return bool(codeocean.get_model_data_asset(self.id, 'dlc_eye'))
+        except(FileNotFoundError, ValueError):
+            return False
+
+    @functools.cached_property
+    def is_dlc_side(self) -> bool:
+        """
+        The dlc side capsule has yield a result for this session
+        >>> get_session_info("676909_2023-12-13").is_dlc_side
+        True
+        """
+
+        if not self.is_ephys:
+            return False
+
+        try:
+            return bool(codeocean.get_model_data_asset(self.id, 'dlc_side'))
+        except(FileNotFoundError, ValueError):
+            return False
+
+    @property
+    def is_dlc_face(self) -> bool:
+        """
+        The dlc face capsule has yield a result for this session
+        >>> get_session_info("676909_2023-12-13").is_dlc_face
+        True
+        """
+
+        if not self.is_ephys:
+            return False
+
+        try:
+            return bool(codeocean.get_model_data_asset(self.id, 'dlc_face'))
+        except(FileNotFoundError, ValueError):
+            return False
+
+    @functools.cached_property
+    def is_facemap(self) -> bool:
+        """
+        The facemap capsule has yield a result for this session
+        >>> get_session_info("676909_2023-12-13").is_facemap
+        True
+        """
+
+        if not self.is_ephys:
+            return False
+
+        try:
+            return bool(codeocean.get_model_data_asset(self.id, 'facemap'))
+        except(FileNotFoundError, ValueError):
+            return False
+
     @functools.cached_property
     def is_sorted(self) -> bool:
         """The AIND sorting pipeline has yielded a Result asset for this