ENH: add get_instance_file_URL() function

Re #161 this will return AWS/GCP URLs for a file given SOPInstanceUID,
only for microscopy instances

Author: fedorov

idc_index/index.py

@@ -676,6 +676,48 @@ def get_series_file_URLs(self, seriesInstanceUID, source_bucket_location="aws"):
 
         return file_names
 
+    def get_instance_file_URL(self, sopInstanceUID, source_bucket_location="aws"):
+        """
+        Get the bucket URL of the file corresponding to a given SOPInstanceUID.
+
+        This function will only return the URL for the Slide Microscopy (SM) instances,
+        which are maintained in the `sm_instance_index` table.
+
+        Args:
+            sopInstanceUID: string containing the value of DICOM SOPInstanceUID
+            source_bucket_location: string containing the source bucket location, either "aws" or "gcp"
+
+        Returns:
+            string containing the bucket URL of the file corresponding to the SOPInstanceUID,
+            or None if the SOPInstanceUID is not recognized
+        """
+
+        # sm_instance_index is required to complete this operation - install it!
+        self.fetch_index("sm_instance_index")
+
+        if sopInstanceUID not in self.sm_instance_index["SOPInstanceUID"].values:
+            raise ValueError("SOPInstanceUID not found in IDC sm_instance_index.")
+
+        # merge with the main index to get series_aws_url
+        selected_instance_df = self.sm_instance_index[
+            self.sm_instance_index["SOPInstanceUID"] == sopInstanceUID
+        ].copy()[["SeriesInstanceUID", "SOPInstanceUID", "crdc_instance_uuid"]]
+        selected_instance_df = pd.merge(
+            selected_instance_df,
+            self.index,
+            on="SeriesInstanceUID",
+            how="left",
+        )
+
+        if source_bucket_location == "gcp":
+            # replace AWS with the GCP bucket
+            self._replace_aws_with_gcp_buckets(selected_instance_df, "series_aws_url")
+
+        # instance files are named using crdc_instance_uuid
+        series_url = selected_instance_df.iloc[0]["series_aws_url"][:-1]
+        instance_uuid = selected_instance_df.iloc[0]["crdc_instance_uuid"]
+        return series_url + instance_uuid + ".dcm"
+
     def get_viewer_URL(
         self, seriesInstanceUID=None, studyInstanceUID=None, viewer_selector=None
     ):

tests/idcindex.py

@@ -597,6 +597,14 @@ def test_series_files_URLs(self):
         assert len(files_aws) > 0
         assert len(files_gcp) == len(files_aws)
 
+    def test_instance_file_URLs(self):
+        c = IDCClient()
+        sopInstanceUID = "1.3.6.1.4.1.5962.99.1.1900325859.924065538.1719887277027.10.0"
+        file_url = "s3://idc-open-data/763fe058-7d25-4ba7-9b29-fd3d6c41dc4b/210f0529-c767-4795-9acf-bad2f4877427.dcm"
+        files_aws = c.get_instance_file_URL(sopInstanceUID, "aws")
+        files_gcp = c.get_instance_file_URL(sopInstanceUID, "gcp")
+        assert files_aws == files_gcp == file_url
+
 
 if __name__ == "__main__":
     unittest.main()