Skip to content

BatchProcessor

Process multiple gene set files in batch mode.

Overview

BatchProcessor enables efficient processing of multiple DEG files with automatic output generation.

Class Reference

gs2txt.batch.BatchProcessor

Process multiple gene sets in batch with progress tracking.

Source code in gs2txt/batch.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class BatchProcessor:
    """Process multiple gene sets in batch with progress tracking."""

    def __init__(self, annotator: GeneSetAnnotator):
        """
        Initialize with a GeneSetAnnotator instance.

        Parameters
        ----------
        annotator : GeneSetAnnotator
            Configured annotator for gene set annotation
        """
        self.annotator = annotator

    def process_grouped_data(
        self,
        df: pd.DataFrame,
        group_column: str,
        **annotate_kwargs
    ) -> pd.DataFrame:
        """
        Process dataframe grouped by a column.

        Parameters
        ----------
        df : pd.DataFrame
            Input data with gene and group columns
        group_column : str
            Column to group by (e.g., 'cluster', 'celltype')
        **annotate_kwargs
            Parameters passed to annotate() method

        Returns
        -------
        pd.DataFrame
            Summary data with columns: gs, annotation, pathways, PPIs, Final_prompt
        """
        # Get unique groups
        groups = df[group_column].unique()

        # Process each group with progress bar
        results = []
        for group in tqdm(groups, desc="Processing gene sets"):
            # Get genes for this group
            group_df = df[df[group_column] == group]

            # Annotate with detailed info
            try:
                detailed = self.annotator.annotate_detailed(group_df, **annotate_kwargs)
                results.append({
                    "gs": group,
                    "annotation": detailed["annotation"],
                    "pathways": detailed["pathways"],
                    "PPIs": detailed["ppis"],
                    "Final_prompt": detailed["final_prompt"],
                })
            except Exception as e:
                print(f"Warning: Failed to annotate {group}: {e}")
                results.append({
                    "gs": group,
                    "annotation": "",
                    "pathways": "",
                    "PPIs": "",
                    "Final_prompt": "",
                })

        return pd.DataFrame(results)

    def process_single_geneset(
        self,
        df: pd.DataFrame,
        gs_name: str = "gene_set",
        **annotate_kwargs
    ) -> pd.DataFrame:
        """
        Process a single gene set.

        Parameters
        ----------
        df : pd.DataFrame
            Input data with gene column
        gs_name : str
            Name for this gene set (used in output 'gs' column)
        **annotate_kwargs
            Parameters passed to annotate() method

        Returns
        -------
        pd.DataFrame
            Summary data with columns: gs, annotation, pathways, PPIs, Final_prompt
        """
        # Annotate the entire dataset as one gene set with detailed info
        try:
            detailed = self.annotator.annotate_detailed(df, **annotate_kwargs)
            return pd.DataFrame([{
                "gs": gs_name,
                "annotation": detailed["annotation"],
                "pathways": detailed["pathways"],
                "PPIs": detailed["ppis"],
                "Final_prompt": detailed["final_prompt"]
            }])
        except Exception as e:
            print(f"Warning: Annotation failed: {e}")
            return pd.DataFrame([{
                "gs": gs_name,
                "annotation": "",
                "pathways": "",
                "PPIs": "",
                "Final_prompt": ""
            }])

    def process_single_file(
        self,
        input_path: str,
        output_path: str,
        group_column: Optional[str] = None,
        **annotate_kwargs
    ):
        """
        Process entire CSV file and save results.

        Parameters
        ----------
        input_path : str
            Input CSV file path
        output_path : str
            Output CSV file path
        group_column : str, optional
            Column to group by. If None, treats all genes as single set.
        **annotate_kwargs
            Parameters passed to annotate() method
        """
        from pathlib import Path

        # Read CSV
        print(f"Reading input file: {input_path}")
        df = CSVReader.read_gene_file(input_path, group_column)

        # Process (grouped or single)
        if group_column:
            print(f"Processing {len(df[group_column].unique())} gene sets grouped by '{group_column}'...")
            results = self.process_grouped_data(df, group_column, **annotate_kwargs)
        else:
            # Use input filename (without extension) as gs name
            gs_name = Path(input_path).stem
            print(f"Processing single gene set '{gs_name}' with {len(df)} genes...")
            results = self.process_single_geneset(df, gs_name=gs_name, **annotate_kwargs)

        # Write results
        print(f"Writing results to: {output_path}")
        CSVWriter.write_results(results, output_path)
        print("Done!")

