graphld.vcf_io

I/O functions for GWAS-VCF format summary statistics.

For broader loading workflows, see the I/O and Merging guide.

vcf_io

validate_vcf_format_columns

validate_vcf_format_columns(columns: list, verbose: bool = False) -> dict

Validate and describe VCF FORMAT columns according to GWAS-VCF specification.

Parameters:

Name	Type	Description	Default
columns	list	List of column names from VCF FORMAT	required
verbose	bool	If True, print detailed information about columns. Defaults to False.	False

Returns:

Name	Type	Description
dict	dict	Dictionary of validated columns with their descriptions and requirements

Source code in src/graphld/vcf_io.py

def validate_vcf_format_columns(columns: list, verbose: bool = False) -> dict:
    """
    Validate and describe VCF FORMAT columns according to GWAS-VCF specification.

    Args:
        columns (list): List of column names from VCF FORMAT
        verbose (bool, optional): If True, print detailed information about columns.
            Defaults to False.

    Returns:
        dict: Dictionary of validated columns with their descriptions and requirements
    """
    # Define the VCF FORMAT column specifications
    vcf_format_spec = {
        'NS': {
            'description': 'Variant-specific number of samples/individuals with called '
                         'genotypes used to test association with specified trait',
            'required': False
        },
        'EZ': {
            'description': 'Z-score provided if it was used to derive the ES and SE fields',
            'required': False
        },
        'SI': {
            'description': 'Accuracy score of association statistics imputation',
            'required': False
        },
        'NC': {
            'description': 'Variant-specific number of cases used to estimate genetic '
                         'effect (binary traits only)',
            'required': False
        },
        'ES': {
            'description': 'Effect size estimate relative to the alternative allele',
            'required': True
        },
        'SE': {
            'description': 'Standard error of effect size estimate',
            'required': True
        },
        'LP': {
            'description': '-log10 p-value for effect estimate',
            'required': True
        },
        'AF': {
            'description': 'Alternative allele frequency in trait subset',
            'required': False
        },
        'AC': {
            'description': 'Alternative allele count in the trait subset',
            'required': False
        }
    }

    # Check for missing required columns
    missing_required = [
        col for col, spec in vcf_format_spec.items()
        if spec['required'] and col not in columns
    ]
    if missing_required:
        raise ValueError(f"Missing required columns: {missing_required}")

    # Check for extra columns not in specification
    extra_columns = [col for col in columns if col not in vcf_format_spec]
    if extra_columns:
        raise ValueError(
            f"Extra columns not in VCF FORMAT specification: {extra_columns}"
        )

    # If verbose, print detailed information
    if verbose:
        print("GWAS-VCF keys and descriptions:")
        for col in columns:
            spec = vcf_format_spec[col]
            req_status = 'Required' if spec['required'] else 'Optional'
            print(f"- {col}: {spec['description']} ({req_status})")

    # Return the specification for the found columns
    return {col: vcf_format_spec[col] for col in columns}

read_gwas_vcf

read_gwas_vcf(file_path: str, num_rows: Optional[int] = None, maximum_missingness: float = 0.1, verbose: bool = False) -> pl.DataFrame

Reads a GWAS-VCF file using Polars and returns a DataFrame.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the VCF file.	required
num_rows	Optional[int]	Number of rows to read. Defaults to None.	None
maximum_missingness	float	Maximum fraction of missing samples allowed. Defaults to 0.1.	0.1
verbose	bool	Print detailed information about FORMAT columns. Defaults to False.	False

Returns:

Type	Description
DataFrame	pl.DataFrame: DataFrame containing the VCF data.

Source code in src/graphld/vcf_io.py

def read_gwas_vcf(
    file_path: str,
    num_rows: Optional[int] = None,
    maximum_missingness: float = 0.1,
    verbose: bool = False
) -> pl.DataFrame:
    """
    Reads a GWAS-VCF file using Polars and returns a DataFrame.

    Args:
        file_path (str): Path to the VCF file.
        num_rows (Optional[int], optional): Number of rows to read. Defaults to None.
        maximum_missingness (float, optional): Maximum fraction of missing samples
            allowed. Defaults to 0.1.
        verbose (bool, optional): Print detailed information about FORMAT columns.
            Defaults to False.

    Returns:
        pl.DataFrame: DataFrame containing the VCF data.
    """
    # Read the VCF file, skipping the header lines
    df = pl.read_csv(
        file_path,
        separator='\t',
        comment_prefix='##',
        has_header=True,
        n_rows=num_rows
    )

    df = split_sample_column(df, verbose=verbose)
    df = process_chromosome_column(df)

    # Filter based on missingness using NS (number of samples)
    if 'NS' in df.columns:
        min_samples = (1 - maximum_missingness) * pl.col('NS').max()
        df = df.filter(pl.col('NS') >= min_samples)

    return df