from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from pyobs.interfaces import IBinning, IFilters, IFlatField, IReady, IRoof, ITelescope
from pyobs.robotic.scripts import Script
from pyobs.robotic.utils.skyflats.priorities.base import SkyflatPriorities
from pyobs.robotic.utils.skyflats.scheduler import Scheduler, SchedulerItem
from pyobs.utils.time import Time
if TYPE_CHECKING:
from pyobs.robotic.task import TaskData
log = logging.getLogger(__name__)
FlatFunctions = str | dict[str, str | dict[str, str]]
[docs]
class SkyFlatsScript(Script):
"""Script for scheduling and running skyflats using an IFlatField module."""
roof: str
telescope: str
flatfield: str
functions: FlatFunctions = {}
priorities: SkyflatPriorities
min_exptime: float = 0.5
max_exptime: float = 5
timespan: float = 7200
filter_change: float = 30
count: int = 20
readout: dict[str, float] | None = None
[docs]
async def can_run(self, data: TaskData | None) -> bool:
"""Whether this config can currently run.
Returns:
True if script can run now.
"""
# get modules
if not await self.comm.has_proxy(self.roof, IRoof):
self._cant_run_reason = "No roof found."
return False
if not await self.comm.has_proxy(self.telescope, ITelescope):
self._cant_run_reason = "No telescope found."
return False
if not await self.comm.has_proxy(self.flatfield, IFlatField):
self._cant_run_reason = "No flatfielder found."
return False
# we need an open roof and a working telescope
async with self.comm.proxy(self.roof, IRoof) as roof:
ready_state = roof.get_state(IReady)
if ready_state is None or not ready_state.ready:
self._cant_run_reason = "Roof not ready."
return False
async with self.comm.proxy(self.telescope, ITelescope) as telescope:
ready_state = telescope.get_state(IReady)
if ready_state is None or not ready_state.ready:
self._cant_run_reason = "Telescope not ready."
return False
# seems alright
self._cant_run_reason = None
return True
[docs]
async def run(self, data: TaskData | None) -> None:
"""Run script.
Raises:
InterruptedError: If interrupted
"""
# create scheduler
scheduler = Scheduler(
self.functions,
self.priorities,
self.observer,
min_exptime=self.min_exptime,
max_exptime=self.max_exptime,
timespan=self.timespan,
filter_change=self.filter_change,
count=self.count,
readout=self.readout,
)
# schedule
log.info("Scheduling flat-fields...")
await scheduler(Time.now())
# log schedule
log.info("Found schedule:")
for sched in scheduler:
log.info("- %s", sched)
# total exposure time in seconds
exptime_done = 0
# do flat fields
item: SchedulerItem
for item in scheduler:
# do flat fields
log.info("Performing flat-fields in %s %dx%d...", item.filter_name, *item.binning)
async with self.comm.proxy(self.flatfield, IBinning) as flatfield:
if flatfield:
await flatfield.set_binning(*item.binning)
async with self.comm.safe_proxy(self.flatfield, IFilters) as flatfield:
if flatfield:
await flatfield.set_filter(item.filter_name)
async with self.comm.proxy(self.flatfield, IFlatField) as flatfield:
done, exp_time = await flatfield.flat_field(self.count)
log.info("Finished flat-fields.")
# increase exposure time
exptime_done += int(exp_time)
# finished
log.info("Finished all scheduled flat-fields.")
[docs]
def estimate_duration(self, data: TaskData | None = None, time: Time | None = None) -> float:
"""Estimate duration of the sky flats.
The actual schedule depends on sky conditions that can only be evaluated at runtime,
so this returns the configured timespan as a conservative upper bound.
"""
return self.timespan
__all__ = ["SkyFlatsScript"]