Functions

__init__(annotator)

Initialize with a GeneSetAnnotator instance.

Parameters:

Name Type Description Default
annotator GeneSetAnnotator

Configured annotator for gene set annotation

 required
Source code in gs2txt/batch.py 
17
18
19
20
21
22
23
24
25
26
def __init__(self, annotator: GeneSetAnnotator):
    """
    Initialize with a GeneSetAnnotator instance.

    Parameters
    ----------
    annotator : GeneSetAnnotator
        Configured annotator for gene set annotation
    """
    self.annotator = annotator
process_grouped_data(df, group_column, **annotate_kwargs)

Process dataframe grouped by a column.

Parameters:

Name Type Description Default
df DataFrame

Input data with gene and group columns

 required
group_column str

Column to group by (e.g., 'cluster', 'celltype')

 required
**annotate_kwargs

Parameters passed to annotate() method

 {}

Returns:

Type Description
DataFrame

Summary data with columns: gs, annotation, pathways, PPIs, Final_prompt
Source code in gs2txt/batch.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def process_grouped_data(
    self,
    df: pd.DataFrame,
    group_column: str,
    **annotate_kwargs
) -> pd.DataFrame:
    """
    Process dataframe grouped by a column.

    Parameters
    ----------
    df : pd.DataFrame
        Input data with gene and group columns
    group_column : str
        Column to group by (e.g., 'cluster', 'celltype')
    **annotate_kwargs
        Parameters passed to annotate() method

    Returns
    -------
    pd.DataFrame
        Summary data with columns: gs, annotation, pathways, PPIs, Final_prompt
    """
    # Get unique groups
    groups = df[group_column].unique()

    # Process each group with progress bar
    results = []
    for group in tqdm(groups, desc="Processing gene sets"):
        # Get genes for this group
        group_df = df[df[group_column] == group]

        # Annotate with detailed info
        try:
            detailed = self.annotator.annotate_detailed(group_df, **annotate_kwargs)
            results.append({
                "gs": group,
                "annotation": detailed["annotation"],
                "pathways": detailed["pathways"],
                "PPIs": detailed["ppis"],
                "Final_prompt": detailed["final_prompt"],
            })
        except Exception as e:
            print(f"Warning: Failed to annotate {group}: {e}")
            results.append({
                "gs": group,
                "annotation": "",
                "pathways": "",
                "PPIs": "",
                "Final_prompt": "",
            })

    return pd.DataFrame(results)
process_single_file(input_path, output_path, group_column=None, **annotate_kwargs)

Process entire CSV file and save results.

Parameters:

Name Type Description Default
input_path str

Input CSV file path

 required
output_path str

Output CSV file path

 required
group_column str

Column to group by. If None, treats all genes as single set.

 None
**annotate_kwargs

Parameters passed to annotate() method

 {}
Source code in gs2txt/batch.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def process_single_file(
    self,
    input_path: str,
    output_path: str,
    group_column: Optional[str] = None,
    **annotate_kwargs
):
    """
    Process entire CSV file and save results.

    Parameters
    ----------
    input_path : str
        Input CSV file path
    output_path : str
        Output CSV file path
    group_column : str, optional
        Column to group by. If None, treats all genes as single set.
    **annotate_kwargs
        Parameters passed to annotate() method
    """
    from pathlib import Path

    # Read CSV
    print(f"Reading input file: {input_path}")
    df = CSVReader.read_gene_file(input_path, group_column)

    # Process (grouped or single)
    if group_column:
        print(f"Processing {len(df[group_column].unique())} gene sets grouped by '{group_column}'...")
        results = self.process_grouped_data(df, group_column, **annotate_kwargs)
    else:
        # Use input filename (without extension) as gs name
        gs_name = Path(input_path).stem
        print(f"Processing single gene set '{gs_name}' with {len(df)} genes...")
        results = self.process_single_geneset(df, gs_name=gs_name, **annotate_kwargs)

    # Write results
    print(f"Writing results to: {output_path}")
    CSVWriter.write_results(results, output_path)
    print("Done!")
process_single_geneset(df, gs_name='gene_set', **annotate_kwargs)

Process a single gene set.

Parameters:

Name Type Description Default
df DataFrame

Input data with gene column

 required
gs_name str

Name for this gene set (used in output 'gs' column)

 'gene_set'
**annotate_kwargs

Parameters passed to annotate() method

 {}

Returns:

Type Description
DataFrame

