Merge pull request #2 from fractal-analytics-platform/Add-sample-data

Add sample data loader from zenodo

setup.cfg

@@ -36,6 +36,7 @@ install_requires =
     qtpy
     scikit-image
     fractal-tasks-core==0.14.0
+    wget
 
 python_requires = >=3.8
 include_package_data = True

src/napari_ome_zarr_navigator/__init__.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 try:
     from ._version import version as __version__
 except ImportError:
@@ -10,3 +12,8 @@
 )
 
 __all__ = ("ROILoader", "ImgBrowser")
+FILE = Path(__file__).resolve()
+_PACKAGE_DIR = FILE.parents[2]
+_MODULE_DIR = FILE.parent
+_TEST_DIR = _MODULE_DIR.joinpath("_tests")
+_TEST_DATA_DIR = _PACKAGE_DIR.joinpath("test_data")

src/napari_ome_zarr_navigator/_tests/test_sample_data.py

@@ -1,7 +1,16 @@
-# from napari_ome_zarr_navigator import make_sample_data
+from pathlib import Path
 
-# add your tests here...
+from napari_ome_zarr_navigator.generate_test_data import (
+    load_ome_zarr_from_zenodo,
+)
 
 
-def test_something():
-    pass
+from napari_ome_zarr_navigator import _TEST_DATA_DIR
+
+
+def test_load_zenodo_data():
+    doi = "10.5281/zenodo.10424292"
+    zarr_url = "20200812-CardiomyocyteDifferentiation14-Cycle1_mip.zarr"
+    doi_path = doi_path = Path(_TEST_DATA_DIR).joinpath(doi.replace("/", "_"))
+    load_ome_zarr_from_zenodo(doi, zarr_url)
+    assert doi_path.is_dir() == True

src/napari_ome_zarr_navigator/generate_test_data/__init__.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+from pathlib import Path
+from typing import Union
+import re
+import requests
+import urllib
+import hashlib
+import wget
+import shutil
+
+
+from napari_ome_zarr_navigator import _TEST_DATA_DIR
+
+
+def load_ome_zarr_from_zenodo(doi: str, zarr_url: Union[str, Path]):
+    doi_path = Path(_TEST_DATA_DIR).joinpath(doi.replace("/", "_"))
+    zarr_path = doi_path.joinpath(zarr_url)
+    if not doi_path.is_dir():
+        download_from_zenodo(doi, directory=doi_path)
+        shutil.unpack_archive(zarr_path.with_suffix(".zarr.zip"), doi_path)
+
+
+def download_from_zenodo(
+    doi: str,
+    overwrite: bool = False,
+    directory: Union[str, Path] = Path(),
+    access_token: str = None,
+):
+    record_id = re.match(r".*/zenodo.(\w+)", doi).group(1)
+    url = "https://zenodo.org/api/records/" + record_id
+    js = requests.get(url).json()
+    doi = js["metadata"]["doi"]
+    print("Title: " + js["metadata"]["title"])
+    print("Publication date: " + js["metadata"]["publication_date"])
+    print("DOI: " + js["metadata"]["doi"])
+    print(
+        "Total file size: {:.1f} MB".format(
+            sum(f["size"] / 10**6 for f in js["files"])
+        )
+    )
+    doi_path = Path(directory)
+    try:
+        doi_path.mkdir(exist_ok=overwrite, parents=True)
+    except FileExistsError:
+        print(f"{doi_path} exists. Don't overwrite.")
+        return
+    for file in js["files"]:
+        file_path = Path(doi_path).joinpath(file["key"])
+        algorithm, checksum = file["checksum"].split(":")
+        try:
+            link = urllib.parse.unquote(file["links"]["self"])
+            wget.download(
+                f"{link}?access_token={access_token}", str(directory)
+            )
+            check_passed, returned_checksum = verify_checksum(
+                file_path, algorithm, checksum
+            )
+            if check_passed:
+                print(f"\nChecksum is correct. ({checksum})")
+            else:
+                print(
+                    f"\nChecksum is incorrect! ({checksum} got: {returned_checksum})"
+                )
+        except urllib.error.HTTPError:
+            pass
+
+
+def verify_checksum(filename: Union[str, Path], algorithm, original_checksum):
+    h = hashlib.new(algorithm)
+    with open(filename, "rb") as f:
+        h.update(f.read())
+        returned_checksum = h.hexdigest()
+    if returned_checksum == original_checksum:
+        return True, returned_checksum
+    else:
+        return False, returned_checksum