Python examples

Measure a time series

The complete source is included below.

python -m nextastro_obs_client.examples.variable_star_series --ra 12.5 --dec -30 --target "Variable field" --filter V --seconds 10 --count 20 --cadence 30

Writes variable_star.csv after each completed frame. Cadence is start-to-start in seconds; acquisition/readout overruns start the next frame immediately. Supply --star X Y and --comparison X Y together to add aperture fluxes and ratios. All coordinates are zero-based pixel centres; choose actual stars in this image. This basic exercise assumes they remain at those positions. It does not register images, correct extinction, convert time scales, or calibrate magnitude.

Student extension: Track centroids across the sequence, add comparison stars, and convert ratios to differential magnitudes.

Python source

"""Acquire a time series with optional differential aperture photometry."""
import time
from nextastro_obs_client.examples._common import parser, target_arguments, point, execute, write_rows


def run(obs, camera, args):
    point(obs, args)
    start = time.monotonic()
    rows = []
    for index in range(args.count):
        time.sleep(max(0, start + index * args.cadence - time.monotonic()))
        image = camera.expose(seconds=args.seconds, filter=args.filter, target_name=args.target)[0]
        row = dict(index=index, started_at=image.started_at, seconds=image.seconds, path=str(image.path))
        if args.star is not None:
            measurements = [image.aperture_photometry(x=x, y=y, radius=args.radius,
                sky_inner=args.sky_inner, sky_outer=args.sky_outer) for x, y in (args.star, args.comparison)]
            star, comparison = measurements
            row.update(star_flux=star.flux, comparison_flux=comparison.flux,
                       flux_ratio=star.flux / comparison.flux if star.flux is not None and comparison.flux else None,
                       edge_clipped=star.edge_clipped or comparison.edge_clipped)
        rows.append(row)
        write_rows(args.output, rows)  # Preserve completed measurements if interrupted.
    return rows


def main(argv=None):
    p = parser("Variable star time series")
    target_arguments(p)
    p.add_argument("--seconds", type=float, required=True)
    p.add_argument("--count", type=int, required=True)
    p.add_argument("--cadence", type=float, default=0, help="Start-to-start seconds; overruns run immediately")
    p.add_argument("--star", type=float, nargs=2, metavar=("X", "Y"))
    p.add_argument("--comparison", type=float, nargs=2, metavar=("X", "Y"))
    p.add_argument("--radius", type=float, default=5)
    p.add_argument("--sky-inner", type=float, default=8)
    p.add_argument("--sky-outer", type=float, default=12)
    p.add_argument("--output", default="variable_star.csv")
    args = p.parse_args(argv)
    if (args.star is None) != (args.comparison is None):
        p.error("--star and --comparison must be supplied together for differential photometry")
    return execute(p, run, argv)


if __name__ == "__main__":
    main()

Shared argument and session helpers: Example support code.