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
Next
Script Ncbi Utils
scripts/utils/file_discovery.py
Python·189 lines·5 KB
15class FileInfo:
16 """Information about a discovered file."""
17 path: str
18 name: str
19 stem: str
20 extension: str
21 size: int
22 file_type: str # fastq, bam, cram
23
24 def __repr__(self):
25 return f"FileInfo({self.name}, type={self.file_type})"
26
27
28# Supported file extensions by type
29EXTENSIONS = {
30 "fastq": [".fastq.gz", ".fq.gz", ".fastq", ".fq"],
31 "bam": [".bam"],
32 "cram": [".cram"],
33}
34
35# Index file extensions
36INDEX_EXTENSIONS = {
37 "bam": [".bam.bai", ".bai"],
38 "cram": [".cram.crai", ".crai"],
39}
40
41
42def discover_files(
43 directory: str,
44 file_type: str = "fastq",
45 follow_symlinks: bool = True
46) -> List[FileInfo]:
47 """
48 Recursively discover files of specified type.
49
50 Args:
51 directory: Root directory to search
52 file_type: One of 'fastq', 'bam', 'cram'
53 follow_symlinks: Whether to follow symbolic links
54
55 Returns:
56 List of FileInfo objects sorted by path
57 """
58 if file_type not in EXTENSIONS:
59 raise ValueError(f"Unknown file type: {file_type}. Supported: {list(EXTENSIONS.keys())}")
60
61 directory = os.path.abspath(directory)
62 if not os.path.isdir(directory):
63 raise ValueError(f"Not a directory: {directory}")
64
65 extensions = EXTENSIONS[file_type]
66 files = []
67 seen_paths = set() # Avoid duplicates from symlinks
68
69 for root, _, filenames in os.walk(directory, followlinks=follow_symlinks):
70 for filename in filenames:
71 # Check each extension
72 for ext in extensions:
73 if filename.lower().endswith(ext.lower()):
74 full_path = os.path.join(root, filename)
75
76 # Resolve to handle symlinks
77 try:
78 real_path = os.path.realpath(full_path)
79 except OSError:
80 real_path = full_path
81
82 if real_path in seen_paths:
83 continue
84 seen_paths.add(real_path)
85
86 try:
87 size = os.path.getsize(full_path)
88 except OSError:
89 size = 0
90
91 # Extract stem (remove extension)
92 stem = filename
93 for e in extensions:
94 if stem.lower().endswith(e.lower()):
95 stem = stem[:-len(e)]
96 break
97
98 files.append(FileInfo(
99 path=full_path,
100 name=filename,
101 stem=stem,
102 extension=ext,
103 size=size,
104 file_type=file_type
105 ))
106 break # Found matching extension, no need to check others
107
108 return sorted(files, key=lambda f: f.path)
109
110
111def count_files_by_type(directory: str) -> Dict[str, int]:
112 """
113 Count files by type in directory.
114
115 Args:
116 directory: Directory to scan
117
118 Returns:
119 Dict mapping file_type to count
120 """
121 counts = {}
122 for file_type in EXTENSIONS:
123 try:
124 files = discover_files(directory, file_type)
125 counts[file_type] = len(files)
126 except (ValueError, PermissionError):
127 counts[file_type] = 0
128 return counts
129
130
131def find_index_file(alignment_file: str) -> Optional[str]:
132 """
133 Find index file for a BAM or CRAM file.
134
135 Args:
136 alignment_file: Path to BAM or CRAM file
137
138 Returns:
139 Path to index file if found, None otherwise
140 """
141 path = Path(alignment_file)
142
143 # Determine file type
144 if path.suffix.lower() == ".bam":
145 index_exts = INDEX_EXTENSIONS["bam"]
146 elif path.suffix.lower() == ".cram":
147 index_exts = INDEX_EXTENSIONS["cram"]
148 else:
149 return None
150
151 # Try common index file patterns
152 for ext in index_exts:
153 # Pattern: file.bam.bai or file.bai
154 if ext.startswith(".bam") or ext.startswith(".cram"):
155 candidate = Path(str(path) + ext.split(".")[-1])
156 else:
157 candidate = path.with_suffix(ext)
158
159 if candidate.exists():
160 return str(candidate)
161
162 # Also try: file.bam -> file.bam.bai
163 candidate = Path(str(path) + "." + ext.lstrip("."))
164 if candidate.exists():
165 return str(candidate)
166
167 return None
168
169
170def detect_input_type(directory: str) -> str:
171 """
172 Auto-detect predominant input file type in directory.
173
174 Prioritizes: FASTQ > BAM > CRAM
175
176 Args:
177 directory: Directory to scan
178
179 Returns:
180 Detected file type ('fastq', 'bam', or 'cram')
181 """
182 counts = count_files_by_type(directory)
183
184 # Prioritize by preference
185 for file_type in ["fastq", "bam", "cram"]:
186 if counts.get(file_type, 0) > 0:
187 return file_type
188
189 return "fastq" # Default