backoffice

bioimage.io collection backoffice

Modules:

• __main__ –
• _cli –
• _config –
• _requests_utils –
• _settings –
• _summarize –
• _version –
• check_compatibility –

  check compatibility of tools against resources in the index.json

• compatibility –

  data models for compatibility reports

• compatibility_pure –

  Pure Python alternative data models for compatibility reports

• index –

  Data models and functions for indexing the bioimage.io collection

• utils –

  utility functions

• utils_pure –

  utility functions available in backoffice without dependencies

Attributes:

• __all__ –
• __version__ –

__all__ module-attribute

__all__ = ['__version__']

__version__ module-attribute

__version__ = '0.1.0'

__main__

Functions:

• main –

main

main()

Source code in src/backoffice/__main__.py

def main():
    cli = Backoffice()  # pyright: ignore[reportCallIssue]
    cli.run()

_cli

Classes:

• Backoffice –

  backoffice - manage the bioimage.io collection

• CmdBase –
• IndexCmd –
• SummarizeCmd –

Backoffice

Bases: BaseSettings

backoffice - manage the bioimage.io collection

Methods:

• run –

Attributes:

• index (CliSubCommand[IndexCmd]) –

  index the bioimage.io collection

• summarize (CliSubCommand[SummarizeCmd]) –

  conflate tool summaries

index instance-attribute

index: CliSubCommand[IndexCmd]

index the bioimage.io collection

summarize instance-attribute

summarize: CliSubCommand[SummarizeCmd]

conflate tool summaries

run

run()

Source code in src/backoffice/_cli.py

def run(self):
    cmd = self.index or self.summarize
    if cmd is None:
        raise ValueError(
            "No command specified. Use 'backoffice --help' to see available commands."
        )
    else:
        sys.exit(cmd.run())

CmdBase pydantic-model

Bases: BaseModel

IndexCmd pydantic-model

Bases: CmdBase

run

run()

Index the bioimage.io collection

Source code in src/backoffice/_cli.py

def run(self):
    """Index the bioimage.io collection"""
    _ = create_index()

SummarizeCmd pydantic-model

Bases: CmdBase

run

run()

Conflate tool summaries

Source code in src/backoffice/_cli.py

def run(self):
    """Conflate tool summaries"""

    summarize_reports()

_config

Modules:

• common –
• id_parts –

  describes a file holding all parts to create resource ids

• reviewers –

Classes:

• CollectionConfig –

CollectionConfig pydantic-model

Bases: ConfigNode

id_parts instance-attribute

id_parts: IdParts

reviewers instance-attribute

reviewers: Reviewers

load cached classmethod

load()

Source code in src/backoffice/_config/__init__.py

@classmethod
@lru_cache
def load(cls):
    if isinstance(settings.collection_config, HttpUrl):
        r = httpx.get(
            str(settings.collection_config), timeout=settings.http_timeout
        )
        raise_for_status_discretely(r)
        data = r.json()
    else:
        with Path(settings.collection_config).open(encoding="utf-8") as f:
            data = json.load(f)

    return cls.model_validate(data)

common

Classes:

• ConfigNode –

Attributes:

• yaml –

yaml module-attribute

yaml = YAML(typ='safe')

ConfigNode pydantic-model

Bases: BaseModel

id_parts

describes a file holding all parts to create resource ids

Classes:

• IdParts –

  parts to create resource ids

• IdPartsEntry –

  parts to create resource ids for a specific resource type

IdParts pydantic-model

Bases: ConfigNode

parts to create resource ids

dataset instance-attribute

dataset: IdPartsEntry

model instance-attribute

model: IdPartsEntry

notebook instance-attribute

notebook: IdPartsEntry

__getitem__

__getitem__(type_: str) -> IdPartsEntry

Source code in src/backoffice/_config/id_parts.py

def __getitem__(self, type_: str) -> IdPartsEntry:
    if type_ == "model":
        return self.model
    elif type_ == "dataset":
        return self.dataset
    elif type_ == "notebook":
        return self.notebook
    else:
        raise NotImplementedError(
            f"handling resource id for type '{type_}' is not yet implemented"
        )

get_icon

get_icon(resource_id: str)

Source code in src/backoffice/_config/id_parts.py

def get_icon(self, resource_id: str):
    for parts in (self.model, self.dataset, self.notebook):
        noun = parts.get_noun(resource_id)
        if noun is not None and noun in parts.nouns:
            return parts.nouns[noun]

    return None

IdPartsEntry pydantic-model

Bases: ConfigNode

parts to create resource ids for a specific resource type

adjectives instance-attribute

adjectives: Sequence[str]

nouns instance-attribute

nouns: Mapping[str, str]

get_noun

get_noun(resource_id: str)

Source code in src/backoffice/_config/id_parts.py

def get_noun(self, resource_id: str):
    if not isinstance(resource_id, str):
        raise TypeError(f"invalid resource_id type: {type(resource_id)}")
    if not resource_id:
        raise ValueError("empty resource_id")

    for adj in self.adjectives:
        if resource_id.startswith(adj + "-"):
            break
    else:
        return None

    return resource_id[len(adj) + 1 :]

validate_concept_id

validate_concept_id(resource_id: str)

Source code in src/backoffice/_config/id_parts.py

def validate_concept_id(self, resource_id: str):
    noun = self.get_noun(resource_id)
    if noun is None:
        raise ValueError(
            f"{resource_id} does not start with a listed adjective"
            + " (or does not follow the pattern 'adjective-noun')"
        )

    if noun not in self.nouns:
        raise ValueError(
            f"{resource_id} does not end with a listed noun"
            + " (or does not follow the pattern 'adjective-noun')"
        )

reviewers

Classes:

• Reviewer –

Attributes:

• Reviewers –

Reviewers module-attribute

Reviewers = Sequence[Reviewer]

Reviewer pydantic-model

Bases: ConfigNode

affiliation instance-attribute

affiliation: str

email instance-attribute

email: str

github_user instance-attribute

github_user: str

id instance-attribute

id: str

hypha id

name instance-attribute

name: str

orcid instance-attribute

orcid: str

_requests_utils

_settings

Classes:

• Settings –

Attributes:

• settings –

settings module-attribute

settings = Settings()

Settings

Bases: BaseSettings

Methods:

• get_hypha_headers –

Attributes:

• collection_config (Annotated[Union[HttpUrl, Path], Field(union_mode=left_to_right)]) –

  collection config

• http_timeout (int) –

  Timeout for HTTP requests in seconds

• hypha_api_token (SecretStr) –
• hypha_base_url (str) –
• model_config –
• summaries (Path) –
• tools (Sequence[str]) –

collection_config class-attribute instance-attribute

collection_config: Annotated[
    Union[HttpUrl, Path], Field(union_mode=left_to_right)
] = parent / "../../bioimageio_collection_config.json"

collection config

http_timeout class-attribute instance-attribute

http_timeout: int = 30

Timeout for HTTP requests in seconds

hypha_api_token instance-attribute

hypha_api_token: SecretStr

hypha_base_url class-attribute instance-attribute

hypha_base_url: str = 'https://hypha.aicell.io'

model_config class-attribute instance-attribute

model_config = SettingsConfigDict(
    env_file=".env", env_file_encoding="utf-8"
)

summaries class-attribute instance-attribute

summaries: Path = Path('summaries')

tools class-attribute instance-attribute

tools: Sequence[str] = (
    "biapy",
    "bioimageio.core",
    "careamics",
    "ilastik",
)

get_hypha_headers

get_hypha_headers()

Source code in src/backoffice/_settings.py

def get_hypha_headers(self):
    return {
        "Authorization": f"Bearer {self.hypha_api_token.get_secret_value()}",
        "Content-Type": "application/json",
    }

_summarize

Functions:

• summarize_reports –

summarize_reports

summarize_reports()

Source code in src/backoffice/_summarize.py

def summarize_reports():
    index = load_index()
    for item in tqdm(index.items):
        for v in item.versions:
            _summarize(item, v)

_version

Attributes:

• VERSION –

VERSION module-attribute

VERSION = '0.1.0'

check_compatibility

check compatibility of tools against resources in the index.json

Functions:

• check_tool_compatibility –

  helper to implement tool compatibility checks

Attributes:

• ItemId –
• ItemVersion –
• Sha256 –
• Url –

ItemId module-attribute

ItemId = str

ItemVersion module-attribute

ItemVersion = str

Sha256 module-attribute

Sha256 = str

Url module-attribute

Url = str

check_tool_compatibility

check_tool_compatibility(
    tool_name: str,
    tool_version: str,
    *,
    index_path: Path = Path("index.json"),
    check_tool_compatibility_impl: Callable[
        [ItemId, ItemVersion, Url, Sha256],
        ToolCompatibilityReport
        | ToolCompatibilityReportDict,
    ],
    applicable_types: set[str],
    id_startswith: str = "",
)

helper to implement tool compatibility checks

Parameters:

• tool_name

  (str) –

  name of the tool (without version), e.g. "ilastik"

• tool_version

  (str) –

  version of the tool, e.g. "1.4"

• index_path

  (Path, default: Path('index.json') ) –

  Path to the index.json file.

• check_tool_compatibility_impl

  (Callable[[ItemId, ItemVersion, Url, Sha256], ToolCompatibilityReport | ToolCompatibilityReportDict]) –

  Function accepting two positional arguments: URL to an rdf.yaml, SHA-256 of that rdf.yaml. And returning a compatibility report.

• applicable_types

  (set[str]) –

  Set of resource types check_tool_compatibility_impl is applicable to.

Source code in src/backoffice/check_compatibility.py

