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 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
# 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))