Implement support for targets defined by TLE (Two-Line Element set)

astroplan/conftest.py

@@ -41,6 +41,7 @@
     PYTEST_HEADER_MODULES['pyephem'] = 'ephem'
     PYTEST_HEADER_MODULES['matplotlib'] = 'matplotlib'
     PYTEST_HEADER_MODULES['pytest-mpl'] = 'pytest_mpl'
+    PYTEST_HEADER_MODULES['skyfield'] = 'skyfield'
     del PYTEST_HEADER_MODULES['h5py']
 except KeyError:
     pass

astroplan/constraints.py

@@ -258,16 +258,17 @@ def __call__(self, observer, targets, times=None,
                                          time_resolution=time_grid_resolution)
 
         if grid_times_targets:
-            targets = get_skycoord(targets)
+            targets = get_skycoord(targets, times)
             # TODO: these broadcasting operations are relatively slow
             # but there is potential for huge speedup if the end user
             # disables gridding and re-shapes the coords themselves
             # prior to evaluating multiple constraints.
-            if targets.isscalar:
-                # ensure we have a (1, 1) shape coord
-                targets = SkyCoord(np.tile(targets, 1))[:, np.newaxis]
-            else:
-                targets = targets[..., np.newaxis]
+            if not observer._is_broadcastable(targets.shape, times.shape):
+                if targets.isscalar:
+                    # ensure we have a (1, 1) shape coord
+                    targets = SkyCoord(np.tile(targets, 1))[:, np.newaxis]
+                else:
+                    targets = targets[..., np.newaxis]
         times, targets = observer._preprocess_inputs(times, targets, grid_times_targets=False)
         result = self.compute_constraint(times, observer, targets)
 

astroplan/exceptions.py

@@ -42,3 +42,8 @@ class PlotBelowHorizonWarning(PlotWarning):
 class MissingConstraintWarning(AstroplanWarning):
     """Triggered when a constraint is expected but not supplied"""
     pass
+
+
+class InvalidTLEDataWarning(AstroplanWarning):
+    """TLE data invalid for the requested time"""
+    pass

astroplan/observer.py

@@ -516,8 +516,8 @@ def _preprocess_inputs(self, time, target=None, grid_times_targets=False):
             return time, None
 
         # convert any kind of target argument to non-scalar SkyCoord
-        target = get_skycoord(target)
-        if grid_times_targets:
+        target = get_skycoord(target, time)
+        if grid_times_targets and not self._is_broadcastable(target.shape, time.shape):
             if target.isscalar:
                 # ensure we have a (1, 1) shape coord
                 target = SkyCoord(np.tile(target, 1))[:, np.newaxis]

astroplan/scheduling.py

@@ -17,7 +17,7 @@
 
 from .utils import time_grid_from_range, stride_array
 from .constraints import AltitudeConstraint
-from .target import get_skycoord
+from .target import get_skycoord, FixedTarget, TLETarget
 
 __all__ = ['ObservingBlock', 'TransitionBlock', 'Schedule', 'Slot',
            'Scheduler', 'SequentialScheduler', 'PriorityScheduler',
@@ -121,7 +121,6 @@ def __init__(self, blocks, observer, schedule, global_constraints=[]):
         self.observer = observer
         self.schedule = schedule
         self.global_constraints = global_constraints
-        self.targets = get_skycoord([block.target for block in self.blocks])
 
     def create_score_array(self, time_resolution=1*u.minute):
         """
@@ -142,6 +141,7 @@ def create_score_array(self, time_resolution=1*u.minute):
         start = self.schedule.start_time
         end = self.schedule.end_time
         times = time_grid_from_range((start, end), time_resolution)
+        targets = get_skycoord([block.target for block in self.blocks], times)
         score_array = np.ones((len(self.blocks), len(times)))
         for i, block in enumerate(self.blocks):
             # TODO: change the default constraints from None to []
@@ -151,7 +151,7 @@ def create_score_array(self, time_resolution=1*u.minute):
                                                times=times)
                     score_array[i] *= applied_score
         for constraint in self.global_constraints:
-            score_array *= constraint(self.observer, self.targets, times,
+            score_array *= constraint(self.observer, targets, times,
                                       grid_times_targets=True)
         return score_array
 
@@ -273,40 +273,49 @@ def to_table(self, show_transitions=True, show_unused=False):
         start_times = []
         end_times = []
         durations = []
-        ra = []
-        dec = []
+        coordiante_type = []
+        coordinate_info = []
         config = []
         for slot in self.slots:
             if hasattr(slot.block, 'target'):
                 start_times.append(slot.start.iso)
                 end_times.append(slot.end.iso)
-                durations.append(slot.duration.to(u.minute).value)
-                target_names.append(slot.block.target.name)
-                ra.append(u.Quantity(slot.block.target.ra))
-                dec.append(u.Quantity(slot.block.target.dec))
+                durations.append('{:.4f}'.format(slot.duration.to(u.minute).value))
+                target = slot.block.target
+                target_names.append(target.name)
+                if isinstance(target, FixedTarget):
+                    coordiante_type.append("RA/Dec")
+                    coordinate_info.append(target.coord.to_string('hmsdms'))
+                elif isinstance(target, TLETarget):
+                    coordiante_type.append("TLE")
+                    coordinate_info.append(
+                        f"#{target.satellite.model.satnum} "
+                        f"epoch {target.satellite.epoch.utc_strftime(format='%Y-%m-%d %H:%M:%S')}"
+                    )
                 config.append(slot.block.configuration)
             elif show_transitions and slot.block:
                 start_times.append(slot.start.iso)
                 end_times.append(slot.end.iso)
-                durations.append(slot.duration.to(u.minute).value)
+                durations.append('{:.4f}'.format(slot.duration.to(u.minute).value))
                 target_names.append('TransitionBlock')
-                ra.append('')
-                dec.append('')
+                coordiante_type.append('')
+                coordinate_info.append('')
                 changes = list(slot.block.components.keys())
                 if 'slew_time' in changes:
                     changes.remove('slew_time')
                 config.append(changes)
             elif slot.block is None and show_unused:
                 start_times.append(slot.start.iso)
                 end_times.append(slot.end.iso)
-                durations.append(slot.duration.to(u.minute).value)
+                durations.append('{:.4f}'.format(slot.duration.to(u.minute).value))
                 target_names.append('Unused Time')
-                ra.append('')
-                dec.append('')
+                coordiante_type.append('')
+                coordinate_info.append('')
                 config.append('')
-        return Table([target_names, start_times, end_times, durations, ra, dec, config],
+        return Table([target_names, start_times, end_times,
+                      durations, coordiante_type, coordinate_info, config],
                      names=('target', 'start time (UTC)', 'end time (UTC)',
-                            'duration (minutes)', 'ra', 'dec', 'configuration'))
+                            'duration (min)', 'type', 'coordinates/info', 'configuration'))
 
     def new_slots(self, slot_index, start_time, end_time):
         """
@@ -999,9 +1008,8 @@ def __call__(self, oldblock, newblock, start_time, observer):
             # use the constraints cache for now, but should move that machinery
             # to observer
             from .constraints import _get_altaz
-            from .target import get_skycoord
             if oldblock.target != newblock.target:
-                targets = get_skycoord([oldblock.target, newblock.target])
+                targets = get_skycoord([oldblock.target, newblock.target], start_time)
                 aaz = _get_altaz(start_time, observer, targets)['altaz']
                 sep = aaz[0].separation(aaz[1])
                 if sep/self.slew_rate > 1 * u.second: