"""
Parser for MiXCR pipeline outputs.
Handles TSV files generated by MiXCR with
platform-specific column mappings.
"""
from pathlib import Path
from typing import Union, List, Optional
import pandas as pd
import polars as pl
from .base_parser import BaseParser, ParserConfig
from .utils_enhanced import standardize_airr_data
class MiXCRParser(BaseParser):
"""
Parser for MiXCR output files.
Handles the specific column format and naming conventions
used by MiXCR analysis pipeline.
"""
def __init__(self, config: Optional[ParserConfig] = None):
"""Initialize MiXCR parser."""
super().__init__(config)
self.platform = "mixcr"
def parse_file(self, file_path: Path) -> Union[pd.DataFrame, pl.DataFrame]:
"""
Parse a single MiXCR TSV file.
Args:
file_path: Path to the TSV file
Returns:
Parsed data frame
"""
delimiter = self.detect_delimiter(file_path)
repertoire_id = self.extract_repertoire_id(file_path)
try:
if self.config.return_type == "polars":
data = pl.read_csv(
file_path,
separator=delimiter,
null_values=["", "NA", "N/A"],
try_parse_dates=False,
ignore_errors=True
)
data = data.with_columns(pl.lit(repertoire_id).alias("repertoire_id"))
else:
data = pd.read_csv(
file_path,
sep=delimiter,
na_values=["", "NA", "N/A"],
low_memory=False,
dtype=str
)
data["repertoire_id"] = repertoire_id
except Exception as e:
raise ValueError(f"Failed to parse {file_path}: {str(e)}")
return data
def standardize_columns(
self,
data: Union[pd.DataFrame, pl.DataFrame]
) -> Union[pd.DataFrame, pl.DataFrame]:
"""
Standardize MiXCR columns to AIRR format.
Args:
data: Raw MiXCR data
Returns:
Data with standardized column names
"""
repertoire_id = None
if isinstance(data, pl.DataFrame):
if "repertoire_id" in data.columns:
repertoire_id = data["repertoire_id"][0]
else:
if "repertoire_id" in data.columns:
repertoire_id = data["repertoire_id"].iloc[0]
return self.standardize_to_airr(data, platform="mixcr", repertoire_id=repertoire_id)
[docs]
def read_mixcr(
path: Union[str, Path, List[str]],
recursive: bool = False,
parallel: bool = True,
threads: Optional[int] = None,
return_type: str = "polars",
validate_airr: bool = True,
enhanced_mappings: bool = True
) -> Union[pd.DataFrame, pl.DataFrame]:
"""
Read MiXCR pipeline output files.
This function imports tab-separated value (.tsv) files generated by
MiXCR and standardizes them to AIRR format.
Args:
path: Path to directory containing TSV files, single file, or list of files
recursive: Search recursively for files in subdirectories
parallel: Process files in parallel
threads: Number of threads to use
return_type: Return format ("polars", "pandas")
validate_airr: Enable AIRR schema validation
enhanced_mappings: Use comprehensive column mappings
Returns:
Parsed and standardized data frame
Examples:
>>> # Read MiXCR output directory
>>> data = read_mixcr("data/mixcr_output/")
>>> # Read specific file
>>> data = read_mixcr("clones.tsv")
>>> # Read with pandas backend
>>> data = read_mixcr("data/", return_type="pandas")
"""
# Create parser configuration
config = ParserConfig(
parallel=parallel,
threads=threads,
return_type=return_type,
validate_airr=validate_airr,
enhanced_mappings=enhanced_mappings
)
# Initialize parser and process data
parser = MiXCRParser(config)
return parser.parse(path, recursive=recursive)