Move tissuecyte fn from npc_sessions

src/npc_lims/paths/s3.py

@@ -10,12 +10,17 @@
 import upath
 
 import npc_lims.metadata.codeocean as codeocean
+import npc_lims.status.tracked_sessions as tracked_sessions
 
 DR_DATA_REPO = upath.UPath(
     "s3://aind-scratch-data/ben.hardcastle/DynamicRoutingTask/Data"
 )
 NWB_REPO = upath.UPath("s3://aind-scratch-data/ben.hardcastle/nwb/nwb")
 
+TISSUECYTE_REPO = upath.UPath(
+    "s3://aind-scratch-data/arjun.sridhar/tissuecyte_cloud_processed"
+)
+
 CODE_OCEAN_DATA_BUCKET = upath.UPath("s3://codeocean-s3datasetsbucket-1u41qdg42ur9")
 
 
@@ -255,6 +260,39 @@ def get_sorted_precurated_paths_from_s3(
     return sorted_precurated_dirs
 
 
+@functools.cache
+def get_tissuecyte_annotation_files_from_s3(
+    session: str | npc_session.SessionRecord,
+) -> tuple[upath.UPath, ...]:
+    """For each probe inserted, get a csv file containing CCF coordinates for each
+    electrode (channel) on the probe.
+
+    >>> electrode_files = get_tissuecyte_annotation_files_from_s3('626791_2022-08-16')
+    >>> assert len(electrode_files) > 0
+    >>> electrode_files[0].name
+    'Probe_A2_channels_626791_warped_processed.csv'
+    """
+    session = npc_session.SessionRecord(session)
+    day = tracked_sessions.get_session_info(session).day
+    subject_electrode_network_path = TISSUECYTE_REPO / str(session.subject.id)
+
+    if not subject_electrode_network_path.exists():
+        raise FileNotFoundError(
+            f"CCF annotations for {session} have not been uploaded to s3"
+        )
+
+    electrode_files = tuple(
+        subject_electrode_network_path.glob(
+            f"Probe_*{day}_channels_{str(session.subject.id)}_warped_processed.csv"
+        )
+    )
+    if not electrode_files:
+        raise FileNotFoundError(
+            f"{subject_electrode_network_path} exists, but no CCF annotation files found matching {day} and {session.subject.id} - check session day"
+        )
+
+    return electrode_files
+
 @dataclasses.dataclass
 class StimFile:
     path: upath.UPath

src/npc_lims/scripts/update_session_status.py

@@ -27,15 +27,16 @@ def main() -> None:
             subject_id VARCHAR(30),
             project VARCHAR DEFAULT NULL,
             is_uploaded BOOLEAN DEFAULT NULL,
-            is_sorted BOOLEAN DEFAULT NULL
+            is_sorted BOOLEAN DEFAULT NULL,
+            is_annotated BOOLEAN DEFAULT NULL,
         );
         """
     )
     statement = (
-        "INSERT INTO status (date, subject_id, project, is_uploaded, is_sorted) VALUES "
+        "INSERT INTO status (date, subject_id, project, is_uploaded, is_sorted, is_annotated) VALUES "
     )
     for s in sorted(npc_lims.get_session_info(), key=lambda s: s.date, reverse=True):
-        statement += f"\n\t('{s.date}', '{s.subject}', '{s.project}', {int(s.is_uploaded)}, {int(s.is_sorted)}),"
+        statement += f"\n\t('{s.date}', '{s.subject}', '{s.project}', {int(s.is_uploaded)}, {int(s.is_sorted)}, {int(s.is_annotated)}),"
     statement = statement[:-1] + ";"
     response = connection.Execute(statement)
     if response[1]:

src/npc_lims/status/tracked_sessions.py

@@ -90,6 +90,18 @@ def is_sorted(self) -> bool:
         except (FileNotFoundError, ValueError):
             return False
 
+    @property
+    def is_annotated(self) -> bool:
+        """The subject associated with the sessions has CCF annotation data for
+        probes available on S3.
+
+        >>> any(session.is_annotated for session in get_session_info())
+        True
+        """
+        try:
+            return bool(codeocean.get_raw_data_root(self.id))
+        except (FileNotFoundError, ValueError):
+            return False
 
 @typing.overload
 def get_session_info() -> tuple[SessionInfo, ...]: