import os
from django.http import HttpResponse, JsonResponse, Http404
from django.shortcuts import get_object_or_404, render
from lofar.sas.tmss.tmss.tmssapp import models
from lofar.common.json_utils import get_default_json_object_for_schema
from lofar.common.datetimeutils import formatDatetime
from lofar.sas.tmss.tmss.tmssapp.adapters.parset import convert_to_parset
from drf_yasg.utils import swagger_auto_schema
from drf_yasg.openapi import Parameter
from rest_framework.permissions import AllowAny
from rest_framework.decorators import authentication_classes, permission_classes
from django.apps import apps
from rest_framework.decorators import api_view
from datetime import datetime
import dateutil.parser
from astropy.coordinates import SkyCoord
from astropy import units as u
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
# Note: Decorate with @api_view to get this picked up by Swagger
def subtask_template_default_specification(request, subtask_template_pk:int):
subtask_template = get_object_or_404(models.SubtaskTemplate, pk=subtask_template_pk)
spec = get_default_json_object_for_schema(subtask_template.schema)
return JsonResponse(spec)
def task_template_default_specification(request, task_template_pk:int):
task_template = get_object_or_404(models.TaskTemplate, pk=task_template_pk)
spec = get_default_json_object_for_schema(task_template.schema)
return JsonResponse(spec)
def subtask_parset(request, subtask_pk:int):
subtask = get_object_or_404(models.Subtask, pk=subtask_pk)
parset = convert_to_parset(subtask)
return HttpResponse(str(parset), content_type='text/plain')
def index(request):
return render(request, os.path.join(os.environ.get('LOFARROOT'), 'SAS/TMSS/frontend','tmss_webapp/build/index.html'))
#return render(request, "../../../frontend/frontend_poc/build/index.html")
def task_specify_observation(request, pk=None):
task = get_object_or_404(models.TaskDraft, pk=pk)
return HttpResponse("response", content_type='text/plain')
# Allow everybody to GET our publicly available template-json-schema's
@permission_classes([AllowAny])
@authentication_classes([AllowAny])
@swagger_auto_schema(method='GET',
responses={200: 'Get the JSON schema from the template with the requested <template>, <name> and <version>',
404: 'the schema with requested <template>, <name> and <version> is not available'},
operation_description="Get the JSON schema for the given <template> with the given <name> and <version> as application/json content response.")
@api_view(['GET'])
def get_template_json_schema(request, template:str, name:str, version:str):
template_model = apps.get_model("tmssapp", template)
template_instance = get_object_or_404(template_model, name=name, version=version)
schema = template_instance.schema
response = JsonResponse(schema, json_dumps_params={"indent":2})
# config Access-Control. Our schemas use $ref url's to other schemas, mainly pointing to our own common schemas with base definitions.
# We instruct the client to allow fetching those.
response["Access-Control-Allow-Origin"] = "*"
response["Access-Control-Allow-Methods"] = "GET, OPTIONS"
return response
# Allow everybody to GET our publicly available station group lookups
@permission_classes([AllowAny])
@authentication_classes([AllowAny])
@swagger_auto_schema(method='GET',
responses={200: 'A JSON object with two properties: group:<the_group_name>, stations:<the_list_of_stations>',
404: 'No such group or template available'},
operation_description="Get a JSON list of stations for the given <station_group> name the the group definitions in the common_schema_template given by <template_name> and <template_version>")
@api_view(['GET'])
def get_stations_in_group(request, template_name:str, template_version:str, station_group:str):
station_schema_template = get_object_or_404(models.CommonSchemaTemplate, name=template_name, version=template_version)
station_schema = station_schema_template.schema
if 'station_group' not in station_schema.get('definitions', {}):
raise Http404('The JSON schema in template %s version %s has no station_group definitions' % (template_name, template_version))
groups = station_schema['definitions']['station_group']['anyOf']
try:
selected_group = next(g for g in groups if g['title'].lower() == station_group.lower())
except StopIteration:
raise Http404('No station_group with name "%s" found in the JSON schema. template=%s version=%s' % (station_group, template_name, template_version))
stations = selected_group['properties']['stations']['enum'][0]
return JsonResponse({'group': station_group,
'stations': stations})
@permission_classes([AllowAny])
@authentication_classes([AllowAny])
@swagger_auto_schema(method='GET',
responses={200: 'An isoformat timestamp of the current UTC clock of the system'},
operation_description="Get the current system time in UTC")
@api_view(['GET'])
def utc(request):
return HttpResponse(datetime.utcnow().isoformat(), content_type='text/plain')
@permission_classes([AllowAny])
@authentication_classes([AllowAny])
@swagger_auto_schema(method='GET',
responses={200: 'The LST time in hms format at the given UTC time and station or longitude'},
operation_description="Get LST time for UTC time and station or longitude",
manual_parameters=[Parameter(name='station', required=False, type='string', in_='query',
description="A station names (defaults to CS002)"),
Parameter(name='timestamp', required=False, type='string', in_='query',
description="A timestamp in isoformat (defaults to utcnow)"),
Parameter(name='longitude', required=False, type='string', in_='query',
description="A longitude as float")
])
@api_view(['GET'])
def lst(request):
# Handling optional parameters via django paths in urls.py is a pain, we access them on the request directly instead.
timestamp = request.GET.get('timestamp', None)
station = request.GET.get('station', None)
longitude = request.GET.get('longitude', None)
# conversions
if timestamp:
timestamp = dateutil.parser.parse(timestamp) # isot to datetime
if longitude:
longitude = float(longitude)
if station:
lst_lon = local_sidereal_time_for_utc_and_station(timestamp, station)
elif longitude:
lst_lon = local_sidereal_time_for_utc_and_longitude(timestamp, longitude)
else:
# fall back to defaults
lst_lon = local_sidereal_time_for_utc_and_station(timestamp)
# todo: do we want to return a dict, so users can make sure their parameters were parsed correctly instead?
return HttpResponse(str(lst_lon), content_type='text/plain')
@permission_classes([AllowAny])
@authentication_classes([AllowAny])
@swagger_auto_schema(method='GET',
responses={200: 'A JSON object with sunrise, sunset, day and night of the given stations at the given timestamps'},
operation_description="Get sunrise, sunset, day and night for stations and timestamps.\n\n"
"Example request: /api/util/sun_rise_and_set?stations=CS002,CS005×tamps=2020-05-01,2020-09-09T11-11-00",
manual_parameters=[Parameter(name='stations', required=False, type='string', in_='query',
description="comma-separated list of station names"),
Parameter(name='timestamps', required=False, type='string', in_='query',
description="comma-separated list of isoformat timestamps")])
@api_view(['GET'])
def get_sun_rise_and_set(request):
"""
returns sunrise and sunset at the given stations and timestamps, or today at LOFAR core if none specified.
example request: /api/util/sun_rise_and_set?stations=CS002,CS005×tamps=2020-05-01,2020-09-09T11-11-00
"""
timestamps = request.GET.get('timestamps', None)
stations = request.GET.get('stations', None)
if timestamps is None:
timestamps = [datetime.utcnow()]
else:
timestamps = timestamps.split(',')
timestamps = [dateutil.parser.parse(timestamp) for timestamp in timestamps] # isot to datetime
if stations is None:
stations = ["CS002"]
else:
stations = stations.split(',')
return JsonResponse(timestamps_and_stations_to_sun_rise_and_set(timestamps, stations))
@permission_classes([AllowAny])
@authentication_classes([AllowAny])
@swagger_auto_schema(method='GET',
responses={200: 'A JSON object with angular distances of the given sky coordinates from the given solar system bodies at the given timestamps and stations. \n'
'Outer list contains results per coordinate in given order, inner list per timestamp.'},
operation_description="Get angular distances of the given sky coordinates from the given solar system bodies at all given timestamps and stations. \n\n"
"Example request: /api/util/sun_rise_and_set?coordinates=,CS005×tamps=2020-05-01,2020-09-09T11-11-00",
manual_parameters=[Parameter(name='ras', required=True, type='string', in_='query',
description="comma-separated list of right ascensions (celestial longitudes)"),
Parameter(name='decs', required=True, type='string', in_='query',
description="comma-separated list of declinations (celestial latitude)"),
Parameter(name='stations', required=False, type='string', in_='query',
description="comma-separated list of station names"),
Parameter(name='timestamps', required=False, type='string', in_='query',
description="comma-separated list of isoformat timestamps"),
Parameter(name='bodies', required=False, type='string', in_='query',
description="comma-separated list of solar system bodies")])
@api_view(['GET'])
def get_angular_separation_from_bodies(request):
'''
returns angular distances of the given sky coordinates from the given astronomical objects at the given timestamps and stations
'''
stations = request.GET.get('stations', None)
timestamps = request.GET.get('timestamps', None)
ras = request.GET.get('ras', None)
decs = request.GET.get('decs', None)
bodies = request.GET.get('bodies', None)
if timestamps is None:
timestamps = [datetime.utcnow()]
else:
timestamps = timestamps.split(',')
timestamps = [dateutil.parser.parse(timestamp) for timestamp in timestamps] # isot to datetime
if bodies is None:
bodies = ['sun', 'moon', 'jupiter']
else:
bodies = bodies.split(',')
if stations is None:
stations = ["CS002"]
else:
stations = stations.split(',')
if ras is None or decs is None:
raise ValueError("Please provide celestial coordinates in radians/J2000 via the 'ras' and 'decs' the properties.")
coords = []
longitudes = ras.split(',')
latitudes = decs.split(',')
for ra, dec in zip(longitudes, latitudes):
coords.append(SkyCoord(ra=ra, dec=dec, unit=u.rad, equinox='J2000'))
sep_dict = coordinates_and_timestamps_to_separation_from_bodies(coords=coords, stations=stations, bodies=bodies, timestamps=timestamps)
for station, v in sep_dict.items():
for coord in v:
for object, angles in coord.items():
coord[object] = [angle.rad for angle in angles]
return JsonResponse(sep_dict)