Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions src/damast/cli/data_converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,14 +42,14 @@ def __init__(self, parser: ArgumentParser):
required=False,
)
parser.add_argument("--output-type",
help="The output file type: .parquet (default) or .hdf5 ",
help="The output file type: .parquet or .hdf5 (default: %(default)s)",
default=".parquet",
required=False,
)
parser.add_argument("--validation-mode",
default="readonly",
default="update_data",
choices=[x.value.lower() for x in ValidationMode],
help="Define the validation mode")
help="Define the validation mode (default: %(default)s)")

def validate(self, adf: AnnotatedDataFrame, args):
"""
Expand Down
47 changes: 46 additions & 1 deletion src/damast/cli/data_inspect.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,50 @@ def __init__(self, parser: ArgumentParser):
choices=[x.value.lower() for x in ValidationMode],
help="Define the validation mode")

def fill_missing_value_stats(self, adf: AnnotatedDataFrame) -> dict[str, set[str]]:
"""
Compute a column's 'value_range'/'value_stats' on the fly wherever the loaded
metadata left them unset, without touching either field where it is already set.

If `adf.metadata_inferred` is True (no metadata spec file was found for the input,
so its metadata was inferred from the data itself - see
`damast.core.dataframe.AnnotatedDataFrame.metadata_inferred`), every populated
'value_range'/'value_stats' is reported as generated too, not just the ones this
call fills in, since none of it came from an actual spec file either way.

Args:
adf: The dataframe whose metadata columns to fill in, updated in place

Returns:
Column name to the set of field names ('value_range' and/or 'value_stats') that
are not backed by an actual metadata spec - for
`damast.core.metadata.MetaData.to_str`'s ``generated_fields``
"""
generated_fields: dict[str, set[str]] = {}
for column_spec in adf.metadata.columns:
original_range = column_spec.value_range
original_stats = column_spec.value_stats

if original_range is None or original_stats is None:
column_spec.update_datarange_and_stats(adf.lazyframe, column_spec.name)

if not (original_range is None and column_spec.value_range is not None):
column_spec.value_range = original_range
if not (original_stats is None and column_spec.value_stats is not None):
column_spec.value_stats = original_stats

computed = set()
for field_name, original_value in (("value_range", original_range), ("value_stats", original_stats)):
if getattr(column_spec, field_name) is None:
continue
if original_value is None or adf.metadata_inferred:
computed.add(field_name)

if computed:
generated_fields[column_spec.name] = computed

return generated_fields

def expand_filter_arg(self, adf: AnnotatedDataFrame, arg: str):
if arg in adf.column_names:
return f"pl.col('{arg}')"
Expand Down Expand Up @@ -128,7 +172,8 @@ def execute(self, args):
# Refresh value_range/value_stats in place to match the filtered rows,
adf.validate_metadata(validation_mode=ValidationMode.UPDATE_METADATA)

print(adf.metadata.to_str(columns=args.columns))
generated_fields = self.fill_missing_value_stats(adf)
print(adf.metadata.to_str(columns=args.columns, generated_fields=generated_fields))
print(f"\n\nFirst {args.head} and last {args.tail} rows:")
df = adf.lazyframe

Expand Down
19 changes: 16 additions & 3 deletions src/damast/core/dataframe.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ class AnnotatedDataFrame(XDataFrame):
valid range (specified in :attr:`metadata`) with missing value.
- If :attr:`damast.core.ValidationMode.UPDATE_METADATA` update the metadata according to the encountered values.
- Else :attr:`damast.core.ValidationMode.READONLY` will throw when encountering inconsistencies
:param metadata_inferred: Whether `metadata` was inferred from the data (e.g. by
:func:`infer_annotation`, when no spec file was found) rather than loaded from an
actual metadata spec - see :attr:`metadata_inferred`
"""

