Coverage for core / src / sensorkit / astro / target.py: 83%
312 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-02 00:03 +0000
1# SPDX-License-Identifier: Apache-2.0
2from __future__ import annotations
4import asyncio
5import collections
6from abc import ABC, abstractmethod
7from collections.abc import AsyncIterator, Callable, Sequence
8from dataclasses import dataclass
9from datetime import UTC, datetime, timedelta
10from typing import TYPE_CHECKING, Annotated, Any, Literal, override
12import astropy.units as u
13import numpy as np
14import satkit
15from astropy.coordinates import (
16 CIRS,
17 GCRS,
18 ICRS,
19 TEME,
20 AltAz,
21 BaseCoordinateFrame,
22 CartesianRepresentation,
23 EarthLocation,
24 SkyCoord,
25 UnitSphericalRepresentation,
26 get_body,
27 get_sun,
28)
29from astropy.coordinates import AltAz as AltAzFrame
30from astropy.time import Time
31from loguru import logger
32from pydantic import BaseModel, Discriminator
34from sensorkit.astro.common import TLE, ReferenceFrame
35from sensorkit.astro.coords import (
36 Coordinates,
37 Equatorial,
38 Geodetic,
39 Horizontal,
40 StateVector,
41)
42from sensorkit.astro.trajectory import OrbitalTrajectory, TLETrajectory, Trajectory
44if TYPE_CHECKING:
45 from sensorkit.astro.observer import EarthObserver
46 from sensorkit.std.weather import BasicWeather
49# Frames built from the site-to-target vector, and so unusable without an observer.
50_OBSERVER_RELATIVE_FRAMES = (
51 ReferenceFrame.ALTAZ,
52 ReferenceFrame.CIRF,
53 ReferenceFrame.ICRF,
54)
56# How far past a requested window a track propagates, so that successive windows are
57# usually served without propagating again.
58_PROPAGATION_HORIZON = timedelta(hours=1)
60# Window covering consumers that ask adapt() for a materialized EphemerisTarget. A track
61# leaves this choice to the consumer; these apply only where one is not accepted.
62_ADAPTED_EPHEMERIS_DURATION = timedelta(minutes=1)
63_ADAPTED_EPHEMERIS_STEP = timedelta(seconds=2)
66def _topocentric_cartesian(gcrs: GCRS, location: EarthLocation) -> CartesianRepresentation:
67 """Return the site-to-target vector in GCRS axes.
69 Relies on *gcrs* carrying the default geocentric origin; a GCRS with a non-zero
70 obsgeoloc is already observer-relative and would be shifted twice.
71 """
72 obsgeoloc, _ = location.get_gcrs_posvel(gcrs.obstime)
73 return gcrs.cartesian.without_differentials() - obsgeoloc
76def _topocentric_icrs(gcrs: GCRS, location: EarthLocation) -> ICRS:
77 return ICRS(
78 _topocentric_cartesian(gcrs, location).represent_as(UnitSphericalRepresentation)
79 )
82def _topocentric_cirs(gcrs: GCRS, location: EarthLocation) -> CIRS:
83 topo = _topocentric_cartesian(gcrs, location)
84 cirs = GCRS(topo, obstime=gcrs.obstime).transform_to(CIRS(obstime=gcrs.obstime))
85 return CIRS(cirs.cartesian, obstime=gcrs.obstime, location=location)
88def _output_frame_transform(
89 frame: ReferenceFrame,
90 location: EarthLocation | None,
91 weather: BasicWeather | None,
92) -> Callable[[GCRS], BaseCoordinateFrame]:
93 """Return the GCRS-to-*frame* conversion used to build ephemeris points.
95 ICRF, CIRF and ALTAZ are observer-relative and are all built from the topocentric
96 vector; anything else is a geocentric frame astropy can reach from GCRS on its own.
97 """
98 if location is None and frame in _OBSERVER_RELATIVE_FRAMES:
99 raise RuntimeError(f"an observer location is required for {frame}")
101 match frame:
102 case ReferenceFrame.ICRF:
103 return lambda gcrs: _topocentric_icrs(gcrs, location)
104 case ReferenceFrame.CIRF:
105 return lambda gcrs: _topocentric_cirs(gcrs, location)
106 case ReferenceFrame.ALTAZ:
107 return lambda gcrs: _topocentric_cirs(gcrs, location).transform_to(
108 AltAz(
109 obstime=gcrs.obstime,
110 location=location,
111 **(weather.to_astropy() if weather else {}),
112 )
113 )
114 case _:
115 output_frame = frame.to_astropy()
117 if "obstime" in output_frame.frame_attributes:
118 return lambda gcrs: gcrs.transform_to(output_frame(obstime=gcrs.obstime))
120 return lambda gcrs: gcrs.transform_to(output_frame())
123def _ephemeris_points(frame: ReferenceFrame, coord: BaseCoordinateFrame) -> list[Coordinates]:
124 """Pack a sampled coordinate into the point type native to *frame*.
126 Equatorial frames expose ra/dec directly, but ITRF and TEME carry cartesian data and
127 have to be resolved first: ITRF to the sub-satellite geodetic point, TEME to the
128 spherical angles its cartesian representation implies.
129 """
130 match frame:
131 case ReferenceFrame.ALTAZ:
132 return [
133 Horizontal(az=az, alt=alt)
134 for az, alt in zip(coord.az.deg, coord.alt.deg, strict=True)
135 ]
136 case ReferenceFrame.ITRF:
137 lons, lats, heights = coord.earth_location.to_geodetic()
138 return [
139 Geodetic(lon=lon, lat=lat, elev=elev)
140 for lon, lat, elev in zip(
141 lons.deg, lats.deg, heights.to_value(u.m), strict=True
142 )
143 ]
144 case ReferenceFrame.TEME:
145 usph = coord.represent_as(UnitSphericalRepresentation)
146 return [
147 Equatorial(ra=ra, dec=dec)
148 for ra, dec in zip(usph.lon.deg, usph.lat.deg, strict=True)
149 ]
150 case _:
151 return [
152 Equatorial(ra=ra, dec=dec)
153 for ra, dec in zip(coord.ra.deg, coord.dec.deg, strict=True)
154 ]
157class TargetTrack:
158 """The unbounded on-sky path of a single propagatable target."""
160 def __init__(
161 self,
162 trajectory: Trajectory,
163 frame: ReferenceFrame,
164 *,
165 observer: EarthObserver | Geodetic | None = None,
166 weather: BasicWeather | None = None,
167 propagation_horizon: timedelta = _PROPAGATION_HORIZON,
168 ):
169 """Bind *trajectory* to the frame and observing context it is sampled in.
171 Args:
172 trajectory: The path to sample, propagated in GCRS.
173 frame: Reference frame the sampled points are returned in.
174 observer: Site position. Required for observer-relative frames.
175 weather: Ambient conditions, applied as refraction in ALTAZ.
176 propagation_horizon: How far past a requested window to propagate.
177 """
178 self.trajectory = trajectory
179 self.frame = frame
180 self.observer = observer
181 self.weather = weather
182 self.propagation_horizon = propagation_horizon
183 self._propagated: Trajectory | None = None
184 self._propagated_until: datetime | None = None
185 self._propagating = asyncio.Lock()
187 async def sample(
188 self,
189 start_time: datetime,
190 duration: timedelta,
191 step: timedelta,
192 ) -> tuple[list[float], list[Coordinates]]:
193 """Sample the track over a window, returning Julian dates and points in `frame`.
195 The first sample sits one *step* after *start_time*, and samples run to the last
196 whole *step* within *duration*, which falls short of the full span where the two
197 do not divide evenly.
198 """
199 trajectory = await self._propagate_through(start_time + duration)
200 location = self.observer.to_astropy() if self.observer else None
201 transform_to_output_frame = _output_frame_transform(self.frame, location, self.weather)
203 def _sample_series():
204 epochs = [start_time + (i + 1) * step for i in range(duration // step)]
206 if not epochs:
207 return [], []
209 gcrs = trajectory.sample(epochs)
210 coord = transform_to_output_frame(gcrs)
211 jds = list(gcrs.obstime.jd)
212 points = _ephemeris_points(self.frame, coord)
213 logger.debug(f"sampled track ending at {points[-1]}")
215 return jds, points
217 return await asyncio.to_thread(_sample_series)
219 async def stream(
220 self,
221 *,
222 window: timedelta,
223 step: timedelta,
224 lead: timedelta,
225 ) -> AsyncIterator[tuple[list[float], list[Coordinates]]]:
226 """Yield consecutive windows of the track, indefinitely.
228 Each iteration yields the next *window* of samples, then waits until *lead*
229 remains before that window elapses. A consumer therefore always holds samples
230 reaching at least *lead* into the future, and both producing the next window
231 and acting on it have to fit within *lead*.
233 Args:
234 window: Time span covered by each batch of samples.
235 step: Interval between samples within a batch.
236 lead: How long before a window elapses to produce the next one.
238 Raises:
239 ValueError: If *lead* leaves no time to consume a window.
240 """
241 # A window that is not a whole number of steps is covered only as far as its
242 # last sample, which is what the next window has to be timed against.
243 covered = (window // step) * step
245 if lead >= covered - step:
246 raise ValueError("lead must leave at least one step of the window to consume")
248 while True:
249 start = datetime.now(UTC) - step
250 yield await self.sample(start, window, step)
252 delay = (start + covered - lead - datetime.now(UTC)).total_seconds()
254 if delay > 0:
255 await asyncio.sleep(delay)
257 async def _propagate_through(self, until: datetime) -> Trajectory:
258 """Return a trajectory propagated at least as far as *until*."""
259 async with self._propagating:
260 if self._propagated is not None and until <= self._propagated_until:
261 return self._propagated
263 horizon = until + self.propagation_horizon
264 # Propagate from the original trajectory rather than the last result, which
265 # would accumulate error over a long-running track.
266 propagated = await self.trajectory.propagate(horizon)
268 # Record the reach only once it is backed by a result, so a failed
269 # propagation leaves the previous one in place to be retried.
270 self._propagated = propagated
271 self._propagated_until = horizon
273 return propagated
276def _accepted_frame(
277 accepted: type,
278 supported: dict[type, list[ReferenceFrame]],
279 observer: EarthObserver | Geodetic | None,
280) -> ReferenceFrame:
281 """Return the single frame requested for *accepted*, rejecting an unusable request."""
282 frames = supported[accepted]
284 if len(frames) != 1:
285 raise RuntimeError(f"exactly one reference frame is required for {accepted.__name__}")
287 frame = frames[0]
289 if observer is None and frame in _OBSERVER_RELATIVE_FRAMES:
290 raise RuntimeError(f"an observer location is required for {frame}")
292 return frame
295class BaseTarget(BaseModel, ABC):
296 """Abstract base for all target types; discriminated on `target_type`."""
297 target_type: Literal[None]
299 async def adapt(
300 self,
301 *accepts: type[BaseTarget | TargetTrack] | tuple,
302 observer: EarthObserver | Geodetic | None = None,
303 weather: BasicWeather | None = None,
304 _catalog: Any = None,
305 ) -> BaseTarget | TargetTrack:
306 """Convert this target to the best matching type from *accepts*, propagating as needed."""
307 mytype = type(self)
308 supported: dict[type[BaseTarget | TargetTrack], list[ReferenceFrame]] = collections.defaultdict(list)
310 for val in accepts:
311 match val:
312 case (obj, *frames):
313 supported[obj].extend(frames)
314 case obj:
315 supported[obj].clear()
317 # Check whether this target is directly supported.
318 if mytype in supported:
319 # If it has a reference frame, check that too.
320 if not supported[mytype] or self.frame in supported[mytype]:
321 return self
323 if mytype is CatalogTarget:
324 # TODO: Do catalog lookup.
325 raise NotImplementedError("Catalog lookup is not yet supported")
327 # Fixed and rate targets have no trajectory and fall through to the error below.
328 trajectory = self.to_trajectory()
330 # Prefer a track over a materialized ephemeris where both are accepted.
331 if trajectory is not None and TargetTrack in supported:
332 frame = _accepted_frame(TargetTrack, supported, observer)
334 logger.debug(f"adapting {mytype.__name__} to TargetTrack in {frame}")
335 return TargetTrack(trajectory, frame, observer=observer, weather=weather)
337 if trajectory is not None and EphemerisTarget in supported:
338 frame = _accepted_frame(EphemerisTarget, supported, observer)
340 logger.debug(f"adapting {mytype.__name__} to EphemerisTarget in {frame}")
341 return await self.to_ephemeris_target(
342 start_time=datetime.now(UTC),
343 duration=_ADAPTED_EPHEMERIS_DURATION,
344 step=_ADAPTED_EPHEMERIS_STEP,
345 frame=frame,
346 observer=observer,
347 weather=weather,
348 )
350 raise RuntimeError("Could not adapt target to a supported type")
352 def to_trajectory(self) -> Trajectory | None:
353 """Return a Trajectory for this target, or None if it cannot be propagated."""
354 return None
356 def to_track(
357 self,
358 frame: ReferenceFrame = ReferenceFrame.GCRF,
359 observer: EarthObserver | Geodetic | None = None,
360 weather: BasicWeather | None = None,
361 ) -> TargetTrack:
362 """Return a TargetTrack sampling this target in the given frame.
364 Raises:
365 TypeError: If this target cannot be propagated.
366 """
367 trajectory = self.to_trajectory()
369 if trajectory is None:
370 raise TypeError(f"{type(self).__name__} cannot be propagated")
372 return TargetTrack(trajectory, frame, observer=observer, weather=weather)
374 async def to_ephemeris_target(
375 self,
376 start_time: datetime,
377 duration: timedelta,
378 step: timedelta,
379 frame: ReferenceFrame = ReferenceFrame.GCRF,
380 observer: EarthObserver | Geodetic | None = None,
381 weather: BasicWeather | None = None,
382 ) -> EphemerisTarget:
383 """Propagate this target into a pre-computed EphemerisTarget over the given time window."""
384 track = self.to_track(frame, observer, weather)
385 jds, points = await track.sample(start_time, duration, step)
387 return EphemerisTarget(
388 frame=frame,
389 jds=jds,
390 points=points,
391 )
394class CompositeTarget(BaseTarget):
395 """A sequence of targets.
397 The timing semantics of constituent targets are undefined.
398 """
399 target_type: Literal["composite"] = "composite"
400 sequence: Sequence[BaseTarget]
403class FrameTarget(BaseTarget):
404 """An unspecified target in a particular reference frame.
406 An object of type `FrameTarget` (as opposed to one of its subclasses) does not specify the
407 position within the target reference frame. The default semantics in this case are that the
408 target refers to the previous position in the user context converted to the target reference
409 frame. If there is no previous position in this context, use of this target is considered an
410 error.
411 """
412 target_type: Literal["frame"] = "frame"
413 frame: ReferenceFrame
416class FixedTarget[T: Coordinates](FrameTarget, ABC):
417 """A target at a fixed position in the given reference frame."""
418 target_type: Literal["fixed"] = "fixed"
419 coords: T
421 @abstractmethod
422 def to_astropy(self, **kwargs) -> SkyCoord:
423 """Convert this fixed target to an astropy SkyCoord."""
424 ...
427class AltAzTarget(FixedTarget[Horizontal]):
428 """A target at a fixed position in the alt-azimuth frame."""
429 frame: Literal[ReferenceFrame.ALTAZ] = ReferenceFrame.ALTAZ
431 @override
432 def to_astropy(
433 self,
434 time: Time | None = None,
435 observer: EarthObserver | Geodetic | None = None,
436 weather: BasicWeather | None = None,
437 wavelength: float | None = None,
438 ):
439 return self.coords.to_astropy(
440 obstime=time,
441 location=EarthLocation(observer.to_astropy()) if observer else None,
442 obswl=wavelength,
443 **(weather.to_astropy() if weather else {}),
444 )
447class ICRSTarget(FixedTarget[Equatorial]):
448 """A target at a fixed position in the International Celestial Reference Frame."""
449 frame: Literal[ReferenceFrame.ICRF] = ReferenceFrame.ICRF
451 @override
452 def to_astropy(
453 self,
454 time: Time | None = None,
455 ):
456 return self.coords.to_astropy(frame=self.frame, obstime=time)
459class RateTarget(FrameTarget):
460 """A target moving at a fixed rate relative to an initial position."""
461 target_type: Literal["rate"] = "rate"
462 rates: Coordinates
463 initial_time: datetime
464 initial_frame: ReferenceFrame
465 initial_coords: Coordinates
468class EphemerisTarget(FrameTarget):
469 """A target moving according to a precomputed ephemeris."""
470 target_type: Literal["ephemeris"] = "ephemeris"
471 jds: Sequence[float]
472 points: Sequence[Coordinates]
473 # FIXME: Needs velocity too.
476class TLETarget(FrameTarget):
477 """A target moving according to the input Two-Line Element set."""
478 target_type: Literal["tle"] = "tle"
479 frame: Literal[ReferenceFrame.TEME] = ReferenceFrame.TEME
480 tle: TLE
482 @override
483 def to_trajectory(self):
484 return TLETrajectory(self.tle)
487class StateVectorTarget(FrameTarget):
488 """A target moving according to the input state vector.
490 Position units must be meters, and velocity units must be meters per second.
491 """
492 target_type: Literal["state_vector"] = "state_vector"
493 sv: StateVector
495 @override
496 def to_trajectory(self):
497 return OrbitalTrajectory(self.sv_gcrf())
499 def sv_gcrf(self):
500 """Return the state vector converted to GCRF, transforming if necessary."""
501 if self.frame == ReferenceFrame.GCRF:
502 return self.sv
504 return StateVector.from_astropy(
505 self.sv.to_astropy(frame=self.frame).transform_to("gcrs")
506 )
509class CatalogTarget(BaseTarget):
510 """A target identified by name in an astronomical catalog."""
511 target_type: Literal["catalog"] = "catalog"
512 object: str
515Target = Annotated[
516 Annotated[AltAzTarget | ICRSTarget, Discriminator("frame")]
517 | RateTarget
518 | EphemerisTarget
519 | TLETarget
520 | StateVectorTarget
521 | CatalogTarget
522 | FrameTarget,
523 Discriminator("target_type"),
524]
527@dataclass
528class ObserveWindow:
529 """A time window for observability calculations, sampled at *step_seconds* intervals."""
530 start_time: datetime
531 end_time: datetime
532 step_seconds: int = 60
535def time_grid(win: ObserveWindow) -> list[datetime]:
536 """Return a list of timezone-aware datetimes spanning the window at the configured step."""
537 if win.start_time.tzinfo is None or win.end_time.tzinfo is None:
538 raise ValueError("start_time and end_time must be timezone-aware (UTC).")
539 if win.end_time < win.start_time:
540 raise ValueError("end_time must be >= start_time.")
541 if win.step_seconds <= 0:
542 raise ValueError("step_seconds must be > 0.")
544 t = win.start_time
545 out: list[datetime] = []
546 while t <= win.end_time:
547 out.append(t)
548 t += timedelta(seconds=win.step_seconds)
549 return out
552def make_altaz_frame(loc: EarthLocation, obstime: Time) -> AltAzFrame:
553 """Return an astropy AltAz frame for the given location and time."""
554 return AltAzFrame(obstime=obstime, location=loc)
556def norm_az_deg(az_deg: float) -> float:
557 """Normalise an azimuth value to the [0, 360) degree range."""
558 return az_deg % 360.0
561def sample_altaz_series(
562 target: Target,
563 loc: EarthLocation,
564 times: list[datetime],
565) -> list[tuple[float, float]]:
566 """Return a list of `(altitude_deg, azimuth_deg)` tuples for the target at each time."""
567 out: list[tuple[float, float]] = []
569 match target:
570 case ICRSTarget():
571 sc = SkyCoord(
572 ra=target.coords.ra * u.deg,
573 dec=target.coords.dec * u.deg,
574 frame=ICRS(),
575 )
576 for dt in times:
577 t_ast = Time(dt, scale="utc")
578 a = sc.transform_to(make_altaz_frame(loc, t_ast))
579 out.append((a.alt.deg, norm_az_deg(a.az.deg)))
580 return out
582 case AltAzTarget():
583 az = norm_az_deg(target.coords.az)
584 return [(target.coords.alt, az) for _ in times]
586 case TLETarget():
587 tle = satkit.TLE.from_lines(target.tle.to_list())
588 for dt in times:
589 t_ast = Time(dt, scale="utc")
590 t_sk = satkit.time.from_datetime(dt)
591 try:
592 teme_p, _teme_v = satkit.sgp4(tle, t_sk)
593 except Exception:
594 out.append((-90.0, 0.0))
595 continue
596 teme = TEME(x=teme_p[0]*u.km, y=teme_p[1]*u.km, z=teme_p[2]*u.km, obstime=t_ast)
597 a = teme.transform_to(make_altaz_frame(loc, t_ast))
598 out.append((a.alt.deg, norm_az_deg(a.az.deg)))
599 return out
601 case EphemerisTarget(frame=ReferenceFrame.ICRF):
602 if not target.jds:
603 raise ValueError("EphemerisTarget has no samples")
604 jds = np.asarray(target.jds, dtype=float)
605 for dt in times:
606 t_ast = Time(dt, scale="utc")
607 i = int(np.argmin(np.abs(jds - t_ast.jd)))
608 pt = target.points[i]
609 sc = SkyCoord(ra=pt.ra * u.deg, dec=pt.dec * u.deg, frame=ICRS())
610 a = sc.transform_to(make_altaz_frame(loc, t_ast))
611 out.append((a.alt.deg, norm_az_deg(a.az.deg)))
612 return out
614 raise TypeError(f"Unsupported target type: {type(target)}")
617def altitude_mask(altaz: list[tuple[float, float]], *, min_altitude_deg: float) -> np.ndarray:
618 """Return a boolean mask that is True where altitude >= *min_altitude_deg*."""
619 return np.array([alt >= min_altitude_deg for alt, _ in altaz], dtype=bool)
621def darkness_mask(
622 times: list[datetime],
623 loc: EarthLocation,
624 max_sun_alt_deg: float,
625) -> np.ndarray:
626 """Return a boolean mask that is True where the Sun is at or below *max_sun_alt_deg*."""
627 mask = np.zeros(len(times), dtype=bool)
628 for i, dt in enumerate(times):
629 t_ast = Time(dt, scale="utc")
630 sun_alt = get_sun(t_ast).transform_to(make_altaz_frame(loc, t_ast)).alt.deg
631 mask[i] = (sun_alt <= max_sun_alt_deg)
632 return mask
635def sun_avoidance_mask(
636 altaz: list[tuple[float, float]],
637 times: list[datetime],
638 loc: EarthLocation,
639 min_sun_sep_deg: float,
640) -> np.ndarray:
641 """Return a boolean mask that is True where the target is at least *min_sun_sep_deg* from the Sun."""
642 mask = np.zeros(len(times), dtype=bool)
643 for i, dt in enumerate(times):
644 alt_deg, az_deg = altaz[i]
645 t_ast = Time(dt, scale="utc")
646 frame = make_altaz_frame(loc, t_ast)
647 tgt = SkyCoord(AltAzFrame(az=az_deg * u.deg, alt=alt_deg * u.deg, obstime=t_ast, location=loc))
648 sun = get_sun(t_ast).transform_to(frame)
649 mask[i] = tgt.separation(sun).deg >= min_sun_sep_deg
650 return mask
652def moon_avoidance_mask(
653 altaz: list[tuple[float, float]],
654 times: list[datetime],
655 loc: EarthLocation,
656 min_moon_sep_deg: float,
657) -> np.ndarray:
658 """Return a boolean mask that is True where the target is at least *min_moon_sep_deg* from the Moon."""
659 mask = np.zeros(len(times), dtype=bool)
660 for i, dt in enumerate(times):
661 alt_deg, az_deg = altaz[i]
662 t_ast = Time(dt, scale="utc")
663 frame = make_altaz_frame(loc, t_ast)
664 tgt = SkyCoord(AltAzFrame(az=az_deg * u.deg, alt=alt_deg * u.deg, obstime=t_ast, location=loc))
665 moon = get_body("moon", t_ast, location=loc).transform_to(frame)
666 mask[i] = tgt.separation(moon).deg >= min_moon_sep_deg
667 return mask
669def is_observable(
670 target: Target,
671 site: Geodetic,
672 start_time: datetime,
673 end_time: datetime,
674 step_seconds: int = 10,
675 min_altitude_deg: float | None = None,
676 sun_max_altitude_deg: float | None = None,
677 sun_separation_deg: float | None = None,
678 moon_separation_deg: float | None = None,
679) -> bool:
680 """Return True if the target satisfies all supplied observability constraints throughout the window."""
681 win = ObserveWindow(start_time, end_time, step_seconds)
683 times = time_grid(win)
684 if not times:
685 return False
687 loc = EarthLocation(site.to_astropy())
689 try:
690 altaz = sample_altaz_series(target, loc, times)
691 except TypeError:
692 logger.warning(f"Observability check for {type(target).__name__} is not implemented!")
693 return True
695 mask = np.ones(len(times), dtype=bool)
697 if min_altitude_deg is not None:
698 mask &= altitude_mask(altaz, min_altitude_deg=min_altitude_deg)
700 if sun_max_altitude_deg is not None:
701 mask &= darkness_mask(times, loc, max_sun_alt_deg=sun_max_altitude_deg)
703 if sun_separation_deg is not None:
704 mask &= sun_avoidance_mask(altaz, times, loc, min_sun_sep_deg=sun_separation_deg)
706 if moon_separation_deg is not None:
707 mask &= moon_avoidance_mask(altaz, times, loc, min_moon_sep_deg=moon_separation_deg)
709 return np.all(mask)