Skip to content

plaid.viewer.services.paraview_artifact_service

plaid.viewer.services.paraview_artifact_service

Produce ParaView-readable artifacts from PLAID samples.

This module is the one place in PLAID that writes CGNS files on disk. It delegates the actual CGNS export to PLAID (Sample.save_to_dir writes one CGNS per timestep under meshes/), then adds:

  • A .cgns.series sidecar JSON file that ParaView's vtkCGNSReader / vtkCGNSFileSeriesReader understands for multi-timestep samples.
  • A deterministic artifact id derived from a SHA256 cache key so the same inputs always resolve to the same folder.
  • An optional scene.pvsm placeholder for future preset work.
  • A metadata.json describing the artifact.

plaid.viewer.services.paraview_artifact_service.ParaviewArtifactService 

ParaviewArtifactService(
    dataset_service,
    cache_root,
    *,
    export_version=EXPORT_VERSION,
    extra_cache_key_fields=None,
)

Create and look up ParaView-readable artifacts in a cache directory.

Parameters:

  • dataset_service (PlaidDatasetService) –

    Used to load :class:plaid.Sample instances.

  • cache_root (Path) –

    Root of the artifact cache. Usually owned by a :class:plaid.viewer.cache.CacheRoot instance.

  • export_version (str, default: EXPORT_VERSION ) –

    Opaque string included in the cache key. Bump this whenever the export logic changes in a backwards-incompatible way.

  • extra_cache_key_fields (dict[str, str] | None, default: None ) –

    Extra fields to mix into the cache key (for example to invalidate artifacts when a preset template changes).

Source code in plaid/viewer/services/paraview_artifact_service.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def __init__(
    self,
    dataset_service: PlaidDatasetService,
    cache_root: Path,
    *,
    export_version: str = EXPORT_VERSION,
    extra_cache_key_fields: dict[str, str] | None = None,
) -> None:
    self._dataset_service = dataset_service
    self._cache_root = Path(cache_root)
    self._cache_root.mkdir(parents=True, exist_ok=True)
    self._export_version = export_version
    self._extra = dict(extra_cache_key_fields or {})
    self._by_id: dict[str, ParaviewArtifact] = {}
    # Path of the most recently ensured artifact. The cache keeps at most
    # one artifact on disk at any time: once VTK has read the CGNS file
    # into memory (``vtkCGNSReader.Update()`` in the trame pipeline), the
    # on-disk copy is no longer needed, so we delete it as soon as the
    # user asks for another sample.
    self._current_root: Path | None = None

plaid.viewer.services.paraview_artifact_service.ParaviewArtifactService.ensure_artifact 

ensure_artifact(ref, *, force=False)

Return a :class:ParaviewArtifact for ref, creating it if needed.

The cache holds at most one artifact: any previously-ensured artifact whose layout root differs from ref's is removed from disk.

Source code in plaid/viewer/services/paraview_artifact_service.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def ensure_artifact(
    self, ref: SampleRef, *, force: bool = False
) -> ParaviewArtifact:
    """Return a :class:`ParaviewArtifact` for ``ref``, creating it if needed.

    The cache holds at most one artifact: any previously-ensured artifact
    whose layout root differs from ``ref``'s is removed from disk.
    """
    cache_key = _build_cache_key(
        ref, export_version=self._export_version, extra=self._extra
    )
    layout = _artifact_layout(self._cache_root, ref, cache_key)

    # Evict the previous artifact (if any) as soon as the user requests
    # a different one. ``force`` always rebuilds the current one.
    if (
        self._current_root is not None
        and self._current_root != layout.root
        and self._current_root.exists()
    ):
        shutil.rmtree(self._current_root, ignore_errors=True)
        self._by_id.clear()
    if force and layout.root.exists():
        shutil.rmtree(layout.root)

    if layout.metadata_path.is_file() and not force:
        artifact = self._load_existing(layout, cache_key)
    else:
        layout.root.mkdir(parents=True, exist_ok=True)
        artifact = self._create(ref, layout, cache_key)
    self._by_id = {artifact.artifact_id: artifact}
    self._current_root = layout.root
    return artifact

plaid.viewer.services.paraview_artifact_service.ParaviewArtifactService.get 

get(artifact_id)

Return a previously-created artifact by id.

Raises:

  • KeyError –

    If no artifact with this id has been created.

Source code in plaid/viewer/services/paraview_artifact_service.py 
194
195
196
197
198
199
200
201
202
def get(self, artifact_id: str) -> ParaviewArtifact:
    """Return a previously-created artifact by id.

    Raises:
        KeyError: If no artifact with this id has been created.
    """
    if artifact_id not in self._by_id:
        raise KeyError(f"Unknown artifact id: {artifact_id}")
    return self._by_id[artifact_id]

plaid.viewer.services.paraview_artifact_service.ensure_paraview_artifact 

ensure_paraview_artifact(
    sample_ref, *, cache_dir, dataset_service, force=False
)

Functional wrapper around :meth:ParaviewArtifactService.ensure_artifact.

Source code in plaid/viewer/services/paraview_artifact_service.py 
285
286
287
288
289
290
291
292
293
294
def ensure_paraview_artifact(
    sample_ref: SampleRef,
    *,
    cache_dir: Path,
    dataset_service: PlaidDatasetService,
    force: bool = False,
) -> ParaviewArtifact:
    """Functional wrapper around :meth:`ParaviewArtifactService.ensure_artifact`."""
    service = ParaviewArtifactService(dataset_service, cache_dir)
    return service.ensure_artifact(sample_ref, force=force)