diff --git a/SAS/TMSS/backend/services/scheduling/lib/constraints/template_constraints_v1.py b/SAS/TMSS/backend/services/scheduling/lib/constraints/template_constraints_v1.py
index 7dcb48aef02d519b0aab7fb8bc620a6eb7028a9d..342b727554e0c3a5ca3212ab4008f8ecd116e752 100644
--- a/SAS/TMSS/backend/services/scheduling/lib/constraints/template_constraints_v1.py
+++ b/SAS/TMSS/backend/services/scheduling/lib/constraints/template_constraints_v1.py
@@ -33,7 +33,7 @@ from astropy.coordinates import Angle
import astropy.units
from lofar.sas.tmss.tmss.tmssapp import models
-from lofar.sas.tmss.tmss.tmssapp.conversions import create_astroplan_observer_for_station, Time, timestamps_and_stations_to_sun_rise_and_set, coordinates_and_timestamps_to_separation_from_bodies, coordinates_timestamps_and_stations_to_target_rise_and_set
+from lofar.sas.tmss.tmss.tmssapp.conversions import create_astroplan_observer_for_station, Time, timestamps_and_stations_to_sun_rise_and_set, coordinates_and_timestamps_to_separation_from_bodies, coordinates_timestamps_and_stations_to_target_rise_and_set, coordinates_timestamps_and_stations_to_target_transit, local_sidereal_time_for_utc_and_station
from lofar.sas.tmss.tmss.exceptions import TMSSException
from . import ScoredSchedulingUnit
@@ -265,7 +265,6 @@ def can_run_anywhere_within_timewindow_with_sky_constraints(scheduling_unit: mod
angle1 = beam['angle1']
angle2 = beam['angle2']
direction_type = beam['direction_type']
-
if 'min_distance' in constraints['sky']:
# currently we only check at bounds, we probably want to add some more samples in between later on
distances = coordinates_and_timestamps_to_separation_from_bodies(angle1=angle1, angle2=angle2, direction_type=direction_type, timestamps=(lower_bound, upper_bound), bodies=tuple(constraints['sky']['min_distance'].keys()))
@@ -299,6 +298,81 @@ def can_run_anywhere_within_timewindow_with_sky_constraints(scheduling_unit: mod
else:
logger.info('min_target_elevation=%s constraint is not met at timestamp=%s' % (min_elevation.rad, timestamps[i]))
return False
+ if 'transit_offset' in constraints['sky'] and 'from' in constraints['sky']['transit_offset'] and task['specifications_template'] == 'target observation':
+ # Check constraint on tile beam for HBA only:
+ if task['specifications_doc']['antenna_set'].startswith('HBA'):
+ # since the constraint only applies to the middle of the obs, consider its duration
+ if 'duration' in task['specifications_doc']:
+ duration = timedelta(seconds=task['specifications_doc']['duration'])
+ timestamps = (lower_bound + 0.5 * duration, upper_bound - 0.5 * duration)
+ else:
+ timestamps = (lower_bound, upper_bound)
+ station_groups = task['specifications_doc']['station_groups']
+ stations = list(set(sum([group['stations'] for group in station_groups], []))) # flatten all station_groups to single list
+ transit_times = coordinates_timestamps_and_stations_to_target_transit(angle1=angle1, angle2=angle2, direction_type=direction_type, timestamps=timestamps, stations=tuple(stations))
+ for station, times in transit_times.items():
+ for i in range(len(timestamps)):
+ offset = (timestamps[i] - times[i]).total_seconds()
+ offset_from = constraints['sky']['transit_offset']['from']
+ offset_to = constraints['sky']['transit_offset']['to']
+ # because the constraint allows specifying a window that reaches past 12h from transit,
+ # the transit that it refers to may not be the nearest transit to the observation time.
+ # Hence we also check if the constraint is met with 24h shift (which is approximately
+ # equivalent to checking the constraint for the previous or next transit)
+ if not ((offset_from < offset < offset_to) or
+ (offset_from+86400 < offset < offset_to+86400) or
+ (offset_from-86400 < offset < offset_to-86400)):
+ logger.info('transit_offset constraint from=%s to=%s is not met by offset=%s at timestamp=%s' % (offset_from, offset_to, offset, timestamps[i]))
+ return False
+
+ if 'SAPs' in task['specifications_doc']:
+ if 'transit_offset' in constraints['sky'] and 'from' in constraints['sky']['transit_offset'] and task['specifications_template'] == 'target observation':
+ # Check constraint on SAPs for LBA only:
+ if task['specifications_doc']['antenna_set'].startswith('LBA'):
+ # since the constraint only applies to the middle of the obs, consider its duration
+ if 'duration' in task['specifications_doc']:
+ duration = timedelta(seconds=task['specifications_doc']['duration'])
+ timestamps = (lower_bound + 0.5 * duration, upper_bound - 0.5 * duration)
+ else:
+ timestamps = (lower_bound, upper_bound)
+
+ # for LBA get transit times for all SAPs...
+ sap_transit_times = []
+ station_groups = task['specifications_doc']['station_groups']
+ stations = list(set(sum([group['stations'] for group in station_groups], []))) # flatten all station_groups to single list
+ for sap in task['specifications_doc']['SAPs']:
+ angle1 = sap['digital_pointing']['angle1']
+ angle2 = sap['digital_pointing']['angle2']
+ direction_type = sap['digital_pointing']['direction_type']
+ sap_transit_times.append(coordinates_timestamps_and_stations_to_target_transit(angle1=angle1, angle2=angle2, direction_type=direction_type, timestamps=timestamps, stations=tuple(stations)))
+
+ # ...then for each station and timestamp, average the transit times we got for the different SAPs
+ transit_times = {}
+ _reference_date = datetime(1900, 1, 1)
+ for station in stations:
+ for j in range(len(timestamps)):
+ sap_datetime_list = [sap_transit_times[i][station][j] for i in range(len(task['specifications_doc']['SAPs']))]
+ average_transit_time = _reference_date + sum([date - _reference_date for date in sap_datetime_list], timedelta()) / len(sap_datetime_list)
+ transit_times.get(station, []).append(average_transit_time)
+
+ logger.warning('##### %s' % transit_times)
+
+ for station, times in transit_times.items():
+ for i in range(len(timestamps)):
+ offset = (timestamps[i] - times[i]).total_seconds()
+ offset_from = constraints['sky']['transit_offset']['from']
+ offset_to = constraints['sky']['transit_offset']['to']
+ # because the constraint allows specifying a window that reaches past 12h from transit,
+ # the transit that it refers to may not be the nearest transit to the observation time.
+ # Hence we also check if the constraint is met with 24h shift (which is approximately
+ # equivalent to checking the constraint for the previous or next transit)
+ if not ((offset_from < offset < offset_to) or
+ (offset_from+86400 < offset < offset_to+86400) or
+ (offset_from-86400 < offset < offset_to-86400)):
+ logger.info('transit_offset constraint from=%s to=%s is not met by offset=%s at timestamp=%s' % (offset_from, offset_to, offset, timestamps[i]))
+ return False
+
+
return True
diff --git a/SAS/TMSS/backend/services/scheduling/test/t_dynamic_scheduling.py b/SAS/TMSS/backend/services/scheduling/test/t_dynamic_scheduling.py
index 77e62b3c4a2b4a9505563747c6f78c5c9fb1eaa9..59e644b4a882c4add00baa7b495fb95f41a524df 100755
--- a/SAS/TMSS/backend/services/scheduling/test/t_dynamic_scheduling.py
+++ b/SAS/TMSS/backend/services/scheduling/test/t_dynamic_scheduling.py
@@ -812,12 +812,19 @@ class TestSkyConstraints(unittest.TestCase):
{"rise": datetime(2020, 1, 1, 8, 0, 0), "set": datetime(2020, 1, 1, 12, 30, 0), "always_above_horizon": False, "always_below_horizon": False}]}
self.target_rise_and_set_data_always_above = {"CS002": [{"rise": None, "set": None, "always_above_horizon": True, "always_below_horizon": False}]}
self.target_rise_and_set_data_always_below = {"CS002": [{"rise": None, "set": None, "always_above_horizon": False, "always_below_horizon": True}]}
-
self.target_rise_and_set_patcher = mock.patch('lofar.sas.tmss.services.scheduling.constraints.template_constraints_v1.coordinates_timestamps_and_stations_to_target_rise_and_set')
self.target_rise_and_set_mock = self.target_rise_and_set_patcher.start()
self.target_rise_and_set_mock.return_value = self.target_rise_and_set_data
self.addCleanup(self.target_rise_and_set_patcher.stop)
+ self.target_transit_data = {"CS002": [datetime(2020, 1, 1, 14, 0, 0), datetime(2020, 1, 1, 14, 0, 0)]}
+ self.target_transit_data_previous = {"CS002": [datetime(2019, 12, 31, 14, 0, 0), datetime(2020, 1, 1, 14, 0, 0)]}
+ self.target_transit_data_saps = [{"CS001": [datetime(2020, 1, 1, 14, 0, 0), datetime(2020, 1, 1, 14, 0, 0)]}, {"CS001": [datetime(2020, 1, 1, 16, 0, 0), datetime(2020, 1, 1, 16, 0, 0)]}]
+ self.target_transit_patcher = mock.patch('lofar.sas.tmss.services.scheduling.constraints.template_constraints_v1.coordinates_timestamps_and_stations_to_target_transit')
+ self.target_transit_mock = self.target_transit_patcher.start()
+ self.target_transit_mock.return_value = self.target_transit_data
+ self.addCleanup(self.target_transit_patcher.stop)
+
# min_distance
def test_can_run_anywhere_within_timewindow_with_sky_constraints_with_min_distance_constraint_returns_true_when_met(self):
@@ -850,6 +857,70 @@ class TestSkyConstraints(unittest.TestCase):
returned_value = tc1.can_run_anywhere_within_timewindow_with_sky_constraints(self.scheduling_unit_blueprint, timestamp, timestamp + timedelta(seconds=self.obs_duration))
self.assertFalse(returned_value)
+ # transit_offset
+
+ def test_can_run_anywhere_within_timewindow_with_sky_constraints_with_transit_offset_constraint_returns_true_when_met(self):
+ # case 1: transits at 14h, obs middle is at 13h, so we have an offset of -3600 seconds
+
+ # big window
+ self.scheduling_unit_blueprint.draft.scheduling_constraints_doc['sky'] = {'transit_offset': {'from': -43200, 'to': 43200}} # todo: use blueprint contraints after TMSS-697 was merged
+ self.scheduling_unit_blueprint.save()
+ timestamp = datetime(2020, 1, 1, 12, 0, 0)
+ returned_value = tc1.can_run_anywhere_within_timewindow_with_sky_constraints(self.scheduling_unit_blueprint, timestamp, timestamp + timedelta(seconds=self.obs_duration))
+ self.assertTrue(returned_value)
+
+ # narrow window
+ self.scheduling_unit_blueprint.draft.scheduling_constraints_doc['sky'] = {'transit_offset': {'from': -3601, 'to': -3599}} # todo: use blueprint contraints after TMSS-697 was merged
+ self.scheduling_unit_blueprint.save()
+ timestamp = datetime(2020, 1, 1, 12, 0, 0)
+ returned_value = tc1.can_run_anywhere_within_timewindow_with_sky_constraints(self.scheduling_unit_blueprint, timestamp, timestamp + timedelta(seconds=self.obs_duration))
+ self.assertTrue(returned_value)
+
+ # case 2: transits at 14h, obs middle is at 2h, so we have an offset of -43200 seconds
+
+ # window spans past 12h, so reference transit is not nearest transit to obs time
+ self.target_transit_mock.return_value = self.target_transit_data_previous
+ self.scheduling_unit_blueprint.draft.scheduling_constraints_doc['sky'] = {'transit_offset': {'from': -43300, 'to': -43100}} # todo: use blueprint contraints after TMSS-697 was merged
+ self.scheduling_unit_blueprint.save()
+ timestamp = datetime(2020, 1, 1, 1, 0, 0)
+ returned_value = tc1.can_run_anywhere_within_timewindow_with_sky_constraints(self.scheduling_unit_blueprint, timestamp, timestamp + timedelta(seconds=self.obs_duration))
+ self.assertTrue(returned_value)
+ self.target_transit_mock.return_value = self.target_transit_data
+
+ def test_can_run_anywhere_within_timewindow_with_sky_constraints_with_transit_offset_constraint_returns_false_when_not_met(self):
+ # transits at 14h, obs middle is at 13h, so we have an offset of -3600 seconds
+
+ # window after
+ self.scheduling_unit_blueprint.draft.scheduling_constraints_doc['sky'] = {'transit_offset': {'from': -3599, 'to': 43200}} # todo: use blueprint contraints after TMSS-697 was merged
+ self.scheduling_unit_blueprint.save()
+ timestamp = datetime(2020, 1, 1, 12, 0, 0)
+ returned_value = tc1.can_run_anywhere_within_timewindow_with_sky_constraints(self.scheduling_unit_blueprint, timestamp, timestamp + timedelta(seconds=self.obs_duration))
+ self.assertFalse(returned_value)
+
+ # window before
+ self.scheduling_unit_blueprint.draft.scheduling_constraints_doc['sky'] = {'transit_offset': {'from': -43200, 'to': -3601}} # todo: use blueprint contraints after TMSS-697 was merged
+ self.scheduling_unit_blueprint.save()
+ timestamp = datetime(2020, 1, 1, 12, 0, 0)
+ returned_value = tc1.can_run_anywhere_within_timewindow_with_sky_constraints(self.scheduling_unit_blueprint, timestamp, timestamp + timedelta(seconds=self.obs_duration))
+ self.assertFalse(returned_value)
+
+
+ def test_can_run_anywhere_within_timewindow_with_sky_constraints_with_transit_offset_constraint_averages_SAPs_for_LBA(self):
+ # sap1 transits at 14h, sap2 transits at 16h, so average transit is at 15h
+ # obs middle is 13h, so we have an offset of -7200 seconds
+
+ self.target_transit_mock.side_effect = self.target_transit_data_saps
+ self.scheduling_unit_blueprint.draft.scheduling_constraints_doc['sky'] = {'transit_offset': {'from': -7201, 'to': -7199}} # todo: use blueprint contraints after TMSS-697 was merged
+ self.scheduling_unit_blueprint.requirements_doc['tasks']['Observation']['specifications_doc']['antenna_set'] = 'LBA_INNER'
+ self.scheduling_unit_blueprint.requirements_doc['tasks']['Observation']['specifications_doc']['SAPs'] = \
+ [{'name': 'CygA', 'target': 'CygA', 'subbands': [0, 1], 'digital_pointing': {'angle1': 5.233660650313663, 'angle2': 0.7109404782526458, 'direction_type': 'J2000'}},
+ {'name': 'CasA', 'target': 'CasA', 'subbands': [2, 3], 'digital_pointing': {'angle1': 6.233660650313663, 'angle2': 0.6109404782526458, 'direction_type': 'J2000'}}]
+ self.scheduling_unit_blueprint.save()
+ timestamp = datetime(2020, 1, 1, 12, 0, 0)
+ returned_value = tc1.can_run_anywhere_within_timewindow_with_sky_constraints(self.scheduling_unit_blueprint, timestamp, timestamp + timedelta(seconds=self.obs_duration))
+ self.assertTrue(returned_value)
+ self.target_transit_mock.side_effect = None
+
class TestTimeConstraints(TestCase):
"""
diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/conversions.py b/SAS/TMSS/backend/src/tmss/tmssapp/conversions.py
index 14b0a38e566666fda10ba8292bb9d4f91525afef..642e7090c070be2033a4af4c8404c137bbe2b771 100644
--- a/SAS/TMSS/backend/src/tmss/tmssapp/conversions.py
+++ b/SAS/TMSS/backend/src/tmss/tmssapp/conversions.py
@@ -35,7 +35,6 @@ SUN_SET_RISE_ANGLE_TO_HORIZON = Angle(10, unit=astropy.units.deg)
# TODO: To be considered, now we store the sunset/sunrise data in advanced, we can increase the number of points!!
SUN_SET_RISE_PRECISION = 30
-
def timestamps_and_stations_to_sun_rise_and_set(timestamps: tuple, stations: tuple, angle_to_horizon: Angle=SUN_SET_RISE_ANGLE_TO_HORIZON,
create_when_not_found=False) -> dict:
"""
@@ -268,6 +267,34 @@ def coordinates_timestamps_and_stations_to_target_rise_and_set(angle1: float, an
return return_dict
+# default n_grid_points; higher is more precise but very costly; astropy defaults to 150, note that errors can be in the minutes with a lower values
+TARGET_TRANSIT_PRECISION = 150
+
+@lru_cache(maxsize=256, typed=False) # does not like lists, so use tuples to allow caching
+def coordinates_timestamps_and_stations_to_target_transit(angle1: float, angle2: float, direction_type: str, timestamps: tuple, stations: tuple) -> dict:
+ """
+ Compute nearest meridian transit times of the given coordinates for each given station and timestamp.
+ :param angle1: first angle of celectial coordinates, e.g. RA
+ :param angle2: second angle of celectial coordinates, e.g. Dec
+ :param direction_type: direction_type of celectial coordinates, e.g. 'J2000'
+ :param timestamps: tuple of datetimes, e.g. (datetime(2020, 1, 1), datetime(2020, 1, 2))
+ :param stations: tuple of station names, e.g. ("CS002",)
+ :return A dict that maps station names to a list of transit times (nearest transit for each requested timestamp).
+ E.g.
+ {"CS002": [datetime(2020, 1, 1, 4, 0, 0), datetime(2020, 1, 2, 4, 0, 0)]}
+ """
+ if direction_type == "J2000":
+ coord = astropy.coordinates.SkyCoord(ra=angle1, dec=angle2, unit=astropy.units.rad)
+ else:
+ raise ValueError("Do not know how to convert direction_type=%s to SkyCoord" % direction_type)
+ return_dict = {}
+ for station in stations:
+ for timestamp in timestamps:
+ # todo: this can probably be made faster by moving the following logic to an own function with single station/timestamp as input and putting the lru_cache on there.
+ observer = create_astroplan_observer_for_station(station)
+ target_transit = observer.target_meridian_transit_time(target=coord, time=Time(timestamp), which='nearest', n_grid_points=TARGET_TRANSIT_PRECISION)
+ return_dict.setdefault(station, []).append(target_transit.to_datetime())
+ return return_dict
def local_sidereal_time_for_utc_and_station(timestamp: datetime = None,
diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/views.py b/SAS/TMSS/backend/src/tmss/tmssapp/views.py
index 85bdfe0de03a90428f85f01fb51264e4b4082b49..c043399964b788b809194e49c1c0b6872e57fdfe 100644
--- a/SAS/TMSS/backend/src/tmss/tmssapp/views.py
+++ b/SAS/TMSS/backend/src/tmss/tmssapp/views.py
@@ -18,7 +18,7 @@ from datetime import datetime
import dateutil.parser
from astropy.coordinates import Angle
import astropy.units
-from lofar.sas.tmss.tmss.tmssapp.conversions import local_sidereal_time_for_utc_and_station, local_sidereal_time_for_utc_and_longitude, timestamps_and_stations_to_sun_rise_and_set, coordinates_and_timestamps_to_separation_from_bodies, coordinates_timestamps_and_stations_to_target_rise_and_set
+from lofar.sas.tmss.tmss.tmssapp.conversions import local_sidereal_time_for_utc_and_station, local_sidereal_time_for_utc_and_longitude, timestamps_and_stations_to_sun_rise_and_set, coordinates_and_timestamps_to_separation_from_bodies, coordinates_timestamps_and_stations_to_target_rise_and_set, coordinates_timestamps_and_stations_to_target_transit
# Note: Decorate with @api_view to get this picked up by Swagger
@@ -278,3 +278,27 @@ def get_target_rise_and_set(request):
rise_set_dict = coordinates_timestamps_and_stations_to_target_rise_and_set(angle1=angle1, angle2=angle2, direction_type=direction_type, angle_to_horizon=horizon, timestamps=timestamps, stations=stations)
return JsonResponse(rise_set_dict)
+
+@api_view(['GET'])
+def get_target_transit(request):
+ '''
+ returns transit times of the given coordinates for each given station and timestamp.
+ '''
+ timestamps = request.GET.get('timestamps', None)
+ angle1 = request.GET.get('angle1')
+ angle2 = request.GET.get('angle2')
+ direction_type = request.GET.get("direction_type", "J2000")
+ stations = tuple(request.GET.get('stations', "CS002").split(','))
+
+ if angle1 is None or angle2 is None:
+ raise ValueError("Please provide celestial coordinates via 'angle1', 'angle2' (and optionally 'direction_type') properties.")
+
+ if timestamps is None:
+ timestamps = (datetime.utcnow(),)
+ else:
+ timestamps = timestamps.split(',')
+ timestamps = tuple([dateutil.parser.parse(timestamp, ignoretz=True) for timestamp in timestamps]) # isot to datetime
+
+ # calculate
+ transit_dict = coordinates_timestamps_and_stations_to_target_transit(angle1=angle1, angle2=angle2, direction_type=direction_type, timestamps=timestamps, stations=stations)
+ return JsonResponse(transit_dict)
diff --git a/SAS/TMSS/backend/src/tmss/urls.py b/SAS/TMSS/backend/src/tmss/urls.py
index 5306787cb405fa524cbb475cc7d7e76d1fe3c561..c077e51431b29da1484c0653421d54c27a7a5f91 100644
--- a/SAS/TMSS/backend/src/tmss/urls.py
+++ b/SAS/TMSS/backend/src/tmss/urls.py
@@ -75,6 +75,7 @@ urlpatterns = [
re_path('util/lst/?', views.lst, name="conversion-lst"),
re_path('util/angular_separation/?', views.get_angular_separation, name='get_angular_separation'),
re_path('util/target_rise_and_set/?', views.get_target_rise_and_set, name='get_target_rise_and_set'),
+ re_path('util/target_transit/?', views.get_target_transit, name='get_target_transit'),
]
if os.environ.get('SHOW_DJANGO_DEBUG_TOOLBAR', False):
diff --git a/SAS/TMSS/backend/test/t_conversions.py b/SAS/TMSS/backend/test/t_conversions.py
index ad873499f1271b74fd75cf7da3f59b39fc1cbecf..942fb172dd634dd99f95732459ac833b0e37a622 100755
--- a/SAS/TMSS/backend/test/t_conversions.py
+++ b/SAS/TMSS/backend/test/t_conversions.py
@@ -354,13 +354,13 @@ class UtilREST(unittest.TestCase):
def test_util_target_rise_and_set_considers_horizon(self):
test_horizons = [0.1, 0.2, 0.3]
+ rise_last = None
for horizon in test_horizons:
r = requests.get(BASE_URL + '/util/target_rise_and_set?angle1=0.5&angle2=0.5&horizon=%s' % horizon, auth=AUTH)
self.assertEqual(r.status_code, 200)
r_dict = json.loads(r.content.decode('utf-8'))
# assert all requested horizons yield a response and times differ
- rise_last = None
rise = r_dict['CS002'][0]['rise']
if rise_last:
self.assertNotEqual(rise, rise_last)
@@ -395,6 +395,76 @@ class UtilREST(unittest.TestCase):
self.assertFalse(r_dict['CS002'][0]['always_above_horizon'])
self.assertTrue(r_dict['CS002'][0]['always_below_horizon'])
+ # target transit
+
+ def test_util_target_transit_returns_json_structure_with_defaults(self):
+ r = requests.get(BASE_URL + '/util/target_transit?angle1=0.5&angle2=0.5', auth=AUTH)
+ self.assertEqual(r.status_code, 200)
+ r_dict = json.loads(r.content.decode('utf-8'))
+
+ # defaults are CS002 and today
+ self.assertIn('CS002', r_dict.keys())
+
+ # assert returned timestamp is no further than 12h away from now
+ expected_time = datetime.datetime.utcnow()
+ returned_time = dateutil.parser.parse(r_dict['CS002'][0])
+ time_diff = abs(expected_time - returned_time)
+ self.assertTrue(time_diff <= datetime.timedelta(days=0.5))
+
+ def test_util_target_transit_considers_stations(self):
+ stations = ['CS005', 'RS305', 'DE609']
+ r = requests.get(BASE_URL + '/util/target_transit?angle1=0.5&angle2=0.5&stations=%s' % ','.join(stations), auth=AUTH)
+ self.assertEqual(r.status_code, 200)
+ r_dict = json.loads(r.content.decode('utf-8'))
+
+ # assert station is included in response and timestamps differ
+ target_transit_last = None
+ for station in stations:
+ self.assertIn(station, r_dict.keys())
+ target_transit = dateutil.parser.parse(r_dict[station][0])
+ if target_transit_last:
+ self.assertNotEqual(target_transit, target_transit_last)
+ target_transit_last = target_transit
+
+ def test_util_target_transit_considers_timestamps(self):
+ timestamps = ['2020-01-01', '2020-02-22T16-00-00', '2020-3-11']
+ r = requests.get(BASE_URL + '/util/target_transit?angle1=0.5&angle2=0.5×tamps=%s' % ','.join(timestamps), auth=AUTH)
+ self.assertEqual(r.status_code, 200)
+ r_dict = json.loads(r.content.decode('utf-8'))
+
+ # assert all requested timestamps yield a different response
+ transit_last = None
+ for i in range(len(timestamps)):
+ transit = r_dict['CS002'][i]
+ if transit_last:
+ self.assertNotEqual(transit, transit_last)
+ transit_last = transit
+
+ def test_util_target_transit_returns_correct_date_of_target_transit(self):
+ timestamps = ['2020-01-01T02-00-00']
+ r = requests.get(BASE_URL + '/util/target_transit?angle1=0.5&angle2=0.5×tamps=%s' % ','.join(timestamps), auth=AUTH)
+ self.assertEqual(r.status_code, 200)
+ r_dict = json.loads(r.content.decode('utf-8'))
+
+ # assert transit time is no further than 12h from requested time
+ requested_time = dateutil.parser.parse(timestamps[0]).replace(tzinfo=None)
+ returned_time = dateutil.parser.parse(r_dict['CS002'][0])
+ time_diff = abs(requested_time - returned_time)
+ self.assertTrue(time_diff <= datetime.timedelta(days=0.5))
+
+ def test_util_target_transit_considers_coordinates(self):
+ test_coords = [(0.5, 0.5, "J2000"), (0.6, 0.5, "J2000"), (0.6, 0.6, "J2000")]
+ transit_last = None
+ for coords in test_coords:
+ r = requests.get(BASE_URL + '/util/target_transit?angle1=%s&angle2=%s&direction_type=%s' % coords, auth=AUTH)
+ self.assertEqual(r.status_code, 200)
+ r_dict = json.loads(r.content.decode('utf-8'))
+
+ # assert all requested coordinates yield a response and times differ
+ transit = r_dict['CS002'][0]
+ if transit_last:
+ self.assertNotEqual(transit, transit_last)
+ transit_last = transit
if __name__ == "__main__":
os.environ['TZ'] = 'UTC'