OME-Zarr Collection v1 export

Export an AcqImageList as a staged v1 collection.

Parameters:

Name	Type	Description	Default
destination	str | Path	New local directory ending in .ome.zarr.	required
name	str | None	Optional human-readable collection name.	None
overwrite	bool	Whether an existing destination may be replaced.	False
zarr_format	int	Zarr format passed unchanged to image writers.	3

Source code in src/acqstore/acq_image/io/ome_zarr_collection_v1/exporter.py

class AcqStoreOmeZarrCollectionExporter:
    """Export an ``AcqImageList`` as a staged v1 collection.

    Args:
        destination: New local directory ending in ``.ome.zarr``.
        name: Optional human-readable collection name.
        overwrite: Whether an existing destination may be replaced.
        zarr_format: Zarr format passed unchanged to image writers.
    """

    def __init__(
        self,
        destination: str | Path,
        *,
        name: str | None = None,
        overwrite: bool = False,
        zarr_format: int = 3,
    ) -> None:
        """Configure one collection export operation.

        Args:
            destination: New local directory ending in ``.ome.zarr``.
            name: Optional human-readable collection name.
            overwrite: Whether an existing destination may be replaced.
            zarr_format: Zarr format passed unchanged to image writers.

        Raises:
            ValueError: If the destination is remote, has the wrong suffix, or
                ``zarr_format`` is unsupported.
        """
        raw_destination = str(destination)
        if '://' in raw_destination:
            raise ValueError('v1 collection export supports local paths only')
        self._destination = Path(destination).expanduser().resolve(strict=False)
        if not self._destination.name.lower().endswith('.ome.zarr'):
            raise ValueError("Collection destination must end in '.ome.zarr'")
        if zarr_format not in {2, 3}:
            raise ValueError(f'zarr_format must be 2 or 3, got {zarr_format!r}')
        self._name = name
        self._overwrite = overwrite
        self._zarr_format = zarr_format

    def export(self, acq_image_list: AcqImageList) -> Path:
        """Export a non-empty acquisition list as one v1 collection.

        Args:
            acq_image_list: Fully loaded source collection.

        Returns:
            Resolved completed collection path.

        Raises:
            FileExistsError: If the destination exists and overwrite is false.
            ValueError: If the collection is empty or fails conformance.
        """
        members = tuple(acq_image_list)
        if not members:
            raise ValueError('Cannot export an empty AcqImageList')
        if self._destination.exists() and not self._overwrite:
            raise FileExistsError(f'Destination already exists: {self._destination}')

        self._destination.parent.mkdir(parents=True, exist_ok=True)
        with tempfile.TemporaryDirectory(
            prefix=f'.{self._destination.name}.staging-',
            dir=self._destination.parent,
        ) as temporary_directory:
            staged = Path(temporary_directory) / self._destination.name
            staged.mkdir()
            self._build(acq_image_list, members, staged)
            validate_collection(staged, get_schema_path())
            self._validate_ome_images(staged)
            self._install(staged)
        return self._destination

    def _build(
        self,
        acq_image_list: AcqImageList,
        members: tuple[AcqImage, ...],
        staged: Path,
    ) -> None:
        """Build a complete collection in a staging directory.

        Args:
            acq_image_list: Source collection owning analysis pools.
            members: Ordered source image snapshot.
            staged: Empty staging root.

        Returns:
            None.
        """
        results = [
            AcqStoreOmeZarrImageExporter(
                staged,
                image_id=str(uuid.uuid4()),
                zarr_format=self._zarr_format,
            ).export(acq_image)
            for acq_image in members
        ]
        table_resources = self._export_collection_tables(acq_image_list, staged)
        document: dict[str, Any] = {
            'format': 'acqstore-ome-zarr-collection',
            'version': 1,
            'id': str(uuid.uuid4()),
            'name': self._name or self._default_name(acq_image_list),
            'created': datetime.now(UTC).replace(microsecond=0).isoformat().replace('+00:00', 'Z'),
            'producer': {'name': 'acqstore'},
            'members': results,
        }
        if table_resources:
            document['resources'] = {'tables': table_resources}
        AcqStoreOmeZarrImageExporter._write_json(
            staged / 'acqstore' / 'collection.json',
            document,
        )

    def _export_collection_tables(
        self,
        acq_image_list: AcqImageList,
        staged: Path,
    ) -> list[dict[str, str]]:
        """Write non-empty generic collection-level CSV resources.

        Args:
            acq_image_list: Source collection owning current analysis pools.
            staged: Collection staging root.

        Returns:
            Generic CSV descriptors for ``collection.json``.
        """
        resources: list[dict[str, str]] = []
        pools = (
            ('velocity', acq_image_list.velocity_analysis_pool.get_dataframe()),
            ('sum_intensity', acq_image_list.sum_intensity_analysis_pool.get_dataframe()),
        )
        for resource_id, dataframe in pools:
            if dataframe.empty:
                continue
            relative = Path('tables') / f'{resource_id}.csv'
            (staged / relative).parent.mkdir(parents=True, exist_ok=True)
            dataframe.to_csv(staged / relative, index=False)
            resources.append(
                {
                    'id': resource_id,
                    'media_type': 'text/csv',
                    'path': relative.as_posix(),
                }
            )
        return resources

    def _install(self, staged: Path) -> None:
        """Atomically install a validated staging directory.

        Args:
            staged: Validated staging collection.

        Returns:
            None.
        """
        if self._destination.exists():
            backup = self._destination.with_name(f'.{self._destination.name}.backup-{uuid.uuid4()}')
            os.replace(self._destination, backup)
            try:
                os.replace(staged, self._destination)
            except Exception:
                os.replace(backup, self._destination)
                raise
            shutil.rmtree(backup)
        else:
            os.replace(staged, self._destination)

    @staticmethod
    def _validate_ome_images(staged: Path) -> None:
        """Open every primary and reference image independently as OME-Zarr.

        Args:
            staged: Schema-valid collection staging root.

        Returns:
            None.

        Raises:
            ValueError: If a declared image cannot be opened as OME-Zarr.
        """
        from acqstore.acq_image.io.ome_zarr import read_acq_pixels_ome_zarr

        collection = json.loads(
            (staged / 'acqstore' / 'collection.json').read_text(encoding='utf-8')
        )
        for member in collection['members']:
            read_acq_pixels_ome_zarr(staged / member['ome_zarr'], lazy=True)
            reference = member.get('reference_image')
            if reference is not None:
                read_acq_pixels_ome_zarr(staged / reference['ome_zarr'], lazy=True)

    def _default_name(self, acq_image_list: AcqImageList) -> str:
        """Return a human-readable name without assigning identity semantics.

        Args:
            acq_image_list: Source collection.

        Returns:
            Source-root name when available, otherwise destination name.
        """
        source_root = getattr(acq_image_list, 'source_root_path', None)
        if source_root:
            name = Path(str(source_root)).expanduser().name
            if name:
                return name
        return self._destination.name

