Skills › Scientific Agent Skills › bioservices › batch_id_converter.py
batch_id_converter.py
333 lines · skills/bioservices/scripts/batch_id_converter.py at 5f2bc03
1#!/usr/bin/env python3
2"""
3Batch Identifier Converter
5This script converts multiple identifiers between biological databases
6using UniProt's mapping service. Supports batch processing with
7automatic chunking and error handling.
9Usage:
10 python batch_id_converter.py INPUT_FILE --from DB1 --to DB2 [options]
12Examples:
13 python batch_id_converter.py uniprot_ids.txt --from UniProtKB_AC-ID --to KEGG
14 python batch_id_converter.py gene_ids.txt --from GeneID --to UniProtKB --output mapping.csv
15 python batch_id_converter.py ids.txt --from UniProtKB_AC-ID --to Ensembl --chunk-size 50
17Input file format:
18 One identifier per line (plain text)
20Common database codes:
21 UniProtKB_AC-ID - UniProt accession/ID
22 KEGG - KEGG gene IDs
23 GeneID - NCBI Gene (Entrez) IDs
24 Ensembl - Ensembl gene IDs
25 Ensembl_Protein - Ensembl protein IDs
26 RefSeq_Protein - RefSeq protein IDs
27 PDB - Protein Data Bank IDs
28 HGNC - HGNC identifiers (not gene symbols)
29"""
31import sys
32import argparse
33import csv
34import time
35from pathlib import Path
36from bioservices import UniProt
39# Common database code mappings
40DATABASE_CODES = {
41 'uniprot': 'UniProtKB_AC-ID',
42 'uniprotkb': 'UniProtKB_AC-ID',
43 'kegg': 'KEGG',
44 'geneid': 'GeneID',
45 'entrez': 'GeneID',
46 'ensembl': 'Ensembl',
47 'ensembl_protein': 'Ensembl_Protein',
48 'ensembl_transcript': 'Ensembl_Transcript',
49 'refseq': 'RefSeq_Protein',
50 'refseq_protein': 'RefSeq_Protein',
51 'pdb': 'PDB',
52 'hgnc': 'HGNC',
53 'mgi': 'MGI',
54 'reactome': 'Reactome',
55 'string': 'STRING',
56 'biogrid': 'BioGRID'
57}
60def normalize_database_code(code, target=False):
61 """Normalize database code to official format."""
62 if target and code.lower() in {"uniprot", "uniprotkb"}:
63 return "UniProtKB"
64 # Try exact match first
65 if code in DATABASE_CODES.values():
66 return code
68 # Try lowercase lookup
69 lowercase = code.lower()
70 if lowercase in DATABASE_CODES:
71 return DATABASE_CODES[lowercase]
73 # Return as-is if not found (may still be valid)
74 return code
77def read_ids_from_file(filename):
78 """Read identifiers from file (one per line)."""
79 print(f"Reading identifiers from {filename}...")
81 ids = []
82 with open(filename, 'r') as f:
83 for line in f:
84 line = line.strip()
85 if line and not line.startswith('#'):
86 ids.append(line)
88 print(f"[OK] Read {len(ids)} identifier(s)")
90 return ids
93def mapping_to_lists(response):
94 """Normalize the 1.16 mapping envelope without losing one-to-many matches.
96 UniProtKB targets are records; KEGG and many other targets are strings.
97 A missing/invalid envelope is a service failure, not an unmapped result.
98 """
99 if not isinstance(response, dict) or not ({"results", "failedIds"} & response.keys()):
100 raise ValueError("UniProt mapping did not return results/failedIds")
101 mapping = {}
102 for row in response.get("results", []):
103 source, target = row["from"], row["to"]
104 if isinstance(target, dict):
105 target = target.get("primaryAccession") or target.get("uniParcId") or target.get("id")
106 if not isinstance(source, str) or not isinstance(target, str) or not target:
107 raise ValueError("Unsupported mapping record; preserve and inspect the raw response")
108 values = mapping.setdefault(source, [])
109 if target not in values:
110 values.append(target)
111 for identifier in response.get("failedIds", []):
112 mapping.setdefault(identifier, [])
113 return mapping
116def batch_convert(ids, from_db, to_db, chunk_size=100, delay=0.5):
117 """Return mapping and unresolved IDs; [] means explicitly unmapped, None failed."""
118 if chunk_size < 1 or chunk_size > 100000 or delay < 0:
119 raise ValueError("chunk_size must be 1..100000 and delay nonnegative")
120 u = UniProt(verbose=False)
121 all_results = {}
122 for i in range(0, len(ids), chunk_size):
123 chunk = ids[i:i + chunk_size]
124 try:
125 result = mapping_to_lists(u.mapping(fr=from_db, to=to_db, query=",".join(chunk)))
126 all_results.update({identifier: result.get(identifier) for identifier in chunk})
127 except Exception as error:
128 print(f"[FAIL] Batch lookup: {error}; retrying individual identifiers")
129 for identifier in chunk:
130 try:
131 result = mapping_to_lists(u.mapping(fr=from_db, to=to_db, query=identifier))
132 all_results[identifier] = result.get(identifier)
133 except Exception as retry_error:
134 print(f"[FAIL] {identifier}: {retry_error}")
135 all_results[identifier] = None
136 if delay:
137 time.sleep(delay)
138 if delay and i + chunk_size < len(ids):
139 time.sleep(delay)
140 failed_ids = [identifier for identifier in dict.fromkeys(ids) if not all_results.get(identifier)]
141 print(f"[OK] Mapped {sum(bool(v) for v in all_results.values())}/{len(all_results)} unique IDs")
142 return all_results, failed_ids
145def save_mapping_csv(mapping, output_file, from_db, to_db):
146 """Save mapping results to CSV."""
147 print(f"\nSaving results to {output_file}...")
149 with open(output_file, 'w', newline='') as f:
150 writer = csv.writer(f)
152 # Header
153 writer.writerow(['Source_ID', 'Source_DB', 'Target_IDs', 'Target_DB', 'Mapping_Status'])
155 # Data
156 for source_id, target_ids in sorted(mapping.items()):
157 if target_ids:
158 target_str = ";".join(target_ids)
159 status = "Success"
160 else:
161 target_str = ""
162 status = "Unmapped" if target_ids == [] else "Failed"
164 writer.writerow([source_id, from_db, target_str, to_db, status])
166 print(f"[OK] Results saved")
169def save_failed_ids(failed_ids, output_file):
170 """Save failed IDs to file."""
171 if not failed_ids:
172 return
174 print(f"\nSaving failed IDs to {output_file}...")
176 with open(output_file, 'w') as f:
177 for id_ in failed_ids:
178 f.write(f"{id_}\n")
180 print(f"[OK] Saved {len(failed_ids)} failed ID(s)")
183def print_mapping_summary(mapping, from_db, to_db):
184 """Print summary of mapping results."""
185 print(f"\n{'='*70}")
186 print("MAPPING SUMMARY")
187 print(f"{'='*70}")
189 total = len(mapping)
190 mapped = len([v for v in mapping.values() if v])
191 failed = total - mapped
193 print(f"\nSource database: {from_db}")
194 print(f"Target database: {to_db}")
195 print(f"\nTotal identifiers: {total}")
196 print(f"Successfully mapped: {mapped} ({mapped/total*100 if total else 0:.1f}%)")
197 print(f"Failed to map: {failed} ({failed/total*100 if total else 0:.1f}%)")
199 # Show some examples
200 if mapped > 0:
201 print(f"\nExample mappings (first 5):")
202 count = 0
203 for source_id, target_ids in mapping.items():
204 if target_ids:
205 target_str = ", ".join(target_ids[:3])
206 if len(target_ids) > 3:
207 target_str += f" ... +{len(target_ids)-3} more"
208 print(f" {source_id} -> {target_str}")
209 count += 1
210 if count >= 5:
211 break
213 # Show multiple mapping statistics
214 multiple_mappings = [v for v in mapping.values() if v and len(v) > 1]
215 if multiple_mappings:
216 print(f"\nMultiple target mappings: {len(multiple_mappings)} ID(s)")
217 print(f" (These source IDs map to multiple target IDs)")
219 print(f"{'='*70}")
222def list_common_databases():
223 """Print list of common database codes."""
224 print("\nCommon Database Codes:")
225 print("-" * 70)
226 print(f"{'Alias':<20} {'Official Code':<30}")
227 print("-" * 70)
229 for alias, code in sorted(DATABASE_CODES.items()):
230 if alias != code.lower():
231 print(f"{alias:<20} {code:<30}")
233 print("-" * 70)
234 print("\nNote: Many other database codes are supported.")
235 print("See UniProt documentation for complete list.")
238def main():
239 """Main conversion workflow."""
240 parser = argparse.ArgumentParser(
241 description="Batch convert biological identifiers between databases",
242 formatter_class=argparse.RawDescriptionHelpFormatter,
243 epilog="""
244Examples:
245 python batch_id_converter.py uniprot_ids.txt --from UniProtKB_AC-ID --to KEGG
246 python batch_id_converter.py ids.txt --from GeneID --to UniProtKB -o mapping.csv
247 python batch_id_converter.py ids.txt --from uniprot --to ensembl --chunk-size 50
249Common database codes:
250 UniProtKB_AC-ID, KEGG, GeneID, Ensembl, Ensembl_Protein,
251 RefSeq_Protein, PDB, HGNC, Reactome
253Use --list-databases to see all supported aliases.
254 """
255 )
256 parser.add_argument("input_file", nargs="?", help="Input file with IDs (one per line)")
257 parser.add_argument("--from", dest="from_db",
258 help="Source database code")
259 parser.add_argument("--to", dest="to_db",
260 help="Target database code")
261 parser.add_argument("-o", "--output", default=None,
262 help="Output CSV file (default: mapping_results.csv)")
263 parser.add_argument("--chunk-size", type=int, default=100,
264 help="Number of IDs per batch (default: 100)")
265 parser.add_argument("--delay", type=float, default=0.5,
266 help="Delay between batches in seconds (default: 0.5)")
267 parser.add_argument("--save-failed", action="store_true",
268 help="Save failed IDs to separate file")
269 parser.add_argument("--list-databases", action="store_true",
270 help="List common database codes and exit")
272 args = parser.parse_args()
274 # List databases and exit
275 if args.list_databases:
276 list_common_databases()
277 sys.exit(0)
279 if not args.input_file or not args.from_db or not args.to_db:
280 parser.error("input_file, --from and --to are required unless --list-databases is used")
281 if not 1 <= args.chunk_size <= 100000 or args.delay < 0:
282 parser.error("--chunk-size must be 1..100000 and --delay nonnegative")
284 print("=" * 70)
285 print("BIOSERVICES: Batch Identifier Converter")
286 print("=" * 70)
288 # Normalize database codes
289 from_db = normalize_database_code(args.from_db)
290 to_db = normalize_database_code(args.to_db, target=True)
292 if from_db != args.from_db:
293 print(f"\nNote: Normalized '{args.from_db}' -> '{from_db}'")
294 if to_db != args.to_db:
295 print(f"Note: Normalized '{args.to_db}' -> '{to_db}'")
297 # Read input IDs
298 try:
299 ids = read_ids_from_file(args.input_file)
300 except Exception as e:
301 print(f"\n[FAIL] Error reading input file: {e}")
302 sys.exit(1)
304 if not ids:
305 print("\n[FAIL] No IDs found in input file")
306 sys.exit(1)
308 # Perform conversion
309 mapping, failed_ids = batch_convert(
310 ids,
311 from_db,
312 to_db,
313 chunk_size=args.chunk_size,
314 delay=args.delay
315 )
317 # Print summary
318 print_mapping_summary(mapping, from_db, to_db)
320 # Save results
321 output_file = args.output or "mapping_results.csv"
322 save_mapping_csv(mapping, output_file, from_db, to_db)
324 # Save failed IDs if requested
325 if args.save_failed and failed_ids:
326 failed_file = str(Path(output_file).with_name(Path(output_file).stem + "_failed.txt"))
327 save_failed_ids(failed_ids, failed_file)
329 print(f"\n[OK] Done!")
332if __name__ == "__main__":
333 main()