Skills › Scientific Agent Skills › flowio › inspect_fcs.py
inspect_fcs.py
465 lines · skills/flowio/scripts/inspect_fcs.py at 5f2bc03
1#!/usr/bin/env python3
2"""Inspect local FCS files with FlowIO without network or environment access.
4Metadata-only parsing is the default. Event statistics are opt-in because
5FlowIO loads each dataset into memory and ``as_array`` allocates another array.
6Statistics first perform a metadata-only pass so an estimated array-memory
7limit can be enforced before DATA is loaded.
8"""
10from __future__ import annotations
12import argparse
13import hashlib
14import json
15import struct
16import sys
17from pathlib import Path
18from typing import Any, Iterator, Optional
20import numpy as np
21import flowio
22from flowio import FlowData
23from flowio.exceptions import FlowIOException, MultipleDataSetsError
26DEFAULT_MAX_BYTES = 2_000_000_000
27DEFAULT_MAX_ARRAY_BYTES = 512_000_000
28DEFAULT_MAX_DATASETS = 128
31def nonnegative_int(value: str) -> int:
32 parsed = int(value)
33 if parsed < 0:
34 raise argparse.ArgumentTypeError("must be zero or greater")
35 return parsed
38def positive_int(value: str) -> int:
39 parsed = int(value)
40 if parsed <= 0:
41 raise argparse.ArgumentTypeError("must be greater than zero")
42 return parsed
45def build_parser() -> argparse.ArgumentParser:
46 parser = argparse.ArgumentParser(
47 description=(
48 "Inspect FCS structure and channels as JSON. Event data is not "
49 "loaded unless --stats is supplied."
50 ),
51 formatter_class=argparse.ArgumentDefaultsHelpFormatter,
52 )
53 parser.add_argument("input", type=Path, help="local FCS file")
54 parser.add_argument(
55 "--output",
56 type=Path,
57 help="write JSON to a new file instead of stdout; refuses overwrite",
58 )
59 parser.add_argument(
60 "--include-text",
61 action="store_true",
62 help="include all normalized TEXT metadata (may contain identifiers)",
63 )
64 parser.add_argument(
65 "--include-analysis",
66 action="store_true",
67 help="include all normalized ANALYSIS metadata (may contain identifiers)",
68 )
69 parser.add_argument(
70 "--stats",
71 action="store_true",
72 help="load events and calculate per-channel finite-value statistics",
73 )
74 parser.add_argument(
75 "--raw",
76 action="store_true",
77 help="with --stats, use encoded values (preprocess=False)",
78 )
79 parser.add_argument(
80 "--sha256",
81 action="store_true",
82 help="calculate a SHA-256 checksum (requires a full additional file read)",
83 )
84 parser.add_argument(
85 "--max-bytes",
86 type=nonnegative_int,
87 default=DEFAULT_MAX_BYTES,
88 help="reject larger inputs before parsing; use 0 to disable",
89 )
90 parser.add_argument(
91 "--max-array-bytes",
92 type=nonnegative_int,
93 default=DEFAULT_MAX_ARRAY_BYTES,
94 help=(
95 "with --stats, reject a dataset whose estimated float64 array "
96 "exceeds this size; use 0 to disable"
97 ),
98 )
99 parser.add_argument(
100 "--max-datasets",
101 type=positive_int,
102 default=DEFAULT_MAX_DATASETS,
103 help="stop suspicious or unexpectedly long multi-dataset chains",
104 )
105 parser.add_argument(
106 "--null-channel",
107 action="append",
108 default=[],
109 metavar="PNN",
110 help=(
111 "PnN label to classify as null; repeat for multiple labels. "
112 "The event column is retained."
113 ),
114 )
115 parser.add_argument(
116 "--ignore-offset-error",
117 action="store_true",
118 help="allow a known off-by-one DATA end offset and review warnings",
119 )
120 parser.add_argument(
121 "--ignore-offset-discrepancy",
122 action="store_true",
123 help="use TEXT offsets when HEADER/TEXT offsets differ",
124 )
125 parser.add_argument(
126 "--use-header-offsets",
127 action="store_true",
128 help="use HEADER DATA offsets and suppress offset-discrepancy errors",
129 )
130 return parser
133def sha256_file(path: Path) -> str:
134 digest = hashlib.sha256()
135 with path.open("rb") as handle:
136 for chunk in iter(lambda: handle.read(1024 * 1024), b""):
137 digest.update(chunk)
138 return digest.hexdigest()
141def iter_datasets(
142 path: Path,
143 *,
144 max_datasets: int,
145 null_channel_list: list[str],
146 ignore_offset_error: bool,
147 ignore_offset_discrepancy: bool,
148 use_header_offsets: bool,
149) -> Iterator[tuple[int, int, FlowData]]:
150 """Yield metadata-only datasets for safe offset and memory inspection."""
152 offset = 0
153 file_size = path.stat().st_size
155 for dataset_index in range(max_datasets):
156 flow = FlowData(
157 path,
158 ignore_offset_error=ignore_offset_error,
159 ignore_offset_discrepancy=ignore_offset_discrepancy,
160 use_header_offsets=use_header_offsets,
161 only_text=True,
162 nextdata_offset=offset,
163 null_channel_list=null_channel_list,
164 )
165 yield dataset_index, offset, flow
167 relative_offset = int(flow.text.get("nextdata", "0"))
168 if relative_offset == 0:
169 return
170 if relative_offset < 0:
171 raise MultipleDataSetsError(
172 "input contains a negative relative offset to the next dataset"
173 )
175 next_offset = offset + relative_offset
176 if next_offset <= offset:
177 raise MultipleDataSetsError(
178 "input contains a non-increasing multi-dataset offset"
179 )
180 if next_offset >= file_size:
181 raise MultipleDataSetsError(
182 "next dataset offset is outside the input file"
183 )
184 offset = next_offset
186 raise MultipleDataSetsError(
187 f"input exceeds the --max-datasets limit ({max_datasets})"
188 )
191def channel_kind(flow: FlowData, array_index: int) -> str:
192 if flow.pnn_labels[array_index] in flow.null_channels:
193 return "null"
194 if array_index == flow.time_index:
195 return "time"
196 if array_index in flow.scatter_indices:
197 return "scatter"
198 if array_index in flow.fluoro_indices:
199 return "fluorescence"
200 return "other"
203def channel_records(flow: FlowData) -> list[dict[str, Any]]:
204 records = []
205 for array_index, pnn in enumerate(flow.pnn_labels):
206 parameter_number = array_index + 1
207 metadata = flow.channels[parameter_number]
208 records.append(
209 {
210 "array_index": array_index,
211 "parameter_number": parameter_number,
212 "pnn": pnn,
213 "pns": flow.pns_labels[array_index],
214 "kind": channel_kind(flow, array_index),
215 "pne": list(metadata["pne"]),
216 "png": metadata["png"],
217 "pnr": metadata["pnr"],
218 }
219 )
220 return records
223def finite_statistics(
224 events: np.ndarray,
225 labels: list[str],
226) -> list[dict[str, Any]]:
227 records = []
228 for array_index, label in enumerate(labels):
229 values = events[:, array_index]
230 finite_mask = np.isfinite(values)
231 finite_values = values[finite_mask]
232 # A direct sum can overflow even when every value and the true mean
233 # are finite (FCS supports double-precision DATA). Scale first.
234 mean = None
235 if finite_values.size:
236 magnitude = float(np.max(np.abs(finite_values)))
237 mean = (
238 float(np.clip(np.mean(finite_values / magnitude), -1.0, 1.0))
239 * magnitude
240 if magnitude
241 else 0.0
242 )
243 records.append(
244 {
245 "array_index": array_index,
246 "pnn": label,
247 "finite_count": int(finite_mask.sum()),
248 "nan_count": int(np.isnan(values).sum()),
249 "positive_infinity_count": int(np.isposinf(values).sum()),
250 "negative_infinity_count": int(np.isneginf(values).sum()),
251 "minimum": (
252 float(finite_values.min()) if finite_values.size else None
253 ),
254 "maximum": (
255 float(finite_values.max()) if finite_values.size else None
256 ),
257 "mean": mean,
258 }
259 )
260 return records
263def dataset_record(
264 dataset_index: int,
265 offset: int,
266 flow: FlowData,
267 *,
268 include_text: bool,
269 include_analysis: bool,
270 stats: bool,
271 raw: bool,
272 estimated_array_bytes: int,
273) -> dict[str, Any]:
274 record: dict[str, Any] = {
275 "dataset_index": dataset_index,
276 "byte_offset": offset,
277 "fcs_version": flow.version,
278 "data_type": flow.data_type,
279 "event_count": flow.event_count,
280 "channel_count": flow.channel_count,
281 "estimated_array_bytes": estimated_array_bytes,
282 "channels": channel_records(flow),
283 "metadata_summary": {
284 "text_key_count": len(flow.text),
285 "analysis_key_count": len(flow.analysis),
286 "has_acquisition_date": "date" in flow.text,
287 "has_instrument": "cyt" in flow.text,
288 "has_spillover": (
289 "spillover" in flow.text or "spill" in flow.text
290 ),
291 "nextdata": flow.text.get("nextdata", "0"),
292 },
293 }
295 if include_text:
296 record["text"] = flow.text
297 if include_analysis:
298 record["analysis"] = flow.analysis
300 if stats:
301 expected_values = flow.event_count * flow.channel_count
302 if flow.events is None or len(flow.events) != expected_values:
303 actual_values = None if flow.events is None else len(flow.events)
304 raise ValueError(
305 f"dataset {dataset_index}: DATA has {actual_values} values, "
306 f"but $TOT * $PAR declares {expected_values}; "
307 "refusing inconsistent event counts before as_array()"
308 )
309 events = flow.as_array(preprocess=not raw)
310 record["event_semantics"] = (
311 "decoded DATA values reshaped; preprocess=False; integer range masks retained"
312 if raw
313 else "gain/log/time scaled; uncompensated; ungated"
314 )
315 record["array_shape"] = list(events.shape)
316 record["statistics"] = finite_statistics(events, flow.pnn_labels)
317 else:
318 record["event_semantics"] = "not loaded; metadata-only inspection"
320 return record
323def inspect_file(args: argparse.Namespace) -> dict[str, Any]:
324 path = args.input.expanduser()
325 if not path.exists():
326 raise FileNotFoundError(path)
327 if not path.is_file():
328 raise ValueError(f"input is not a regular file: {path}")
330 size_bytes = path.stat().st_size
331 if args.max_bytes and size_bytes > args.max_bytes:
332 raise ValueError(
333 f"input is {size_bytes} bytes, above --max-bytes={args.max_bytes}"
334 )
336 datasets = []
337 for dataset_index, offset, metadata_flow in iter_datasets(
338 path,
339 max_datasets=args.max_datasets,
340 null_channel_list=args.null_channel,
341 ignore_offset_error=args.ignore_offset_error,
342 ignore_offset_discrepancy=args.ignore_offset_discrepancy,
343 use_header_offsets=args.use_header_offsets,
344 ):
345 if metadata_flow.event_count < 0 or metadata_flow.channel_count <= 0:
346 raise ValueError("$TOT must be nonnegative and $PAR must be positive")
347 if metadata_flow.text.get("mode", "").upper() != "L":
348 raise ValueError("only list-mode ($MODE=L) DATA is supported")
349 if metadata_flow.data_type.upper() not in {"I", "F", "D"}:
350 raise ValueError("only integer, float, and double DATA are supported")
351 estimated_array_bytes = (
352 metadata_flow.event_count * metadata_flow.channel_count * 8
353 )
354 if (
355 args.stats
356 and args.max_array_bytes
357 and estimated_array_bytes > args.max_array_bytes
358 ):
359 raise ValueError(
360 f"dataset {dataset_index} needs an estimated "
361 f"{estimated_array_bytes} bytes for as_array(), above "
362 f"--max-array-bytes={args.max_array_bytes}"
363 )
365 flow = metadata_flow
366 if args.stats:
367 flow = FlowData(
368 path,
369 ignore_offset_error=args.ignore_offset_error,
370 ignore_offset_discrepancy=args.ignore_offset_discrepancy,
371 use_header_offsets=args.use_header_offsets,
372 only_text=False,
373 nextdata_offset=offset,
374 null_channel_list=args.null_channel,
375 )
377 datasets.append(
378 dataset_record(
379 dataset_index,
380 offset,
381 flow,
382 include_text=args.include_text,
383 include_analysis=args.include_analysis,
384 stats=args.stats,
385 raw=args.raw,
386 estimated_array_bytes=estimated_array_bytes,
387 )
388 )
390 report: dict[str, Any] = {
391 "schema_version": "1.0",
392 "flowio_version": flowio.__version__,
393 "file_name": path.name,
394 "size_bytes": size_bytes,
395 "parse_options": {
396 "metadata_only": not args.stats,
397 "ignore_offset_error": args.ignore_offset_error,
398 "ignore_offset_discrepancy": args.ignore_offset_discrepancy,
399 "use_header_offsets": args.use_header_offsets,
400 "max_bytes": args.max_bytes,
401 "max_array_bytes": args.max_array_bytes,
402 "max_datasets": args.max_datasets,
403 "null_channels": args.null_channel,
404 },
405 "dataset_count": len(datasets),
406 "datasets": datasets,
407 }
408 if args.sha256:
409 report["sha256"] = sha256_file(path)
410 return report
413def write_report(report: dict[str, Any], output: Optional[Path]) -> None:
414 text = json.dumps(
415 report,
416 indent=2,
417 ensure_ascii=False,
418 allow_nan=False,
419 )
420 if output is None:
421 sys.stdout.write(text + "\n")
422 return
424 output = output.expanduser()
425 with output.open("x", encoding="utf-8") as handle:
426 handle.write(text + "\n")
429def main(argv: Optional[list[str]] = None) -> int:
430 parser = build_parser()
431 args = parser.parse_args(argv)
432 if args.raw and not args.stats:
433 parser.error("--raw requires --stats")
435 if args.output is not None:
436 try:
437 if args.output.expanduser().resolve() == args.input.expanduser().resolve():
438 parser.error("--output must not overwrite the input FCS file")
439 except OSError:
440 pass
442 try:
443 report = inspect_file(args)
444 write_report(report, args.output)
445 except (
446 FlowIOException,
447 EOFError,
448 FileExistsError,
449 IndexError,
450 KeyError,
451 MemoryError,
452 NotImplementedError,
453 OSError,
454 OverflowError,
455 struct.error,
456 TypeError,
457 ValueError,
458 ) as error:
459 print(f"inspect_fcs: {error}", file=sys.stderr)
460 return 2
461 return 0
464if __name__ == "__main__":
465 raise SystemExit(main())