API reference¶
The reference below is autogenerated from the docstrings in
vmpt/. Most users only need the vmpt.optimizer module — the
rest are the building blocks the Bokeh app drives internally.
vmpt.optimizer¶
The MSA-pointing search + the shutter-collision protection rules.
MSA pointing optimizer.
Searches for (RA, Dec, V3 PA) maximising the number — or weighted flux — of catalog sources that fall in operable, well-centred MSA shutters.
This module is inspired by hMPT — a lightweight Python script for optimizing MSA pointing and roll angles by Daniel Eisenstein, Samuel McCarty, and Zihao Wu (Harvard / CfA), which is itself inspired by ESA’s eMPT (Bonaventura et al. 2023, A&A 672, A40): <https://github.com/zihaowu-astro/hMPT>.
vMPT is not a direct port of hMPT or eMPT. The MSA shutter geometry handling, V2/V3 ↔ (s, d) coordinate transforms, gnomonic projection, centration check, and per-target constraint machinery were all written fresh — the differences are visible side-by-side between this file and hMPT’s. The search algorithm is a simpler variant (single-stage DE refine of the top grid candidates) than hMPT’s. The module composes cleanly with our existing MSA grid (vmpt/data/nirspec_msa_v2v3.npz), CRDS operability loader, and Bokeh UI.
Algorithm summary¶
`radec_to_axy` — vectorised gnomonic projection of source (RA, Dec) onto the MSA aperture plane (ax, ay), with optional differential-velocity-aberration scaling and the PA rotation.
`axy_to_shutter` — per-quadrant CloughTocher2D interpolation maps (ax, ay) → fractional shutter indices (quad, s_row, d_col). Built lazily from the shutter centres vMPT already loads.
`PointingEvaluator.evaluate` — combines the above with the operability mask (incl. a 3-shutter vertical slit constraint), a configurable APT-style centration buffer, and a Gaussian-PSF throughput fraction.
`grid_search` — brute-force ranking over a (ΔRA, ΔDec, ΔPA) cube.
`refine_top` — scipy.optimize.differential_evolution polish of the top-N grid candidates inside a small box.
- class vmpt.optimizer.PointingEvaluator(ra_sources, dec_sources, flux_sources=None, sigma_arcsec: float = 0.06, centration: str = 'UNCONSTRAINED', slit_length: int = 3, operable: ndarray | None = None, *, protect_mask: ndarray | None = None, priorities: ndarray | None = None, weights: ndarray | None = None, disperser: str | None = None, filt: str | None = None, reason: ndarray | None = None, required_lam: ndarray | None = None, no_gap: ndarray | None = None, extend_blue: ndarray | None = None, extend_red: ndarray | None = None, protect: ndarray | None = None, centration_per_target: ndarray | None = None, budget_remaining: ndarray | None = None)[source]¶
Bases:
objectOne catalog × one MSA = a re-usable per-pointing scorer.
Caches the interpolators and operability mask so repeated
evaluate(ra, dec, pa)calls are fast — the grid search runs this hundreds-of-thousands of times.- Parameters:
ra_sources (array-like) – Source positions in degrees.
dec_sources (array-like) – Source positions in degrees.
flux_sources (array-like, optional) – Source fluxes (linear units). Used for the
"flux"objective.sigma_arcsec (float) – Gaussian PSF σ for the throughput integration.
centration (str) – One of the keys in
CENTRATION_BUFFERS.slit_length (int) – Vertical extent of the slitlet (1, 2, 3 or 5 shutters); every shutter in the slitlet must be operable for the source to count.
operable (ndarray, optional) – Pre-loaded (4, 171, 365) operability mask. Loaded lazily if None.
protect_mask (ndarray of bool, optional) – Parallel to
ra_sources; True marks a source whose spectrum must be protected from same-row collisions under the current (disperser, filter). Requiresdisperser+filtto be meaningful. When None or all-False, no protection is applied andevaluatebehaves exactly as before.priorities (ndarray, optional) – Per-source priority / weight, used only to break ties when two protected sources collide (lower priority number wins; on tie, higher weight wins). NaN-tolerant. Falls back to source index order if neither is provided.
weights (ndarray, optional) – Per-source priority / weight, used only to break ties when two protected sources collide (lower priority number wins; on tie, higher weight wins). NaN-tolerant. Falls back to source index order if neither is provided.
disperser (str, optional) – e.g.
"PRISM"and"CLEAR". Only consulted whenprotect_maskflags any source. The V2 half-extent of the spectrum is looked up fromv2_overlap_distance().filt (str, optional) – e.g.
"PRISM"and"CLEAR". Only consulted whenprotect_maskflags any source. The V2 half-extent of the spectrum is looked up fromv2_overlap_distance().reason (ndarray, optional) – (4, 171, 365) operability-reason array from
vmpt.msa.load_operability(). Cells equal to 2 are stuck-open shutters, which act as always-on dispersion sources even when no slitlet is opened there. When provided AND protection is enabled, a protected source landing on a row colliding with any stuck-open shutter is dropped (its spectrum is unavoidably contaminated).
- evaluate(ra_p: float, dec_p: float, pa_v3: float, theta_deg: float = 90.0) tuple[ndarray, ndarray, tuple[ndarray, ndarray, ndarray]][source]¶
Return
(detected_bool, throughput, (quad, s, d))per source.When collision protection was configured at construction time,
detectedis the kept mask — sources dropped by the protection rules are zeroed in bothdetectedandthroughput. The raw pre-drop mask is not returned; useevaluate_with_stats()if you also need the drop count.
- evaluate_with_reasons(ra_p: float, dec_p: float, pa_v3: float, theta_deg: float = 90.0) tuple[ndarray, ndarray, tuple[ndarray, ndarray, ndarray], dict][source]¶
Like
evaluate_with_stats()but returns a dict of per-reason drop counts instead of just a scalar.Keys are the constants in
DROP_REASONS(collision,required_lam,no_gap,extend_blue,extend_red). Values sum to the same scalarevaluate_with_statsreturns.Empty dict when neither protection nor constraints are configured.
- evaluate_with_stats(ra_p: float, dec_p: float, pa_v3: float, theta_deg: float = 90.0) tuple[ndarray, ndarray, tuple[ndarray, ndarray, ndarray], int][source]¶
Like
evaluate()plus ann_droppedcount of sources that landed in operable, centred shutters but were excluded by either the collision-protection rules (v1.2.0+) or the per- target spectral constraints (v1.3.0+) at this pointing.n_dropped == 0when neither protection nor constraints are configured. Seeevaluate_with_reasons()for a per-reason breakdown.
- vmpt.optimizer.axy_to_shutter(axy: ndarray, interpolators: list[dict] | None = None) tuple[ndarray, ndarray, ndarray][source]¶
Return
(quad, s_frac, d_frac)per source.quadis 1–4 for sources inside a quadrant, 0 for outside.s_fracandd_fracare fractional shutter indices (s ∈ [0, 170],d ∈ [0, 364]). NaN wherequad == 0.Vignetting cutoffs at each quadrant’s inner corner mirror hMPT’s find_shutter_from_Axy (lines 437–445 of msa_planner.py).
Uses the degree-
_POLY_DEGREEpolynomial inverse (fast path, default) whenUSE_POLY_INVERSEis set and the fit passed its self-check; otherwise the CloughTocher2DInterpolator (interpolatorsis built lazily only on this fallback path).
- vmpt.optimizer.grid_search(evaluator: PointingEvaluator, ra0: float, dec0: float, pa0: float, *, dra_arcsec: float = 30.0, ddec_arcsec: float = 30.0, dpa_deg: float = 30.0, n_ra: int = 20, n_dec: int = 20, n_pa: int = 20, weights: ndarray | None = None, objective: str = 'number', progress_cb: Callable[[int, int], None] | None = None) dict[source]¶
Brute-force ranking over a (ΔRA, ΔDec, ΔPA) cube.
The ΔRA / ΔDec arguments are in arcseconds; the ΔRA span is automatically scaled by 1/cos(Dec) so the box is roughly square on the sky.
progress_cb(done, total)is invoked at ~2 % increments so the UI can report progress.If any of
dra_arcsec,ddec_arcsec,dpa_degis ≤ 0, that axis is FROZEN at the central value (nis forced to 1, no sweep). This is the convention the UI uses to mean “keep this coordinate at its current value.”
- vmpt.optimizer.radec_to_axy(ra: ndarray, dec: ndarray, ra_p: float, dec_p: float, pa_v3_deg: float, theta_deg: float = 90.0) ndarray[source]¶
Project (RA, Dec) onto MSA aperture coords (ax, ay) in arcsec.
theta_degis the APT differential-velocity-aberration parameter (date-dependent — exported from APT’s XML). The default 90° is the no-correction case used by hMPT during planning, which agrees with APT to ≲ 1 mas at typical pointings.
- vmpt.optimizer.refine_top(evaluator: PointingEvaluator, grid_results: dict, *, n_top: int = 10, dra_arcsec: float = 2.0, ddec_arcsec: float = 2.0, dpa_deg: float = 2.0, maxiter: int = 200, weights: ndarray | None = None, objective: str = 'number', progress_cb: Callable[[int, int], None] | None = None, dedup_tol: tuple[float, float, float] = (0.3, 0.3, 0.05)) dict[source]¶
Differential-evolution polish of the top-N grid candidates.
Each candidate is refined inside a small (dra, ddec, dpa) box. Returns a fresh ranked dict in the same schema as grid_search.
dedup_tolis(arcsec_ra, arcsec_dec, deg_pa): refined solutions within these tolerances of an earlier (higher-scoring) solution are dropped. Without this the user often sees N near-identical rows when the score landscape has a wide plateau.Any of
dra_arcsec,ddec_arcsec,dpa_degthat is ≤ 0 freezes the corresponding axis: scipy’sdifferential_evolutiondoesn’t accept zero-width bounds, so we drop the frozen variable from the optimisation and patch it back in afterwards.
vmpt.msa¶
Loads the per-quadrant shutter centres in V2/V3 and parses CRDS’s
msaoper JSON for operability + stuck-open flags.
NIRSpec MSA shutter grid loader and operability parser.
- vmpt.msa.ensure_current_operability(timeout: float | None = 8.0, force: bool = False, verbose: bool = False) str | None[source]¶
Best-effort refresh of the MSA operability reference at startup.
Make sure the current MOS
msaoperreference — the one the operational CRDS context selects forEXP_TYPE = NRS_MSASPECby USEAFTER, i.e. the same operability APT/MPT uses — is present in the local CRDS cache, fetching it from CRDS if it isn’t. vMPT then loads it on startup, so the failed-/stuck-shutter map stays current.Non-fatal and offline-safe: if
crdsisn’t installed, there’s no network, or the check runs longer thantimeoutseconds, it quietly gives up and vMPT uses whatever is already cached. Runs at most once per process unlessforce=True. SetVMPT_OPERABILITY_AUTOUPDATE=0to disable. Returns the resolved local path, orNone.
vmpt.wavelengths¶
Per-shutter wavelength endpoints for every supported (disperser, filter) combination, plus the V2 overlap distance used by the collision-protection rules.
Per-shutter NIRSpec dispersion model for MSA shutters.
All supported (disperser, filter) combinations use a per-shutter lookup table derived from spacetelescope/msaviz’s numerical integration of the pipeline dispersion polynomials. The table lives in data/dispersion_cutoffs.npz and is regenerated by scripts/precompute_dispersion_cutoffs.py (re-run when the underlying msaviz reference files change).
For each (disperser, filter) the table stores four (4, 171, 365) float32 arrays — one slice per quadrant — under keys
{DISPERSER}_{FILTER}_blue_edge {DISPERSER}_{FILTER}_gap_lo {DISPERSER}_{FILTER}_gap_hi {DISPERSER}_{FILTER}_red_edge
NaN means the shutter’s spectrum doesn’t reach the corresponding detector. A linear V2-shift fallback is retained for the case where the table is missing on disk.
- vmpt.wavelengths.cutoffs(v2_arcsec: float, v3_arcsec: float, disperser: str, filt: str, *, q: int | None = None, s: int | None = None, d: int | None = None) dict[source]¶
Wavelength endpoints of the dispersed spectrum on the detector for a shutter at (V2, V3).
If q, s, d are supplied AND the precomputed dispersion table is available for the requested (disperser, filter), returns the per-shutter values from data/dispersion_cutoffs.npz (derived from msaviz’s integration of the pipeline dispersion models). This is the accurate path; the gap location and spectrum edges vary substantially across the MSA for every disperser, but especially for PRISM (non-linear dispersion).
Without shutter indices, or if the table is missing, the function falls back to a linear V2-shift model. The fallback exists so vMPT still loads on a fresh checkout that hasn’t run the precompute, and so the existing test suite (which calls cutoffs(V2, V3, …) without indices) keeps working.
- vmpt.wavelengths.disperser_max_lambda(disperser: str, filt: str) float | None[source]¶
Reddest wavelength the (disperser, filter) combo can deliver.
Nonewhen the combination isn’t recognised.
- vmpt.wavelengths.disperser_min_lambda(disperser: str, filt: str) float | None[source]¶
Bluest wavelength the (disperser, filter) combo can deliver.
Nonewhen the combination isn’t recognised.
- vmpt.wavelengths.disperser_range(disperser: str, filt: str) tuple[float, float] | None[source]¶
Nominal
(lam_min, lam_max)in μm for a (disperser, filter) combination.Returns the wavelength range the chosen mode actually delivers to the detector (the values in
GRATING_RANGES) — the same numbers shown in the JDox “useful range” docs, except where msaviz / the pipeline reference disagree (we follow the pipeline). ReturnsNonewhen the combination isn’t supported (e.g.G140H+F290LP); the optimizer treats that as “no constraint can pass” and drops the affected sources.
- vmpt.wavelengths.grating_adjacency_min_colsep(disperser: str) int | None[source]¶
Minimum column separation that demotes a no-buffer ADJACENCY conflict (rows exactly 1 apart, no real row overlap) from purple (Mask Conflict) to orange (Masked) for this disperser.
1for any M/H grating — any nonzero column offset is a deliberate diagonal step (see module note); only exact same-column stacking (Δd=0) stays purple. ReturnsNonewhen adjacency must always stay purple — PRISM and anything not an M/H grating.
- vmpt.wavelengths.interval_covered(lo: float, hi: float, blue: float, gap_lo: float, gap_hi: float, red: float) bool[source]¶
Does the spectrum
[blue, red](with the detector gap[gap_lo, gap_hi]excluded) fully cover the requested[lo, hi]range?Used by the per-target “required wavelength range” constraint.
- Parameters:
lo (float) – Requested interval, in μm. Must satisfy
lo <= hi.hi (float) – Requested interval, in μm. Must satisfy
lo <= hi.blue (float) – Bluest / reddest λ the centre shutter delivers to the detector. Both NaN when the source doesn’t reach the detector at all (cutoffs returns NaN for off-grid shutters).
red (float) – Bluest / reddest λ the centre shutter delivers to the detector. Both NaN when the source doesn’t reach the detector at all (cutoffs returns NaN for off-grid shutters).
gap_lo (float) – NRS1/NRS2 detector-gap wavelength bounds for this shutter. Both NaN ⇒ no gap (e.g. M-grating modes within their nominal range).
gap_hi (float) – NRS1/NRS2 detector-gap wavelength bounds for this shutter. Both NaN ⇒ no gap (e.g. M-grating modes within their nominal range).
- Returns:
True iff every wavelength in
[lo, hi]lands somewhere on the detector (i.e. inside[blue, gap_lo] ∪ [gap_hi, red], where the gap is skipped iff it has finite bounds).- Return type:
- vmpt.wavelengths.primary_detector(disperser: str, filt: str, q: int, s: int, d: int) int[source]¶
Per-shutter “primary detector” lookup. Returns:
0 → spectrum's main body is on NRS1 1 → spectrum's main body is on NRS2 -1 → table not loaded / not in shipped combo / off-detector
The main body is the detector that carries the larger wavelength range of the dispersed spectrum. For shutters whose spectrum is fully on one detector this is trivially that detector; for gap-spanning shutters it’s whichever side covers more wavelength.
Derived from the per-detector wavelength bounds in
data/dispersion_cutoffs.npz(populated byscripts/precompute_dispersion_cutoffs.py).The spec-overlap check uses this lookup to skip cross-detector pairs — two shutters whose spectra don’t share a detector main body cannot collide on the pipeline, regardless of how close they are in V2.
- vmpt.wavelengths.primary_detector_grid(disperser: str, filt: str) ndarray | None[source]¶
Return the full (4, 171, 365) int8 array of primary detectors for the (disperser, filter) combo. 0=NRS1, 1=NRS2, -1=neither/unknown.
Returns
Noneif the table or per-detector arrays aren’t available; the live overlay falls back to static quadrant-pairing in that case.
- vmpt.wavelengths.shutter_xy_grids(disperser: str, filt: str) dict | None[source]¶
Return interpolated per-shutter on-detector x/y ranges for the given (disperser, filter), or
Noneif the precomputed sample grid isn’t present.Result is a dict of six (4, 171, 365) float32 arrays:
x_lo_nrs1, x_hi_nrs1, y_nrs1x_lo_nrs2, x_hi_nrs2, y_nrs2NaN at a shutter index means the spectrum doesn’t reach that detector. The 10×10 quadrant grid samples from
precompute_trace_tilt.pyare bilinearly interpolated to every (s, d) so the spec-overlap check can do detector-pixel-accurate interval intersection without re-runningAssignWcsStep.Cached per (disperser, filter); call cost is one quick bilinear pass the first time and then a dict lookup.
- vmpt.wavelengths.tilt_slope_for_shutter(disperser: str, filt: str, q: int, s: int, d: int) float[source]¶
Bilinear-interpolated tilt slope (rows/arcsec V2) for a single open shutter at (quadrant, s-row, d-col). Returns 0.0 if the table is unavailable, the combo isn’t covered, or all four grid corners around (s, d) are NaN. Callers can use the returned value unconditionally — a zero slope reduces the tilt-aware check back to the flat row check.
- vmpt.wavelengths.tilt_slope_grid(disperser: str, filt: str) ndarray | None[source]¶
Return the full (4, 171, 365) tilt slope grid for (disperser, filter), bilinearly interpolated from the 10×10 quadrant sample. Returns
Noneif the table or this combo isn’t available; the caller is expected to fall back to 0-slope (flat) in that case.The cached interpolated grid lets the runtime overlap check estimate each candidate shutter’s spectrum-y-drift (used to widen the y tolerance for cross-quadrant pairs), without recomputing the per-shutter bilinear from scratch.
- vmpt.wavelengths.tilt_slope_map(disperser: str, filt: str) tuple[ndarray, ndarray, ndarray] | None[source]¶
Return the precomputed tilt-slope map for a (disperser, filter) combo, or
Noneif the table doesn’t ship it.The map describes how much the spectrum’s cross-dispersion row drifts as a function of V2 distance from the open shutter:
s_expected(ΔV2_arcsec) = s_open + slope * ΔV2_arcsec
where s_open is the MSA s-row (1-based) of the dispersing shutter and ΔV2_arcsec is the candidate shutter’s V2 offset from it. The slope is in [MSA rows per arcsec V2] and is computed from slit_frame → detector traces on a 10×10 grid per MSA quadrant (see
scripts/precompute_trace_tilt.py).- Returns:
(grid_rows, grid_cols, slope)— its parts:grid_rows— (Ngrid_r,) int16, vMPT row centres (1..171)grid_cols— (Ngrid_c,) int16, vMPT col centres (1..365)slope— (4, Ngrid_r, Ngrid_c) float32, tilt slope in rows per arcsec V2; NaN where the shutter projects onto neither detector for this combo.
Nonewhen the precomputed table is missing or ships no tilt arrays for this combo (older builds) — callers should fall back to slope = 0 (flat-row overlap check).- Return type:
tuple or None
- vmpt.wavelengths.v2_overlap_distance(disperser: str, filt: str) float[source]¶
Full V2 extent (arcsec) of the spectrum on the detector. Two same-row shutters whose spectra land on the SAME detector and whose V2 separation is less than this value share at least one detector pixel — i.e. their spectra collide on the pipeline.
The same-detector test is the responsibility of the caller (use
primary_detector()). This function only returns the distance.Looked up per (disperser, filter) combo because the filter sets the usable wavelength range, which controls the spectrum’s detector x-span. Falls back to the disperser-only value if the combo isn’t in the measured table, and to a safe upper bound (300″) for unknown dispersers.
vmpt.catalog¶
Loader + in-memory Catalog dataclass.
Target catalog loader (CSV / ASCII / FITS table).
- class vmpt.catalog.Catalog(ids: 'np.ndarray', ra_deg: 'np.ndarray', dec_deg: 'np.ndarray', priority: 'np.ndarray', mag: 'np.ndarray', z: 'np.ndarray', label: 'np.ndarray', source_path: 'str', weight: 'np.ndarray' = <factory>, required_lam: 'np.ndarray' = <factory>, no_gap: 'np.ndarray' = <factory>, extend_blue: 'np.ndarray' = <factory>, extend_red: 'np.ndarray' = <factory>, protect: 'np.ndarray' = <factory>, centration: 'np.ndarray' = <factory>, max_configs: 'np.ndarray' = <factory>, extras: 'dict' = <factory>)[source]¶
- vmpt.catalog.evaluate_catalog_condition(expr: str, columns: dict) ndarray[source]¶
Evaluate a boolean row-selection expression over catalog columns.
columnsmaps column name → per-row values. Column names become variables (numeric columns are floats, blanks → NaN; otherwise the column stays string). Returns a booleannp.ndarrayof lengthn_rows.Raises
ValueErrorwith a short, user-facing message on empty input, syntax error, unknown column, a disallowed construct, an evaluation error, or a non-boolean / wrong-length result.
- vmpt.catalog.save_catalog(cat: Catalog, path: str, *, include_constraints: str = 'auto') None[source]¶
Write a
Catalogback to CSV.Emits the standard eight columns (
ID, RA, DEC, priority, weight, mag, z, label) followed optionally by the six v1.3.x per-target constraint columns (lam_req, no_gap, extend_blue, extend_red, protect, centration), then anyextrascolumns the catalog is carrying. The output is round-trip-compatible withload_catalog()— write, reload, and the resultingCatalogmatches the input modulo dtype.- Parameters:
cat (Catalog) – The catalog to save.
path (str) – Destination CSV path. Parent directories are NOT created automatically — callers should ensure the parent exists.
include_constraints ({"auto", "always", "never"}) –
Controls whether the six constraint columns appear in the output:
"auto"(default): emit the columns iff at least one row has a non-default value (the same rule the catalog editor’s Save-as-CSV button uses, so v1.2.x catalogs that never picked up constraints get the same CSV format they had before)."always": always emit the columns, even when every row is at defaults. Useful when you want a “template” CSV the user can hand-edit."never": omit them. Use when you specifically want to drop the constraint metadata.
Notes
Wavelength-range cells round-trip via the same string format the catalog editor uses (
"1.0-1.3; 1.5-1.8"), parsed back by_parse_lam_req_str()on the next load.NaN / missing values render as empty cells in the CSV; on reload they come back as NaN (for float columns) or empty string (for label / extras).
vmpt.catalog_ops¶
Helpers for deriving weight ↔ priority used by the catalog editor.
Pure helpers for the catalog editor — computing weight from priority and vice versa. Lives outside main.py so unit tests don’t need to import Bokeh.
- vmpt.catalog_ops.compute_priorities_from_weights(weights) list[str] | None[source]¶
Group rows by unique weight value, descending. Largest weight → priority 1; next → 2; … Rows with non-finite weight get “” (NaN priority).
- vmpt.catalog_ops.compute_weights_from_priorities(priorities) list[str] | None[source]¶
Iterative weight formula.
For sources at each priority class (smaller p = higher priority), find the smallest INTEGER w(p) such that simultaneously
w(p) > w(p+1) N(p) * w(p) > N(p+1) * w(p+1)
iterating from the LOWEST priority class (largest p) upward. Returns a per-row list of stringified ints (NaN priority → “”) or None if no finite priorities exist.
vmpt.coords¶
Coordinate-frame helpers (V2/V3 ↔ RA/Dec, shutter polygons, PA rotation matrix).
V2/V3 <-> RA/Dec coordinate transforms ported from footprint_emerald.ipynb.
- vmpt.coords.fixed_slit_corners_v2v3() dict[str, ndarray][source]¶
Return {slit_name: (N, 2) corners in V2/V3 arcsec} for the five fixed slits.
- vmpt.coords.shutter_corners_v2v3(v2c: float, v3c: float, w: float = 0.2, h: float = 0.46) ndarray[source]¶
- vmpt.coords.v2v3_to_radec(coord_c: astropy.coordinates.SkyCoord, pa_v3: float, corners_v2v3: ndarray) astropy.coordinates.SkyCoord[source]¶
vmpt.empt_io / vmpt.mpt_io / vmpt.session_io / vmpt.image_io¶
I/O modules — eMPT bundle writer, APT MPT plan reader, vMPT workspace save/load, and image (FITS / JPG + sidecar) loaders respectively.
eMPT-compatible exporters (observed_targets.cat, pointing_summary.txt, shutter_mask.csv).
The shutter_mask.csv tiling reproduces what make_csv_file in
refs/eMPT_v1/reference_files/shutter_routines_new.f90 writes (lines 535-678).
Fortran code summary (1-indexed throughout):
msamap(kk, ii, jj) ! kk=quadrant 1..4, ii=dispersion 1..365, jj=shutter row 1..171
- do ir = 1, 365 ! csv data row, top half
kk=1, ii=ir, jj=1..171 -> chars 1..341 step 2 (cells 1..171) Q1 kk=2, ii=ir, jj=1..171 -> chars 343..683 step 2 (cells 172..342) Q2
- do ir = 366, 730 ! csv data row, bottom half
kk=3, ii=ir-365, jj=1..171 -> cells 1..171 Q3 kk=4, ii=ir-365, jj=1..171 -> cells 172..342 Q4
So with our (q, s, d) convention (q in 1..4, s in 1..171, d in 1..365):
- csv_row (1..730), csv_col (1..342):
top half (csv_row 1..365): d = csv_row; q = 1 if csv_col<=171 else 2 bottom half (csv_row 366..730): d = csv_row - 365; q = 3 if csv_col<=171 else 4 s = csv_col if csv_col <= 171 s = csv_col - 171 otherwise
i.e. each CSV data row holds a single dispersion column d; within that row the
171 cells of Q{1,3} sit side-by-side with the 171 cells of Q{2,4}, indexed by
shutter-row s. There is no transpose or reverse.
Cell alphabet (precedence top-down):
‘x’ failed-closed (operability) ‘s’ failed-open (operability) ‘0’ commanded open (user’s pick) ‘1’ commanded closed / functional
Each line (header + 730 data rows) is exactly 683 characters wide; the header text is padded with spaces to that width to match the reference byte-for-byte.
- class vmpt.empt_io.OpenShutter(q: int, s: int, d: int, target_id: int | str | None = None, role: str = 'target', target_ids: list = <factory>)[source]¶
A user-commanded open shutter.
target_idis the slitlet’s primary source (the anchor) and is kept for backward compatibility / APT primaryIds.target_idslists every catalog source whose footprint falls inside this one shutter — usually 0 or 1, but two very close sources can share a shutter, and v1.4.0 records all of them (one output row each). Snapshotted at open-time so it is stable even after the pointing later changes.
- class vmpt.empt_io.Pointing(ra_deg: float, dec_deg: float, apa_v3_deg: float, pa_ap_deg: float | None = None)[source]¶
Single pointing for the export bundle.
- vmpt.empt_io.parse_pointing_summary_txt(path: str) dict[source]¶
Tiny round-trip helper used by the test suite.
- vmpt.empt_io.parse_shutter_mask_csv(path: str) tuple[ndarray, ndarray, list[OpenShutter]][source]¶
Inverse of
write_shutter_mask_csv()— used by the test suite.Returns
(operable, reason, open_shutters)with the same shapes/conventions as the writer expects.
- vmpt.empt_io.write_mpt_catalog(path: str, targets: list[dict]) None[source]¶
Write a target list in APT MPT-importable format (tab-separated).
- Each dict in targets:
No_cat(int ID) — requiredra_deg,dec_deg— required, decimal degreesWeight— APT’s ranking weight (int). Falls back toPr(priority) then 1. (APT’s MSA catalog format has aWeightcolumn but noPrioritycolumn, so priority rides here when no separate weight exists.)Primary— 1 = source from the user’s input catalog; 0 = a slitlet vMPT opened that had no catalog match. Defaults 1.Magnitude/Redshift— optional floats carried from the input catalog. The column is emitted only when at least one row has a finite value; missing cells are filled with a finite sentinel (99.9mag /-1redshift, NOTNaN— APT rejects NaN) so the whitespace importer keeps a constant column count.label— Label column text; defaults to “real”. “vMPT_synth” marks an entry we made up for an unmatched slitlet.
- vmpt.empt_io.write_observed_targets_cat(path: str, targets: list[dict]) None[source]¶
Write eMPT-style observed_targets.cat.
Each
targetsdict must contain at leastNo_cat,Pr,ra_deganddec_deg.No_subis optional (defaults to the runningNo).
- vmpt.empt_io.write_pointing_summary_txt(path: str, pointing: Pointing, disperser: str, filter_name: str, n_targets_total: int = 0, n_targets_accepted: int = 0) None[source]¶
Write a pointing_summary.txt matching the reference layout.
- vmpt.empt_io.write_shutter_mask_csv(path: str, open_shutters: list[OpenShutter], operable: ndarray, reason: ndarray) None[source]¶
Write the 730 x 342 shutter-mask grid.
- Parameters:
path – Output path.
open_shutters – Commanded-open shutters; written as
'0'(overrides'1'but not operability failures).operable –
(4, 171, 365)bool array.True= functional.reason –
(4, 171, 365)int8 array.1= failed-closed (-> ‘x’),2= failed-open (-> ‘s’); other values fall back to operable/open.
Import MSA plans from APT/MPT exports (JSON plan files, shutter CSVs, .aptx archives).
- class vmpt.mpt_io.MPTPlan(name: str, aperture_pa_deg: float, v3_pa_deg: float, ra_deg: float | None = None, dec_deg: float | None = None, grating: str | None = None, filter_name: str | None = None, slitlets: list[MPTSlitlet] = None, catalog_name: str | None = None, primary_ids: list[int] = None, n_open_shutters: int = 0)[source]¶
One MSA plan / config from an APT MPT JSON file.
- to_open_shutters() list[OpenShutter][source]¶
Unfold slitlets into a flat list of OpenShutter entries.
Each slitlet at (q, s, d, h) maps to h shutters at rows s, s+1, …, s+h-1 in column d of quadrant q. The middle shutter (s + h//2) is the “target” row; the others are “sky”.
- class vmpt.mpt_io.MPTSlitlet(q: int, s: int, d: int, h: int = 1, primary_id: int | None = None)[source]¶
A single slitlet entry from MPT JSON: starting shutter (q, s, d) and slitlet height h. Unfolds to h shutters at rows s, s+1, …, s+h-1.
- vmpt.mpt_io.download_apt_program(program_id: int | str, dest_path: str | None = None) str[source]¶
Fetch <program_id>.aptx from STScI’s public proposal-info URL.
Returns the local filesystem path of the downloaded archive. If dest_path is None, writes to a temp file. Raises ValueError if the fetch fails or the response isn’t a recognizable .aptx.
- vmpt.mpt_io.list_mpt_plans_in_aptx(aptx_path: str) list[str][source]¶
Return the names of MPT-style JSON files embedded in an .aptx archive.
.aptx files are zip archives containing one or more <plan>.json files in MPT format, alongside the proposal XML, manifest, and pointing files. We filter to JSON files that look like MPT plans (top-level keys include ‘configs’ and ‘aperturePA’).
- vmpt.mpt_io.parse_mpt_json(path: str) list[MPTPlan][source]¶
Parse an APT/MPT JSON plan file into one MPTPlan per configs entry.
APT’s MSA-Planner exports JSON with this shape:
{ "aperturePA": <degrees>, # NIRSpec aperture PA on sky "theta": <degrees>, # MSA roll (we don't use it directly) "catalog": {"name": ..., "primariesName": ..., ...}, "configs": [ {"name": "c1 : foo plan 1", "slitlets": [{"q":1,"d":12,"s":50,"h":3}, ...], "exposures": [{"ra": ..., "dec": ..., "gratingFilter": "PRISM_CLEAR", "sourceIds": [...], ...}], "primaryIds": [...], "fillerIds": [...] }, ... ] }
Returns one MPTPlan per config. The plan’s RA/Dec/grating/filter come from the first exposure of the config. The V3 PA is derived from aperturePA - V3IdlYAngle so vMPT can drive the overlay directly.
Raises ValueError on malformed input.
- vmpt.mpt_io.parse_mpt_json_in_aptx(aptx_path: str, member_name: str) list[MPTPlan][source]¶
Extract one MPT JSON entry from an .aptx archive into a list of MPTPlan (delegating to parse_mpt_json after extraction).
- vmpt.mpt_io.parse_shutter_csv(path: str) list[OpenShutter][source]¶
Parse a shutter_mask.csv exported by APT/MPT (or eMPT).
Format: 731 lines (1 header + 730 data rows × 342 cells). Cells encode shutter state: ‘0’ = commanded open (user’s pick), ‘1’ = closed, ‘x’ = failed-closed (operability), ‘s’ = failed-open. We return only the ‘0’ cells.
Tiling matches our writer: CSV row 1..365 = Q1 + Q2 by d, CSV row 366..730 = Q3 + Q4 by d (d = row - 365); CSV col 1..171 = s for Q1/Q3, CSV col 172..342 = s for Q2/Q4 (s = col or col - 171).
Session JSON save/load.
session.json is written as a pure APT MPT plan JSON — no vMPT-only keys, no file paths — so APT’s MPT loader accepts it directly. A sibling vmpt_workspace.json (same parent directory) carries the bits MPT doesn’t preserve: per-shutter target_id / role, highlighted set, image + catalog paths, slitlet height. vMPT reads both on import; APT only sees the MPT file.
Old-style sessions (single file with a flat top-level open_shutters list and a pointing block) are still accepted on import.
- class vmpt.session_io.Session(pointing_ra_deg: 'float', pointing_dec_deg: 'float', pa_v3_deg: 'float', disperser: 'str', filter_name: 'str', slitlet_height: 'int', open_shutters: 'list[OpenShutter]', highlighted: 'list[tuple[int, int, int]]'=<factory>, image_path: 'Optional[str]' = None, wcs_sidecar_path: 'Optional[str]' = None, catalog_path: 'Optional[str]' = None, catalog_paths: 'list' = <factory>, tool_version: 'str' = '1.4', created: 'Optional[str]' = None, name: 'Optional[str]' = None, configs: 'list' = <factory>, active_config: 'int' = 0, region_paths: 'list' = <factory>, contour_specs: 'list' = <factory>, overlays: 'list' = <factory>)[source]¶
- vmpt.session_io.apt_catalog_basename(catalog_path: str | None) str[source]¶
Basename (no extension) shared by the APT-importable .cat file AND the plan JSON’s
catalog.name/primariesName, kept identical so APT’s filename-derived default lines up with the plan. Target-prefixed and_APT_catalog-suffixed so users see at a glance it’s for APT.
- vmpt.session_io.apt_plan_basename(catalog_path: str | None) str[source]¶
Basename (no extension) for the MPT plan JSON — target-prefixed and
_MPT_plan-suffixed, e.g. ‘rxcj2211_targets_MPT_plan’.
- vmpt.session_io.bundle_readme_text(*, catalog_filename: str, catalog_name: str, plan_filename: str, n_configs: int = 1) str[source]¶
Markdown README placed at the root of an exported bundle, telling a user exactly how to load it into APT / MPT. Filenames + catalog name are filled in so the steps are copy-paste accurate for this bundle.
- vmpt.session_io.export_session_json(session: Session, path: str) None[source]¶
Write the session as an MPT-format plan JSON at path, AND a sibling vmpt_workspace.json carrying the vMPT-only extras.
- vmpt.session_io.import_session_json(path: str) Session[source]¶
Parse a session JSON back into a Session. The user can point at EITHER file in a bundle:
session_MPT_plan.json → pure MPT plan; we look for a sibling vmpt_workspace.json to merge in target_ids, roles, image path.
vmpt_workspace.json → vMPT extras; we look for a sibling session_MPT_plan.json (or any *plan*.json matching MPT shape) to pull pointing / PA / disperser / slitlet geometry.
Legacy single-file sessions (open_shutters at top level) still load.
Image loaders (FITS / JPG+sidecar) and display stretching.
- class vmpt.image_io.LoadedImage(data: 'np.ndarray', wcs: 'WCS', shape: 'tuple', source_path: 'str', mode: 'str', wcs_sidecar_path: 'Optional[str]' = None, factor: 'int' = 1, full_shape: 'Optional[tuple]' = None, hdu: 'int' = 0)[source]¶
- vmpt.image_io.compute_histogram(arr: ndarray, nbins: int = 50, clip: tuple | None = None) dict[source]¶
Pixel-value histogram for the Image display dialog.
Bins the finite values over the FULL data range
[min, max](so the binning is fixed once an image is loaded and the whole range is draggable), split intonbinsequal-width bins. An RGB image bins luminance over[0, 255]. Passclip=(lo_pct, hi_pct)to bin over a robust percentile window instead. Values are sub-sampled above ~1M points to keep it fast. Returns a dict withedges(nbins+1),counts(nbins),lo/hi(the binned range), andstats(min/median/max).
- vmpt.image_io.compute_interval(arr: ndarray, scale_mode: str = 'percentile', percentile: float = 99.5, vmin: float | None = None, vmax: float | None = None) tuple[source]¶
Return the
(lo, hi)intensity limits for a FITS array under the given scaling mode. Computed once from the base tier and reused for every on-demand zoom crop, so the image brightness stays stable as you zoom (a percentile taken per-crop would flicker).
- vmpt.image_io.load_fits(path: str, hdu: int | None = None, max_dim: int = 4000) LoadedImage[source]¶
Load a 2D FITS image for display.
Uses a memory-mapped read and, when the image is larger than
max_dimon a side, strided-decimates it on load (data[::f, ::f]) so GB-scale FITS read quickly (~1/f² of pages touched) and produce a display-sized array instead of a browser-overflowing full-res RGBA. The WCS is scaled to match, and the downsamplefactoris recorded for image-coordinate overlays.
- vmpt.image_io.lod_view_factor(visible_full_px: float, target_px: float, base_factor: int) int[source]¶
Choose a decimation factor for an on-demand FITS zoom crop.
visible_full_pxis the longer side of the visible region measured in ORIGINAL (full-res) pixels;target_pxis the on-screen pixel budget (~canvas frame size). Returns a power-of-two factor in[1, base_factor]so the rendered crop is ≳target_px— capped at the base tier (never coarser) and floored at 1 (native resolution). Powers of two give a small, stable set of tiers (e.g. base_factor 4 → {1, 2, 4}) so we re-render only when crossing a tier.
- vmpt.image_io.read_fits_region(path: str, hdu: int, y0: int, y1: int, x0: int, x1: int, view_factor: int) tuple[source]¶
Read a decimated crop
data[y0:y1:f, x0:x1:f](full-res pixel bounds) from the memory-mapped FITS. The read origin is aligned DOWN to a multiple offso the sampled pixel grid is stable across pans (no shimmer). Returns(crop_float32, ay0, ax0)where(ay0, ax0)is the aligned top-left origin actually used.
- vmpt.image_io.stretch_curve(stretch: str, vmin: float, vmax: float, n: int = 48) tuple[source]¶
Trace how data values in
[vmin, vmax]map to display brightness[0, 1]under a tone curve — for overlaying on the pixel histogram so the user sees the stretch they’re looking at. Returns(xs, ys).
- vmpt.image_io.stretch_for_display(arr: ndarray, stretch: str = 'asinh', *, scale_mode: str = 'percentile', percentile: float = 99.5, vmin: float | None = None, vmax: float | None = None, brightness: float = 0.0, contrast: float = 1.0, cmap: str = 'gray', invert: bool = False, nan_white: bool = False) ndarray[source]¶
Convert raw pixels to a uint32 RGBA array for Bokeh’s
image_rgba.RGB images (JPG/PNG, ndim==3): the legacy asinh tone-curve is re-applied (so the image looks exactly as it did pre-v1.8.0), then
brightnessandcontrastadjust on top (fitsmap-style) —out = (tone(rgb)-0.5) * contrast + 0.5 + brightness. Defaults (0, 1) reproduce the original look.Grayscale images (FITS, ndim==2): normalise with
astropy.visualization.ImageNormalizeusingscale_mode(‘percentile’ → centralpercentile%, ‘manual’ →[vmin, vmax], ‘zscale’) and astretchtone curve (linear/sqrt/asinh/log), then apply acmapcolormap (invertflips dark↔bright). Whennan_whiteis set, non-finite (NaN/blank/±inf) pixels are painted white instead of taking the colormap’s zero colour.