def check_tool_compatibility(
    tool_name: str,
    tool_version: str,
    *,
    index_path: Path = Path("index.json"),
    check_tool_compatibility_impl: Callable[
        [ItemId, ItemVersion, Url, Sha256],
        "ToolCompatibilityReport | ToolCompatibilityReportDict",
    ],
    applicable_types: set[str],
    id_startswith: str = "",
):
    """helper to implement tool compatibility checks

    Args:
        tool_name: name of the tool (without version), e.g. "ilastik"
        tool_version: version of the tool, e.g. "1.4"
        index_path: Path to the `index.json` file.
        check_tool_compatibility_impl:
            Function accepting two positional arguments:
            URL to an rdf.yaml, SHA-256 of that rdf.yaml.
            And returning a compatibility report.
        applicable_types: Set of resource types
            **check_tool_compatibility_impl** is applicable to.
    """
    with index_path.open() as f:
        items = json.load(f)["items"]

    filtered_items = [
        item
        for item in items
        if item["type"] in applicable_types and item["id"].startswith(id_startswith)
    ]
    print(f"found {len(filtered_items)} starting with '{id_startswith}'")

    for item in tqdm(filtered_items):
        for version in item["versions"]:
            rdf_url = version["source"]
            sha256 = version["sha256"]

            report_path = get_tool_report_path(
                item["id"], version["version"], tool_name, tool_version
            )
            if report_path.exists():
                logger.info("found existing report at {}", report_path)
                continue

            try:
                report = check_tool_compatibility_impl(
                    item["id"], version["version"], rdf_url, sha256
                )
            except Exception as e:
                traceback.print_exc()
                warnings.warn(f"failed to check '{rdf_url}': {e}")
            else:
                if not isinstance(report, dict):
                    report = report.model_dump(mode="json")

                report_path.parent.mkdir(parents=True, exist_ok=True)
                with report_path.open("wt", encoding="utf-8") as f:
                    json.dump(report, f, indent=4, sort_keys=True, ensure_ascii=False)

            _total, _used, free = shutil.disk_usage(".")
            if free < 7_000_000_000:
                raise RuntimeError("less than 7GB disk space left, stopping now")

compatibility

data models for compatibility reports

Classes:

• Badge –
• CompatibilityScores –
• CompatibilitySummary –
• InitialSummary –
• Node –

  Base data model with common config

• ToolCompatibilityReport –

  Used to report on the compatibility of resource description

• ToolReportDetails –

Attributes:

• PARTNER_TOOL_NAMES –
• PartnerToolName –
• TOOL_NAMES –
• ToolName –
• ToolNameVersioned –

PARTNER_TOOL_NAMES module-attribute

PARTNER_TOOL_NAMES = (
    "ilastik",
    "deepimagej",
    "icy",
    "biapy",
    "careamics",
)

PartnerToolName module-attribute

PartnerToolName = Literal[
    "ilastik", "deepimagej", "icy", "biapy", "careamics"
]

TOOL_NAMES module-attribute

TOOL_NAMES = ('bioimageio.core', *PARTNER_TOOL_NAMES)

ToolName module-attribute

ToolName = Literal['bioimageio.core', PartnerToolName]

ToolNameVersioned module-attribute

ToolNameVersioned = str

Badge pydantic-model

Bases: Node

Show JSON schema:

{
  "properties": {
    "icon": {
      "format": "uri",
      "maxLength": 2083,
      "minLength": 1,
      "title": "Icon",
      "type": "string"
    },
    "label": {
      "title": "Label",
      "type": "string"
    },
    "url": {
      "format": "uri",
      "maxLength": 2083,
      "minLength": 1,
      "title": "Url",
      "type": "string"
    }
  },
  "required": [
    "icon",
    "label",
    "url"
  ],
  "title": "Badge",
  "type": "object"
}

Fields:

• icon (HttpUrl)
• label (str)
• url (HttpUrl)

icon pydantic-field

icon: HttpUrl

label pydantic-field

label: str

url pydantic-field

url: HttpUrl

CompatibilityScores pydantic-model

Bases: Node

Show JSON schema:

{
  "properties": {
    "tool_compatibility_version_specific": {
      "additionalProperties": {
        "maximum": 1.0,
        "minimum": 0,
        "type": "number"
      },
      "title": "Tool Compatibility Version Specific",
      "type": "object"
    },
    "metadata_completeness": {
      "default": 0.0,
      "maximum": 1.0,
      "minimum": 0,
      "title": "Metadata Completeness",
      "type": "number"
    },
    "metadata_format": {
      "default": 0.0,
      "maximum": 1.0,
      "minimum": 0,
      "title": "Metadata Format",
      "type": "number"
    }
  },
  "required": [
    "tool_compatibility_version_specific"
  ],
  "title": "CompatibilityScores",
  "type": "object"
}

Fields:

• tool_compatibility_version_specific (Mapping[ToolNameVersioned, Annotated[float, Interval(ge=0, le=1.0)]])
• metadata_completeness (Annotated[float, Interval(ge=0, le=1.0)])
• metadata_format (Annotated[float, Interval(ge=0, le=1.0)])

core_compatibility property

core_compatibility: float

metadata_completeness pydantic-field

metadata_completeness: Annotated[
    float, Interval(ge=0, le=1.0)
] = 0.0

Score for metadata completeness.

A measure of how many optional fields in the resource RDF are filled out.

metadata_format pydantic-field

metadata_format: Annotated[
    float, Interval(ge=0, le=1.0)
] = 0.0

