torchsky.map¶
Stability: core. Names on this page follow the deprecation policy in contributing.
Lean mapmaking facade.
Guides: coadds, cubes, healpix-mapmaking
This reference is generated from the package source. Names are grouped by
the module that defines them; import every name from the public
namespace torchsky.map unless the group text says otherwise.
Top-level namespace¶
torchsky.map.plan ¶
Validated, content-addressed map execution plans.
_PLAN_BACKGROUND_MODES
module-attribute
¶
_PLAN_BACKGROUND_MODES = ("none", "per-exposure", "overlap")
_PLAN_FIELD_PRESETS
module-attribute
¶
_PLAN_FIELD_PRESETS = (
"fast",
"balanced",
"spectral",
"operator_constrained",
)
_PLAN_PRODUCTS
module-attribute
¶
_PLAN_PRODUCTS = (
"SCI",
"WHT",
"IVAR",
"DQ",
"CTX",
"MASK",
"NEXP",
"NCORR",
)
__all__
module-attribute
¶
__all__ = [
"MapPlanV2",
"build_map_input_identity",
"load_map_plan",
"verify_map_input_identity",
]
MapPlanV2
dataclass
¶
MapPlanV2(payload: Mapping[str, Any])
_require_integer ¶
_require_integer(
value: Any,
label: str,
*,
positive: bool = False,
nonnegative: bool = False,
) -> None
_require_number ¶
_require_number(
value: Any,
label: str,
*,
positive: bool = False,
nonnegative: bool = False,
) -> None
_validate_input_identity ¶
_validate_input_identity(
identity: Mapping[str, Any], label: str
) -> None
_validate_optional_identity_pair ¶
_validate_optional_identity_pair(
item: Mapping[str, Any],
path_key: str,
identity_key: str,
) -> None
build_map_input_identity ¶
build_map_input_identity(
path: str | Path, *, hdu: int | str | None = 0
) -> dict[str, Any]
Build a replay identity without parsing FITS inside torchsky.
verify_map_input_identity ¶
verify_map_input_identity(
identity: Mapping[str, Any],
) -> None
Reject a missing, replaced, or modified input recorded by a plan.
Cube¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
coadd_cube ¶
coadd_cube(
observations: Sequence[Observation | IFUObservation],
target: CubeTarget
| Target
| Mapping[str, Any]
| WCS
| None = None,
*,
resample: ResampleConfig | None = None,
stack: StackConfig | None = None,
kernel: MapResamplingKernel | None = None,
pixfrac: float | None = None,
semantics: MapValueSemantics | None = None,
method: MapStackingMethod | None = None,
spectral_kernel: Literal[
"linear", "drizzle", "emsm", "msm", "nearest"
]
| None = None,
spectral_pixfrac: float | None = None,
emsm_scale: float | None = None,
spatial_roi: Any | None = None,
spectral_roi: Any | None = None,
resolving_power: Any | None = None,
response_chunk_size: int | None = None,
max_response_entries: int | None = None,
footprint_tolerance: float | None = None,
footprint_max_depth: int | None = None,
footprint_order: int | None = None,
spectral_axis: SpectralAxis
| Sequence[float]
| Tensor
| None = None,
wavelengths: SpectralAxis
| Sequence[float]
| Tensor
| None = None,
spectral_bin_edges: Sequence[float]
| Tensor
| None = None,
spectral_unit: str = "Angstrom",
spectral_type: str = "WAVE",
background_estimator: Callable[
[Tensor, Tensor | None], Tensor
]
| None = None,
device: device | str | None = None,
) -> CubeResult
Coadd filter images or IFU slices into an image-plane spectral cube.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
observations
|
Sequence[Observation | IFUObservation]
|
Filter-image or IFU observations to combine. |
required |
target
|
CubeTarget | Target | Mapping[str, Any] | WCS | None
|
Spatial and optional spectral target geometry. |
None
|
resample
|
ResampleConfig | None
|
Structured spatial and spectral resampling configuration. |
None
|
stack
|
StackConfig | None
|
Structured stacking configuration. |
None
|
spectral_axis
|
SpectralAxis | Sequence[float] | Tensor | None
|
Output spectral-axis definition. |
None
|
spectral_bin_edges
|
Sequence[float] | Tensor | None
|
Explicit output spectral-bin edges. |
None
|
spectral_unit
|
str
|
Unit label for a constructed output spectral axis. |
'Angstrom'
|
spectral_type
|
str
|
FITS-style spectral coordinate type. |
'WAVE'
|
device
|
device | str | None
|
Optional compute device. |
None
|
Returns:
| Type | Description |
|---|---|
CubeResult
|
Coadded cube, weights, coverage, mask, and target metadata. |
Examples:
>>> from torchsky.map import coadd_cube
>>> callable(coadd_cube)
True
HEALPix coadd¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
coadd_healpix ¶
coadd_healpix(
observations: Sequence[Observation],
nside: int,
*,
nest: bool = False,
sparse: bool = False,
nside_coverage: int = 32,
kernel: Literal[
"bilinear",
"nearest",
"bicubic",
"lanczos2",
"lanczos3",
"lanczos4",
] = "bilinear",
sky: SkyConfig | None = None,
outliers: OutlierConfig | None = None,
device: device | str | None = None,
reduction_chunk_size: int = 262144,
) -> HealpixResult
Coadd 2D WCS observations directly onto a dense or sparse HEALPix grid.
reduction_chunk_size bounds sampled contributions retained before each
incremental reduction. Sparse accumulation retains merged output support;
dense accumulation writes each chunk directly into the output tensors.
Values are surface brightness. Each exposure is combined in one of two
regimes, chosen from the cell size L (nside2resol) and the image
pixel size p (square root of the central pixel's solid angle):
L <= support(kernel) * p: cells are sampled at their centres withkernel(support 1 for nearest, 2 bilinear, 4 bicubic/lanczos2, 6 lanczos3, 8 lanczos4 pixels).- otherwise, or when the footprint contains no cell centre: each cell gets
the exact overlap integral
sum_i A_ip w_i d_iandsum_i A_ip w_i, withA_ipthe area of image pixeliinside cellpin units of the cell area (drizzle's convention atpixfrac = 1). Partially covered cells get the mean over the covered part and a weight scaled by the covered fraction;kernelis not used.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
observations
|
Sequence[Observation]
|
Image observations to project and combine. |
required |
nside
|
int
|
Output HEALPix resolution parameter. |
required |
nest
|
bool
|
Use nested rather than ring pixel ordering. |
False
|
sparse
|
bool
|
Return sparse coverage rather than a full-sky tensor. |
False
|
nside_coverage
|
int
|
Sparse-map coverage resolution. |
32
|
kernel
|
Literal['bilinear', 'nearest', 'bicubic', 'lanczos2', 'lanczos3', 'lanczos4']
|
Image-plane sampling kernel. |
'bilinear'
|
sky
|
SkyConfig | None
|
Optional background-estimation configuration. |
None
|
outliers
|
OutlierConfig | None
|
Optional outlier-rejection configuration. |
None
|
device
|
device | str | None
|
Optional compute device. |
None
|
reduction_chunk_size
|
int
|
Maximum sampled contributions per reduction. |
262144
|
Returns:
| Type | Description |
|---|---|
HealpixResult
|
Dense or sparse HEALPix data, weights, coverage, and metadata. |
Examples:
>>> from torchsky.map import coadd_healpix
>>> callable(coadd_healpix)
True
resample_healpix ¶
resample_healpix(
healpix_map: Tensor | HealpixResult | Any,
target: Target | Mapping[str, Any] | WCS,
*,
nside: int | None = None,
nest: bool = False,
method: Literal[
"bilinear", "nearest", "harmonic"
] = "bilinear",
device: device | str | None = None,
) -> MapResult
Project a dense or sparse HEALPix map (or HealpixResult) onto a 2D target WCS.
Ifu¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
build_ifu_cube ¶
build_ifu_cube(
observations: Sequence[IFUObservation],
target: CubeTarget,
**plan_options: Any,
) -> CubeResult
Execute one frozen physical IFU plan for science and all side products.
plan_ifu_cube_response ¶
plan_ifu_cube_response(
observations: Sequence[IFUObservation],
target: CubeTarget,
*,
kernel: IFUKernel = "drizzle",
pixfrac: float = 1.0,
spectral_pixfrac: float = 1.0,
spatial_roi: ScaleModel | None = None,
spectral_roi: ScaleModel | None = None,
resolving_power: ScaleModel | None = None,
emsm_scale: float = 1.0,
chunk_size: int = 65536,
max_response_entries: int | None = None,
footprint_tolerance: float = 0.02,
footprint_max_depth: int = 10,
device: device | str | None = None,
) -> IFUCubeResponse
Freeze one sparse physical response used by all IFU cube products.
Image plane coadd¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
aperture_uncertainty ¶
aperture_uncertainty(
observations: Sequence[Observation],
result: MapResult,
aperture_weights: Tensor,
*,
resample: ResampleConfig | None = None,
kernel: MapResamplingKernel | None = None,
pixfrac: float | None = None,
semantics: MapValueSemantics | None = None,
spatial_tile_hw: tuple[int, int] | None = None,
target_cache_capacity: int = _DEFAULT_TARGET_CACHE_CAPACITY,
surface_brightness_area_scaling: bool | None = None,
device: device | str | None = None,
) -> ApertureUncertainty
Propagate independent detector noise through one finite map aperture.
blot ¶
blot(
coadd_or_data: MapResult | Tensor,
source_or_target: Observation
| Sequence[Observation]
| Target
| Mapping[str, Any]
| WCS,
*,
target: Observation
| Sequence[Observation]
| Target
| Mapping[str, Any]
| WCS
| None = None,
kernel: Literal[
"bilinear",
"bicubic",
"lanczos2",
"lanczos3",
"lanczos4",
"nearest",
] = "bilinear",
semantics: MapValueSemantics = "surface_brightness",
device: device | str | None = None,
) -> Tensor | list[Tensor]
Inverse-project (blot) a clean coadd back into one or more distorted exposure frames.
Supports both:
- blot(coadd_result, observations, kernel=..., semantics=...)
- blot(coadd_tensor, coadd_wcs_or_header, target=observations, ...)
coadd ¶
coadd(
observations: Sequence[Observation],
target: Target | Mapping[str, Any] | WCS | None = None,
*,
resample: ResampleConfig | None = None,
sky: SkyConfig | None = None,
outliers: OutlierConfig | None = None,
stack: StackConfig | None = None,
kernel: MapResamplingKernel | None = None,
pixfrac: float | None = None,
semantics: MapValueSemantics | None = None,
method: MapStackingMethod | None = None,
clip_sigma: float | None = None,
clip_iterations: int | None = None,
footprint_order: int | None = None,
tile_size: int | tuple[int, int] | None = None,
spatial_tile_hw: tuple[int, int] | None = None,
target_cache_capacity: int = _DEFAULT_TARGET_CACHE_CAPACITY,
surface_brightness_area_scaling: bool | None = None,
output_path: str | None = None,
scratch_dir: str | Path | None = None,
background_estimator: Callable[
[Tensor, Tensor | None], Tensor
]
| None = None,
cross_exposure_sky_corrector: CrossExposureSkyCorrector
| None = None,
device: device | str | None = None,
) -> MapResult
Unified 2D image-plane mosaic / coadd entry point.
Accepts either structured configuration objects (resample, sky, outliers, stack)
or direct keyword arguments (kernel, pixfrac, semantics, method, tile_size).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
observations
|
Sequence[Observation]
|
Image observations to resample and combine. |
required |
target
|
Target | Mapping[str, Any] | WCS | None
|
Explicit target geometry, WCS, header, or |
None
|
resample
|
ResampleConfig | None
|
Structured resampling configuration. |
None
|
sky
|
SkyConfig | None
|
Background-estimation configuration. |
None
|
outliers
|
OutlierConfig | None
|
Outlier-rejection configuration. |
None
|
stack
|
StackConfig | None
|
Structured stacking configuration. |
None
|
device
|
device | str | None
|
Optional compute device. |
None
|
Returns:
| Type | Description |
|---|---|
MapResult
|
Coadded data, weights, coverage, mask, and target metadata. |
Examples:
>>> from torchsky.map import coadd
>>> callable(coadd)
True
coadd_image_plane ¶
coadd_image_plane(
observations: Sequence[ImageObservation],
*,
target_header: dict | None = None,
output_path: str | None = None,
footprint_order: int | None = None,
resampling_kernel: MapResamplingKernel = "bilinear",
value_semantics: MapValueSemantics = "surface_brightness",
stacking_method: MapStackingMethod = "weighted_mean",
drizzle_pixfrac: float = 1.0,
clip_sigma: float = 4.0,
clip_iterations: int = 3,
spatial_tile_hw: tuple[int, int] | None = None,
target_cache_capacity: int = _DEFAULT_TARGET_CACHE_CAPACITY,
surface_brightness_area_scaling: bool | None = None,
device: device | str | None = None,
scratch_dir: str | Path | None = None,
background_estimator: Callable[
[Tensor, Tensor | None], Tensor
]
| None = None,
cross_exposure_sky_corrector: CrossExposureSkyCorrector
| None = None,
propagate_variance: bool = True,
auto_tile: bool = True,
tile_memory_budget_bytes: int | None = None,
_inner_tile: bool = False,
) -> MapResult
Coadd image-plane observations onto a common target grid.
Parameters beyond the resampling and stacking options:
spatial_tile_hw
(height, width) of the tiles the target is processed in. None
(default) chooses the tiling automatically from a memory budget (see
auto_tile); an explicit value is always honoured. Tiling is
incompatible with footprint_order.
auto_tile
When spatial_tile_hw is None, estimate the peak working set of the
single-pass coadd from the kernel, value semantics and target size
(bytes per output pixel measured on the samplers, see
:func:torchsky.map.operators.execution.plan_spatial_tiles) and tile the
target only if it would not fit in tile_memory_budget_bytes. Targets
that fit are not tiled. False forces the single-pass path. Ignored
with footprint_order (the MOC footprint is not defined per tile).
tile_memory_budget_bytes
Memory budget of the automatic planner. The default is
execution.TILE_MEMORY_FRACTION (30%) of the free device memory on CUDA
or of the available host memory (cgroup limit aware) otherwise; the
TORCHSKY_TILE_MEMORY_BUDGET_BYTES environment variable and the
execution.TILE_MEMORY_BUDGET_BYTES constant override that default.
When the target is tiled on an accelerator the full-size accumulators are
host tensors and only tile windows visit the device; the returned maps
(and context words) are moved back to the requested device at the end,
exactly like an untiled coadd. Only if that move would not fit the free
device memory (or runs out of memory) do the affected tensors stay on the
host, and a RuntimeWarning is emitted.
propagate_variance
True (default) propagates the exact diagonal variance of the
resampling (inverse_variance and correlation_factor on the result)
for observations with explicit noise information. False skips that
second sampling pass: data, weight, overlap_count and
context are unchanged, inverse_variance and
correlation_factor are None, and the coadd is faster. Sigma-clipped
stacking weights its samples with the propagated variance, so it still
computes it internally (results unchanged) and only drops the outputs.
coadd_multiband ¶
coadd_multiband(
observations: Sequence[Observation],
target: Target | Mapping[str, Any] | WCS | None = None,
*,
resample: ResampleConfig | None = None,
stack: StackConfig | None = None,
kernel: MapResamplingKernel | None = None,
semantics: MapValueSemantics | None = None,
method: MapStackingMethod | None = None,
footprint_order: int | None = None,
spectral_unit: str = "Angstrom",
spectral_type: str = "WAVE",
device: device | str | None = None,
) -> MultibandResult
Clean multi-band image-plane coadd entry point.
resample ¶
resample(
observation: Observation | Sequence[Observation],
target: Target | Mapping[str, Any] | WCS | None = None,
*,
config: ResampleConfig | None = None,
kernel: MapResamplingKernel | None = None,
pixfrac: float | None = None,
semantics: MapValueSemantics | None = None,
footprint_order: int | None = None,
device: device | str | None = None,
) -> MapResult | list[MapResult]
Resample one or more observations onto a common target WCS grid without stacking.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
observation
|
Observation | Sequence[Observation]
|
One observation or a sequence of observations. |
required |
target
|
Target | Mapping[str, Any] | WCS | None
|
Explicit target geometry, WCS, header, or |
None
|
config
|
ResampleConfig | None
|
Structured resampling configuration. |
None
|
kernel
|
MapResamplingKernel | None
|
Resampling kernel override. |
None
|
pixfrac
|
float | None
|
Drizzle footprint fraction. |
None
|
semantics
|
MapValueSemantics | None
|
Pixel-value semantics used for area scaling. |
None
|
footprint_order
|
int | None
|
Boundary sampling order used while planning a target. |
None
|
device
|
device | str | None
|
Optional compute device. |
None
|
Returns:
| Type | Description |
|---|---|
MapResult | list[MapResult]
|
One map result, or one result per input observation. |
Examples:
>>> from torchsky.map import resample
>>> callable(resample)
True
Observations and targets¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
ApertureUncertainty
dataclass
¶
ApertureUncertainty(
variance: Tensor, diagonal_variance: Tensor
)
CubeResult
dataclass
¶
CubeResult(
spectral_axis: SpectralAxis,
band_metadata: BandMetadataTable,
data: Tensor,
weight: Tensor,
header: dict,
spatial_wcs: WCS,
inverse_variance: Tensor | None = None,
correlation_factor: Tensor | None = None,
overlap_count: Tensor | None = None,
coverage_mocs: tuple[MOC | None, ...] | None = None,
variance_components: Mapping[str, Tensor] = dict(),
dq: Tensor | None = None,
usability_mask: Tensor | None = None,
context: Tensor | None = None,
covariance: Any | None = None,
response_plan: Any | None = None,
provenance: Mapping[str, Any] | None = None,
value_semantics: MapValueSemantics = "surface_brightness",
)
coverage_mocs
class-attribute
instance-attribute
¶
coverage_mocs: tuple[MOC | None, ...] | None = None
value_semantics
class-attribute
instance-attribute
¶
value_semantics: MapValueSemantics = 'surface_brightness'
variance_components
class-attribute
instance-attribute
¶
variance_components: Mapping[str, Tensor] = field(
default_factory=dict
)
aperture_uncertainty ¶
aperture_uncertainty(
spatial_aperture: Tensor,
spectral_weights: Tensor | None = None,
) -> ApertureUncertainty
CubeTarget
dataclass
¶
CubeTarget(
spatial_wcs: WCS
| dict[str, Any]
| PlannedImagePlaneTarget,
spectral_bin_edges: Tensor
| list[float]
| tuple[float, ...],
*,
shape: tuple[int, int] | None = None,
spectral_unit: str = "Angstrom",
spectral_type: str = "WAVE",
value_semantics: MapValueSemantics = "surface_brightness",
)
Physical IFU cube grid: a spatial WCS and explicit spectral bin edges.
HealpixResult
dataclass
¶
HealpixResult(
data: Any,
weight: Any,
nside: int,
nest: bool = False,
sparse: bool = False,
overlap_count: Any | None = None,
coverage_moc: MOC | None = None,
)
Result of direct-to-HEALPix coaddition.
IFUObservation
dataclass
¶
IFUObservation(
sci: Tensor,
detector_to_world: Any,
dq: Tensor | None = None,
variance_components: Mapping[str, Tensor] = dict(),
slice_id: Tensor | None = None,
detector_pixel_footprint: Tensor | None = None,
spectral_unit: str = "Angstrom",
spectral_type: str = "WAVE",
value_semantics: MapValueSemantics = "surface_brightness",
weight: Tensor | float | None = None,
dq_policy: DQPolicy | None = None,
provenance: Mapping[str, Any] = dict(),
aperture_diameter_arcsec: Tensor | float | None = None,
)
In-memory detector product for runtime-independent IFU cube building.
Each detector sample (one pixel of sci) is mapped to (ra, dec,
wavelength) by detector_to_world. The drizzle kernel needs the sky
footprint of every sample, given in one of three ways:
- by default, the detector pixel square mapped through
detector_to_world, which is only meaningful when neighbouring detector pixels are neighbours on the sky. When slice ids are known (slice_idor the transform'sslice_id_map), a sample without a same-slice neighbour along a detector axis has no grid footprint and planning raises; detector_pixel_footprint: explicit per-sample polygons, shape(H, W, V, 2)in detector pixels or(H, W, V, 3)in(ra, dec, wavelength);aperture_diameter_arcsec: a circular aperture (an optical fibre) centred on each sample's sky position, scalar or one value per sample. Its exact area overlap with every output spaxel is used, so fibre-fed (row-stacked) spectra need no pixel-grid adjacency.
aperture_diameter_arcsec
class-attribute
instance-attribute
¶
aperture_diameter_arcsec: Tensor | float | None = None
detector_pixel_footprint
class-attribute
instance-attribute
¶
detector_pixel_footprint: Tensor | None = None
provenance
class-attribute
instance-attribute
¶
provenance: Mapping[str, Any] = field(default_factory=dict)
value_semantics
class-attribute
instance-attribute
¶
value_semantics: MapValueSemantics = 'surface_brightness'
variance_components
class-attribute
instance-attribute
¶
variance_components: Mapping[str, Tensor] = field(
default_factory=dict
)
ImageObservation
dataclass
¶
ImageObservation(
path: str | None = None,
hdu: int | str | None = None,
weight: float = 1.0,
photometric_scale: float = 1.0,
background_offset: float = 0.0,
wavelength: float | None = None,
wavelength_map: Tensor | None = None,
ifu_wcs: Any | None = None,
spectral_min: float | None = None,
spectral_max: float | None = None,
transmission_center: float | None = None,
transmission_sigma: float | None = None,
band_name: str | None = None,
weight_path: str | None = None,
weight_hdu: int | str | None = 0,
mask_path: str | None = None,
mask_hdu: int | str | None = 0,
primary_beam_path: str | None = None,
primary_beam_hdu: int | str | None = None,
weight_is_inverse_variance: bool = False,
data: Tensor | None = None,
header: dict[str, Any] | None = None,
wcs: WCS | dict[str, Any] | Any | None = None,
weight_map: Tensor | None = None,
mask: Tensor | None = None,
dq_path: str | None = None,
dq_hdu: int | str | None = "DQ",
dq: Tensor | None = None,
dq_policy: DQPolicy | None = None,
primary_beam: Tensor | None = None,
weight_type: str = "weight",
weight_thresh: float | None = None,
good_bits: int | None = None,
var_poisson: Tensor | None = None,
var_rnoise: Tensor | None = None,
var_flat: Tensor | None = None,
flux_scale_keyword: str | None = None,
exposure_time_keyword: str | None = None,
gain_keyword: str | None = None,
exposure_time: float | None = None,
head_path: str | None = None,
use_sidecar_head: bool = False,
dcr_zenith_deg: float | None = None,
dcr_parallactic_deg: float = 0.0,
dcr_ref_wavelength: float = 5000.0,
dcr_pressure_hpa: float = 600.0,
dcr_temperature_c: float = 7.0,
tracking_rate: tuple[float, float] | None = None,
mjd_obs: float | None = None,
reference_mjd: float | None = None,
)
Describe one image exposure and its calibration metadata.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | None
|
Optional input path consumed through the torchfits boundary. |
None
|
data
|
Tensor | None
|
Optional in-memory image tensor, used instead of |
None
|
header
|
dict[str, Any] | None
|
FITS-style header mapping for in-memory data. |
None
|
wcs
|
WCS | dict[str, Any] | Any | None
|
Explicit WCS object or header mapping. |
None
|
weight
|
float
|
Scalar exposure weight. |
1.0
|
photometric_scale
|
float
|
Multiplicative flux calibration. |
1.0
|
background_offset
|
float
|
Additive background removed before resampling. |
0.0
|
weight_map
|
Tensor | None
|
Optional per-pixel relative weight tensor (an inverse
variance only when |
None
|
mask
|
Tensor | None
|
Optional Boolean valid-pixel mask. |
None
|
Examples:
>>> import torch
>>> from torchsky.map import Observation
>>> observation = Observation(data=torch.ones(2, 2), header={})
>>> tuple(observation.data.shape)
(2, 2)
weight_is_inverse_variance
class-attribute
instance-attribute
¶
weight_is_inverse_variance: bool = False
MapResult
dataclass
¶
MapResult(
data: Tensor,
weight: Tensor,
header: dict,
inverse_variance: Tensor | None = None,
correlation_factor: Tensor | None = None,
context: Tensor | None = None,
provenance: dict[str, Any] | None = None,
overlap_count: Tensor | None = None,
coverage_moc: MOC | None = None,
observation_mocs: tuple[MOC, ...] | None = None,
var_poisson: Tensor | None = None,
var_rnoise: Tensor | None = None,
var_flat: Tensor | None = None,
dq: Tensor | None = None,
usability_mask: Tensor | None = None,
rejection_log: tuple[dict[str, Any], ...] = (),
dq_policy: DQPolicy | None = None,
)
MultibandResult
dataclass
¶
MultibandResult(
spectral_axis: SpectralAxis,
data: Tensor,
weight: Tensor,
header: dict,
inverse_variance: Tensor | None = None,
correlation_factor: Tensor | None = None,
overlap_count: Tensor | None = None,
coverage_mocs: tuple[MOC | None, ...] | None = None,
band_observation_counts: Tensor | None = None,
)
OutlierConfig
dataclass
¶
OutlierConfig(
enabled: bool = True,
pixfrac: float = 1.0,
combine: str = "median",
blot_kernel: Literal[
"bilinear",
"bicubic",
"lanczos2",
"lanczos3",
"lanczos4",
"nearest",
] = "bilinear",
snr: tuple[float, float] = (5.0, 4.0),
scale: tuple[float, float] = (1.2, 0.7),
min_floor: float = 0.0,
grow: int = 1,
mode: Literal[
"mask", "weight_zero", "flag_only"
] = "mask",
parity_mode: bool = False,
dq_policy: DQPolicy | None = None,
)
Configuration for AstroDrizzle / stcal-compatible cosmic-ray outlier detection.
ResampleConfig
dataclass
¶
ResampleConfig(
kernel: MapResamplingKernel = "bilinear",
pixfrac: float = 1.0,
semantics: MapValueSemantics = "surface_brightness",
area_scaling: bool | None = None,
spectral_kernel: Literal[
"linear", "drizzle", "emsm", "msm", "nearest"
] = "linear",
spectral_pixfrac: float = 1.0,
emsm_scale: float = 1.0,
spatial_roi: Any | None = None,
spectral_roi: Any | None = None,
resolving_power: Any | None = None,
response_chunk_size: int = 65536,
max_response_entries: int | None = None,
footprint_tolerance: float = 0.02,
footprint_max_depth: int = 10,
)
Configuration for single-exposure or mosaic resampling.
spectral_kernel
class-attribute
instance-attribute
¶
spectral_kernel: Literal[
"linear", "drizzle", "emsm", "msm", "nearest"
] = "linear"
SkyConfig
dataclass
¶
SkyConfig(
mode: str = "none",
stat: str = "huber",
mesh_size: int = 64,
smooth_scale: int = 3,
regularization: float = 0.0001,
clip_sigma: float = 3.0,
clip_iterations: int = 3,
subtract: bool = True,
)
Configuration for per-exposure, overlap, planar, or mesh sky matching.
StackConfig
dataclass
¶
StackConfig(
method: MapStackingMethod = "weighted_mean",
clip_sigma: float = 4.0,
clip_iterations: int = 3,
tile_hw: tuple[int, int] | None = None,
tile_size: int | tuple[int, int] | None = None,
footprint_order: int | None = None,
scratch_dir: str | Path | None = None,
)
Configuration for multi-exposure stacking and uncertainty propagation.
Adaptive¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
AdaptiveResampleConfig
dataclass
¶
AdaptiveResampleConfig(
value_semantics: AdaptiveValueSemantics = "surface_brightness",
max_edge_error_px: float = 0.02,
max_jacobian_variation: float = 0.02,
max_source_step_px: float = 0.5,
min_abs_jacobian_determinant: float = 1e-08,
max_depth: int = 8,
planning_chunk_size: int = 4096,
default_chunk_size: int = 65536,
max_plan_entries: int = 50000000,
)
adaptive_reprojection_response ¶
adaptive_reprojection_response(
source_wcs: WCS,
target_wcs: WCS,
source_shape: tuple[int, int],
target_shape: tuple[int, int],
*,
config: AdaptiveResampleConfig | None = None,
dtype: dtype = float32,
device: device | str | None = None,
) -> IndexedResponsePlan
Build one sparse adaptive response for science, variance, DQ, context, and adjoint.
Planning¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
plan_cube ¶
plan_cube(
observations: Sequence[ImageObservation],
*,
spectral_axis: SpectralAxis | None = None,
spectral_bin_edges: Tensor | None = None,
spectral_header: dict | None = None,
spectral_unit: str = "Angstrom",
spectral_type: str = "WAVE",
reference_index: int | None = None,
header_overrides: dict | None = None,
padding: float = 1.0,
max_pixels: int | None = None,
device: device | str | None = None,
) -> PlannedImagePlaneCubeTarget
Plan a 3D spatial-spectral target cube from a set of observations.
Products and stacking¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
BandpassSpec
dataclass
¶
BandpassSpec(
label: str | None = None,
spectral_min: float | None = None,
spectral_max: float | None = None,
transmission_center: float | None = None,
transmission_sigma: float | None = None,
spectral_unit: str = "Angstrom",
passband: Passband | None = None,
)
ProductResult
dataclass
¶
ProductResult(
product: ProductSpec,
data: Tensor,
weight: Tensor | None,
header: dict[str, Any],
overlap_count: Tensor | None = None,
coverage_mocs: tuple[MOC | None, ...] | None = None,
uncertainty: UncertaintyProduct | None = None,
mask: Tensor | None = None,
context: Tensor | None = None,
provenance: Mapping[str, Any] | None = None,
quality: QualityProduct | None = None,
)
coverage_mocs
class-attribute
instance-attribute
¶
coverage_mocs: tuple[MOC | None, ...] | None = None
target_render_spec ¶
target_render_spec(
*, band_index: int = 0, psf_spec: object | None = None
) -> TargetRenderSpec
target_render_specs ¶
target_render_specs(
*, psf_specs: Sequence[object | None] | None = None
) -> tuple[TargetRenderSpec, ...]
ProductSpec
dataclass
¶
ProductSpec(
target_header: dict[str, Any],
bands: tuple[BandpassSpec, ...],
value_semantics: MapValueSemantics = "surface_brightness",
footprint_order: int | None = None,
)
value_semantics
class-attribute
instance-attribute
¶
value_semantics: MapValueSemantics = 'surface_brightness'
target_render_spec ¶
target_render_spec(
*, band_index: int = 0, psf_spec: object | None = None
) -> TargetRenderSpec
target_render_specs ¶
target_render_specs(
*, psf_specs: Sequence[object | None] | None = None
) -> tuple[TargetRenderSpec, ...]
UncertaintyProduct
dataclass
¶
UncertaintyProduct(
variance: Tensor,
kind: MapUncertaintyKind,
components: Mapping[str, Tensor] = dict(),
correlation_factor: Tensor | None = None,
)
Labeled diagonal uncertainty without pretending it is full covariance.
components
class-attribute
instance-attribute
¶
components: Mapping[str, Tensor] = field(
default_factory=dict
)
plan_product ¶
plan_product(
observations: Sequence[ImageObservation],
*,
reference_index: int = 0,
header_overrides: dict[str, Any] | None = None,
bands: Sequence[BandpassSpec] | None = None,
spectral_axis: SpectralAxis | None = None,
spectral_bin_edges: Tensor | None = None,
spectral_unit: str = "Angstrom",
spectral_type: str = "WAVE",
value_semantics: MapValueSemantics = "surface_brightness",
footprint_order: int | None = None,
union_footprint: bool = False,
footprint_padding: float = 1.0,
footprint_tolerance_px: float = 0.25,
footprint_max_depth: int = 20,
max_footprint_pixels: int | None = None,
device: device | str | None = None,
) -> ProductSpec
Plan a map product from a sequence of image-plane observations.
When union_footprint=True, the target WCS is computed automatically from
the union bounding box of all input footprints (like SWarp's automatic
output grid). Set --reference to a FITS file path to override the
spatial target completely.
stack_product ¶
stack_product(
observations: Sequence[ImageObservation],
product: ProductSpec,
*,
resampling_kernel: MapResamplingKernel = "bilinear",
method: str = "coadd",
stacking_method: MapStackingMethod = "weighted_mean",
drizzle_pixfrac: float = 1.0,
clip_sigma: float = 4.0,
clip_iterations: int = 3,
spatial_tile_hw: tuple[int, int] | None = None,
device: device | str | None = None,
scratch_dir: str | Path | None = None,
background_estimator: Callable[
[Tensor, Tensor | None], Tensor
]
| None = None,
cross_exposure_sky_corrector: CrossExposureSkyCorrector
| None = None,
) -> ProductResult
Resample observations and stack them into a requested map product.
Quality¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
DQPolicy
dataclass
¶
DQPolicy(
reject_bits: int | None = None,
propagate_bits: int | None = None,
cosmic_ray_bit: int = int(COSMIC_RAY),
unknown_bit_policy: Literal[
"reject", "preserve", "ignore"
] = "reject",
)
Declare detector DQ rejection, propagation, and cosmic-ray semantics.
unknown_bit_policy
class-attribute
instance-attribute
¶
unknown_bit_policy: Literal[
"reject", "preserve", "ignore"
] = "reject"
DQRejectionLog
dataclass
¶
DQRejectionLog(
observation_index: int,
primary_count: int,
grown_count: int,
rejected_count: int,
cosmic_ray_bit: int,
)
Sky matching¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
detect_outliers ¶
detect_outliers(
observations: Sequence[Observation],
coadd_result: MapResult | None = None,
*,
config: OutlierConfig | None = None,
snr_primary: float | None = None,
snr_neighbor: float | None = None,
scale_primary: float | None = None,
scale_neighbor: float | None = None,
min_floor: float | None = None,
gain: float | None = 1.0,
grow: int | None = None,
kernel: Literal[
"bilinear",
"bicubic",
"lanczos2",
"lanczos3",
"lanczos4",
"nearest",
]
| None = None,
update_masks: bool = True,
return_logs: Literal[False] = False,
device: device | str | None = None,
) -> tuple[list[Tensor], list[Observation]]
detect_outliers(
observations: Sequence[Observation],
coadd_result: MapResult | None = None,
*,
config: OutlierConfig | None = None,
snr_primary: float | None = None,
snr_neighbor: float | None = None,
scale_primary: float | None = None,
scale_neighbor: float | None = None,
min_floor: float | None = None,
gain: float | None = 1.0,
grow: int | None = None,
kernel: Literal[
"bilinear",
"bicubic",
"lanczos2",
"lanczos3",
"lanczos4",
"nearest",
]
| None = None,
update_masks: bool = True,
return_logs: Literal[True],
device: device | str | None = None,
) -> tuple[
list[Tensor],
list[Observation],
tuple[dict[str, int], ...],
]
detect_outliers(
observations: Sequence[Observation],
coadd_result: MapResult | None = None,
*,
config: OutlierConfig | None = None,
snr_primary: float | None = None,
snr_neighbor: float | None = None,
scale_primary: float | None = None,
scale_neighbor: float | None = None,
min_floor: float | None = None,
gain: float | None = 1.0,
grow: int | None = None,
kernel: Literal[
"bilinear",
"bicubic",
"lanczos2",
"lanczos3",
"lanczos4",
"nearest",
]
| None = None,
update_masks: bool = True,
return_logs: bool = False,
device: device | str | None = None,
) -> (
tuple[list[Tensor], list[Observation]]
| tuple[
list[Tensor],
list[Observation],
tuple[dict[str, int], ...],
]
)
Detect cosmic rays and transient outliers via AstroDrizzle / JWST blot + spatial derivative test.
Returns (outlier_masks, updated_observations) where outlier_masks[i] is a boolean
tensor with True at detected cosmic-ray/outlier pixels in exposure i.
When no variance/weight plane is declared, the per-pixel noise falls back to
sqrt(|blotted| / gain + MAD**2) in calibrated data units: a Poisson term
plus the robust residual scatter. gain is the number of electrons per
calibrated data unit (1.0, the default and previous behaviour, means the
data are already in electrons; ADU data need the detector gain in e-/ADU).
gain=None drops the Poisson term. The decisions are invariant under
data -> k * data with gain -> gain / k. The MAD floor is
eps(dtype) * max|finite data|: the smallest scatter distinguishable from
float rounding at the image's own scale (unit-covariant, unlike an absolute
constant). min_floor remains an explicit threshold floor in data units.
match_sky ¶
match_sky(
observations: Sequence[Observation],
*,
target: Target | Mapping[str, Any] | WCS | None = None,
config: SkyConfig | None = None,
mode: Literal[
"constant",
"planar",
"mesh",
"per_exposure",
"local",
"global",
"match",
"global+match",
"overlap_constant",
"overlap_planar",
"overlap_mesh",
"none",
]
| None = None,
mesh_size: int | None = None,
regularization: float | None = None,
apply: bool | None = None,
device: device | str | None = None,
) -> tuple[list[Observation], dict[str, Any]]
Solve and optionally subtract sky background across exposures.
Supports single-exposure mesh background estimation ("per_exposure" / "local"),
JWST/Roman stcal.skymatch global modes ("global", "match", "global+match"),
and global overlap-graph rectification ("constant", "planar", "mesh").
Render targets¶
Lean mapmaking facade.
Implementation modules remain available under torchsky.map.*. The flat
torchsky.map namespace is intentionally limited to promoted coadd/product
entry points and result/specification objects.
TargetRenderSpec
dataclass
¶
TargetRenderSpec(
header: dict[str, Any],
shape: tuple[int, int],
value_semantics: str = "surface_brightness",
band_label: str | None = None,
spectral_min: float | None = None,
spectral_max: float | None = None,
transmission_center: float | None = None,
transmission_sigma: float | None = None,
psf_spec: object | None = None,
spectral_unit: str = "Angstrom",
passband: Passband | None = None,
)