Summary data with columns: gs, annotation, pathways, PPIs, Final_prompt
Source code in gs2txt/batch.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def process_single_geneset(
    self,
    df: pd.DataFrame,
    gs_name: str = "gene_set",
    **annotate_kwargs
) -> pd.DataFrame:
    """
    Process a single gene set.

    Parameters
    ----------
    df : pd.DataFrame
        Input data with gene column
    gs_name : str
        Name for this gene set (used in output 'gs' column)
    **annotate_kwargs
        Parameters passed to annotate() method

    Returns
    -------
    pd.DataFrame
        Summary data with columns: gs, annotation, pathways, PPIs, Final_prompt
    """
    # Annotate the entire dataset as one gene set with detailed info
    try:
        detailed = self.annotator.annotate_detailed(df, **annotate_kwargs)
        return pd.DataFrame([{
            "gs": gs_name,
            "annotation": detailed["annotation"],
            "pathways": detailed["pathways"],
            "PPIs": detailed["ppis"],
            "Final_prompt": detailed["final_prompt"]
        }])
    except Exception as e:
        print(f"Warning: Annotation failed: {e}")
        return pd.DataFrame([{
            "gs": gs_name,
            "annotation": "",
            "pathways": "",
            "PPIs": "",
            "Final_prompt": ""
        }])

Constructor

BatchProcessor(
    llm_provider: BaseLLMProvider,
    enrichment_method: Optional[str] = "pathway",
    prompt_builder: Optional[PromptBuilder] = None
)

Parameters

Parameter Type Default Description
llm_provider BaseLLMProvider Required LLM provider instance
enrichment_method str or None "pathway" Enrichment method
prompt_builder PromptBuilder None Custom prompt builder

Methods

process_directory()

Process all CSV files in a directory.

def process_directory(
    self,
    input_dir: str,
    output_dir: str,
    max_gene_num: int = 60,
    max_pathway_num: int = 10,
    pvalue_threshold: float = 0.05,
    log2fc_threshold: float = 1.0
) -> Dict[str, str]

Parameters

Parameter Type Default Description
input_dir str Required Directory with DEG CSV files
output_dir str Required Output directory
max_gene_num int 60 Maximum genes per file
max_pathway_num int 10 Maximum pathways
pvalue_threshold float 0.05 P-value filter
log2fc_threshold float 1.0 Log2FC filter

Returns

Dict[str, str] - Mapping of input filename to annotation result

Example

from gs2txt.batch import BatchProcessor
from gs2txt.llm import OpenAIProvider

provider = OpenAIProvider(api_key="...", model_id="gpt-4")
processor = BatchProcessor(llm_provider=provider)

results = processor.process_directory(
    input_dir="data/deg/",
    output_dir="results/",
    max_gene_num=100,
    pvalue_threshold=0.01
)

for filename, annotation in results.items():
    print(f"{filename}: {annotation[:100]}...")

process_file()

Process a single DEG file.

def process_file(
    self,
    input_file: str,
    output_file: Optional[str] = None,
    **kwargs
) -> str

Parameters

Parameter Type Default Description
input_file str Required Path to DEG CSV file
output_file str None Output path (optional)
**kwargs Same as process_directory()

Returns

str - Annotation result

Example

result = processor.process_file(
    input_file="data/sample1.csv",
    output_file="results/sample1_annotation.txt"
)

Input File Format

CSV files must have a gene column. Optional columns:

Column Description
gene Gene symbols (required)
pvalue P-values for filtering
logFC Log2 fold change

Example

gene,pvalue,logFC
TP53,0.001,2.5
BRCA1,0.002,2.3
MYC,0.003,-1.8

Output Format

Directory Processing

Creates one output file per input file:

results/
├── sample1_annotation.csv
├── sample2_annotation.csv
└── sample3_annotation.csv

Each output file contains:

gene_set,annotation
sample1,"This gene set is primarily involved in..."

Single File Processing

Returns annotation as string, optionally saves to file.

Error Handling

try:
    results = processor.process_directory(
        input_dir="data/",
        output_dir="results/"
    )
except FileNotFoundError as e:
    print(f"Directory not found: {e}")
except ValueError as e:
    print(f"Invalid file format: {e}")

Thread Safety

BatchProcessor processes files sequentially by default. For parallel processing, create separate instances:

from concurrent.futures import ThreadPoolExecutor

def process_single(filepath):
    processor = BatchProcessor(llm_provider=provider)
    return processor.process_file(filepath)

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(process_single, file_list))