export

export(acq_image_list: AcqImageList) -> Path

Export a non-empty acquisition list as one v1 collection.

Parameters:

Name	Type	Description	Default
acq_image_list	AcqImageList	Fully loaded source collection.	required

Returns:

Type	Description
Path	Resolved completed collection path.

Raises:

Type	Description
FileExistsError	If the destination exists and overwrite is false.
ValueError	If the collection is empty or fails conformance.

Source code in src/acqstore/acq_image/io/ome_zarr_collection_v1/exporter.py

def export(self, acq_image_list: AcqImageList) -> Path:
    """Export a non-empty acquisition list as one v1 collection.

    Args:
        acq_image_list: Fully loaded source collection.

    Returns:
        Resolved completed collection path.

    Raises:
        FileExistsError: If the destination exists and overwrite is false.
        ValueError: If the collection is empty or fails conformance.
    """
    members = tuple(acq_image_list)
    if not members:
        raise ValueError('Cannot export an empty AcqImageList')
    if self._destination.exists() and not self._overwrite:
        raise FileExistsError(f'Destination already exists: {self._destination}')

    self._destination.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(
        prefix=f'.{self._destination.name}.staging-',
        dir=self._destination.parent,
    ) as temporary_directory:
        staged = Path(temporary_directory) / self._destination.name
        staged.mkdir()
        self._build(acq_image_list, members, staged)
        validate_collection(staged, get_schema_path())
        self._validate_ome_images(staged)
        self._install(staged)
    return self._destination

