Use symlink to Xenium output morphology/transcripts files to avoid duplicating data

Author: quentinblampey

CHANGELOG.md

@@ -5,6 +5,7 @@
 
 ### Added
 - Run `module load baysor` in Snakemake pipeline if the module is available.
+- Use symlink to Xenium output morphology/transcripts files to avoid duplicating data
 
 ## [2.0.2] - 2025-02-21
 

sopa/_constants.py

@@ -53,6 +53,7 @@ class SopaAttrs:
     TRANSCRIPTS = "transcripts_dataframe"
     BOUNDARIES = "boundaries_shapes"
     UID = "sopa_uid"
+    XENIUM_OUTPUT_PATH = "xenium_output_path"
 
 
 class SopaFiles:

sopa/io/explorer/converter.py

@@ -150,7 +150,7 @@ def write(
     if len(sdata.points):
         df = get_spatial_element(sdata.points, key=points_key or sdata.attrs.get(SopaAttrs.TRANSCRIPTS))
 
-    if _should_save(mode, "t") and df is not None:
+    if _should_save(mode, "t") and not _use_symlink(path, sdata, "transcripts*") and df is not None:
         gene_column = gene_column or get_feature_key(df)
         if gene_column is not None:
             df = to_intrinsic(sdata, df, image_key)
@@ -159,7 +159,7 @@ def write(
             log.warning("The argument 'gene_column' has to be provided to save the transcripts")
 
     ### Saving image
-    if _should_save(mode, "i"):
+    if _should_save(mode, "i") and not _use_symlink(path, sdata, "morphology*"):
         write_image(
             path,
             sdata[image_key],
@@ -176,6 +176,26 @@ def write(
     log.info(f"You can open the experiment with 'open {path / FileNames.METADATA}'")
 
 
+def _use_symlink(path: Path, sdata: SpatialData, pattern: str) -> bool:
+    """Try using the Xenium output files when existing to avoid re-generating large files."""
+    if SopaAttrs.XENIUM_OUTPUT_PATH not in sdata.attrs:
+        return False
+
+    files = list(Path(sdata.attrs[SopaAttrs.XENIUM_OUTPUT_PATH]).glob(pattern))
+    for file in files:
+        target = path / file.name
+
+        if target.exists():
+            if not target.is_symlink():  # avoid removing non-symlink files
+                return False
+            target.unlink()
+
+        target.symlink_to(file)
+        log.info(f"Created symlink {target} -> {file}")
+
+    return len(files) > 0
+
+
 def _get_n_obs(sdata: SpatialData, geo_df: gpd.GeoDataFrame, table_key: str) -> int:
     if table_key in sdata.tables:
         return sdata.tables[table_key].n_obs

sopa/io/reader/xenium.py

@@ -92,4 +92,6 @@ def xenium(
                 sdata.points["transcripts"]["qv"] < qv_threshold
             )
 
+    sdata.attrs[SopaAttrs.XENIUM_OUTPUT_PATH] = str(Path(path).resolve())
+
     return sdata