Skip to content
Snippets Groups Projects
Commit f08d0d08 authored by Oriol Tintó's avatar Oriol Tintó
Browse files

Refactoring encodings and renaming main classes for clarity.

parent c815193f
No related branches found
No related tags found
1 merge request!6Get rid of Enums and use a dictionary to define available lossy compressors,...
Hide whitespace changes
Inline Side-by-side
enstools/encoding/compressors/__init__.py deleted 100644 → 0
+ 0
− 0
View file @ c815193f
enstools/encoding/dataset_encoding.py 0 → 100644
+ 155
− 0
View file @ f08d0d08
import os
from copy import deepcopy
from pathlib import Path
from typing import Union, Dict
import xarray
import yaml
from . import rules
from .errors import InvalidCompressionSpecification
from .variable_encoding import _Mapping, parse_variable_specification, Encoding, \
NullEncoding
def compression_dictionary_to_string(compression_dictionary: dict) -> str:
"""
Convert a dictionary containing multiple entries to a single line specification
"""
return rules.VARIABLE_SEPARATOR.join(
[f"{key}{rules.VARIABLE_NAME_SEPARATOR}{value}" for key, value in compression_dictionary.items()])
def parse_full_specification(spec: str) -> Dict[str, Encoding]:
from enstools.encoding.rules import VARIABLE_SEPARATOR, VARIABLE_NAME_SEPARATOR, \
DATA_DEFAULT_LABEL, DATA_DEFAULT_VALUE, COORD_LABEL, COORD_DEFAULT_VALUE
result = {}
if spec is None:
spec = "None"
parts = spec.split(VARIABLE_SEPARATOR)
for p in parts:
# For each part, check if there's a variable name.
# If there's a variable name, split the name and the specification
if VARIABLE_NAME_SEPARATOR in p:
var_name, var_spec = p.split(VARIABLE_NAME_SEPARATOR)
# Otherwise, it corresponds to the default.
else:
var_name = DATA_DEFAULT_LABEL
var_spec = p
# If the variable name was already in the dictionary, raise an error.
if var_name in result:
raise InvalidCompressionSpecification(f"Variable {var_name} has multiple definitions."
f"")
# Parse the variable specification
result[var_name] = parse_variable_specification(var_spec)
# In case values for default and coordinates haven't been provided, use the default values.
if DATA_DEFAULT_LABEL not in result:
result[DATA_DEFAULT_LABEL] = parse_variable_specification(DATA_DEFAULT_VALUE)
if COORD_LABEL not in result:
# If the default is a NullSpecification, we'll use the same for the coordinates
if isinstance(result[DATA_DEFAULT_LABEL], NullEncoding):
result[COORD_LABEL] = result[DATA_DEFAULT_LABEL]
else:
result[COORD_LABEL] = parse_variable_specification(COORD_DEFAULT_VALUE)
# For each specification, check that the specifications are valid.
for key, spec in result.items():
spec.check_validity()
return result
class DatasetEncoding(_Mapping):
"""
Class to encapsulate compression specification parameters corresponding to a full dataset.
The kind of encoding that xarray expects is a mapping between the variables and their corresponding h5py encoding.
"""
def __init__(self, dataset: xarray.Dataset, compression: Union[str, dict, None]):
self.dataset = dataset
# Process the compression argument to get a single string with per-variable specifications
compression = self.get_a_single_compression_string(compression)
# Save it
self.variable_encodings = parse_full_specification(compression)
@staticmethod
def get_a_single_compression_string(compression: Union[str, dict, Path]) -> str:
# The compression parameter can be a string or a dictionary.
# In case it is a string, it can be directly a compression specification or a yaml file.
# If it is a file, convert it to a Path
if isinstance(compression, str) and os.path.exists(compression):
compression = Path(compression)
if isinstance(compression, dict):
# Just to make sure that we have all the mandatory fields (default, coordinates), we will convert
# the input dictionary to a single specification string and convert it back.
return compression_dictionary_to_string(compression)
elif isinstance(compression, Path):
with compression.open("r") as stream:
dict_of_strings = yaml.safe_load(stream)
return compression_dictionary_to_string(dict_of_strings)
elif isinstance(compression, str):
# Convert the single string in a dictionary with an entry for each specified variable plus the defaults
# for data and coordinates
return compression
elif compression is None:
return None
else:
raise InvalidCompressionSpecification(
f"The argument 'compression' should be a string, a dictionary or a Path. It is {type(compression)!r}-")
def encoding(self):
# Get the defaults
data_default = self.variable_encodings[rules.DATA_DEFAULT_LABEL]
coordinates_default = self.variable_encodings[rules.COORD_LABEL]
# Set encoding for coordinates
coordinate_encodings = {coord: deepcopy(coordinates_default) for coord in self.dataset.coords}
# Set encoding for data variables
data_variable_encodings = {
str(var): deepcopy(self.variable_encodings[str(var)]) if var in self.variable_encodings else deepcopy(
data_default) for
var
in self.dataset.data_vars}
# Add chunking?
for variable in self.dataset.data_vars:
chunks = {k: v if k != "time" else 1 for k, v in self.dataset[variable].sizes.items()}
chunk_sizes = tuple(chunks.values())
# Ugly python magic to add chunk sizes into the encoding mapping object.
data_variable_encodings[variable]._kwargs._kwargs["chunksizes"] = chunk_sizes # noqa
# Merge
all_encodings = {**coordinate_encodings, **data_variable_encodings}
return all_encodings
@property
def _kwargs(self):
return self.encoding()
def add_metadata(self):
"""
Add the corresponding compression metadata to the dataset.
"""
for variable, encoding in self._kwargs.items():
# If its a case without compression, don't write the metadata.
if isinstance(encoding, NullEncoding):
continue
self.dataset[variable].attrs["compression"] = encoding.description()
def is_a_valid_dataset_compression_specification(specification):
try:
_ = parse_full_specification(specification)
return True
except InvalidCompressionSpecification:
return False
enstools/encoding/encodings.py deleted 100644 → 0
+ 0
− 349
View file @ c815193f
import os.path
from typing import Union, Any, Mapping
import xarray
import yaml
from . import rules
from hdf5plugin import SZ, Zfp, Blosc
try:
from hdf5plugin import SZ3
except ImportError:
pass
from .compressors.no_compressor import NoCompression
from .definitions import Compressors, CompressionModes
from .errors import WrongCompressionSpecificationError, WrongCompressionModeError, WrongCompressorError, \
WrongParameterError
from copy import deepcopy
from pathlib import Path
import logging
# Change hdf5plugin logging levels to Warning
loggers = {name: logging.getLogger(name) for name in logging.root.manager.loggerDict}
hdf5plugin_loggers = [key for key in loggers.keys() if key.count("hdf5plugin")]
for key in hdf5plugin_loggers:
loggers[key].setLevel(logging.WARNING)
class _Mapping(Mapping):
"""
Subclass to implement dunder methods that are mandatory for Mapping to avoid repeating the code everywhere.
"""
_kwargs: Mapping
def __getitem__(self, item):
return self._kwargs[item]
def __setitem__(self, key, value):
self._kwargs[key] = value
def __iter__(self):
return iter(self._kwargs)
def __len__(self):
return len(self._kwargs)
class FilterEncodingForH5py(_Mapping):
"""
Class to encapsulate compression specification parameters for a single variable.
It stores the compressor, the mode and the parameter.
It has a method to create a new instance from a specification string,
a method to get the corresponding specification string from an existing object
and a method to obtain the corresponding mapping expected by h5py.
"""
def __init__(self,
compressor: Union[Compressors, str],
mode: Union[CompressionModes, str, None],
parameter: Union[float, int, None],
):
# Init basic components
self.set_compressor(compressor)
self.set_mode(mode)
self.set_parameter(parameter)
self._kwargs = self.filter_mapping()
def set_compressor(self, compressor: Union[Compressors, str]):
if isinstance(compressor, Compressors):
self.compressor = compressor
elif isinstance(compressor, str):
self.compressor = Compressors[compressor.upper()]
else:
raise NotImplementedError
def set_mode(self, mode: Union[CompressionModes, str, None]):
if isinstance(mode, CompressionModes):
self.mode = mode
elif isinstance(mode, str):
self.mode = CompressionModes[mode.upper()]
elif mode is None:
self.mode = None
else:
raise NotImplementedError
def set_parameter(self, parameter: [float, int, None]):
# Should I add further checks here?
self.parameter = parameter
@staticmethod
def from_string(string: str) -> Any:
"""
Method to create a specification object from a specification string
"""
return compression_string_to_object(string)
def to_string(self) -> str:
"""
Method to obtain a specification string from a specification object
"""
return compression_object_to_string(self)
def filter_mapping(self):
"""
Method to get the corresponding FilterRefBase expected by h5py/xarray
"""
if self.compressor is Compressors.BLOSC:
return Blosc(clevel=9)
elif self.compressor is Compressors.ZFP:
mode = str(self.mode).lower().split('.')[-1]
options = {mode: self.parameter}
return Zfp(**options)
elif self.compressor is Compressors.SZ:
mode = str(self.mode).lower().split('.')[-1]
# In the hdf5plugin implementation of SZ we ended up using more self-explanatory names so we need to
# map the names here
sz_mode_map = {
"abs": "absolute",
"rel": "relative",
"pw_rel": "pointwise_relative"
}
options = {sz_mode_map[mode]: self.parameter}
return SZ(**options)
elif self.compressor is Compressors.SZ3:
mode = str(self.mode).lower().split('.')[-1]
# In the hdf5plugin implementation of SZ3 were also using more self-explanatory names
# We need to map the names as well.
sz3_mode_map = {
"abs": "absolute",
"rel": "relative",
"psnr": "peak_signal_to_noise_ratio",
"norm2": "norm2",
}
options = {sz3_mode_map[mode]: self.parameter}
return SZ3(**options)
elif self.compressor is Compressors.NONE:
return NoCompression()
else:
raise NotImplementedError
def description(self):
return f"Compressed with HDF5 filter: {self.to_string()} " \
f"(Using {self.compressor} " \
f"filter with id: {self.compressor.value})"
class FilterEncodingForXarray(_Mapping):
"""
Class to encapsulate compression specification parameters corresponding to a full dataset.
The kind of encoding that xarray expects is a mapping between the variables and their corresponding h5py encoding.
"""
def __init__(self, dataset: xarray.Dataset, compression: Union[str, dict, None]):
self.dataset = dataset
# Process the compression argument to get a dictionary with a specification for each variable
dict_of_strings = self.get_dictionary_of_specifications(compression)
# Convert each string specification into a CompressionDataArraySpecification
variable_encodings = {key: FilterEncodingForH5py.from_string(value) for key, value in
dict_of_strings.items()}
# Save it
self.variable_encodings = variable_encodings
@staticmethod
def get_dictionary_of_specifications(compression: Union[str, dict, Path]):
# The compression parameter can be a string or a dictionary.
# In case it is a string, it can be directly a compression specification or a yaml file.
# If it is a file, convert it to a Path
if isinstance(compression, str) and os.path.exists(compression):
compression = Path(compression)
if isinstance(compression, dict):
# Just to make sure that we have all the mandatory fields (default, coordinates), we will convert
# the input dictionary to a single specification string and convert it back.
dict_of_strings = compression_string_to_dictionary(compression_dictionary_to_string(compression))
elif isinstance(compression, Path):
if compression.exists():
with compression.open("r") as stream:
dict_of_strings = yaml.safe_load(stream)
dict_of_strings = compression_string_to_dictionary(compression_dictionary_to_string(dict_of_strings))
elif isinstance(compression, str):
# Convert the single string in a dictionary with an entry for each specified variable plus the defaults
# for data and coordinates
dict_of_strings = compression_string_to_dictionary(compression)
elif compression is None:
dict_of_strings = {rules.DATA_DEFAULT_LABEL: None, rules.COORD_LABEL: None}
else:
raise TypeError(
f"The argument 'compression' should be a string or a dictionary. It is {type(compression)!r}-")
return dict_of_strings
@property
def _kwargs(self):
return self.encoding()
def encoding(self):
# Get the defaults
data_default = self.variable_encodings[rules.DATA_DEFAULT_LABEL]
coordinates_default = self.variable_encodings[rules.COORD_LABEL]
# Set encoding for coordinates
coordinate_encodings = {coord: deepcopy(coordinates_default) for coord in self.dataset.coords}
# Set encoding for data variables
data_variable_encodings = {
str(var): deepcopy(self.variable_encodings[str(var)]) if var in self.variable_encodings else deepcopy(data_default) for
var
in self.dataset.data_vars}
# Add chunking?
for variable in self.dataset.data_vars:
chunks = {k: v if k != "time" else 1 for k, v in self.dataset[variable].sizes.items()}
chunk_sizes = tuple(chunks.values())
# Ugly python magic to add chunk sizes into the encoding mapping object.
data_variable_encodings[variable]._kwargs._kwargs["chunksizes"] = chunk_sizes # noqa
# Merge
all_encodings = {**coordinate_encodings, **data_variable_encodings}
return all_encodings
def add_metadata(self):
"""
Add the corresponding compression metadata to the dataset.
"""
for variable, encoding in self.encoding().items():
if encoding.compressor != Compressors.NONE:
self.dataset[variable].attrs["compression"] = encoding.description()
def compression_string_to_object(compression: str) -> FilterEncodingForH5py:
"""
Receive a CFS string and return a CompressionDataArraySpecification
"""
if compression in [None, "None", "none"]:
return FilterEncodingForH5py(compressor=Compressors.NONE, mode=CompressionModes.NONE, parameter=None)
arguments = compression.split(rules.COMPRESSION_SPECIFICATION_SEPARATOR)
mode = None
parameter = 0.0
if len(arguments) == 4:
keyword, compressor, mode, parameter = arguments
# In case the keyword doesn't correspond to lossy most probably the user has used the wrong separator
if keyword != "lossy":
raise WrongCompressionSpecificationError(
f"Wrong separator {keyword!r}, "
f"please use {rules.VARIABLE_NAME_SEPARATOR!r}"
)
try:
compressor = Compressors[compressor.upper()]
except KeyError:
raise WrongCompressorError(f"Lossy compressor {compressor} is not valid")
try:
mode = CompressionModes[mode.upper()]
except KeyError:
raise WrongCompressionModeError(f"Compression mode {mode} is not valid")
try:
parameter = float(parameter)
except ValueError:
raise WrongParameterError
elif arguments[0] == "lossless":
compressor = Compressors.BLOSC
elif arguments[0].lower() == "none":
compressor = Compressors.NONE
else:
raise WrongCompressionSpecificationError(
f"Problem parsing compression specification: {compression!r}")
return FilterEncodingForH5py(compressor, mode, parameter)
def compression_object_to_string(compression: FilterEncodingForH5py) -> str:
"""
From a CompressionDataArraySpecification to a CFS compliant string
:param compression:
:return:
"""
# Separator
s = rules.COMPRESSION_SPECIFICATION_SEPARATOR
if compression.compressor is Compressors.BLOSC:
# For now we are ignoring BLOSC backends
return f"lossless"
else:
# Convert the compressor
compressor = str(compression.compressor).lower().split('.')[-1]
mode = str(compression.mode).lower().split('.')[-1]
parameter = str(compression.parameter)
return f"lossy{s}{compressor}{s}{mode}{s}{parameter}"
def compression_dictionary_to_string(compression_dictionary: dict) -> str:
"""
Convert a dictionary containing multiple entries to a single line specification
"""
return rules.VARIABLE_SEPARATOR.join(
[f"{key}{rules.VARIABLE_NAME_SEPARATOR}{value}" for key, value in compression_dictionary.items()])
def compression_string_to_dictionary(compression_string: str) -> dict:
"""
This function splits a single string containing multiple compression specifications into a dictionary.
This dictionary will contain a default entry, a coordinates entry and an additional entry for each variable
explicitly mentioned.
----------
compression_specification: str
Returns
-------
compression_specification: dict
"""
compression_dictionary = {}
cases = compression_string.split(rules.VARIABLE_SEPARATOR)
for case in cases:
if case.strip():
if rules.VARIABLE_NAME_SEPARATOR in case:
variable, spec = case.split(rules.VARIABLE_NAME_SEPARATOR)
else:
variable = rules.DATA_DEFAULT_LABEL
spec = case
compression_dictionary[variable] = spec
# Default for data variables if not explicitly specified
if rules.DATA_DEFAULT_LABEL not in compression_dictionary:
compression_dictionary[rules.DATA_DEFAULT_LABEL] = rules.DATA_DEFAULT_VALUE
# Default for coordinates will be set if it was not explicitly specified in the string
if rules.COORD_LABEL not in compression_dictionary:
compression_dictionary[rules.COORD_LABEL] = rules.COORD_DEFAULT_VALUE
return compression_dictionary
enstools/encoding/variable_encoding.py 0 → 100644
+ 325
− 0
View file @ f08d0d08
from abc import ABC
from .definitions import lossy_compressors_and_modes
import logging
from typing import Mapping, Union, Protocol
import hdf5plugin
from enstools.encoding import rules, definitions
from enstools.encoding.errors import InvalidCompressionSpecification
from .rules import LOSSLESS_DEFAULT_BACKEND, LOSSLESS_DEFAULT_COMPRESSION_LEVEL
# Change logging level for the hdf5plugin to avoid unnecessary warnings
loggers = {name: logging.getLogger(name) for name in logging.root.manager.loggerDict}
hdf5plugin_loggers = [key for key in loggers.keys() if key.count("hdf5plugin")]
for key in hdf5plugin_loggers:
loggers[key].setLevel(logging.WARNING)
class _Mapping(Mapping):
"""
Subclass to implement dunder methods that are mandatory for Mapping to avoid repeating the code everywhere.
"""
_kwargs: Mapping
def __getitem__(self, item):
return self._kwargs[item]
def __setitem__(self, key, value):
self._kwargs[key] = value
def __iter__(self):
return iter(self._kwargs)
def __len__(self):
return len(self._kwargs)
class Encoding(ABC):
def check_validity(self) -> bool:
...
def to_string(self) -> str:
...
def encoding(self) -> Mapping:
...
def description(self) -> str:
...
def __repr__(self):
return f"{self.__class__.__name__}({self.to_string()})"
class VariableEncoding(_Mapping):
"""
Factory class to get the proper encoding depending on the arguments provided.
To get an encoding from a specification:
>>> VariableEncoding("lossy,zfp,rate,3.2")
or
>>> VariableEncoding(specification="lossy,zfp,rate,3.2")
To get a lossy compression encoding specifing things separatelly:
>>> VariableEncoding(compressor="zfp", mode="rate", parameter="3.2")
To get a lossless compression encoding:
>>> VariableEncoding("lossless")
Or it is possible to specify the backend and the compression level.
>>> VariableEncoding(backend="snappy")
>>> VariableEncoding(backend="snappy", compression_level=9)
"""
def __new__(cls,
specification: str = None,
compressor: str = None,
mode: str = None,
parameter: Union[str, float, int] = None,
backend: str = None,
compression_level: int = None,
) -> Encoding:
return get_variable_encoding(specification=specification,
compressor=compressor,
mode=mode,
parameter=parameter,
backend=backend,
compression_level=compression_level,
)
class NullEncoding(Encoding):
def check_validity(self) -> bool:
return True
def to_string(self) -> str:
return "None"
def encoding(self) -> Mapping:
return {}
def description(self) -> str:
return ""
class LosslessEncoding(Encoding):
def __init__(self, backend: str, compression_level: int):
self.backend = backend if backend is not None else rules.LOSSLESS_DEFAULT_BACKEND
self.compression_level = compression_level if compression_level is not None \
else rules.LOSSLESS_DEFAULT_COMPRESSION_LEVEL
self.check_validity()
self._kwargs = self.encoding()
def check_validity(self) -> bool:
if self.backend not in definitions.lossless_backends:
raise InvalidCompressionSpecification(f"Backend {self.backend!r} is not a valid backend.")
elif not (1 <= self.compression_level <= 9):
raise InvalidCompressionSpecification(f"Compression level {self.compression_level} must be within 1 and 9.")
else:
return True
def to_string(self) -> str:
return rules.COMPRESSION_SPECIFICATION_SEPARATOR.join(["lossless", self.backend, str(self.compression_level)])
def encoding(self) -> Mapping:
return hdf5plugin.Blosc(cname=self.backend, clevel=self.compression_level)
def description(self) -> str:
return f"Losslessly compressed with the HDF5 Blosc filter: {self.to_string()} " \
f"(Using {self.backend!r} with compression level {self.compression_level})"
def __repr__(self):
return f"{self.__class__.__name__}(backend={self.backend}, compression_level={self.compression_level})"
class LossyEncoding(Encoding):
def __init__(self, compressor: str, mode: str, parameter: Union[float, int]):
self.compressor = compressor
self.mode = mode
self.parameter = parameter
self.check_validity()
self._kwargs = self.encoding()
def check_validity(self):
# Check compressor validity
if self.compressor not in definitions.lossy_compressors_and_modes:
raise InvalidCompressionSpecification(f"Invalid compressor {self.compressor}")
# Check compression mode validity
if self.mode not in definitions.lossy_compressors_and_modes[self.compressor]:
raise InvalidCompressionSpecification(f"Invalid mode {self.mode!r} for compressor {self.compressor!r}")
# Get parameter range and type
mode_range = definitions.lossy_compressors_and_modes[self.compressor][self.mode]["range"]
mode_type = definitions.lossy_compressors_and_modes[self.compressor][self.mode]["type"]
# Check type
if not isinstance(self.parameter, mode_type):
try:
self.parameter = mode_type(self.parameter)
except TypeError:
raise InvalidCompressionSpecification(f"Invalid parameter type {self.parameter!r}")
# Check range
if self.parameter <= mode_range[0] or self.parameter >= mode_range[1]:
raise InvalidCompressionSpecification(f"Parameter out of range {self.parameter!r}")
return True
def to_string(self) -> str:
return rules.COMPRESSION_SPECIFICATION_SEPARATOR.join(
["lossy", self.compressor, self.mode, str(self.parameter)])
def encoding(self) -> Mapping:
mode = definitions.sz_mode_map[self.mode] if self.mode in definitions.sz_mode_map else self.mode
arguments = {mode: self.parameter}
return definitions.compressor_map[self.compressor](**arguments)
def description(self) -> str:
return f"Lossy compressed using the HDF5 filters with specification: {self.to_string()} " \
f"(Using {self.compressor!r} with mode {self.mode!r} and parameter {self.parameter})"
def __repr__(self):
return f"{self.__class__.__name__}(compressor={self.compressor}, mode={self.mode}, parameter={self.parameter})"
def parse_variable_specification(var_spec: str) -> Encoding:
"""
Parse a string following the compression specification format for a single variable
and return a Specification object.
Parameters
----------
var_spec
Returns
-------
"""
if var_spec in (None, "None", "none"):
return NullEncoding()
from enstools.encoding.rules import COMPRESSION_SPECIFICATION_SEPARATOR
# Split the specification in the different parts.
var_spec_parts = var_spec.split(COMPRESSION_SPECIFICATION_SEPARATOR)
# Treatment for lossless
if var_spec_parts[0] == "lossless":
backend = var_spec_parts[1] if len(var_spec_parts) > 1 else None
compression_level = int(var_spec_parts[2]) if len(var_spec_parts) > 2 else None
return LosslessEncoding(backend, compression_level)
# Treatment for lossy
elif var_spec_parts[0] == "lossy":
# Lossy specifications must have 4 elements (lossy,compressor,mode,parameter)
if len(var_spec_parts) != 4:
raise InvalidCompressionSpecification(f"Invalid specification {var_spec!r}")
# Get the different components.
compressor, mode, specification = var_spec_parts[1:]
# Check that the compressor is a valid option.
if compressor not in lossy_compressors_and_modes:
raise InvalidCompressionSpecification(f"Invalid compressor {compressor!r} in {var_spec!r}")
# Check that the mode is a valid option.
if mode not in lossy_compressors_and_modes[compressor]:
raise InvalidCompressionSpecification(
f"Invalid mode {mode!r} for compressor {compressor!r} in {var_spec!r}")
# Cast the specification to the proper type.
specification_type = lossy_compressors_and_modes[compressor][mode]["type"]
try:
specification = specification_type(specification)
except ValueError:
raise InvalidCompressionSpecification(f"Could not cast {specification!r} to type {specification_type!r}")
return LossyEncoding(compressor, mode, specification)
else:
# In case its not lossy nor lossless, raise an exception.
raise InvalidCompressionSpecification(f"Invalid specification {var_spec!r}")
# class VariableEncoding(_Mapping):
# """
# Class to encapsulate compression specification parameters for a single variable.
#
# It stores the compressor, the mode and the parameter.
#
# It has a method to create a new instance from a specification string,
# a method to get the corresponding specification string from an existing object
# and a method to obtain the corresponding mapping expected by h5py.
#
# """
#
# def __init__(self, specification: Specification):
# # Init basic components
# self.specification = specification
#
# self._kwargs = self.filter_mapping()
#
# @staticmethod
# def from_string(string: str) -> 'VariableEncoding':
# specification = parse_variable_specification(string)
# """
# Method to create a specification object from a specification string
# """
# return VariableEncoding(specification)
#
# def to_string(self) -> str:
# """
# Method to obtain a specification string from a specification object
# """
# return self.specification.to_string()
#
# def filter_mapping(self) -> Mapping:
# """
# Method to get the corresponding FilterRefBase expected by h5py/xarray
# """
#
# return self.specification.encoding()
#
# def description(self):
# self.specification.description()
def get_variable_encoding(
specification: str = None,
compressor: str = None,
mode: str = None,
parameter: Union[str, float, int] = None,
backend: str = None,
compression_level: int = None,
) -> Encoding:
"""
Wildcard entry point for all ways of specifying an encoding:
- Using a string specification
- For lossy compression, specifying the compressor, the mode and the parameter
- For lossless compression, specifying the backend and the compression level
"""
# Two special cases:
if (specification is None) and (compressor is None) and (backend is None):
return LosslessEncoding(backend=LOSSLESS_DEFAULT_BACKEND, compression_level=LOSSLESS_DEFAULT_COMPRESSION_LEVEL)
assert (specification is None) + (compressor is None) + (backend is None) == 2, \
"Only one of the options can be used to create an Encoding"
if specification:
return parse_variable_specification(specification)
elif compressor:
return LossyEncoding(compressor=compressor, mode=mode, parameter=parameter)
elif backend:
if compression_level is None:
compression_level = LOSSLESS_DEFAULT_COMPRESSION_LEVEL
return LosslessEncoding(backend=backend, compression_level=compression_level)
def is_valid_variable_compression_specification(specification):
try:
_ = parse_variable_specification(specification)
return True
except InvalidCompressionSpecification:
return False
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment