liblaf.melon.io ¶

Bases: Protocol

              flowchart TD
              liblaf.melon.io.AbstractConverter[AbstractConverter]

              

              click liblaf.melon.io.AbstractConverter href "" "liblaf.melon.io.AbstractConverter"
            

Callable that converts one object into another mesh representation.

Methods:

• __call__ –

__call__(obj: F, /, **kwargs) -> T

16

def __call__(self, obj: F, /, **kwargs) -> T: ...

Bases: Protocol

              flowchart TD
              liblaf.melon.io.AbstractReader[AbstractReader]

              

              click liblaf.melon.io.AbstractReader href "" "liblaf.melon.io.AbstractReader"
            

Callable that loads an object from a path.

Methods:

• __call__ –

__call__(path: Path, /, **kwargs) -> T

17

def __call__(self, path: Path, /, **kwargs) -> T: ...

Bases: Protocol

              flowchart TD
              liblaf.melon.io.AbstractWriter[AbstractWriter]

              

              click liblaf.melon.io.AbstractWriter href "" "liblaf.melon.io.AbstractWriter"
            

Callable that writes an object to a path.

Methods:

• __call__ –

__call__(obj: T, path: Path, /, **kwargs) -> None

19

def __call__(self, obj: T, path: Path, /, **kwargs) -> None: ...

Single-dispatch conversion registry with identity conversion built in.

The target type is registered as an identity conversion, so callers can pass through objects that are already in the requested representation.

Examples:

>>> converter = ConverterDispatcher(str)
>>> converter("mesh")
'mesh'
>>> @converter.register(int)
... def _from_int(obj, /, **kwargs):
...     return str(obj)
>>> converter(42)
'42'

Parameters:

• to_type (type[T]) –

  Target type returned unchanged when the input already has that runtime type.

• registry (_SingleDispatchCallable[T], default: <dynamic> ) –

  Underlying functools.singledispatch registry.

Methods:

• __call__ –
• register –

Attributes:

• registry (_SingleDispatchCallable[T]) –

  Underlying functools.singledispatch registry.

• to_type (type[T]) –

  Target type returned unchanged when the input already has that runtime type.

registry: _SingleDispatchCallable[T] = attrs.field(
    default=attrs.Factory(
        _default_registry, takes_self=True
    )
)

to_type: type[T]

__call__(obj: Any, /, **kwargs) -> T

60
61

def __call__(self, obj: Any, /, **kwargs) -> T:
    return self.registry(obj, **kwargs)

register[F](
    cls: _RegType, converter: AbstractConverter[F, T]
) -> AbstractConverter[F, T]

register[F](
    cls: _RegType, converter: None = None
) -> Callable[
    [AbstractConverter[F, T]],
    AbstractConverter[F, T],
]

71
72
73
74
75
76
77

def register[F](
    self, cls: _RegType, converter: AbstractConverter[F, T] | None = None
) -> Callable[..., Any]:
    if converter is None:
        return functools.partial(self.register, cls)
    self.registry.register(cls, converter)
    return converter

Dispatch readers by file suffix with an optional fallback reader.

Raises:

• KeyError –

  If the path suffix is unknown and no fallback reader exists.

Attributes:

Parameters:

• to_type (type[T]) –
• fallback (AbstractReader[T] | None, default: None ) –
• registry (dict[str, AbstractReader[T]], default: <class 'dict'> ) –

  dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Methods:

• __call__ –
• register –
• register_fallback –

fallback: AbstractReader[T] | None = None

registry: dict[str, AbstractReader[T]] = attrs.field(
    factory=dict
)

to_type: type[T]

__call__(path: StrPath, /, **kwargs) -> T

37
38
39
40
41
42

def __call__(self, path: StrPath, /, **kwargs) -> T:
    path: Path = Path(path)
    reader: AbstractReader[T] | None = self.registry.get(path.suffix, self.fallback)
    if reader is None:
        raise KeyError(path.suffix)
    return reader(path, **kwargs)

register(
    suffixes: Iterable[str], reader: AbstractReader[T]
) -> AbstractReader[T]

register(
    suffixes: Iterable[str], reader: None = None
) -> Callable[[AbstractReader[T]], AbstractReader[T]]

52
53
54
55
56
57
58
59

def register(
    self, suffixes: Iterable[str], reader: AbstractReader[T] | None = None
) -> Callable[..., Any]:
    if reader is None:
        return functools.partial(self.register, suffixes)
    for s in suffixes:
        self.registry[s] = reader
    return reader

register_fallback(
    reader: AbstractReader[T],
) -> AbstractReader[T]

61
62
63

def register_fallback(self, reader: AbstractReader[T]) -> AbstractReader[T]:
    self.fallback = reader
    return reader

Methods:

• __enter__ –
• __exit__ –
• append –
• dataset –
• dataset_offsets –
• dataset_tail_equal –
• group –

Attributes:

• file (File) –
• n_steps (int) –

file: File

n_steps: int

__enter__() -> Self

62
63
64
65
66
67
68
69

def __enter__(self) -> Self:
    group: h5.Group = self.group("/VTKHDF")
    group.attrs["Version"] = (2, 5)
    group.attrs["Type"] = "UnstructuredGrid"
    steps: h5.Group = self.group("/VTKHDF/Steps")
    if "NSteps" not in steps.attrs:
        steps.attrs["NSteps"] = 0
    return self