The collection exporter delegates each member to the focused image exporter:

Export one AcqImage into an in-progress v1 collection.

Parameters:

Name	Type	Description	Default
collection_root	str | Path	Staging root for the complete collection.	required
image_id	str	Newly allocated opaque v1 member identifier.	required
zarr_format	int	Zarr format passed unchanged to public OME-Zarr writers.	3

Source code in src/acqstore/acq_image/io/ome_zarr_collection_v1/exporter.py

class AcqStoreOmeZarrImageExporter:
    """Export one ``AcqImage`` into an in-progress v1 collection.

    Args:
        collection_root: Staging root for the complete collection.
        image_id: Newly allocated opaque v1 member identifier.
        zarr_format: Zarr format passed unchanged to public OME-Zarr writers.
    """

    def __init__(
        self,
        collection_root: str | Path,
        *,
        image_id: str,
        zarr_format: int = 3,
    ) -> None:
        """Create an exporter for one v1 member.

        Args:
            collection_root: Staging root for the complete collection.
            image_id: Newly allocated opaque v1 member identifier.
            zarr_format: Zarr format passed to public OME-Zarr writers.
        """
        self._root = Path(collection_root)
        self._image_id = image_id
        self._zarr_format = zarr_format
        self._roi_ids: dict[int, str] = {}

    def export(self, acq_image: AcqImage) -> dict[str, Any]:
        """Write one primary image and its declared AcqStore resources.

        Args:
            acq_image: Fully loaded source acquisition image.

        Returns:
            Collection member descriptor for ``collection.json``.

        Raises:
            ValueError: If required pixels or analysis tables are not loaded,
                an ROI is unsupported, or source values cannot satisfy v1.
        """
        if not acq_image.images_loaded:
            raise ValueError(f'AcqImage pixels are not loaded: {acq_image.name}')
        if not acq_image.analysis_csv_loaded:
            raise ValueError(f'AcqImage analysis tables are not loaded: {acq_image.name}')

        primary_path = Path('images') / self._image_id
        acq_image.save_as_ome_zarr(
            self._root / primary_path,
            overwrite=False,
            zarr_format=self._zarr_format,
        )

        metadata_dir = self._root / 'metadata' / self._image_id
        metadata_dir.mkdir(parents=True, exist_ok=False)
        rois = self._build_rois(acq_image)
        acqimage_relative = Path('metadata') / self._image_id / 'acqimage.json'
        self._write_json(
            self._root / acqimage_relative,
            self._build_acqimage_document(acq_image, rois),
        )

        resources: dict[str, str] = {'acqimage': acqimage_relative.as_posix()}
        analyses = self._export_analyses(acq_image)
        if analyses:
            analyses_relative = Path('metadata') / self._image_id / 'analyses.json'
            self._write_json(
                self._root / analyses_relative,
                {
                    'format': 'acqstore-analyses',
                    'version': 1,
                    'image_id': self._image_id,
                    'analyses': analyses,
                },
            )
            resources['analyses'] = analyses_relative.as_posix()

        member: dict[str, Any] = {
            'id': self._image_id,
            'name': str(acq_image.name),
            'ome_zarr': primary_path.as_posix(),
            'resources': resources,
            'summary': self._build_summary(acq_image, analyses),
        }
        reference = self._export_reference(acq_image)
        if reference is not None:
            member['reference_image'] = reference
        return member

    def _build_rois(self, acq_image: AcqImage) -> list[dict[str, Any]]:
        """Build v1 ROI documents from existing public ROI objects.

        Args:
            acq_image: Source acquisition image.

        Returns:
            Schema-ready ROI objects.

        Raises:
            TypeError: If the image contains an unsupported ROI class.
        """
        output: list[dict[str, Any]] = []
        for roi in acq_image.rois:
            roi_id = str(uuid.uuid4())
            self._roi_ids[int(roi.roi_id)] = roi_id
            common: dict[str, Any] = {
                'id': roi_id,
                'name': str(roi.name) or f'ROI {roi.roi_id}',
                'coordinate_space': 'primary-image-full-resolution-pixels',
            }
            if roi.note:
                common['metadata'] = {'note': str(roi.note)}
            if isinstance(roi, RectROI):
                output.append(
                    {
                        **common,
                        'type': 'rectangle',
                        'start': [int(roi.bounds.dim1_start), int(roi.bounds.dim0_start)],
                        'stop': [int(roi.bounds.dim1_stop), int(roi.bounds.dim0_stop)],
                    }
                )
            elif isinstance(roi, LineROI):
                output.append(
                    {
                        **common,
                        'type': 'line',
                        'start': [int(roi.endpoints.col0), int(roi.endpoints.row0)],
                        'stop': [int(roi.endpoints.col1), int(roi.endpoints.row1)],
                    }
                )
            else:
                raise TypeError(f'Unsupported v1 ROI class: {type(roi).__name__}')
        return output

    def _build_acqimage_document(
        self,
        acq_image: AcqImage,
        rois: list[dict[str, Any]],
    ) -> dict[str, Any]:
        """Build one schema-ready ``acqimage.json`` object.

        Args:
            acq_image: Source acquisition image.
            rois: Previously allocated v1 ROI documents.

        Returns:
            Complete v1 AcqImage document.
        """
        document: dict[str, Any] = {
            'format': 'acqstore-acqimage',
            'version': 1,
            'image_id': self._image_id,
            'accepted': bool(acq_image.get_schema_row()['accept']),
            'rois': rois,
            'experiment_metadata': self._json_value(
                acq_image.get_metadata_section('experiment_metadata').get_values()
            ),
            'image_metadata': self._json_value(
                acq_image.get_metadata_section('acq_image_header').get_values()
            ),
        }
        axis_display = self._build_axis_display(acq_image)
        if axis_display:
            document['axis_display'] = axis_display
        return document

    def _build_axis_display(self, acq_image: AcqImage) -> dict[str, Any]:
        """Build sparse display overrides for exceptional raster axes.

        Args:
            acq_image: Source acquisition image.

        Returns:
            Empty object for ordinary images or a sparse axis mapping.
        """
        header = acq_image.pixels.header
        overrides: dict[str, Any] = {}
        temporal_units = {
            's': 'second',
            'sec': 'second',
            'second': 'second',
            'seconds': 'second',
            'ms': 'millisecond',
            'millisecond': 'millisecond',
            'milliseconds': 'millisecond',
            'us': 'microsecond',
            'µs': 'microsecond',
            'microsecond': 'microsecond',
            'microseconds': 'microsecond',
        }
        for index, axis in enumerate(header.dims):
            raw_unit = str(header.physical_units_labels[index]).strip().lower()
            if str(axis).upper() != 'Y' or raw_unit not in temporal_units:
                continue
            overrides['y'] = {
                'type': 'time',
                'unit': temporal_units[raw_unit],
                'scale': float(header.physical_units[index]),
            }
        return overrides

    def _export_analyses(self, acq_image: AcqImage) -> list[dict[str, Any]]:
        """Write analysis CSVs and build analysis envelopes.

        Args:
            acq_image: Source acquisition image.

        Returns:
            Analysis documents for instances with result tables.

        Raises:
            ValueError: If an analysis references an unknown ROI.
        """
        output: list[dict[str, Any]] = []
        analysis_dir = self._root / 'analysis' / self._image_id
        for analysis in acq_image.analysis_set.as_list():
            table = analysis.table_with_bookkeeping()
            if table is None:
                if analysis.result.summary:
                    raise ValueError(
                        f'Completed analysis {analysis.key.analysis_name!r} has a summary but no '
                        'CSV table resource required by Collection v1'
                    )
                continue
            source_roi_id = int(analysis.key.roi_id)
            if source_roi_id not in self._roi_ids:
                raise ValueError(
                    f'Analysis {analysis.key.analysis_name!r} references unknown ROI {source_roi_id}'
                )
            analysis_id = str(uuid.uuid4())
            analysis_dir.mkdir(parents=True, exist_ok=True)
            csv_relative = Path('analysis') / self._image_id / f'{analysis_id}.csv'
            table.to_csv(self._root / csv_relative, index=False)
            output.append(
                self._build_analysis_document(
                    analysis,
                    analysis_id=analysis_id,
                    roi_id=self._roi_ids[source_roi_id],
                    csv_path=csv_relative.as_posix(),
                )
            )
        return output

    def _build_analysis_document(
        self,
        analysis: BaseAnalysis,
        *,
        analysis_id: str,
        roi_id: str,
        csv_path: str,
    ) -> dict[str, Any]:
        """Build one typed analysis envelope.

        Args:
            analysis: Source analysis instance.
            analysis_id: Newly allocated opaque analysis identifier.
            roi_id: Owning opaque v1 ROI identifier.
            csv_path: Collection-root-relative result table path.

        Returns:
            Complete schema-ready analysis object.
        """
        return {
            'id': analysis_id,
            'type': str(analysis.key.analysis_name),
            'roi_id': roi_id,
            'channel': int(analysis.key.channel),
            'parameters': self._json_value(dict(analysis.detection_params)),
            'summary': self._json_value(dict(analysis.result.summary)),
            'resources': [
                {
                    'id': 'table',
                    'media_type': 'text/csv',
                    'path': csv_path,
                }
            ],
        }

    def _export_reference(self, acq_image: AcqImage) -> dict[str, str] | None:
        """Write an optional reference image and metadata document.

        Args:
            acq_image: Source acquisition image.

        Returns:
            Reference-image link for ``collection.json``, or ``None``.

        Raises:
            ValueError: If scan-path points are fractional or malformed.
        """
        if not acq_image.images.has_reference_image:
            return None
        reference_relative = Path('images') / f'{self._image_id}-reference'
        acq_image.save_reference_as_ome_zarr(
            self._root / reference_relative,
            overwrite=False,
            zarr_format=self._zarr_format,
        )
        values = self._json_value(
            acq_image.get_metadata_section('reference_image_metadata').get_values()
        )
        x_values = values.pop('scan_path_x_pixels', [])
        y_values = values.pop('scan_path_y_pixels', [])
        has_scan_path = bool(values.pop('has_scan_path', False))
        values.pop('scan_path_num_points', None)
        metadata_relative = Path('metadata') / self._image_id / 'reference-image.json'
        document: dict[str, Any] = {
            'format': 'acqstore-reference-image',
            'version': 1,
            'image_id': self._image_id,
            'metadata': values,
        }
        if has_scan_path:
            if len(x_values) != len(y_values) or len(x_values) < 2:
                raise ValueError('Reference scan path must contain matching X/Y arrays with at least two points')
            points = [
                [self._integer_coordinate(x), self._integer_coordinate(y)]
                for x, y in zip(x_values, y_values, strict=True)
            ]
            document['scan_path'] = {
                'coordinate_space': 'reference-image-full-resolution-pixels',
                'points': points,
            }
        self._write_json(self._root / metadata_relative, document)
        return {
            'ome_zarr': reference_relative.as_posix(),
            'metadata': metadata_relative.as_posix(),
        }

    def _build_summary(
        self,
        acq_image: AcqImage,
        analyses: list[dict[str, Any]],
    ) -> dict[str, Any]:
        """Build deliberately small non-authoritative discovery metadata.

        Args:
            acq_image: Source acquisition image.
            analyses: Exported analysis envelopes.

        Returns:
            Untyped summary object for collection discovery.
        """
        header = acq_image.pixels.header
        return {
            'shape': [int(value) for value in header.shape],
            'dims': [str(dim).lower() for dim in header.dims],
            'dtype': str(header.dtype),
            'num_channels': int(header.num_channels),
            'num_rois': int(acq_image.rois.num_rois),
            'analysis_types': sorted({str(item['type']) for item in analyses}),
            'accepted': bool(acq_image.get_schema_row()['accept']),
            'has_reference_image': bool(acq_image.images.has_reference_image),
        }

    @staticmethod
    def _integer_coordinate(value: object) -> int:
        """Return an integer-valued pixel coordinate without rounding.

        Args:
            value: Numeric coordinate supplied by the public metadata API.

        Returns:
            Exact integer coordinate.

        Raises:
            ValueError: If the value is negative, non-finite, or fractional.
        """
        numeric = float(value)
        if not np.isfinite(numeric) or numeric < 0 or not numeric.is_integer():
            raise ValueError(f'v1 scan-path coordinate must be a non-negative integer, got {value!r}')
        return int(numeric)

    @classmethod
    def _json_value(cls, value: Any) -> Any:
        """Convert common scientific scalar/container values to JSON values.

        Args:
            value: Value returned by an existing public AcqStore API.

        Returns:
            Recursively converted JSON-compatible value.
        """
        if isinstance(value, dict):
            return {str(key): cls._json_value(item) for key, item in value.items()}
        if isinstance(value, (list, tuple)):
            return [cls._json_value(item) for item in value]
        if isinstance(value, np.ndarray):
            return cls._json_value(value.tolist())
        if isinstance(value, np.generic):
            return value.item()
        if value is pd.NA:
            return None
        return value

    @staticmethod
    def _write_json(path: Path, document: dict[str, Any]) -> None:
        """Write one deterministic UTF-8 JSON document.

        Args:
            path: Destination file path.
            document: JSON-compatible object.

        Returns:
            None.
        """
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(document, indent=2) + '\n', encoding='utf-8')