#: Metadata associated with the dataframe
Expand All @@ -54,6 +57,7 @@ def __init__(
dataframe: polars.DataFrame | polars.LazyFrame | XDataFrame,
metadata: MetaData,
validation_mode: ValidationMode = ValidationMode.READONLY,
metadata_inferred: bool = False,
):
if isinstance(dataframe, XDataFrame):
dataframe = dataframe.lazyframe
Expand All @@ -76,6 +80,7 @@ def __init__(
super().__init__(df=dataframe)

self._metadata = metadata
self._metadata_inferred = metadata_inferred

# Ensure conformity of the metadata with the dataframe
self.validate_metadata(validation_mode=validation_mode)
Expand All @@ -85,6 +90,11 @@ def metadata(self) -> MetaData:
"""Get the metadata for this dataframe"""
return self._metadata

@property
def metadata_inferred(self) -> bool:
"""Whether `metadata` was inferred from the data rather than loaded from a spec file."""
return self._metadata_inferred

@classmethod
def ensure_type(cls, obj: any):
if not isinstance(obj, cls):
Expand Down Expand Up @@ -289,14 +299,16 @@ def from_files(cls,
_log.info("Metadata is not available and not required, so inferring annotation")
metadata = cls.infer_annotation(df)
_log.info("Metadata inferring completed")
return cls(dataframe=df, metadata=metadata, validation_mode=ValidationMode.IGNORE)
return cls(dataframe=df, metadata=metadata, validation_mode=ValidationMode.IGNORE,
metadata_inferred=True)
else:
_log.info(f"Found metadata: {[str(x) for x in metadata_file_candidates]}")
elif len(metadata) != len(files):
_log.info("Metadata is not available for all files, so inferring annotation")
metadata = cls.infer_annotation(df)
_log.info("Metadata inferring completed")
return cls(dataframe=df, metadata=metadata, validation_mode=ValidationMode.IGNORE)
return cls(dataframe=df, metadata=metadata, validation_mode=ValidationMode.IGNORE,
metadata_inferred=True)
elif len(metadata) == len(files):
metadata_list = list(metadata.values())
metadata = MetaData(
Expand Down Expand Up @@ -473,4 +485,5 @@ def shape(self):

def __deepcopy__(self, memo=None):
# ignore validation, also erroneous frames should be copyable
return AnnotatedDataFrame(self.lazyframe.clone(), copy.deepcopy(self._metadata), validation_mode=ValidationMode.IGNORE)
return AnnotatedDataFrame(self.lazyframe.clone(), copy.deepcopy(self._metadata), validation_mode=ValidationMode.IGNORE,
metadata_inferred=self._metadata_inferred)
107 changes: 75 additions & 32 deletions src/damast/core/metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -801,6 +801,13 @@ def merge(self, other: DataSpecification, strategy: MergeStrategy | None = None)

ds = DataSpecification(name=self.name)

# value_range/value_stats describe the data actually observed for this column. If
# either side never computed them (e.g. a spec written without stats), the merged
# value must not silently fall back to just the other side's - that range was never
# checked against the side with no recorded stats and can understate the true
# combined range once both sides' data is considered together.
data_dependent_keys = (self.Key.value_range, self.Key.value_stats)

for key in self.Key:
if key == self.Key.name:
continue
Expand All @@ -810,42 +817,51 @@ def merge(self, other: DataSpecification, strategy: MergeStrategy | None = None)

if this_value == other_value:
setattr(ds, key.value, this_value)
elif this_value is None:
continue

if key in data_dependent_keys and (this_value is None or other_value is None):
setattr(ds, key.value, None)
continue

if this_value is None:
setattr(ds, key.value, other_value)
elif other_value is None:
continue
if other_value is None:
setattr(ds, key.value, this_value)
else:
if key == self.Key.representation_type:
if hasattr(this_value, "to_python"):
this_value = this_value.to_python()
if hasattr(other_value, "to_python"):
other_value = other_value.to_python()
continue

try:
if this_value == other_value:
setattr(ds, key.value, this_value)
return ds
elif hasattr(this_value, "merge"):
merged_value = this_value.merge(other_value)
setattr(ds, key.value, merged_value)
return ds
except Exception as e:
logger.warning(f"Merge failed: {e}")
if key == self.Key.representation_type:
if hasattr(this_value, "to_python"):
this_value = this_value.to_python()
if hasattr(other_value, "to_python"):
other_value = other_value.to_python()

if not strategy:
raise ValueError(
f"{self.__class__.__name__}.merge cannot merge specs for '{self.name}': value for '{key.value}' differs: "
try:
if this_value == other_value:
setattr(ds, key.value, this_value)
continue
elif hasattr(this_value, "merge"):
merged_value = this_value.merge(other_value)
setattr(ds, key.value, merged_value)
continue
except Exception as e:
logger.warning(f"Merge failed: {e}")

f" on self: '{this_value}' vs. other: '{other_value}'"
)
if not strategy:
raise ValueError(
f"{self.__class__.__name__}.merge cannot merge specs for '{self.name}': value for '{key.value}' differs: "

f" on self: '{this_value}' vs. other: '{other_value}'"
)

logger.info(f"{self.__class__.__name__}.merge: using merge strategy {strategy} for {key.value}: this={this_value} -- other={other_value}")
if strategy == DataSpecification.MergeStrategy.OTHER:
setattr(ds, key.value, other_value)
elif strategy == DataSpecification.MergeStrategy.THIS:
setattr(ds, key.value, this_value)
else:
raise RuntimeError(f"{self.__class__.__name__}.merge: Invalid merge strategy {strategy} provided")

logger.info(f"{self.__class__.__name__}.merge: using merge strategy {strategy} for {key.value}: this={this_value} -- other={other_value}")
if strategy == DataSpecification.MergeStrategy.OTHER:
setattr(ds, key.value, other_value)
elif strategy == DataSpecification.MergeStrategy.THIS:
setattr(ds, key.value, this_value)
else:
raise RuntimeError(f"{self.__class__.__name__}.merge: Invalid merge strategy {strategy} provided")
return ds

@classmethod
Expand Down Expand Up @@ -1043,7 +1059,29 @@ def __iter__(self):
annotations[key] = dict(value)[key]
yield "annotations", annotations

def to_str(self, columns: list[str] | None = None, indent: int = 0, default_indent: str = " " * 4) -> str:
def to_str(
self,
columns: list[str] | None = None,
indent: int = 0,
default_indent: str = " " * 4,
generated_fields: Dict[str, set] | None = None,
) -> str:
"""
Render this metadata as a human-readable string.

Args:
columns: Only render these columns; default: all columns
indent: Number of leading spaces before every line
default_indent: Indent added per nesting level, below `indent`
generated_fields: Column name to the set of that column's field names (e.g.
``{"value_stats"}``) that were computed on the fly for this call rather than
loaded from the metadata itself - marked as e.g. ``value_stats*`` in the
output, with a legend appended if any field is marked

Returns:
The rendered metadata
"""
generated_fields = generated_fields or {}
hspace = " " * indent
txt_repr = [f"{hspace}Annotations:"]
for name, annotation in self.annotations.items():
Expand All @@ -1055,13 +1093,18 @@ def to_str(self, columns: list[str] | None = None, indent: int = 0, default_inde
continue

txt_repr.append(hspace + default_indent + f"{spec_dict['name']}:")
column_generated_fields = generated_fields.get(spec_dict['name'], set())
for field_name, value in spec_dict.items():
if field_name == "name":
continue
marker = "*" if field_name in column_generated_fields else ""
txt_repr.append(
hspace + default_indent + default_indent + f"{field_name}: {value}"
hspace + default_indent + default_indent + f"{field_name}{marker}: {value}"
)

if any(generated_fields.values()):
txt_repr.append(f"{hspace}* computed on the fly for display, not part of the loaded metadata")

return "\n".join(txt_repr)

@classmethod
Expand Down
Loading
Loading