Score for metadata formatting.

• 1.0: resource RDF conforms to the latest spec version
• 0.5: resource RDF conforms to an older spec version
• 0.0: resource RDF does not conform to any known spec version

overall_compatibility property

overall_compatibility: Annotated[
    float, Interval(ge=0, le=1.0)
]

Weighted, overall score between 0 and 1. Note: The scoring scheme is subject to change in the future.

overall_partner_tool_compatibility property

overall_partner_tool_compatibility: Annotated[
    float, Interval(ge=0, le=1.0)
]

Overall partner tool compatibility score. Note: - Currently implemented as: Average of the top 3 partner tool compatibility scores. - Implementation is subject to change in the future.

tool_compatibility property

tool_compatibility: Mapping[
    ToolName, Annotated[float, Interval(ge=0, le=1.0)]
]

Aggregated tool compatibility score

tool_compatibility_version_specific pydantic-field

tool_compatibility_version_specific: Mapping[
    ToolNameVersioned,
    Annotated[float, Interval(ge=0, le=1.0)],
]

CompatibilitySummary pydantic-model

Bases: InitialSummary

Show JSON schema:

{
  "$defs": {
    "Badge": {
      "properties": {
        "icon": {
          "format": "uri",
          "maxLength": 2083,
          "minLength": 1,
          "title": "Icon",
          "type": "string"
        },
        "label": {
          "title": "Label",
          "type": "string"
        },
        "url": {
          "format": "uri",
          "maxLength": 2083,
          "minLength": 1,
          "title": "Url",
          "type": "string"
        }
      },
      "required": [
        "icon",
        "label",
        "url"
      ],
      "title": "Badge",
      "type": "object"
    },
    "CompatibilityScores": {
      "properties": {
        "tool_compatibility_version_specific": {
          "additionalProperties": {
            "maximum": 1.0,
            "minimum": 0,
            "type": "number"
          },
          "title": "Tool Compatibility Version Specific",
          "type": "object"
        },
        "metadata_completeness": {
          "default": 0.0,
          "maximum": 1.0,
          "minimum": 0,
          "title": "Metadata Completeness",
          "type": "number"
        },
        "metadata_format": {
          "default": 0.0,
          "maximum": 1.0,
          "minimum": 0,
          "title": "Metadata Format",
          "type": "number"
        }
      },
      "required": [
        "tool_compatibility_version_specific"
      ],
      "title": "CompatibilityScores",
      "type": "object"
    },
    "ToolCompatibilityReport": {
      "additionalProperties": true,
      "description": "Used to report on the compatibility of resource description\nin the bioimageio collection for a version specific tool.",
      "properties": {
        "tool": {
          "enum": [
            "bioimageio.core",
            "ilastik",
            "deepimagej",
            "icy",
            "biapy",
            "careamics"
          ],
          "title": "Tool",
          "type": "string"
        },
        "tool_version": {
          "pattern": "^[a-z0-9\\.-]+$",
          "title": "Tool Version",
          "type": "string"
        },
        "status": {
          "enum": [
            "passed",
            "failed",
            "not-applicable"
          ],
          "title": "Status",
          "type": "string"
        },
        "score": {
          "maximum": 1.0,
          "minimum": 0,
          "title": "Score",
          "type": "number"
        },
        "error": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "title": "Error"
        },
        "details": {
          "anyOf": [
            {
              "$ref": "#/$defs/ToolReportDetails"
            },
            {
              "type": "string"
            },
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Details"
        },
        "badge": {
          "anyOf": [
            {
              "$ref": "#/$defs/Badge"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "links": {
          "default": [],
          "items": {
            "type": "string"
          },
          "title": "Links",
          "type": "array"
        }
      },
      "required": [
        "tool",
        "tool_version",
        "status",
        "score",
        "error"
      ],
      "title": "ToolCompatibilityReport",
      "type": "object"
    },
    "ToolReportDetails": {
      "additionalProperties": true,
      "properties": {
        "traceback": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Traceback"
        },
        "warnings": {
          "anyOf": [
            {
              "additionalProperties": true,
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Warnings"
        },
        "metadata_completeness": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Metadata Completeness"
        },
        "status": {
          "anyOf": [
            {
              "enum": [
                "passed",
                "valid-format",
                "failed"
              ],
              "type": "string"
            },
            {}
          ],
          "default": null,
          "title": "Status"
        }
      },
      "title": "ToolReportDetails",
      "type": "object"
    }
  },
  "properties": {
    "rdf_content": {
      "additionalProperties": true,
      "title": "Rdf Content",
      "type": "object"
    },
    "rdf_yaml_sha256": {
      "title": "Rdf Yaml Sha256",
      "type": "string"
    },
    "status": {
      "enum": [
        "passed",
        "failed",
        "untested"
      ],
      "title": "Status",
      "type": "string"
    },
    "scores": {
      "$ref": "#/$defs/CompatibilityScores"
    },
    "tests": {
      "additionalProperties": {
        "additionalProperties": {
          "$ref": "#/$defs/ToolCompatibilityReport"
        },
        "type": "object"
      },
      "propertyNames": {
        "enum": [
          "bioimageio.core",
          "ilastik",
          "deepimagej",
          "icy",
          "biapy",
          "careamics"
        ]
      },
      "title": "Tests",
      "type": "object"
    }
  },
  "required": [
    "rdf_content",
    "rdf_yaml_sha256",
    "status",
    "scores",
    "tests"
  ],
  "title": "CompatibilitySummary",
  "type": "object"
}

Fields:

• rdf_content (dict[str, Any])
• rdf_yaml_sha256 (str)
• status (Literal['passed', 'failed', 'untested'])
• scores (CompatibilityScores)
• tests (Mapping[ToolName, Mapping[str, ToolCompatibilityReport]])

rdf_content pydantic-field

rdf_content: dict[str, Any]

The RDF content of the original rdf.yaml file.

rdf_yaml_sha256 pydantic-field

rdf_yaml_sha256: str

SHA-256 of the original RDF YAML file.

scores pydantic-field

scores: CompatibilityScores

Scores for compatibility with the bioimage.io community tools.

status pydantic-field

status: Literal['passed', 'failed', 'untested']

status of the bioimageio.core reproducibility tests.

tests pydantic-field

tests: Mapping[
    ToolName, Mapping[str, ToolCompatibilityReport]
]

Compatibility reports for each tool for each version.

InitialSummary pydantic-model

Bases: Node

Show JSON schema:

{
  "properties": {
    "rdf_content": {
      "additionalProperties": true,
      "title": "Rdf Content",
      "type": "object"
    },
    "rdf_yaml_sha256": {
      "title": "Rdf Yaml Sha256",
      "type": "string"
    },
    "status": {
      "enum": [
        "passed",
        "failed",
        "untested"
      ],
      "title": "Status",
      "type": "string"
    }
  },
  "required": [
    "rdf_content",
    "rdf_yaml_sha256",
    "status"
  ],
  "title": "InitialSummary",
  "type": "object"
}

Fields:

• rdf_content (dict[str, Any])
• rdf_yaml_sha256 (str)
• status (Literal['passed', 'failed', 'untested'])

rdf_content pydantic-field

rdf_content: dict[str, Any]

The RDF content of the original rdf.yaml file.

rdf_yaml_sha256 pydantic-field

rdf_yaml_sha256: str

SHA-256 of the original RDF YAML file.

status pydantic-field

status: Literal['passed', 'failed', 'untested']

status of the bioimageio.core reproducibility tests.

Node pydantic-model

Bases: BaseModel

Base data model with common config

Show JSON schema:

{
  "description": "Base data model with common config",
  "properties": {},
  "title": "Node",
  "type": "object"
}

ToolCompatibilityReport pydantic-model

Bases: Node

Used to report on the compatibility of resource description in the bioimageio collection for a version specific tool.

Show JSON schema:

{
  "$defs": {
    "Badge": {
      "properties": {
        "icon": {
          "format": "uri",
          "maxLength": 2083,
          "minLength": 1,
          "title": "Icon",
          "type": "string"
        },
        "label": {
          "title": "Label",
          "type": "string"
        },
        "url": {
          "format": "uri",
          "maxLength": 2083,
          "minLength": 1,
          "title": "Url",
          "type": "string"
        }
      },
      "required": [
        "icon",
        "label",
        "url"
      ],
      "title": "Badge",
      "type": "object"
    },
    "ToolReportDetails": {
      "additionalProperties": true,
      "properties": {
        "traceback": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Traceback"
        },
        "warnings": {
          "anyOf": [
            {
              "additionalProperties": true,
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Warnings"
        },
        "metadata_completeness": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Metadata Completeness"
        },
        "status": {
          "anyOf": [
            {
              "enum": [
                "passed",
                "valid-format",
                "failed"
              ],
              "type": "string"
            },
            {}
          ],
          "default": null,
          "title": "Status"
        }
      },
      "title": "ToolReportDetails",
      "type": "object"
    }
  },
  "additionalProperties": true,
  "description": "Used to report on the compatibility of resource description\nin the bioimageio collection for a version specific tool.",
  "properties": {
    "tool": {
      "enum": [
        "bioimageio.core",
        "ilastik",
        "deepimagej",
        "icy",
        "biapy",
        "careamics"
      ],
      "title": "Tool",
      "type": "string"
    },
    "tool_version": {
      "pattern": "^[a-z0-9\\.-]+$",
      "title": "Tool Version",
      "type": "string"
    },
    "status": {
      "enum": [
        "passed",
        "failed",
        "not-applicable"
      ],
      "title": "Status",
      "type": "string"
    },
    "score": {
      "maximum": 1.0,
      "minimum": 0,
      "title": "Score",
      "type": "number"
    },
    "error": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "title": "Error"
    },
    "details": {
      "anyOf": [
        {
          "$ref": "#/$defs/ToolReportDetails"
        },
        {
          "type": "string"
        },
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Details"
    },
    "badge": {
      "anyOf": [
        {
          "$ref": "#/$defs/Badge"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "links": {
      "default": [],
      "items": {
        "type": "string"
      },
      "title": "Links",
      "type": "array"
    }
  },
  "required": [
    "tool",
    "tool_version",
    "status",
    "score",
    "error"
  ],
  "title": "ToolCompatibilityReport",
  "type": "object"
}

Fields:

• tool (Annotated[ToolName, Field(exclude=True, pattern='^[a-zA-Z0-9-\\.]+$')])
• tool_version (Annotated[str, Field(exclude=True, pattern='^[a-z0-9\\.-]+$')])
• status (Literal['passed', 'failed', 'not-applicable'])
• score (Annotated[float, Interval(ge=0, le=1.0)])
• error (Optional[str])
• details (Union[ToolReportDetails, str, List[str], None])
• badge (Optional[Badge])
• links (Sequence[str])

Validators:

• _set_default_score

badge pydantic-field

badge: Optional[Badge] = None

status badge with a resource specific link to the tool

details pydantic-field

details: Union[ToolReportDetails, str, List[str], None] = (
    None
)

details to explain the status

error pydantic-field

error: Optional[str]

error message if status=='failed'

links pydantic-field

links: Sequence[str] = ()

the checked resource should link these other bioimage.io resources

report_name property

report_name: str

score pydantic-field

score: Annotated[float, Interval(ge=0, le=1.0)]

score for the compatibility of this tool with the resource

status pydantic-field

status: Literal['passed', 'failed', 'not-applicable']

status of this tool for this resource

tool pydantic-field

tool: Annotated[
    ToolName,
    Field(exclude=True, pattern="^[a-zA-Z0-9-\\.]+$"),
]

tool name

tool_version pydantic-field

tool_version: Annotated[
    str, Field(exclude=True, pattern="^[a-z0-9\\.-]+$")
]

tool version, ideally in SemVer 2.0 format

ToolReportDetails pydantic-model

Bases: Node

Show JSON schema:

{
  "additionalProperties": true,
  "properties": {
    "traceback": {
      "anyOf": [
        {
          "items": {
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Traceback"
    },
    "warnings": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Warnings"
    },
    "metadata_completeness": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Metadata Completeness"
    },
    "status": {
      "anyOf": [
        {
          "enum": [
            "passed",
            "valid-format",
            "failed"
          ],
          "type": "string"
        },
        {}
      ],
      "default": null,
      "title": "Status"
    }
  },
  "title": "ToolReportDetails",
  "type": "object"
}

Fields:

• traceback (Optional[Sequence[str]])
• warnings (Optional[Mapping[str, Any]])
• metadata_completeness (Optional[float])
• status (Union[Literal['passed', 'valid-format', 'failed'], Any])

metadata_completeness pydantic-field

metadata_completeness: Optional[float] = None

status pydantic-field

status: Union[
    Literal["passed", "valid-format", "failed"], Any
] = None

traceback pydantic-field

traceback: Optional[Sequence[str]] = None

warnings pydantic-field

warnings: Optional[Mapping[str, Any]] = None

compatibility_pure

Pure Python alternative data models for compatibility reports

Usable when installing backoffice without dependencies.

Classes:

• BadgeDict –
• ToolCompatibilityReportDict –

BadgeDict

Bases: TypedDict

Attributes:

• icon (str) –
• label (str) –
• url (str) –

icon instance-attribute

icon: str

label instance-attribute

label: str

url instance-attribute

url: str

ToolCompatibilityReportDict

Bases: TypedDict

Attributes:

• badge (Optional[BadgeDict]) –

  status badge with a resource specific link to the tool

• details (Any) –

  details to explain the status

• error (Optional[str]) –

  error message if status=='failed'

• links (Sequence[str]) –

  the checked resource should link these other bioimage.io resources

• status (Literal['passed', 'failed', 'not-applicable']) –

  status of this tool for this resource

• tool (str) –

  toolname (including version separated by an underscore)

badge instance-attribute

badge: Optional[BadgeDict]

status badge with a resource specific link to the tool

details instance-attribute

details: Any

details to explain the status

error instance-attribute

error: Optional[str]

error message if status=='failed'

links instance-attribute

links: Sequence[str]

the checked resource should link these other bioimage.io resources

status instance-attribute

status: Literal['passed', 'failed', 'not-applicable']

status of this tool for this resource

tool instance-attribute

tool: str

toolname (including version separated by an underscore)

index

Data models and functions for indexing the bioimage.io collection

Classes:

• Index –
• IndexItem –
• IndexItemVersion –
• Node –
• Response –

  Response from Hypha list endpoint

• ResponseItem –
• ResponseItemVersion –

Functions:

• create_index –

  Index the bioimage.io collection

• load_index –

_ module-attribute

_ = create_index()

Index pydantic-model

Bases: Node

count_per_type instance-attribute

count_per_type: dict[str, int]

items instance-attribute

items: list[IndexItem]

timestamp class-attribute instance-attribute

timestamp: datetime = Field(default_factory=now)

total instance-attribute

total: int

IndexItem pydantic-model

Bases: Node

id instance-attribute

id: str

type instance-attribute

type: str

versions instance-attribute

versions: Sequence[IndexItemVersion]

IndexItemVersion pydantic-model

Bases: Node

comment instance-attribute

comment: Optional[str]

created_at instance-attribute

created_at: datetime

sha256 instance-attribute

sha256: str

source instance-attribute

source: str

version instance-attribute

version: str

Node pydantic-model

Bases: BaseModel

Response pydantic-model

Bases: Node

Response from Hypha list endpoint

items instance-attribute

items: list[ResponseItem]

limit instance-attribute

limit: int

offset instance-attribute

offset: int

total instance-attribute

total: int

ResponseItem pydantic-model

Bases: Node

id instance-attribute

id: str

type instance-attribute

type: str

versions instance-attribute

versions: Sequence[ResponseItemVersion]

ResponseItemVersion pydantic-model

Bases: Node

comment instance-attribute

comment: Optional[str]

created_at instance-attribute

created_at: datetime

version instance-attribute

version: str

create_index

create_index() -> Index

Index the bioimage.io collection

Source code in src/backoffice/index.py

def create_index() -> Index:
    """Index the bioimage.io collection"""

    index_path = Path("index.json")
    if index_path.exists():
        index = load_index(index_path)
    else:
        url = f"{settings.hypha_base_url}/public/services/artifact-manager/list"

        def request(offset: int) -> Response:
            r = httpx.get(
                url,
                params=dict(
                    parent_id="bioimage-io/bioimage.io",
                    offset=offset,
                    pagination=True,
                    limit=10000,
                ),
                headers=settings.get_hypha_headers(),
                timeout=settings.http_timeout,
            )
            try:
                _ = r.raise_for_status()
            except Exception as e:
                logger.error(r.json())
                raise e
            else:
                return Response.model_validate_json(r.content)

        items: list[ResponseItem] = []
        for page in range(100):
            response = request(len(items))
            logger.info("Page {}: {} entries", page, len(response.items))
            items.extend(response.items)
            if response.total <= len(items):
                if response.total != len(items):
                    logger.error(
                        "response.total {} != len(items) {}", response.total, len(items)
                    )
                break

        index_items: list[IndexItem] = []
        for item in items:
            domain, item_id_wo_domain = item.id.split("/", 1)
            versions: list[IndexItemVersion] = []
            for v in item.versions:
                url = f"{settings.hypha_base_url}/{domain}/artifacts/{item_id_wo_domain}/files/rdf.yaml?version={v.version}"
                sha256 = _initialize_report_directory(item, v, url)
                versions.append(
                    IndexItemVersion(
                        version=v.version,
                        comment=v.comment,
                        created_at=v.created_at,
                        source=url,
                        sha256=sha256,
                    )
                )
            index_items.append(IndexItem(id=item.id, versions=versions, type=item.type))

        count_per_type = defaultdict[str, int](int)
        for item in index_items:
            count_per_type[item.type] += 1

        index = Index(
            items=index_items,
            total=len(index_items),
            count_per_type=dict(count_per_type),
        )

        json_dict = index.model_dump(mode="json")
        with index_path.open("wt", encoding="utf-8") as f:
            json.dump(json_dict, f, indent=4, sort_keys=True, ensure_ascii=False)
        # TODO: use .model_dump_json once it supports 'sort_keys' argument for a potential speed gain
        # _ = index_path.write_text(index.model_dump_json(indent=4), encoding="utf-8")

        logger.info("saved index to {}", index_path)

    logger.info(
        "loaded index with {} ids and {} versions",
        len(index.items),
        sum(len(item.versions) for item in index.items),
    )
    return index

load_index

load_index(path: Path = Path('index.json')) -> Index

Source code in src/backoffice/index.py

def load_index(path: Path = Path("index.json")) -> Index:
    logger.info("loading index from {}", path)
    return Index.model_validate_json(path.read_text(encoding="utf-8"))

utils

utility functions

Functions:

• get_rdf_content_from_url –
• get_summary –

  Retrieve the summary for a specific item and version.

Attributes:

• yaml –

yaml module-attribute

yaml = YAML(typ='safe')

get_rdf_content_from_url

get_rdf_content_from_url(
    url: str, sha256: str
) -> dict[str, Any]

Source code in src/backoffice/utils.py

def get_rdf_content_from_url(url: str, sha256: str) -> dict[str, Any]:
    local_path = cached_download(url, sha256)
    return yaml.load(local_path)

get_summary

get_summary(
    item_id: str, version: str
) -> InitialSummary | CompatibilitySummary

Retrieve the summary for a specific item and version.

Source code in src/backoffice/utils.py

def get_summary(item_id: str, version: str) -> "InitialSummary | CompatibilitySummary":
    """Retrieve the summary for a specific item and version."""
    summary_path = get_summary_file_path(item_id, version)
    if not summary_path.exists():
        return InitialSummary(
            rdf_content={},
            rdf_yaml_sha256="",
            status="untested",
        )

    with summary_path.open(encoding="utf-8") as f:
        data = json.load(f)

    try:
        return CompatibilitySummary.model_validate(data)
    except ValidationError:
        return InitialSummary.model_validate(data)

utils_pure

utility functions available in backoffice without dependencies

Functions:

• cached_download –

  Download a file from the given URL and cache it locally.

• get_all_tool_report_paths –
• get_log_file –
• get_rdf_content_from_id –

  Get the RDF file content of a specific item version.

• get_report_path –
• get_summary_data –

  Get the summary data of a specific item version.

• get_summary_file_path –
• get_tool_report_path –

  Get the path to the report for a specific item version and tool.

• raise_for_status_discretely –

  Raises :class:httpx.HTTPError for 4xx or 5xx responses,

_ module-attribute

_ = load_dotenv()

cached_download

cached_download(url: str, sha256: str) -> Path

Download a file from the given URL and cache it locally.

Source code in src/backoffice/utils_pure.py

def cached_download(url: str, sha256: str) -> Path:
    """Download a file from the given URL and cache it locally."""
    import httpx

    local_path = Path("cache") / sha256
    if not local_path.exists():
        local_path.parent.mkdir(parents=True, exist_ok=True)
        response = httpx.get(
            url, timeout=float(os.environ.get("HTTP_TIMEOUT", "30"))
        ).raise_for_status()
        with local_path.open("wb") as f:
            _ = f.write(response.content)

    return local_path

get_all_tool_report_paths

get_all_tool_report_paths(item_id: str, version: str)

Source code in src/backoffice/utils_pure.py

def get_all_tool_report_paths(
    item_id: str,
    version: str,
):
    return list((get_report_path(item_id, version) / "reports").glob("*.json"))

get_log_file

get_log_file(item_id: str, version: str) -> Path

Source code in src/backoffice/utils_pure.py

def get_log_file(item_id: str, version: str) -> Path:
    return get_report_path(item_id, version) / "log.txt"

get_rdf_content_from_id

get_rdf_content_from_id(
    item_id: str, version: str
) -> dict[str, Any]

Get the RDF file content of a specific item version.

Source code in src/backoffice/utils_pure.py

def get_rdf_content_from_id(item_id: str, version: str) -> dict[str, Any]:
    """Get the RDF file content of a specific item version."""
    with get_summary_file_path(item_id, version).open() as f:
        return json.load(f)["rdf_content"]

get_report_path

get_report_path(item_id: str, version: str) -> Path

Source code in src/backoffice/utils_pure.py

def get_report_path(
    item_id: str,
    version: str,
) -> Path:
    return Path(os.getenv("REPORTS", "reports")) / item_id.replace(":", "_") / version

get_summary_data

get_summary_data(
    item_id: str, version: str
) -> Optional[dict[str, Any]]

Get the summary data of a specific item version.

Source code in src/backoffice/utils_pure.py

def get_summary_data(item_id: str, version: str) -> Optional[dict[str, Any]]:
    """Get the summary data of a specific item version."""
    summary_file_path = get_summary_file_path(item_id, version)
    if not summary_file_path.exists():
        return None

    with summary_file_path.open(encoding="utf-8") as f:
        return json.load(f)

get_summary_file_path

get_summary_file_path(item_id: str, version: str) -> Path

Source code in src/backoffice/utils_pure.py

def get_summary_file_path(item_id: str, version: str) -> Path:
    return get_report_path(item_id, version) / "summary.json"

get_tool_report_path

get_tool_report_path(
    item_id: str,
    version: str,
    tool_name: str,
    tool_version: str,
)

Get the path to the report for a specific item version and tool.

Source code in src/backoffice/utils_pure.py

def get_tool_report_path(
    item_id: str,
    version: str,
    tool_name: str,
    tool_version: str,
):
    """Get the path to the report for a specific item version and tool."""
    if "_" in tool_name:
        raise ValueError("Underscore not allowed in tool_name")

    if "_" in tool_version:
        raise ValueError("Underscore not allowed in tool_version")

    return (
        get_report_path(item_id, version)
        / "reports"
        / f"{tool_name}_{tool_version}.json"
    )

raise_for_status_discretely

raise_for_status_discretely(response: Response)

Raises :class:httpx.HTTPError for 4xx or 5xx responses, but hides any query and userinfo from url to avoid leaking sensitive data.

Source code in src/backoffice/utils_pure.py

def raise_for_status_discretely(response: "httpx.Response"):
    """Raises :class:`httpx.HTTPError` for 4xx or 5xx responses,
    **but** hides any query and userinfo from url to avoid leaking sensitive data.
    """
    import httpx

    http_error_msg = ""
    reason = response.reason_phrase

    discrete_url = response.url.copy_with(
        query=(b"***query*hidden***" if response.url.query else b""),
        userinfo=(b"***userinfo*hidden***" if response.url.userinfo else b""),
    )

    if 400 <= response.status_code < 500:
        http_error_msg = (
            f"{response.status_code} Client Error: {reason} for url: {discrete_url}"
        )

    elif 500 <= response.status_code < 600:
        http_error_msg = (
            f"{response.status_code} Server Error: {reason} for url: {discrete_url}"
        )

    if http_error_msg:
        raise httpx.HTTPError(http_error_msg)