export

export(acq_image: AcqImage) -> dict[str, Any]

Write one primary image and its declared AcqStore resources.

Parameters:

Name	Type	Description	Default
acq_image	AcqImage	Fully loaded source acquisition image.	required

Returns:

Type	Description
dict[str, Any]	Collection member descriptor for collection.json.

Raises:

Type	Description
ValueError	If required pixels or analysis tables are not loaded, an ROI is unsupported, or source values cannot satisfy v1.

Source code in src/acqstore/acq_image/io/ome_zarr_collection_v1/exporter.py

def export(self, acq_image: AcqImage) -> dict[str, Any]:
    """Write one primary image and its declared AcqStore resources.

    Args:
        acq_image: Fully loaded source acquisition image.

    Returns:
        Collection member descriptor for ``collection.json``.

    Raises:
        ValueError: If required pixels or analysis tables are not loaded,
            an ROI is unsupported, or source values cannot satisfy v1.
    """
    if not acq_image.images_loaded:
        raise ValueError(f'AcqImage pixels are not loaded: {acq_image.name}')
    if not acq_image.analysis_csv_loaded:
        raise ValueError(f'AcqImage analysis tables are not loaded: {acq_image.name}')

    primary_path = Path('images') / self._image_id
    acq_image.save_as_ome_zarr(
        self._root / primary_path,
        overwrite=False,
        zarr_format=self._zarr_format,
    )

    metadata_dir = self._root / 'metadata' / self._image_id
    metadata_dir.mkdir(parents=True, exist_ok=False)
    rois = self._build_rois(acq_image)
    acqimage_relative = Path('metadata') / self._image_id / 'acqimage.json'
    self._write_json(
        self._root / acqimage_relative,
        self._build_acqimage_document(acq_image, rois),
    )

    resources: dict[str, str] = {'acqimage': acqimage_relative.as_posix()}
    analyses = self._export_analyses(acq_image)
    if analyses:
        analyses_relative = Path('metadata') / self._image_id / 'analyses.json'
        self._write_json(
            self._root / analyses_relative,
            {
                'format': 'acqstore-analyses',
                'version': 1,
                'image_id': self._image_id,
                'analyses': analyses,
            },
        )
        resources['analyses'] = analyses_relative.as_posix()

    member: dict[str, Any] = {
        'id': self._image_id,
        'name': str(acq_image.name),
        'ome_zarr': primary_path.as_posix(),
        'resources': resources,
        'summary': self._build_summary(acq_image, analyses),
    }
    reference = self._export_reference(acq_image)
    if reference is not None:
        member['reference_image'] = reference
    return member