__exit__(
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None

71
72
73
74
75
76
77
78
79

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: types.TracebackType | None,
) -> None:
    if self._file is not None:
        self._file.close()
    self._tmp_path.unlink(missing_ok=True)

append(
    mesh: UnstructuredGrid,
    time: SupportsFloat | None = None,
) -> None

240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259

def append(
    self, mesh: pv.UnstructuredGrid, time: SupportsFloat | None = None
) -> None:
    if time is None:
        if "/VTKHDF/Steps/Values" in self.file:
            time: float = self.dataset("/VTKHDF/Steps/Values")[-1] + 1.0
        else:
            time: float = 0.0
    else:
        time: float = float(time)
    mesh._store_metadata()  # noqa: SLF001
    self._append_topology(mesh)
    self._append_points(mesh)
    self._append_point_data(mesh)
    self._append_cell_data(mesh)
    self._append_field_data(mesh)
    self.dataset("/VTKHDF/Steps/Values", time)
    self.n_steps += 1
    self.file.close()
    shutil.copy2(self._tmp_path, self._path)

dataset(
    name: str, data: ArrayLike | None = None
) -> Dataset

106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

def dataset(self, name: str, data: ArrayLike | None = None) -> h5.Dataset:
    if data is None:
        return self.file[name]
    data: np.ndarray = _sanitize_array(data)
    kwargs: dict[str, object] = DEFAULT_DATASET_OPTIONS.copy()
    kwargs["chunks"] = _chunk_shape(data)
    dataset: h5.Dataset = self.file.require_dataset(
        name=name,
        shape=(0, *data.shape[1:]),
        dtype=data.dtype,
        maxshape=(None, *data.shape[1:]),
        **kwargs,
    )
    if data.size > 0:
        dataset.resize(dataset.len() + data.shape[0], axis=0)
        dataset[-data.shape[0] :] = data
    return dataset

dataset_offsets(
    dataset_name: str, offsets_name: str, data: ArrayLike
) -> tuple[Dataset, Dataset]

124
125
126
127
128
129
130
131
132
133

def dataset_offsets(
    self, dataset_name: str, offsets_name: str, data: ArrayLike
) -> tuple[h5.Dataset, h5.Dataset]:
    data: np.ndarray = _sanitize_array(data)
    if not self.dataset_tail_equal(dataset_name, data):
        dataset: h5.Dataset = self.dataset(dataset_name, data)
    else:
        dataset: h5.Dataset = self.dataset(dataset_name)
    offsets: h5.Dataset = self.dataset(offsets_name, dataset.len() - data.shape[0])
    return dataset, offsets

dataset_tail_equal(name: str, data: ArrayLike) -> bool

135
136
137
138
139
140
141
142
143
144
145
146
147

def dataset_tail_equal(self, name: str, data: ArrayLike) -> bool:
    data: np.ndarray = _sanitize_array(data)
    if name not in self.file:
        return False
    dataset: h5.Dataset = self.file[name]
    if dataset.len() == 0 or data.size == 0:
        return False
    tail: np.ndarray = dataset[-data.shape[0] :]
    tail: np.ndarray = _sanitize_array(tail)
    if tail.dtype != data.dtype:
        return False
    numeric: bool = dataset.dtype.kind in {"b", "i", "u", "f", "c"}
    return np.array_equal(tail, data, equal_nan=numeric)

group(name: str) -> Group

103
104

def group(self, name: str) -> h5.Group:
    return self.file.require_group(name)

Dispatch writers first by file suffix and then by object type.

Registered writers receive a pathlib.Path. Parent directories are created before the writer is called, which keeps concrete writer implementations focused on serialization.

Raises:

• KeyError –

  If no writer registry exists for the path suffix.

• NotImplementedError –

  If the suffix exists but the object type is not registered for that suffix.

Parameters:

• registry (dict[str, _SingleDispatchCallable[None]], default: <class 'dict'> ) –

  dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Methods:

• __call__ –
• register –

Attributes:

• registry (dict[str, _SingleDispatchCallable[None]]) –

registry: dict[str, _SingleDispatchCallable[None]] = (
    attrs.field(factory=dict)
)

__call__(obj: T, path: StrPath, /, **kwargs) -> None

42
43
44
45
46

def __call__(self, obj: T, path: StrPath, /, **kwargs) -> None:
    path: Path = Path(path)
    writer: _SingleDispatchCallable[None] = self.registry[path.suffix]
    path.parent.mkdir(parents=True, exist_ok=True)
    return writer(obj, path, **kwargs)

register(
    cls: _RegType,
    suffixes: Iterable[str],
    writer: AbstractWriter[T],
) -> AbstractWriter[T]

register(
    cls: _RegType,
    suffixes: Iterable[str],
    writer: None = None,
) -> Callable[[AbstractWriter[T]], AbstractWriter[T]]

56
57
58
59
60
61
62
63
64
65
66
67
68

def register(
    self,
    cls: _RegType,
    suffixes: Iterable[str],
    writer: AbstractWriter[T] | None = None,
) -> Callable[..., Any]:
    if writer is None:
        return functools.partial(self.register, cls, suffixes)
    for suffix in suffixes:
        if suffix not in self.registry:
            self.registry[suffix] = functools.singledispatch(_default_writer)
        self.registry[suffix].register(cls, writer)
    return writer