Coverage for core / src / sensorkit / data / fits.py: 97%
290 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-02 00:03 +0000
1# SPDX-License-Identifier: Apache-2.0
2import array
3import asyncio
4import io
5from collections.abc import Buffer, Iterable
6from typing import Any, Literal, NamedTuple, Protocol, Self, overload, runtime_checkable
8import numpy as np
9from astropy.io import fits
10from astropy.io.fits.card import UNDEFINED
11from loguru import logger
12from pydantic import BaseModel, Field
14from sensorkit.common.keyword import declare_keyword, is_keyword
15from sensorkit.data.context import Context
16from sensorkit.data.graph import DataFlow, DataOp
18# `array.array` typecodes to numpy dtype names, for cameras whose SDK hands back a
19# flat `array.array` rather than an ndarray. (Platform note: `l`/`L` are at least
20# 32 bits and platform-dependent in width; the mapping assumes the 64-bit form.)
21_ARRAY_TYPECODE_TO_DTYPE = {
22 "b": "int8",
23 "B": "uint8",
24 "h": "int16",
25 "H": "uint16",
26 "i": "int32",
27 "I": "uint32",
28 "l": "int64",
29 "L": "uint64",
30 "f": "float32",
31 "d": "float64",
32}
35@declare_keyword
36class ArrayInfo(BaseModel):
37 """Metadata describing the shape, dtype, and memory order of a raw array buffer."""
38 shape: tuple[int, ...]
39 dtype: str
40 order: Literal["C", "F"] = "C"
42 @overload
43 @classmethod
44 def from_array(cls, source: np.ndarray) -> Self: ...
46 @overload
47 @classmethod
48 def from_array(
49 cls,
50 source: array.array,
51 *,
52 shape: tuple[int, ...],
53 order: Literal["C", "F"] = "C",
54 ) -> Self: ...
56 @classmethod
57 def from_array(
58 cls,
59 source: np.ndarray | array.array,
60 *,
61 shape: tuple[int, ...] | None = None,
62 order: Literal["C", "F"] | None = None,
63 ) -> Self:
64 """Build an `ArrayInfo` describing *source*.
66 For an ndarray, `shape`, `dtype`, and memory `order` are all read from the array
67 itself, because it already knows its own geometry.
69 For an `array.array`, the dtype is derived from its typecode and *shape* is required,
70 because the buffer carries no 2-D geometry.
72 Args:
73 source: The source ndarray, or a flat `array.array` buffer.
74 shape: Image shape as `(rows, cols)`. Required for an `array.array`; rejected for
75 an ndarray.
76 order: Memory order the resulting metadata describes. Applies to an `array.array`
77 (default `"C"`); rejected for an ndarray.
79 Raises:
80 TypeError: The source is neither an ndarray nor an `array.array` of a recognized
81 typecode.
82 ValueError: A `shape` or `order` was given for an ndarray, or `shape` was omitted
83 for an `array.array`.
84 """
85 match source:
86 case np.ndarray():
87 if shape is not None or order is not None:
88 raise ValueError(
89 "shape and order do not apply to an ndarray; it already carries its "
90 "own shape and memory order"
91 )
93 derived_order = (
94 "F" if source.flags.f_contiguous and not source.flags.c_contiguous else "C"
95 )
97 return cls(shape=source.shape, dtype=str(source.dtype), order=derived_order)
99 case array.array():
100 dtype = _ARRAY_TYPECODE_TO_DTYPE.get(source.typecode)
102 if dtype is None:
103 raise TypeError(
104 f"Cannot infer a dtype from an array.array of typecode {source.typecode!r}"
105 )
107 if shape is None:
108 raise ValueError(
109 "shape is required to build ArrayInfo from a flat array.array buffer"
110 )
112 return cls(shape=shape, dtype=dtype, order=order or "C")
114 case _:
115 raise TypeError(f"Cannot infer a dtype from a {type(source).__name__} buffer")
117 @property
118 def bit_length(self):
119 """Return the number of bits per element in the array."""
120 return np.dtype(self.dtype).itemsize * 8
122 def ndarray_from_buffer(self, buffer: Buffer, allow_copy: bool = False) -> np.ndarray:
123 """Interpret *buffer* as an ndarray with this object's shape and dtype."""
124 arr = np.frombuffer(buffer, dtype=np.dtype(self.dtype))
126 if arr.size != np.prod(self.shape):
127 raise ValueError(f"Array size {arr.size} does not match shape {self.shape}")
129 return arr.reshape(self.shape, order=self.order, copy=None if allow_copy else False)
132class ReshapeArray(DataOp):
133 """Convert the input buffer into an ndarray."""
134 op: Literal["reshape_array"] = "reshape_array"
135 array: ArrayInfo
137 async def process(self, incoming: list[DataFlow], outgoing: list[DataFlow]):
138 context, buffer = await incoming[0].receive("buffer")
140 if _info := context.get(ArrayInfo):
141 # TODO: handle pre-existing ArrayInfo
142 pass
144 arr = await asyncio.to_thread(self.array.ndarray_from_buffer, buffer)
145 context.set(self.array)
146 await outgoing[0].send(context, arr)
149# FITS scalar types and 2-tuple (value, comment) form for header cards.
150type FITSCardScalar = str | int | float | bool | complex | None
151type FITSCardValue = FITSCardScalar | tuple[FITSCardScalar, str]
154class FITSCardValueWithComment(NamedTuple):
155 """Model for parsing a FITS card value with a comment.
157 This exists only to enable a dictionary input form in user configuration. Instances
158 will match the bare tuple form `(value, comment)` thus are included in `FITSCardValue`.
159 """
161 value: FITSCardScalar
162 comment: str
165# Type for parsing FITS card input.
166type FITSCardInput = FITSCardValue | FITSCardValueWithComment
169@declare_keyword
170class FITSHeader(dict[str, FITSCardValue]):
171 """A FITS header as a dictionary of keyword-value pairs.
173 Each value is either a bare scalar card value or a `(value, comment)` tuple carrying
174 an associated comment.
175 """
177 @classmethod
178 def from_astropy_header(cls, header: fits.Header) -> Self:
179 """Build a header from an astropy `Header`, keyed by card keyword.
181 This is the foreign-data path, taking values as they appear in the file: a valueless
182 card becomes `None` and a complex-valued card stays a Python `complex`. The result may
183 therefore hold values outside `FITSCardValue`, so it is served as-is rather than
184 validated as a keyword.
186 TODO: Repeated cards (e.g. COMMENT, HISTORY) are not supported here; the last card of
187 a given keyword wins. Supporting multi-valued cards likely means backing this type with
188 an ordered list plus an index rather than a plain dict, which would also open the door
189 to user-facing card reordering.
190 """
191 result = cls()
193 for card in header.cards:
194 if not card.keyword:
195 continue
197 value: Any = None if card.value is UNDEFINED else card.value
198 result[card.keyword] = value
200 return result
202 def write_to(self, header: fits.Header) -> None:
203 """Write these cards into an astropy *header*, preserving comments."""
204 for keyword, card in self.items():
205 header[keyword] = card
207 def resolve_from_context(
208 self,
209 context: Context,
210 keyword: str,
211 card: FITSCardValue,
212 *,
213 suppress_missing: bool = False,
214 ) -> None:
215 """Resolve *card* and set it on this header under *keyword* unless it resolves to None."""
216 resolved = resolve_fits_card(card, context, suppress_missing=suppress_missing)
218 if resolved is not None:
219 self[keyword] = resolved
222@runtime_checkable
223class FITSCardProvider(Protocol):
224 """Protocol for objects that can provide FITS header cards."""
226 def get_fits_cards(self) -> Iterable[tuple[str, FITSCardValue]]: ...
229# Color filter array pattern, named by the 2x2 filter tile at the image origin. A monochrome
230# sensor, or one whose driver has already debayered into color planes, has no pattern.
231type BayerPattern = Literal["RGGB", "BGGR", "GRBG", "GBRG"]
234@declare_keyword
235class ImageInfo(BaseModel):
236 """Structure and pixel encoding of a captured image.
238 Describes what is needed to interpret the pixels: the buffer they arrive in, how color
239 is encoded, and where on the sensor they were read from.
241 Composes: `ArrayInfo`
243 Attributes:
244 array: The raw buffer this image arrives in.
245 bayer: Color filter array pattern, or None for monochrome or already-debayered data.
246 bayer_offset: `(x, y)` shift of the bayer pattern, which a subframe read from an odd
247 origin displaces. Meaningless without `bayer`.
248 binning: `(x, y)` on-sensor binning factors.
249 origin: `(x, y)` start position of the subframe on the sensor.
250 pixel_size: `(x, y)` pixel pitch in microns, after binning.
251 top_down: Whether row 0 is the top of the image. Defaults to False (bottom-up).
252 """
254 array: ArrayInfo
255 bayer: BayerPattern | None = None
256 bayer_offset: tuple[int, int] = (0, 0)
257 binning: tuple[int, int] = (1, 1)
258 origin: tuple[int, int] = (0, 0)
259 pixel_size: tuple[float, float] | None = None
260 top_down: bool = False
262 @property
263 def width(self) -> int:
264 """Return the number of columns in the image."""
265 return self.array.shape[-1]
267 @property
268 def height(self) -> int:
269 """Return the number of rows in the image."""
270 return self.array.shape[-2]
272 def get_fits_cards(self) -> Iterable[tuple[str, FITSCardValue]]:
273 """Yield the FITS cards describing this image.
275 The structural and scaling keywords (SIMPLE, BITPIX, NAXIS, NAXISn, BSCALE, BZERO) are
276 owned by the writer: astropy derives them from the physical array itself — including the
277 half-range BZERO for unsigned data — so they are not yielded here.
278 """
279 yield "XBINNING", (self.binning[0], "Binning factor in X")
280 yield "YBINNING", (self.binning[1], "Binning factor in Y")
281 yield "XORGSUBF", (self.origin[0], "Subframe origin in X")
282 yield "YORGSUBF", (self.origin[1], "Subframe origin in Y")
283 yield "ROWORDER", ("TOP-DOWN" if self.top_down else "BOTTOM-UP", "Row order of the data")
285 if self.bayer is not None:
286 yield "BAYERPAT", (self.bayer, "Color filter array pattern")
287 yield "XBAYROFF", (self.bayer_offset[0], "Bayer pattern offset in X")
288 yield "YBAYROFF", (self.bayer_offset[1], "Bayer pattern offset in Y")
290 if self.pixel_size is not None:
291 yield "XPIXSZ", (self.pixel_size[0], "Pixel pitch in X [micron]")
292 yield "YPIXSZ", (self.pixel_size[1], "Pixel pitch in Y [micron]")
294 def composed_keywords(self) -> Iterable[object]:
295 yield self.array
298def resolve_fits_card(
299 card: FITSCardValue,
300 context: Context,
301 *,
302 suppress_missing: bool = False,
303) -> FITSCardValue | None:
304 """Resolve a FITS card's value against *context*.
306 The card is either a bare value or a `(value, comment)` tuple whose value is resolved
307 and whose comment is a literal kept verbatim. A string value is resolved with
308 `Context.resolve`: `=expr` evaluates a Python expression, `f"..."` and text containing
309 `{...}` interpolate, and anything else is literal text. A non-string scalar (int, float,
310 or bool, e.g. a YAML number or boolean) is a literal carried through unresolved.
312 Args:
313 card: The card to resolve, as a bare scalar value or a `(value, comment)` tuple.
314 context: The context the value is resolved against.
315 suppress_missing: When true, a reference to a name absent from the context resolves
316 to `None` instead of raising `NameError`.
318 Returns:
319 The resolved card (a bare value, or a `(value, comment)` tuple when a comment is
320 present), or `None` when the value resolves to `None`.
321 """
322 if isinstance(card, tuple):
323 source, comment = card[0], card[1]
324 else:
325 source, comment = card, ""
327 if isinstance(source, str):
328 value = context.resolve(source, default=None) if suppress_missing else context.resolve(source)
329 else:
330 # A non-string scalar (int/float/bool) is a literal value; pass it through unresolved.
331 value = source
333 if value is None:
334 return None
336 return FITSCardValueWithComment(value, comment) if comment else value
339class BuildFITSHeader(DataOp):
340 """Build a FITS header from a context.
342 This DataOp constructs FITS header cards by resolving card values against the context
343 and applying various transformations. Card values are resolved with `resolve_fits_card`
344 (via `Context.resolve`): `=expr` evaluates an expression, `f"..."` and `{...}` forms
345 interpolate, and anything else is literal text. A card carrying a comment is written
346 either as `[value, comment]` or as `{value: ..., comment: ...}`. It supports multiple
347 ways to populate header keywords, applied *in order* as shown below:
349 - **Include** — `FITSCardProvider` objects looked up from the context by keyword, either
350 a given set of keywords or every such provider in the context.
351 - **Rename** — change keyword names.
352 - **Remove** — delete specific keywords.
353 - **Define** — set keywords that are not already set.
354 - **Option** — set keywords only if values resolve to non-`None` (suppresses `NameError`).
355 - **Mutate** — replace the value of keywords that already exist.
357 Attributes:
358 op: Operation type identifier, fixed as `"fits_header"`.
359 include: Set of keyword keys, or `"all"`. Keywords that implement the `FITSCardProvider`
360 protocol are queried for FITS cards, and they are included in the header. Invalid
361 keywords or keywords that do not implement the protocol result in warnings. If "all"
362 is specified, the inclusion order is defined by the configured import order.
363 rename: Dictionary mapping old FITS keyword names to new names.
364 remove: Set of FITS keyword names to remove from the header.
365 define: Dictionary mapping FITS keyword names to card values resolved against the
366 context; only keywords not already present are set.
367 option: Like `define`, but the keyword is added only if the value resolves to a
368 non-`None` value, and a reference to an absent context name is suppressed.
369 mutate: Dictionary mapping existing FITS keyword names to card values that replace
370 their current values (keywords absent from the header are ignored).
371 """
373 op: Literal["fits_header"] = "fits_header"
374 include: set[str] | Literal["all"] = Field(default_factory=set)
375 rename: dict[str, str] = Field(default_factory=dict)
376 remove: set[str] = Field(default_factory=set)
377 define: dict[str, FITSCardInput] = Field(default_factory=dict)
378 option: dict[str, FITSCardInput] = Field(default_factory=dict)
379 mutate: dict[str, FITSCardInput] = Field(default_factory=dict)
381 async def process(self, incoming: list[DataFlow], outgoing: list[DataFlow]):
382 context, buffer = await incoming[0].receive("buffer")
384 # Respect a pre-existing FITSHeader keyword, otherwise start a fresh one.
385 header = context.get(FITSHeader)
387 if header is None:
388 header = FITSHeader()
390 self._populate(header, context)
391 context.set(header)
393 # Pass the buffer through unchanged; this op only builds the header.
394 await outgoing[0].send(context, buffer)
396 def _populate(self, header: FITSHeader, context: Context):
397 """Apply each population step against *header* in documented order."""
398 # Include: add cards from each context-resolved `FITSCardProvider`.
399 self._apply_includes(header, context)
401 # Rename: move a keyword's card to a new name.
402 for old, new in self.rename.items():
403 if old in header:
404 header[new] = header.pop(old)
406 # Remove: drop keywords entirely.
407 for keyword in self.remove:
408 header.pop(keyword, None)
410 # Define: set keywords that are not already present.
411 for keyword, card in self.define.items():
412 if keyword not in header:
413 header.resolve_from_context(context, keyword, card)
415 # Option: set only when the value resolves to non-None; missing names are ignored.
416 for keyword, card in self.option.items():
417 header.resolve_from_context(context, keyword, card, suppress_missing=True)
419 # Mutate: replace the value of keywords that already exist.
420 for keyword, card in self.mutate.items():
421 if keyword in header:
422 header.resolve_from_context(context, keyword, card)
424 def _apply_includes(self, header: FITSHeader, context: Context):
425 """Add cards from each context-resolved `FITSCardProvider`."""
426 match self.include:
427 case "all":
428 include = (
429 key for key, value in context.items() if isinstance(value, FITSCardProvider)
430 )
431 case _:
432 include = self.include
434 for key in include:
435 if not is_keyword(key):
436 logger.warning(f"fits_header include '{key}' is not a declared keyword")
437 continue
439 provider = context.get(key)
441 # Includes are optional and will not warn if a valid keyword is not present.
442 if provider is None:
443 continue
445 if not isinstance(provider, FITSCardProvider):
446 logger.warning(f"fits_header include '{key}' is not a FITSCardProvider")
447 continue
449 for kw, card in provider.get_fits_cards():
450 header[kw] = card
453class ArrayToFITS(DataOp):
454 """Converts an input array to FITS format."""
455 op: Literal["array_to_fits"] = "array_to_fits"
456 header: dict[str, FITSCardInput] = Field(default_factory=dict)
458 async def process(self, incoming: list[DataFlow], outgoing: list[DataFlow]):
459 context, buffer = await incoming[0].receive("buffer")
461 # This op requires the ArrayInfo keyword to be present.
462 array = context[ArrayInfo]
464 # Reshape the input buffer. This should always be zero-copy in cases where ReshapeArray
465 # preceded this op, because if memory reallocation was required, it will have already been
466 # done. Note in that case the buffer is actually already an ndarray, but we cannot assume
467 # that.
468 image_ndarray: np.ndarray = await asyncio.to_thread(array.ndarray_from_buffer, buffer)
470 # Here we stipulate that this is a 2D image. FITS data cube support could be added here.
471 if image_ndarray.ndim != 2:
472 raise RuntimeError("array_to_fits only supports 2D arrays")
474 # Set the ImageInfo context if it is not already present.
475 if context.get(ImageInfo) is None:
476 context.set(ImageInfo(array=array))
478 # Respect a pre-existing FITSHeader keyword, otherwise start a fresh one, then add this
479 # op's own keywords by evaluating the input patterns against the context.
480 fits_header = context.get(FITSHeader)
482 if fits_header is None:
483 fits_header = FITSHeader()
485 for kw, card in self.header.items():
486 fits_header.resolve_from_context(context, kw, card)
488 context.set(fits_header)
490 # Build primary HDU. astropy populates SIMPLE, NAXIS, and NAXIS{n} from the
491 # array shape (NAXIS1 = fastest-varying axis = number of columns).
492 primary_hdu = fits.PrimaryHDU(image_ndarray)
494 # The accumulated FITSHeader is the source of truth for the written header.
495 fits_header.write_to(primary_hdu.header)
497 # Build the output FITS bytes and send it along the graph.
498 hdul = fits.HDUList([primary_hdu])
499 bio = io.BytesIO()
500 hdul.writeto(bio)
502 await outgoing[0].send(context, bio.getvalue())
505class CompressFITS(DataOp):
506 """Tile-compress a FITS image buffer using the FITS tiled-image convention.
508 Wraps the primary image HDU in a CompImageHDU. If the image data exceeds
509 32 bits per pixel (e.g. int64, float64), compression falls back to GZIP_1
510 (lossless for all dtypes) and logs a warning, since RICE_1 only supports
511 ≤32-bit values.
512 """
513 op: Literal["compress_fits"] = "compress_fits"
514 algorithm: Literal["RICE_1", "GZIP_1", "GZIP_2", "HCOMPRESS_1"] = Field(
515 "RICE_1",
516 description="Tile compression algorithm. RICE_1 is lossless for ≤32-bit integer data.",
517 )
518 quantize_level: float = Field(
519 0.0,
520 description=(
521 "Floating-point quantization level. 0 disables quantization (lossless). "
522 "Values > 0 enable lossy quantization for floating-point data."
523 ),
524 )
526 async def process(self, incoming: list[DataFlow], outgoing: list[DataFlow]):
527 context, buffer = await incoming[0].receive("buffer")
528 compressed = await asyncio.to_thread(self._compress, buffer)
529 await outgoing[0].send(context, compressed)
531 def _compress(self, buffer: bytes) -> bytes:
532 with fits.open(io.BytesIO(buffer)) as hdul:
533 data = hdul[0].data
535 # CompImageHDU derives the compressed HDU's structure from data, so the
536 # source header must contribute observational cards only. Without strip(),
537 # astropy carries the primary array's SIMPLE over as ZSIMPLE and the
538 # reconstructed image header declares itself a primary array rather than
539 # an IMAGE extension.
540 header = hdul[0].header.copy()
541 header.strip()
543 algorithm = self.algorithm
545 # Safety check: RICE_1 and HCOMPRESS_1 only support ≤32-bit data.
546 if data.dtype.itemsize > 4 and algorithm in ("RICE_1", "HCOMPRESS_1"):
547 logger.warning(
548 f"compress_fits: image dtype {data.dtype} exceeds 32 bits; "
549 f"falling back to GZIP_1 for lossless compression."
550 )
551 algorithm = "GZIP_1"
553 compressed_hdu = fits.CompImageHDU(
554 data=data,
555 header=header,
556 compression_type=algorithm,
557 quantize_level=self.quantize_level,
558 )
559 new_hdul = fits.HDUList([fits.PrimaryHDU(), compressed_hdu])
561 bio = io.BytesIO()
562 new_hdul.writeto(bio)
563 return bio.getvalue()
566class ContextFromFITS(DataOp):
567 """DataGraph node that populates the context with values from a FITS header."""
568 op: Literal["context_from_fits"] = "context_from_fits"
569 keyword_map: dict = Field(default_factory=dict)
571 async def process(self, incoming: list[DataFlow], outgoing: list[DataFlow]):
572 from astropy.io import fits
574 # Receive the FITS data as a buffer
575 context, buffer = await incoming[0].receive("buffer")
577 # Read FITS header from buffer
578 with io.BytesIO(buffer) as bio:
579 hdul = fits.open(bio)
580 header = hdul[0].header
582 # Map FITS keywords to context based on keyword map
583 for meta_key, fits_key in self.keyword_map.items():
584 if fits_key in header:
585 context.set_value(meta_key, header[fits_key])
587 # Pass through the original buffer
588 await outgoing[0].send(context, buffer)
593@declare_keyword
594class DarkInfo(BaseModel):
595 """Information about dark frame subtraction."""
596 applied: bool = False
597 dark_path: str | None = None
598 dark_exposure: float | None = None
599 image_exposure: float | None = None
602class ApplyDark(DataOp):
603 """Subtract a dark frame from the input image, matching by exposure time.
605 Scans a directory for master dark FITS files and selects the one with the
606 closest exposure time to the input image. Dark frames should have EXPTIME
607 or EXPOSURE in their headers.
608 """
609 op: Literal["apply_dark"] = "apply_dark"
610 dark_directory: str = Field(..., description="Directory containing master dark FITS files")
611 pattern: str = Field("*.fits", description="Glob pattern for dark frame files")
613 _dark_library: dict[float, tuple[str, np.ndarray]] | None = None
615 async def process(self, incoming: list[DataFlow], outgoing: list[DataFlow]):
616 context, buffer = await incoming[0].receive("buffer")
618 # Load dark library on first use
619 if self._dark_library is None:
620 self._dark_library = await asyncio.to_thread(self._load_dark_library)
622 # Apply dark subtraction
623 result_buffer, dark_info = await asyncio.to_thread(
624 self._apply_dark, buffer
625 )
627 context.set(dark_info)
628 await outgoing[0].send(context, result_buffer)
630 def _load_dark_library(self) -> dict[float, tuple[str, np.ndarray]]:
631 """Load all dark frames from directory, indexed by exposure time."""
632 import glob
633 import os
635 library: dict[float, tuple[str, np.ndarray]] = {}
636 pattern_path = os.path.join(self.dark_directory, self.pattern)
638 for filepath in glob.glob(pattern_path):
639 try:
640 with fits.open(filepath) as hdul:
641 header = hdul[0].header
642 exposure = header.get('EXPTIME') or header.get('EXPOSURE')
643 if exposure is not None:
644 data = hdul[0].data.astype(np.float64)
645 library[float(exposure)] = (filepath, data)
646 except Exception:
647 continue
649 return library
651 def _find_closest_dark(self, target_exposure: float) -> tuple[float, str, np.ndarray] | None:
652 """Find the dark frame with closest exposure time."""
653 if not self._dark_library:
654 return None
656 exposures = list(self._dark_library.keys())
657 closest_exp = min(exposures, key=lambda x: abs(x - target_exposure))
658 filepath, data = self._dark_library[closest_exp]
659 return closest_exp, filepath, data
661 def _apply_dark(self, buffer: bytes) -> tuple[bytes, DarkInfo]:
662 """Subtract matching dark frame from image data."""
663 with io.BytesIO(buffer) as bio_in:
664 with fits.open(bio_in) as hdul:
665 image_data = hdul[0].data.astype(np.float64)
666 header = hdul[0].header.copy()
668 # Get image exposure time
669 img_exposure = header.get('EXPTIME') or header.get('EXPOSURE')
671 if img_exposure is None:
672 # No exposure info, pass through unchanged
673 return buffer, DarkInfo(applied=False, image_exposure=None)
675 # Find closest matching dark
676 match = self._find_closest_dark(float(img_exposure))
677 if match is None:
678 # No darks available, pass through unchanged
679 return buffer, DarkInfo(applied=False, image_exposure=float(img_exposure))
681 dark_exposure, dark_path, dark_data = match
683 # Subtract dark frame
684 calibrated = image_data - dark_data
686 # Update header
687 header['DARKFILE'] = (dark_path, 'Dark frame applied')
688 header['DARKEXP'] = (dark_exposure, 'Dark frame exposure time')
690 # Create output FITS
691 primary_hdu = fits.PrimaryHDU(calibrated.astype(image_data.dtype), header=header)
692 hdul_out = fits.HDUList([primary_hdu])
694 bio_out = io.BytesIO()
695 hdul_out.writeto(bio_out)
697 dark_info = DarkInfo(
698 applied=True,
699 dark_path=dark_path,
700 dark_exposure=dark_exposure,
701 image_exposure=float(img_exposure),
702 )
704 return bio_out.getvalue(), dark_info