Validation uses the published Draft 2020-12 schema plus cross-document checks:

Validate one complete AcqStore v1 additive resource graph.

Parameters:

Name	Type	Description	Default
root	Path	Collection root containing acqstore/collection.json.	required
schema_path	Path	Canonical bundled schema path.	required

Returns:

Type	Description
None	None.

Raises:

Type	Description
ConformanceError	If any schema or cross-document rule fails.

Source code in src/acqstore/acq_image/io/ome_zarr_collection_v1/validator.py

def validate_collection(root: Path, schema_path: Path) -> None:
    """Validate one complete AcqStore v1 additive resource graph.

    Args:
        root: Collection root containing ``acqstore/collection.json``.
        schema_path: Canonical bundled schema path.

    Returns:
        None.

    Raises:
        ConformanceError: If any schema or cross-document rule fails.
    """
    schema = load_json(schema_path)
    Draft202012Validator.check_schema(schema)
    validator = Draft202012Validator(schema, format_checker=FormatChecker())

    manifest_path = root / "acqstore" / "collection.json"
    collection = load_json(manifest_path)
    schema_validate(validator, collection, "acqstore/collection.json")

    members = collection["members"]
    require_unique([member["id"] for member in members], "member IDs")

    collection_tables = collection.get("resources", {}).get("tables", [])
    require_unique([resource["id"] for resource in collection_tables], "collection table resource IDs")
    for resource in collection_tables:
        resolve_existing(root, resource["path"], "collection CSV resource")

    for member in members:
        image_id = member["id"]
        resolve_existing(root, member["ome_zarr"], f"primary OME-Zarr path for {image_id}")

        acqimage_path = resolve_existing(root, member["resources"]["acqimage"], "acqimage.json")
        acqimage = load_json(acqimage_path)
        schema_validate(validator, acqimage, str(acqimage_path.relative_to(root)))
        if acqimage["format"] != "acqstore-acqimage":
            raise ConformanceError(f"Expected acqimage document: {acqimage_path.relative_to(root)}")
        if acqimage["image_id"] != image_id:
            raise ConformanceError(f"acqimage image_id does not match member {image_id}")

        rois = acqimage["rois"]
        roi_ids = [roi["id"] for roi in rois]
        require_unique(roi_ids, f"ROI IDs for {image_id}")
        roi_id_set = set(roi_ids)
        for roi in rois:
            if roi["type"] == "rectangle":
                start, stop = roi["start"], roi["stop"]
                if not (stop[0] > start[0] and stop[1] > start[1]):
                    raise ConformanceError(
                        f"Rectangle ROI {roi['id']} stop must be greater than start on both axes"
                    )

        analyses_relative = member["resources"].get("analyses")
        if analyses_relative is not None:
            analyses_path = resolve_existing(root, analyses_relative, "analyses.json")
            analyses_document = load_json(analyses_path)
            schema_validate(validator, analyses_document, str(analyses_path.relative_to(root)))
            if analyses_document["format"] != "acqstore-analyses":
                raise ConformanceError(f"Expected analyses document: {analyses_path.relative_to(root)}")
            if analyses_document["image_id"] != image_id:
                raise ConformanceError(f"analyses image_id does not match member {image_id}")
            analyses = analyses_document["analyses"]
            require_unique([analysis["id"] for analysis in analyses], f"analysis IDs for {image_id}")
            for analysis in analyses:
                roi_id = analysis.get("roi_id")
                if roi_id is not None and roi_id not in roi_id_set:
                    raise ConformanceError(
                        f"Analysis {analysis['id']} refers to unknown ROI {roi_id}"
                    )
                resources = analysis["resources"]
                require_unique(
                    [resource["id"] for resource in resources],
                    f"resource IDs for analysis {analysis['id']}",
                )
                for resource in resources:
                    resolve_existing(root, resource["path"], "analysis CSV resource")

        reference_link = member.get("reference_image")
        if reference_link is not None:
            resolve_existing(root, reference_link["ome_zarr"], f"reference OME-Zarr path for {image_id}")
            reference_path = resolve_existing(root, reference_link["metadata"], "reference-image.json")
            reference = load_json(reference_path)
            schema_validate(validator, reference, str(reference_path.relative_to(root)))
            if reference["format"] != "acqstore-reference-image":
                raise ConformanceError(f"Expected reference-image document: {reference_path.relative_to(root)}")
            if reference["image_id"] != image_id:
                raise ConformanceError(f"reference-image image_id does not match member {image_id}")