Setting the file. One moment. 22
Validate Data
Tech Debt
62
Recruiting Pipeline
71
Vendor Check
125
Zoom Meeting SDK Web
88
Vendor Review
181
Create An Asset
Video Sdk/web
Script Ncbi Utils
scripts/utils/sample_inference.py
Python·290 lines·8 KB
(
r
'_R1_
\d
{3}
'
,
10
),
# _R1_001 (Illumina standard)
16 (r'_R1[_.]', 8), # _R1. or _R1_
17 (r'\.R1[_.]', 8), # .R1. or .R1_
18 (r'_1[_.]', 5), # _1. or _1_
19 (r'_R1\.f', 6), # _R1.fastq
20 (r'_1\.f', 4), # _1.fastq
21]
22
23R2_PATTERNS = [
24 (r'_R2_\d{3}', 10), # _R2_001 (Illumina standard)
25 (r'_R2[_.]', 8), # _R2. or _R2_
26 (r'\.R2[_.]', 8), # .R2. or .R2_
27 (r'_2[_.]', 5), # _2. or _2_
28 (r'_R2\.f', 6), # _R2.fastq
29 (r'_2\.f', 4), # _2.fastq
30]
31
32# Tumor/normal keywords
33TUMOR_KEYWORDS = [
34 r'\btumou?r\b',
35 r'\bmetastasis\b',
36 r'\bmet\b',
37 r'\bprimary\b',
38 r'\bcancer\b',
39 r'\bmalignant\b',
40 r'[-_]T[-_]',
41 r'[-_]T\d*$',
42 r'^T\d*[-_]',
43]
44
45NORMAL_KEYWORDS = [
46 r'\bnormal\b',
47 r'\bgermline\b',
48 r'\bblood\b',
49 r'\bpbmc\b',
50 r'\bcontrol\b',
51 r'\bhealthy\b',
52 r'\bmatched\b',
53 r'[-_]N[-_]',
54 r'[-_]N\d*$',
55 r'^N\d*[-_]',
56]
57
58# Lane pattern
59LANE_PATTERN = r'[_.]L(\d{3})[_.]'
60
61# Patient/sample extraction patterns
62PATIENT_PATTERNS = [
63 r'^(P\d+)[-_]', # P001_sample
64 r'^(patient\d+)[-_]', # patient1_sample
65 r'^(TCGA-\w+-\w+)', # TCGA format
66 r'^([A-Z]{2,3}\d{3,})[-_]', # AB123_sample
67]
68
69# Replicate patterns
70REPLICATE_PATTERNS = [
71 r'[_.]rep(\d+)', # _rep1, .rep2
72 r'[_.]replicate(\d+)', # _replicate1
73 r'[_.]R(\d+)[_.]', # _R1_ (but not R1/R2 for reads!)
74 r'[-_](\d+)$', # sample_1 (last resort)
75]
76
77
78def extract_sample_info(filepath: str) -> Dict[str, str]:
79 """
80 Extract sample metadata from filepath.
81
82 Args:
83 filepath: Path to sequencing file
84
85 Returns:
86 Dict with: sample, patient, lane (if detectable)
87 """
88 filename = os.path.basename(filepath)
89
90 # Remove extensions
91 stem = filename
92 for ext in ['.fastq.gz', '.fq.gz', '.fastq', '.fq', '.bam', '.cram', '.bai', '.crai']:
93 if stem.lower().endswith(ext):
94 stem = stem[:-len(ext)]
95 break
96
97 info = {}
98
99 # Extract lane
100 lane_match = re.search(LANE_PATTERN, stem)
101 info['lane'] = f"L{lane_match.group(1)}" if lane_match else "L001"
102
103 # Remove lane from stem
104 clean_stem = re.sub(LANE_PATTERN, '_', stem)
105
106 # Remove R1/R2 indicators and everything after
107 for pattern, _ in R1_PATTERNS + R2_PATTERNS:
108 clean_stem = re.sub(pattern + r'.*', '', clean_stem, flags=re.IGNORECASE)
109
110 # Clean up trailing/multiple underscores and dots
111 clean_stem = re.sub(r'[_.-]+$', '', clean_stem)
112 clean_stem = re.sub(r'[_.-]{2,}', '_', clean_stem)
113
114 # Try to extract patient ID
115 for pattern in PATIENT_PATTERNS:
116 match = re.match(pattern, clean_stem, re.IGNORECASE)
117 if match:
118 info['patient'] = match.group(1)
119 break
120
121 # Sample is the cleaned stem
122 info['sample'] = clean_stem if clean_stem else filename.split('.')[0]
123
124 # Default patient to sample if not extracted
125 if 'patient' not in info:
126 info['patient'] = info['sample']
127
128 return info
129
130
131def infer_tumor_normal_status(sample_name: str) -> Optional[int]:
132 """
133 Infer tumor (1) or normal (0) status from sample name.
134
135 Args:
136 sample_name: Sample identifier
137
138 Returns:
139 1 for tumor, 0 for normal, None if cannot determine
140 """
141 name_lower = sample_name.lower()
142
143 # Check tumor indicators
144 for pattern in TUMOR_KEYWORDS:
145 if re.search(pattern, name_lower, re.IGNORECASE):
146 return 1
147
148 # Check normal indicators
149 for pattern in NORMAL_KEYWORDS:
150 if re.search(pattern, name_lower, re.IGNORECASE):
151 return 0
152
153 return None
154
155
156def extract_replicate_number(sample_name: str) -> Optional[int]:
157 """
158 Extract replicate number from sample name.
159
160 Args:
161 sample_name: Sample identifier
162
163 Returns:
164 Replicate number if found, None otherwise
165 """
166 for pattern in REPLICATE_PATTERNS:
167 match = re.search(pattern, sample_name, re.IGNORECASE)
168 if match:
169 try:
170 return int(match.group(1))
171 except ValueError:
172 continue
173 return None
174
175
176def _get_pattern_score(filename: str, patterns: List[Tuple[str, int]]) -> int:
177 """Get highest matching pattern score."""
178 max_score = 0
179 for pattern, score in patterns:
180 if re.search(pattern, filename, re.IGNORECASE):
181 max_score = max(max_score, score)
182 return max_score
183
184
185def _get_sample_key(filepath: str) -> str:
186 """Generate a key for grouping related files."""
187 info = extract_sample_info(filepath)
188 sample = info['sample']
189 lane = info.get('lane', 'L001')
190
191 # Include lane in key for multi-lane samples
192 if lane != "L001":
193 return f"{sample}_{lane}"
194 return sample
195
196
197def match_read_pairs(files) -> Dict[str, Dict]:
198 """
199 Match R1/R2 read pairs using scored pattern matching.
200
201 Args:
202 files: List of FileInfo objects (from file_discovery)
203
204 Returns:
205 Dict mapping sample_key to {'r1': path, 'r2': path, 'info': dict}
206 """
207 # Classify files
208 r1_files = []
209 r2_files = []
210
211 for file in files:
212 filename = file.name if hasattr(file, 'name') else os.path.basename(str(file))
213 filepath = file.path if hasattr(file, 'path') else str(file)
214
215 r1_score = _get_pattern_score(filename, R1_PATTERNS)
216 r2_score = _get_pattern_score(filename, R2_PATTERNS)
217
218 if r2_score > r1_score and r2_score > 0:
219 r2_files.append((filepath, r2_score))
220 elif r1_score > 0:
221 r1_files.append((filepath, r1_score))
222 else:
223 # No clear indicator - assume R1 (single-end or non-standard naming)
224 r1_files.append((filepath, 0))
225
226 # Build pairs by matching sample keys
227 pairs = {}
228
229 # Process R1 files first
230 for r1_path, score in r1_files:
231 key = _get_sample_key(r1_path)
232 info = extract_sample_info(r1_path)
233
234 if key not in pairs:
235 pairs[key] = {
236 'r1': r1_path,
237 'r2': None,
238 'info': info,
239 'score': score
240 }
241 else:
242 # Multiple R1 files for same sample (should not happen)
243 pairs[key]['r1'] = r1_path
244
245 # Match R2 files
246 for r2_path, score in r2_files:
247 key = _get_sample_key(r2_path)
248 info = extract_sample_info(r2_path)
249
250 if key in pairs:
251 pairs[key]['r2'] = r2_path
252 else:
253 # R2 without matching R1
254 pairs[key] = {
255 'r1': None,
256 'r2': r2_path,
257 'info': info,
258 'score': score
259 }
260
261 return pairs
262
263
264def infer_patient_groupings(sample_names: List[str]) -> Dict[str, str]:
265 """
266 Infer patient groupings from sample names.
267
268 Groups samples that share a common prefix pattern.
269
270 Args:
271 sample_names: List of sample identifiers
272
273 Returns:
274 Dict mapping sample_name to patient_id
275 """
276 patient_map = {}
277
278 for sample in sample_names:
279 # Try to find a patient pattern
280 for pattern in PATIENT_PATTERNS:
281 match = re.match(pattern, sample, re.IGNORECASE)
282 if match:
283 patient_map[sample] = match.group(1)
284 break
285
286 if sample not in patient_map:
287 # Default: each sample is its own patient
288 patient_map[sample] = sample
289
290 return patient_map