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

  1. `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.

  2. `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.

  3. `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.

  4. `grid_search` — brute-force ranking over a (ΔRA, ΔDec, ΔPA) cube.

  5. `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: object

One 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). Requires disperser + filt to be meaningful. When None or all-False, no protection is applied and evaluate behaves 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 when protect_mask flags any source. The V2 half-extent of the spectrum is looked up from v2_overlap_distance().

  • filt (str, optional) – e.g. "PRISM" and "CLEAR". Only consulted when protect_mask flags any source. The V2 half-extent of the spectrum is looked up from v2_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, detected is the kept mask — sources dropped by the protection rules are zeroed in both detected and throughput. The raw pre-drop mask is not returned; use evaluate_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 scalar evaluate_with_stats returns.

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 an n_dropped count 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 == 0 when neither protection nor constraints are configured. See evaluate_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.

quad is 1–4 for sources inside a quadrant, 0 for outside. s_frac and d_frac are fractional shutter indices (s [0, 170], d [0, 364]). NaN where quad == 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_DEGREE polynomial inverse (fast path, default) when USE_POLY_INVERSE is set and the fit passed its self-check; otherwise the CloughTocher2DInterpolator (interpolators is built lazily only on this fallback path).

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_deg is ≤ 0, that axis is FROZEN at the central value (n is 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_deg is 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_tol is (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_deg that is ≤ 0 freezes the corresponding axis: scipy’s differential_evolution doesn’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 msaoper reference — the one the operational CRDS context selects for EXP_TYPE = NRS_MSASPEC by 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 crds isn’t installed, there’s no network, or the check runs longer than timeout seconds, it quietly gives up and vMPT uses whatever is already cached. Runs at most once per process unless force=True. Set VMPT_OPERABILITY_AUTOUPDATE=0 to disable. Returns the resolved local path, or None.

vmpt.msa.load_msa_grid() tuple[ndarray, ndarray][source]
vmpt.msa.load_operability(crds_path: str | None = None) tuple[ndarray, ndarray][source]
vmpt.msa.operability_is_current() bool[source]

True only when the map came from a live CRDS reference (not the bundled snapshot or the all-operable degenerate fallback).

vmpt.msa.shutter_center_v2v3(q: int, s: int, d: int) tuple[float, float][source]
vmpt.msa.shutters_in_bbox(v2_min: float, v2_max: float, v3_min: float, v3_max: float) ndarray[source]

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. None when 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. None when 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). Returns None when 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. 1 for 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. Returns None when 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:

bool

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 by scripts/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 None if 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 None if 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_nrs1 x_lo_nrs2, x_hi_nrs2, y_nrs2

NaN at a shutter index means the spectrum doesn’t reach that detector. The 10×10 quadrant grid samples from precompute_trace_tilt.py are bilinearly interpolated to every (s, d) so the spec-overlap check can do detector-pixel-accurate interval intersection without re-running AssignWcsStep.

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 None if 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 None if 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.

None when 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]
centration: ndarray
dec_deg: ndarray
extend_blue: ndarray
extend_red: ndarray
extras: dict
ids: ndarray
label: ndarray
mag: ndarray
max_configs: ndarray
no_gap: ndarray
priority: ndarray
protect: ndarray
ra_deg: ndarray
required_lam: ndarray
source_path: str
weight: ndarray
z: ndarray
vmpt.catalog.catalog_in_view(cat: Catalog, ra_min, ra_max, dec_min, dec_max) ndarray[source]
vmpt.catalog.evaluate_catalog_condition(expr: str, columns: dict) ndarray[source]

Evaluate a boolean row-selection expression over catalog columns.

columns maps column name → per-row values. Column names become variables (numeric columns are floats, blanks → NaN; otherwise the column stays string). Returns a boolean np.ndarray of length n_rows.

Raises ValueError with 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.load_catalog(path: str) Catalog[source]
vmpt.catalog.save_catalog(cat: Catalog, path: str, *, include_constraints: str = 'auto') None[source]

Write a Catalog back 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 any extras columns the catalog is carrying. The output is round-trip-compatible with load_catalog() — write, reload, and the resulting Catalog matches 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.rot_matrix(rotation: float = 30.0) ndarray[source]
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_id is the slitlet’s primary source (the anchor) and is kept for backward compatibility / APT primaryIds. target_ids lists 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) — required

  • ra_deg, dec_deg — required, decimal degrees

  • Weight — APT’s ranking weight (int). Falls back to Pr (priority) then 1. (APT’s MSA catalog format has a Weight column but no Priority column, 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.9 mag / -1 redshift, NOT NaN — 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 targets dict must contain at least No_cat, Pr, ra_deg and dec_deg. No_sub is optional (defaults to the running No).

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 into nbins equal-width bins. An RGB image bins luminance over [0, 255]. Pass clip=(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 with edges (nbins+1), counts (nbins), lo/hi (the binned range), and stats (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_dim on 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 downsample factor is 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_px is the longer side of the visible region measured in ORIGINAL (full-res) pixels; target_px is 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 of f so 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 brightness and contrast adjust 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.ImageNormalize using scale_mode (‘percentile’ → central percentile %, ‘manual’ → [vmin, vmax], ‘zscale’) and a stretch tone curve (linear/sqrt/asinh/log), then apply a cmap colormap (invert flips dark↔bright). When nan_white is set, non-finite (NaN/blank/±inf) pixels are painted white instead of taking the colormap’s zero colour.