formats

FormatSpec dataclass

Specification for a data format

Parameters:

Name	Type	Description	Default
name	str	Name of the format.	required
required_columns	list[str]	List of columns required to identify this format.	required
filter_columns	list[str]	List of columns that can be used to filter data.	required
converter	Callable[[DataFrame], DataFrame]	Function to convert a DataFrame to OpenHDX format.	required
aggregated	bool | Callable[[DataFrame], bool]	Whether the format is aggregated, or a function to determine if a DataFrame is aggregated.	required

Source code in hdxms_datasets/formats.py

@dataclass(frozen=True)
class FormatSpec:
    """Specification for a data format

    Args:
        name: Name of the format.
        required_columns: List of columns required to identify this format.
        filter_columns: List of columns that can be used to filter data.
        converter: Function to convert a DataFrame to OpenHDX format.
        aggregated: Whether the format is aggregated, or a function to determine if a DataFrame is aggregated.

    """

    name: str
    required_columns: list[str]
    filter_columns: list[str]
    converter: Callable[[nw.DataFrame], nw.DataFrame]
    aggregated: bool | Callable[[nw.DataFrame], bool]

    def matches(self, df: nw.DataFrame) -> bool:
        """Check if a DataFrame matches this format."""
        df_cols = set(df.columns)
        required_cols = set(self.required_columns)
        return required_cols.issubset(df_cols)

    def convert(self, df: nw.DataFrame) -> nw.DataFrame:
        """Convert DataFrame to OpenHDX format."""
        return self.converter(df)

    def is_aggregated(self, df: nw.DataFrame | None = None) -> bool:
        """Check if a DataFrame is aggregated."""
        if self.aggregated is True:
            return True
        if callable(self.aggregated):
            if df is None:
                raise ValueError("DataFrame must be provided to check aggregation")
            return self.aggregated(df)
        return False

convert(df)

Convert DataFrame to OpenHDX format.

Source code in hdxms_datasets/formats.py

def convert(self, df: nw.DataFrame) -> nw.DataFrame:
    """Convert DataFrame to OpenHDX format."""
    return self.converter(df)

is_aggregated(df=None)

Check if a DataFrame is aggregated.

Source code in hdxms_datasets/formats.py

def is_aggregated(self, df: nw.DataFrame | None = None) -> bool:
    """Check if a DataFrame is aggregated."""
    if self.aggregated is True:
        return True
    if callable(self.aggregated):
        if df is None:
            raise ValueError("DataFrame must be provided to check aggregation")
        return self.aggregated(df)
    return False

matches(df)

Check if a DataFrame matches this format.

Source code in hdxms_datasets/formats.py

def matches(self, df: nw.DataFrame) -> bool:
    """Check if a DataFrame matches this format."""
    df_cols = set(df.columns)
    required_cols = set(self.required_columns)
    return required_cols.issubset(df_cols)

identify_format(df)

Identify format from DataFrame columns

Source code in hdxms_datasets/formats.py

def identify_format(df: nw.DataFrame) -> Optional[FormatSpec]:
    """Identify format from DataFrame columns"""
    for fmt in FORMATS:
        if fmt.matches(df):
            return fmt
    return None