Get started with new S3 collection backend

.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+.env

.vscode/launch.json

@@ -0,0 +1,14 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "python",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "${file}",
+            "console": "integratedTerminal",
+            "args": [],
+            "justMyCode": false,
+        }
+    ]
+}

.vscode/settings.json

@@ -0,0 +1,14 @@
+{
+    "python.analysis.typeCheckingMode": "strict",
+    "notebook.codeActionsOnSave": {
+        "source.organizeImports": true
+    },
+    "editor.codeActionsOnSave": {
+        "source.organizeImports": "explicit"
+    },
+    "python.testing.pytestArgs": [
+        "tests"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true,
+}

pyproject.toml

@@ -0,0 +1,23 @@
+[tool.black]
+target-version = ["py312"]
+
+[tool.pyright]
+include = ["scripts"]
+pythonPlatform = "All"
+pythonVersion = "3.12"
+reportIncompatibleMethodOverride = true
+reportMissingSuperCall = "error"
+reportMissingTypeArgument = true
+reportMissingTypeStubs = "warning"
+reportUninitializedInstanceVariable = "error"
+reportUnknownMemberType = false
+reportUnnecessaryIsInstance = false
+reportUnnecessaryTypeIgnoreComment = "error"
+reportUnusedCallResult = "error"
+reportUnusedVariable = "error"
+typeCheckingMode = "strict"
+useLibraryCodeForTypes = true
+
+[tool.pytest.ini_options]
+addopts = "--capture=no --doctest-modules --failed-first"
+testpaths = ["scripts"]

scripts/stage.py

@@ -0,0 +1,14 @@
+import typer
+from utils.remote_resource import RemoteResource
+from utils.s3_client import Client
+from utils.validate_format import validate_format
+
+
+def stage(resource_id: str, package_url: str):
+    resource = RemoteResource(client=Client(), id=resource_id)
+    staged = resource.stage_new_version(package_url)
+    validate_format(staged)
+
+
+if __name__ == "__main__":
+    typer.run(stage)

scripts/test_dynamically.py

@@ -0,0 +1,87 @@
+import traceback
+from functools import partialmethod
+from pathlib import Path
+from typing import Optional
+
+import typer
+from bioimageio.spec import load_raw_resource_description
+from bioimageio.spec.shared import yaml
+from utils.remote_resource import StagedVersion
+from utils.s3_client import Client
+
+try:
+    from tqdm import tqdm
+except ImportError:
+    pass
+else:
+    tqdm.__init__ = partialmethod(tqdm.__init__, disable=True)  # silence tqdm
+
+
+def test_summary_from_exception(name: str, exception: Exception):
+    return dict(
+        name=name,
+        status="failed",
+        error=str(exception),
+        traceback=traceback.format_tb(exception.__traceback__),
+    )
+
+
+def test_dynamically(
+    resource_id: str,
+    version: int,
+    weight_format: Optional[str] = typer.Argument(
+        ..., help="weight format to test model with."
+    ),
+    create_env_outcome: str = "success",
+):
+    staged = StagedVersion(client=Client(), id=resource_id, version=version)
+    rdf_source = staged.get_rdf_url()
+    if weight_format is None:
+        # no dynamic tests for non-model resources yet...
+        return
+
+    if create_env_outcome == "success":
+        try:
+            from bioimageio.core.resource_tests import test_resource
+        except Exception as e:
+            summaries = [
+                test_summary_from_exception(
+                    "import test_resource from test environment", e
+                )
+            ]
+        else:
+            try:
+                rdf = yaml.load(rdf_source)
+                test_kwargs = (
+                    rdf.get("config", {})
+                    .get("bioimageio", {})
+                    .get("test_kwargs", {})
+                    .get(weight_format, {})
+                )
+            except Exception as e:
+                summaries = [test_summary_from_exception("check for test kwargs", e)]
+            else:
+                try:
+                    rd = load_raw_resource_description(rdf_source)
+                    summaries = test_resource(
+                        rd, weight_format=weight_format, **test_kwargs
+                    )
+                except Exception as e:
+                    summaries = [test_summary_from_exception("call 'test_resource'", e)]
+
+    else:
+        env_path = Path(f"conda_env_{weight_format}.yaml")
+        if env_path.exists():
+            error = "Failed to install conda environment:\n" + env_path.read_text()
+        else:
+            error = f"Conda environment yaml file not found: {env_path}"
+
+        summaries = [
+            dict(name="install test environment", status="failed", error=error)
+        ]
+
+    staged.add_log_entry("validation_summaries", summaries)
+
+
+if __name__ == "__main__":
+    typer.run(test_dynamically)

scripts/update_status.py

@@ -0,0 +1,37 @@
+import datetime
+from typing import Optional
+
+from loguru import logger
+from typer import Argument, run
+from typing_extensions import Annotated
+from utils.s3_client import Client
+
+
+def update_status(
+    resource_id: str,
+    version: int,
+    message: Annotated[str, Argument(help="status message")],
+    step: Annotated[
+        Optional[int], Argument(help="optional step in multi-step process")
+    ] = None,
+    num_steps: Annotated[int, Argument(help="total steps of multi-step process")] = 6,
+):
+    assert step is None or step <= num_steps, (step, num_steps)
+    timenow = datetime.datetime.now().isoformat()
+
+    resource_path, version = version_from_resource_path_or_s3(resource_path, client)
+    status = client.get_status(resource_path, version)
+
+    if "messages" not in status:
+        status["messages"] = []
+    if step is not None:
+        status["step"] = step
+    if num_steps is not None:
+        status["num_steps"] = num_steps
+    status["last_message"] = message
+    status["messages"].append({"timestamp": timenow, "text": message})
+    client.put_status(resource_path, version, status)
+
+
+if __name__ == "__main__":
+    run(update_status)