| name | tooluniverse-crispr-screen-analysis |
| description | Comprehensive CRISPR screen analysis for functional genomics. Analyze pooled or arrayed CRISPR screens (knockout, activation, interference) to identify essential genes, synthetic lethal interactions, and drug targets. Perform sgRNA count processing, gene-level scoring (MAGeCK, BAGEL), quality control, pathway enrichment, and drug target prioritization. Use for CRISPR screen analysis, gene essentiality studies, synthetic lethality detection, functional genomics, drug target validation, or identifying genetic vulnerabilities. |
ToolUniverse CRISPR Screen Analysis
Comprehensive skill for analyzing CRISPR-Cas9 genetic screens to identify essential genes, synthetic lethal interactions, and therapeutic targets through robust statistical analysis and pathway enrichment.
Overview
CRISPR screens enable genome-wide functional genomics by systematically perturbing genes and measuring fitness effects. This skill provides an 8-phase workflow for:
- Processing sgRNA count matrices
- Quality control and normalization
- Gene-level essentiality scoring (MAGeCK-like and BAGEL-like approaches)
- Synthetic lethality detection
- Pathway enrichment analysis
- Drug target prioritization with DepMap integration
- Integration with expression and mutation data
Core Workflow
Phase 1: Data Import & sgRNA Count Processing
Load sgRNA Count Matrix
import pandas as pd
import numpy as np
def load_sgrna_counts(counts_file):
"""
Load sgRNA count matrix from MAGeCK format or generic TSV.
Expected format:
sgRNA | Gene | Sample1 | Sample2 | Sample3 | ...
sgRNA_1 | BRCA1 | 1500 | 1200 | 1100 | ...
sgRNA_2 | BRCA1 | 1800 | 1500 | 1400 | ...
"""
counts = pd.read_csv(counts_file, sep='\t')
required_cols = ['sgRNA', 'Gene']
if not all(col in counts.columns for col in required_cols):
raise ValueError(f"Missing required columns: {required_cols}")
sample_cols = [col for col in counts.columns if col not in ['sgRNA', 'Gene']]
count_matrix = counts[sample_cols].copy()
count_matrix.index = counts['sgRNA']
sgrna_to_gene = dict(zip(counts['sgRNA'], counts['Gene']))
metadata = {
'n_sgrnas': len(counts),
'n_genes': counts['Gene'].nunique(),
'n_samples': len(sample_cols),
'sample_names': sample_cols,
'sgrna_to_gene': sgrna_to_gene
}
return count_matrix, metadata
counts, meta = load_sgrna_counts("sgrna_counts.txt")
print(f"Loaded {meta['n_sgrnas']} sgRNAs targeting {meta['n_genes']} genes across {meta['n_samples']} samples")
Create Experimental Design Table
def create_design_matrix(sample_names, conditions, timepoints=None):
"""
Create experimental design linking samples to conditions.
Example:
Sample | Condition | Timepoint | Replicate
T0_rep1 | baseline | 0 | 1
T14_rep1 | treatment | 14 | 1
"""
design = pd.DataFrame({
'Sample': sample_names,
'Condition': conditions
})
if timepoints is not None:
design['Timepoint'] = timepoints
design['Replicate'] = design.groupby('Condition').cumcount() + 1
return design
sample_names = ['T0_rep1', 'T0_rep2', 'T14_rep1', 'T14_rep2', 'T14_rep3']
conditions = ['baseline', 'baseline', 'treatment', 'treatment', 'treatment']
design = create_design_matrix(sample_names, conditions)
Phase 2: Quality Control & Filtering
Assess sgRNA Distribution
def qc_sgrna_distribution(count_matrix, min_reads=30, min_samples=2):
"""
Quality control for sgRNA distribution.
- Remove sgRNAs with low read counts
- Check for outlier samples
- Assess library representation
"""
results = {}
library_sizes = count_matrix.sum(axis=0)
results['library_sizes'] = library_sizes
results['median_library_size'] = library_sizes.median()
zero_counts = (count_matrix == 0).sum(axis=1)
results['zero_counts'] = zero_counts
results['sgrnas_with_zeros'] = (zero_counts > 0).sum()
low_count_mask = (count_matrix < min_reads).sum(axis=1) > (len(count_matrix.columns) - min_samples)
results['low_count_sgrnas'] = low_count_mask.sum()
def gini_coefficient(counts):
sorted_counts = np.sort(counts)
n = len(counts)
cumsum = np.cumsum(sorted_counts)
return (2 * np.sum((np.arange(1, n+1)) * sorted_counts)) / (n * cumsum[-1]) - (n + 1) / n
results['gini_per_sample'] = {col: gini_coefficient(count_matrix[col].values)
for col in count_matrix.columns}
results[] = {
: min_reads,
: min_samples,
: low_count_mask.()
}
results
qc_results = qc_sgrna_distribution(counts, min_reads=, min_samples=)
()
()
Filter Low-Count sgRNAs
def filter_low_count_sgrnas(count_matrix, sgrna_to_gene, min_reads=30, min_samples=2):
"""
Remove sgRNAs with insufficient read counts.
"""
keep_mask = (count_matrix >= min_reads).sum(axis=1) >= min_samples
filtered_counts = count_matrix[keep_mask].copy()
filtered_mapping = {k: v for k, v in sgrna_to_gene.items() if k in filtered_counts.index}
print(f"Filtered: {(~keep_mask).sum()} sgRNAs removed, {keep_mask.sum()} retained")
return filtered_counts, filtered_mapping
filtered_counts, filtered_mapping = filter_low_count_sgrnas(counts, meta['sgrna_to_gene'])
Phase 3: Normalization
Library Size Normalization
def normalize_counts(count_matrix, method='median'):
"""
Normalize sgRNA counts to account for library size differences.
Methods:
- 'median': Median ratio normalization (like DESeq2)
- 'total': Total count normalization (CPM-like)
"""
if method == 'median':
pseudo_ref = np.exp(np.log(count_matrix + 1).mean(axis=1)) - 1
size_factors = {}
for col in count_matrix.columns:
ratios = count_matrix[col] / pseudo_ref
ratios = ratios[ratios > 0]
size_factors[col] = ratios.median()
normalized = count_matrix.div(pd.Series(size_factors), axis=1)
elif method == 'total':
size_factors = count_matrix.sum(axis=0) / 1e6
normalized = count_matrix.div(size_factors, axis=1)
else:
raise ValueError(f"Unknown normalization method: {method}")
return normalized, size_factors
norm_counts, size_factors = normalize_counts(filtered_counts, method='median')
Log-Fold Change Calculation
def calculate_lfc(norm_counts, design, control_condition='baseline', treatment_condition='treatment'):
"""
Calculate log2 fold changes between treatment and control.
"""
control_samples = design[design['Condition'] == control_condition]['Sample'].tolist()
treatment_samples = design[design['Condition'] == treatment_condition]['Sample'].tolist()
control_mean = norm_counts[control_samples].mean(axis=1)
treatment_mean = norm_counts[treatment_samples].mean(axis=1)
lfc = np.log2((treatment_mean + 1) / (control_mean + 1))
return lfc, control_mean, treatment_mean
lfc, control_mean, treatment_mean = calculate_lfc(norm_counts, design)
Phase 4: Gene-Level Scoring (MAGeCK-like)
Aggregate sgRNA Scores to Gene Level
def mageck_gene_scoring(lfc, sgrna_to_gene, method='rra'):
"""
Gene-level essentiality scoring using MAGeCK-like approach.
Methods:
- 'rra': Robust Rank Aggregation (identify genes with consistently low-ranking sgRNAs)
- 'mean': Simple mean LFC across sgRNAs
"""
gene_lfc = {}
for sgrna, gene in sgrna_to_gene.items():
if sgrna in lfc.index:
if gene not in gene_lfc:
gene_lfc[gene] = []
gene_lfc[gene].append(lfc[sgrna])
if method == 'rra':
ranked_sgrnas = lfc.sort_values()
ranks = {sgrna: rank for rank, sgrna in enumerate(ranked_sgrnas.index, 1)}
gene_scores = {}
for gene, sgrna_list in gene_lfc.items():
gene_ranks = [ranks[sgrna] for sgrna in sgrna_list if sgrna in ranks]
if len(gene_ranks) > 0:
gene_scores[gene] = {
'score': np.mean(gene_ranks),
'n_sgrnas': len(gene_ranks),
: np.mean([lfc[sg] sg sgrna_list sg lfc.index])
}
gene_df = pd.DataFrame(gene_scores).T
gene_df[] = gene_df[].rank()
method == :
gene_df = pd.DataFrame({
gene: {
: np.mean(sgrna_lfcs),
: (sgrna_lfcs),
: np.mean(sgrna_lfcs)
}
gene, sgrna_lfcs gene_lfc.items()
}).T
gene_df = gene_df.sort_values()
gene_df
gene_scores = mageck_gene_scoring(lfc, filtered_mapping, method=)
()
Bayes Factor Scoring (BAGEL-like)
def bagel_bayes_factor(lfc, sgrna_to_gene, essential_genes=None, nonessential_genes=None):
"""
BAGEL-like Bayes Factor calculation for gene essentiality.
Uses reference sets of known essential and non-essential genes to
calculate likelihood ratios.
"""
if essential_genes is None:
essential_genes = ['RPL5', 'RPS6', 'POLR2A', 'PSMC2', 'PSMD14']
if nonessential_genes is None:
nonessential_genes = ['AAVS1', 'ROSA26', 'HPRT1']
essential_lfc = [lfc[sg] for sg, g in sgrna_to_gene.items()
if g in essential_genes and sg in lfc.index]
nonessential_lfc = [lfc[sg] for sg, g in sgrna_to_gene.items()
if g in nonessential_genes and sg in lfc.index]
if len(essential_lfc) < 3 or len(nonessential_lfc) < 3:
print("Warning: Insufficient reference genes for BAGEL scoring")
return
essential_mean, essential_std = np.mean(essential_lfc), np.std(essential_lfc)
nonessential_mean, nonessential_std = np.mean(nonessential_lfc), np.std(nonessential_lfc)
gene_bf = {}
gene_lfc_map = {}
sgrna, gene sgrna_to_gene.items():
sgrna lfc.index:
gene gene_lfc_map:
gene_lfc_map[gene] = []
gene_lfc_map[gene].append(lfc[sgrna])
gene, sgrna_lfcs gene_lfc_map.items():
mean_lfc = np.mean(sgrna_lfcs)
scipy.stats norm
l_essential = norm.pdf(mean_lfc, essential_mean, essential_std)
l_nonessential = norm.pdf(mean_lfc, nonessential_mean, nonessential_std)
bf = l_essential / (l_nonessential + )
gene_bf[gene] = {
: bf,
: mean_lfc,
: (sgrna_lfcs)
}
bf_df = pd.DataFrame(gene_bf).T
bf_df = bf_df.sort_values(, ascending=)
bf_df
bf_scores = bagel_bayes_factor(lfc, filtered_mapping)
bf_scores :
()
Phase 5: Synthetic Lethality Detection
Identify Context-Specific Essential Genes
def detect_synthetic_lethality(gene_scores_wildtype, gene_scores_mutant,
lfc_threshold=-1.0, rank_diff_threshold=100):
"""
Identify genes that are selectively essential in mutant context
(synthetic lethal interactions).
Compare essentiality scores between wildtype and mutant cell lines.
"""
comparison = pd.merge(
gene_scores_wildtype[['mean_lfc', 'rank']],
gene_scores_mutant[['mean_lfc', 'rank']],
left_index=True,
right_index=True,
suffixes=('_wt', '_mut')
)
comparison['delta_lfc'] = comparison['mean_lfc_mut'] - comparison['mean_lfc_wt']
comparison['delta_rank'] = comparison['rank_wt'] - comparison['rank_mut']
sl_candidates = comparison[
(comparison['mean_lfc_mut'] < lfc_threshold) &
(comparison['mean_lfc_wt'] > -0.5) &
(comparison['delta_rank'] > rank_diff_threshold)
].copy()
sl_candidates = sl_candidates.sort_values('delta_lfc')
return sl_candidates
Query DepMap for Known Dependencies
def query_depmap_dependencies(gene_symbol):
"""
Query DepMap database for known gene dependencies.
ToolUniverse doesn't have direct DepMap tools, but we can use
STRING or literature tools to find dependency information.
"""
from tooluniverse import ToolUniverse
tu = ToolUniverse()
result = tu.run_one_function({
"name": "PubMed_search",
"arguments": {
"query": f'("{gene_symbol}"[Gene]) AND ("CRISPR screen" OR "gene essentiality" OR "DepMap")',
"max_results": 20
}
})
if 'data' in result and 'papers' in result['data']:
papers = result['data']['papers']
print(f"Found {len(papers)} papers on {gene_symbol} essentiality")
return papers
return []
Phase 6: Pathway Enrichment Analysis
Enrichment of Essential Genes
def enrich_essential_genes(gene_scores, top_n=100, databases=['KEGG_2021_Human', 'GO_Biological_Process_2021']):
"""
Perform pathway enrichment on top essential genes.
"""
from tooluniverse import ToolUniverse
tu = ToolUniverse()
top_genes = gene_scores.head(top_n).index.tolist()
print(f"Enriching {len(top_genes)} top essential genes...")
result = tu.run_one_function({
"name": "Enrichr_submit_genelist",
"arguments": {
"gene_list": top_genes,
"description": "CRISPR_screen_essential_genes"
}
})
if 'data' not in result or 'userListId' not in result['data']:
print("Failed to submit gene list to Enrichr")
return None
user_list_id = result['data']['userListId']
all_results = {}
for db in databases:
enrich_result = tu.run_one_function({
"name": "Enrichr_get_results",
"arguments": {
"userListId": user_list_id,
: db
}
})
enrich_result db enrich_result[]:
all_results[db] = pd.DataFrame(enrich_result[][db])
()
all_results
Phase 7: Drug Target Prioritization
Integrate with Expression & Mutation Data
def prioritize_drug_targets(gene_scores, expression_data=None, mutation_data=None):
"""
Prioritize CRISPR hits as drug targets based on:
1. Essentiality score (from CRISPR screen)
2. Expression level in disease vs normal (if provided)
3. Mutation frequency in tumors (if provided)
4. Druggability (query DGIdb)
"""
from tooluniverse import ToolUniverse
tu = ToolUniverse()
candidates = gene_scores.head(50).copy()
if expression_data is not None:
candidates = candidates.merge(expression_data, left_index=True, right_index=True, how='left')
if mutation_data is not None:
candidates = candidates.merge(mutation_data, left_index=True, right_index=True, how='left')
druggability_scores = {}
for gene in candidates.index[:20]:
result = tu.run_one_function({
"name": "DGIdb_query_gene",
"arguments": {"gene_symbol": gene}
})
if 'data' in result and 'matchedTerms' in result['data']:
matches = result[][]
(matches) > :
n_drugs = (matches[].get(, []))
druggability_scores[gene] = n_drugs
:
druggability_scores[gene] =
:
druggability_scores[gene] =
candidates[] = pd.Series(druggability_scores)
candidates[] = (candidates[].() - candidates[]) / \
(candidates[].() - candidates[].())
candidates.columns:
candidates[] = (candidates[] - candidates[].()) / \
(candidates[].() - candidates[].())
:
candidates[] =
candidates[] = candidates[] / (candidates[].() + )
candidates[] = (
* candidates[] +
* candidates[] +
* candidates[]
)
candidates = candidates.sort_values(, ascending=)
candidates
Query Existing Drugs for Top Targets
def find_drugs_for_targets(target_genes, max_per_gene=5):
"""
Find existing drugs targeting top candidate genes.
"""
from tooluniverse import ToolUniverse
tu = ToolUniverse()
drug_results = {}
for gene in target_genes[:10]:
print(f"Searching drugs for {gene}...")
result = tu.run_one_function({
"name": "DGIdb_query_gene",
"arguments": {"gene_symbol": gene}
})
if 'data' in result and 'matchedTerms' in result['data']:
matches = result['data']['matchedTerms']
if len(matches) > 0:
interactions = matches[0].get('interactions', [])
drugs = []
for interaction in interactions[:max_per_gene]:
drugs.append({
'drug_name': interaction.get('drugName', 'Unknown'),
'interaction_type': interaction.get('interactionTypes', ['Unknown'])[0],
'source': interaction.get('source', )
})
drug_results[gene] = drugs
drug_results
Phase 8: Report Generation
Comprehensive CRISPR Screen Report
def generate_crispr_report(gene_scores, enrichment_results, drug_targets,
output_file="crispr_screen_report.md"):
"""
Generate comprehensive CRISPR screen analysis report.
"""
with open(output_file, 'w') as f:
f.write("# CRISPR Screen Analysis Report\n\n")
f.write("## Summary\n\n")
f.write(f"- **Total genes analyzed**: {len(gene_scores)}\n")
f.write(f"- **Essential genes** (LFC < -1): {(gene_scores['mean_lfc'] < -1).sum()}\n")
f.write(f"- **Non-essential genes** (LFC > -0.5): {(gene_scores['mean_lfc'] > -0.5).sum()}\n\n")
f.write("## Top 20 Essential Genes\n\n")
f.write("| Rank | Gene | Mean LFC | sgRNAs | Score |\n")
f.write("|------|------|----------|--------|-------|\n")
for idx, (gene, row) in enumerate(gene_scores.head(20).iterrows(), 1):
f.write(f"| {idx} | {gene} | {row['mean_lfc']:.3f} | {int(row['n_sgrnas'])} | {row['score']:.2f} |\n")
f.write()
enrichment_results:
f.write()
db, results enrichment_results.items():
f.write()
f.write()
f.write()
_, row results.head().iterrows():
term = row.get(, )
pval = row.get(, )
adj_pval = row.get(, )
genes = row.get(, )
f.write()
f.write()
drug_targets :
f.write()
f.write()
f.write()
idx, (gene, row) (drug_targets.head().iterrows(), ):
ess = row[]
expr = row.get(, )
drugs = (row.get(, ))
priority = row[]
f.write()
f.write()
f.write()
f.write()
f.write()
f.write()
f.write()
f.write()
()
output_file
Advanced Use Cases
Use Case 1: Genome-Wide Essentiality Screen
counts, meta = load_sgrna_counts("genome_wide_screen.txt")
design = create_design_matrix(
sample_names=['T0_1', 'T0_2', 'T14_1', 'T14_2', 'T14_3'],
conditions=['baseline', 'baseline', 'treatment', 'treatment', 'treatment']
)
qc_results = qc_sgrna_distribution(counts)
filtered_counts, filtered_mapping = filter_low_count_sgrnas(counts, meta['sgrna_to_gene'])
norm_counts, size_factors = normalize_counts(filtered_counts, method='median')
lfc, control_mean, treatment_mean = calculate_lfc(norm_counts, design)
gene_scores = mageck_gene_scoring(lfc, filtered_mapping, method='rra')
enrichment = enrich_essential_genes(gene_scores, top_n=100)
report = generate_crispr_report(gene_scores, enrichment, None)
Use Case 2: Synthetic Lethality Screen (KRAS)
counts_wt, meta_wt = load_sgrna_counts("kras_wildtype_screen.txt")
counts_mut, meta_mut = load_sgrna_counts("kras_mutant_screen.txt")
gene_scores_wt = mageck_gene_scoring(lfc_wt, filtered_mapping_wt)
gene_scores_mut = mageck_gene_scoring(lfc_mut, filtered_mapping_mut)
sl_hits = detect_synthetic_lethality(gene_scores_wt, gene_scores_mut)
print(f"Identified {len(sl_hits)} synthetic lethal candidates with KRAS mutation")
print(sl_hits.head(10))
drug_targets = prioritize_drug_targets(sl_hits)
Use Case 3: Drug Target Discovery Pipeline
gene_scores = mageck_gene_scoring(lfc, filtered_mapping)
highly_essential = gene_scores[gene_scores['mean_lfc'] < -1.5]
drug_targets = prioritize_drug_targets(highly_essential, expression_data=tumor_expression)
drug_candidates = find_drugs_for_targets(drug_targets.index.tolist())
report = generate_crispr_report(gene_scores, None, drug_targets)
print(f"Identified {len(drug_candidates)} druggable targets with {sum(len(v) for v in drug_candidates.values())} total drug candidates")
Use Case 4: Integration with Expression Data
rna_results = pd.read_csv("deseq2_results.csv", index_col=0)
integrated = gene_scores.merge(
rna_results[['log2FoldChange', 'padj']],
left_index=True,
right_index=True,
how='inner'
)
targets = integrated[
(integrated['mean_lfc'] < -1) &
(integrated['log2FoldChange'] > 1) &
(integrated['padj'] < 0.05)
]
print(f"Identified {len(targets)} genes essential and overexpressed in disease")
ToolUniverse Tool Integration
Key Tools Used:
PubMed_search - Literature search for gene essentiality
Enrichr_submit_genelist - Pathway enrichment submission
Enrichr_get_results - Retrieve enrichment results
DGIdb_query_gene - Drug-gene interactions and druggability
STRING_get_network - Protein interaction networks
KEGG_get_pathway - Pathway visualization
Expression Integration:
GEO_get_dataset - Download expression data
ArrayExpress_get_experiment - Alternative expression source
Variant Integration:
ClinVar_query_gene - Known pathogenic variants
gnomAD_get_gene - Population allele frequencies
Best Practices
-
sgRNA Design Quality: Ensure library uses validated sgRNA designs (e.g., Brunello, Avana libraries)
-
Replicates: Minimum 2 biological replicates per condition; 3+ preferred
-
Sequencing Depth: Aim for 500-1000 reads per sgRNA at T0; 200+ at final timepoint
-
Reference Genes: Include positive (essential) and negative (non-essential) control genes
-
Timepoint Selection: Balance cell doublings (14-21 days) vs. sgRNA dropout
-
Normalization: Use median ratio normalization for count data (more robust than CPM)
-
Multiple Testing: Apply FDR correction when calling essential genes (padj < 0.05)
-
Validation: Validate top hits with orthogonal methods (siRNA, small molecule inhibitors)
-
Context Matters: Gene essentiality is context-dependent (cell line, tissue, genetic background)
-
Druggability: Essential genes are not always druggable; check DGIdb early in prioritization
Troubleshooting
Problem: Low library representation (many zero-count sgRNAs)
- Solution: Increase sequencing depth; check for PCR biases in library prep
Problem: High Gini coefficient (skewed distribution)
- Solution: Optimize PCR cycles; consider using unique molecular identifiers (UMIs)
Problem: No strong essential genes detected
- Solution: Check timepoint (may be too early); verify cell viability; confirm sgRNA cutting efficiency
Problem: Too many essential genes (>500)
- Solution: Timepoint may be too late; adjust LFC threshold; check for batch effects
Problem: Discordant sgRNAs for same gene
- Solution: Check for off-target effects; verify sgRNA sequences; consider removing outlier sgRNAs
References
- Li W, et al. (2014) MAGeCK enables robust identification of essential genes from genome-scale CRISPR/Cas9 knockout screens. Genome Biology
- Hart T, et al. (2015) High-Resolution CRISPR Screens Reveal Fitness Genes and Genotype-Specific Cancer Liabilities. Cell
- Meyers RM, et al. (2017) Computational correction of copy number effect improves specificity of CRISPR-Cas9 essentiality screens. Nature Genetics
- Tsherniak A, et al. (2017) Defining a Cancer Dependency Map. Cell (DepMap)
Quick Start
import pandas as pd
from tooluniverse import ToolUniverse
counts, meta = load_sgrna_counts("sgrna_counts.txt")
design = create_design_matrix(['T0_1', 'T0_2', 'T14_1', 'T14_2'],
['baseline', 'baseline', 'treatment', 'treatment'])
filtered_counts, filtered_mapping = filter_low_count_sgrnas(counts, meta['sgrna_to_gene'])
norm_counts, _ = normalize_counts(filtered_counts)
lfc, _, _ = calculate_lfc(norm_counts, design)
gene_scores = mageck_gene_scoring(lfc, filtered_mapping)
enrichment = enrich_essential_genes(gene_scores, top_n=100)
drug_targets = prioritize_drug_targets(gene_scores)
report = generate_crispr_report(gene_scores, enrichment, drug_targets)