""" Views ===== Views reacting to Http Requests by interfacing between backend and frontend. Functions --------- index(request) Home page """ from django.shortcuts import render from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseNotAllowed, HttpRequest from django.core.exceptions import BadRequest, ObjectDoesNotExist from .models import * from .forms import * import json from datetime import datetime from django.views.decorators.csrf import csrf_exempt from .class_names import * def get_timetable(r, trips_r, stop_sequence): """ Given a pt_map.models.Route, calculate the timetable for all its stops. Parameters ---------- r : pt_map.models.Route Route, the timetable should be calculated for trips : dict(str, list(pt_map.Trip)) Dictionary mapping all trips to route_ids they travel on stop_sequences : dict(str, list(str)) Dict mapping route_ids to lists of stop_ids they serve. Currently the first trip is taken as reference for stops and sequence. Returns ------- dict{"stop_sequence": list(str), "stop_times": dict(str, list(str)} Dict containing two elements: "stop_sequence" : list(str) list of stop_ids the route serves "stop_times" : dict(str, list(str)) dict mapping stop_ids from stop_sequence to time strings the route is serving the stop at """ timetable = {"stop_sequence": stop_sequence} sts = {} for stop in stop_sequence: times = [] for t in trips_r: for st in StopTime.objects.filter(trip_id=t.trip_id): times.append(st.departure_time.strftime("%H:%M")) sts[stop] = times timetable["stop_times"] = sts return timetable def index(request): stops = {s.stop_id: {name: getattr(s, name) for name in ['stop_name', 'stop_lat', 'stop_lon']} for s in Stop.objects.all()} route_name = lambda r : r.route_short_name if r.route_short_name else r.route_long_name routes = [{"route_id": r.route_id, "route_type": r.route_type, "route_name": route_name(r), "agency_id": r.agency_id.agency_id} for r in Route.objects.all()] trips = {r["route_id"]: [t for t in Trip.objects.filter(route_id_id=r["route_id"])] for r in routes} stop_sequences = {} for r in routes: seq = [] t = trips[r["route_id"]] for s in StopTime.objects.filter(trip_id_id__exact=t[0].trip_id): seq.append(s) stop_sequences[r["route_id"]] = [s.stop_id.stop_id for s in sorted(seq, key=lambda st : st.stop_sequence)] timetable = {} if request.GET.get("timetable"): try: r = Route.objects.get(route_id=request.GET.get("timetable")) timetable = get_timetable(r, trips[r.route_id], stop_sequences[r.route_id]) except Route.DoesNotExist: print(f"Invalid request for Route with id {request.GET['timetable']}") context = {"stops": json.dumps(stops), "routes": json.dumps(routes), "timetable": json.dumps(timetable)} return render(request,"map.html", context) def get_field_names(model: models.Model): return [field.name for field in model._meta.fields] def timetable(request): if request.method == "GET": try: r = Route.objects.get(route_id=request.GET["route_id"]) trips_r = [t for t in Trip.objects.filter(route_id_id=r.route_id)] stop_sequence = [s.stop_id.stop_id for s in sorted([s for s in StopTime.objects.filter(trip_id_id__exact=trips_r[0].trip_id)], key=lambda st : st.stop_sequence)] timetable = get_timetable(r, trips_r, stop_sequence) return HttpResponse(json.dumps(timetable), content_type="text/json") except KeyError: return HttpResponseBadRequest("route_id missing or malformed.") except Route.DoesNotExist: return HttpResponseBadRequest("Route not found.") return HttpResponseNotAllowed(["GET"]) def get_pks_from_get(req_get): result = {} for k in req_get.keys(): if k in classes_by_primary_keys.keys(): result[classes_by_primary_keys[k]] = req_get.getlist(k) return result def get_obj_by_pk(mdl: models.Model, pks: list[str]): return [obj for obj in [mdl.objects.get(**{primary_keys[mdl]: pk}) for pk in pks] if obj] def obj_from_get(req_get) -> str: return {mdl: get_obj_by_pk(mdl, keys) for mdl, keys in get_pks_from_get(req_get)} def mdl_to_jsnzbl_dict(cls, obj: models.Model) -> dict: result = {} fields = get_field_names(cls) for fk in fk_dict[cls]: fields.remove(fk[1]) if getattr(obj, fk[1]): result[fk[1]] = getattr(getattr(obj, fk[1]), primary_keys[fk[0]]) for field in fields: if getattr(obj, field): result[field] = getattr(obj, field) return result def json_from_pk(req_get) -> str: for cpk in classes_by_primary_keys: if req_get.get(cpk): for key in (req_get[cpk] if isinstance(req_get[cpk], list) else [req_get[cpk]]): obj = classes_by_primary_keys[cpk].objects.get(**{cpk: key}) for field in get_field_names(obj): v = getattr(obj, field) if v: if field in [f[1] for f in foreign_keys[classes_by_primary_keys[cpk]]]: fk = get_attr(v, field) return json.dumps(obj) def jsnzbl_from_get(req_get) -> str: return {mdl._meta.object_name: [mdl_to_jsnzbl_dict(mdl, o) for o in get_obj_by_pk(mdl, keys)] for mdl, keys in get_pks_from_get(req_get).items()} def jsnz(jsnzbl: dict): return json.dumps(jsnzbl) def rsp_for_json_from_get(req_get): try: return HttpResponse(jsnz(jsnzbl_from_get(req_get))) except ObjectDoesNotExist: return HttpResponseBadRequest("Object(s) not found.") def data(request): """ Handle database requests from the frontend. Using Http semantics to specify what to do with the data. Request ------- PUT Create a new object if no object with the given primary key exists in the database or delete and replace an existing object. Body must be a json dict of lists of fully specified, valid models. If primary keys are given as GET keys, they are applied to the first elements in their corresponding object lists. Primary keys in body will be ignored. RETURN 400 if primary keys not existing. PATCH Modify an existing objects given the instructions in the body. Body must be a json dict of lists of fields to change and their valid values existing objects in the database, identified by their valid primary keys as GET keys. Primary keys in body will be ignored. GET Return json of models identified by primary keys. DELETE Delete models with given primary keys if they exist. Returns number of deleted models. """ if request.method == "PUT": if not request.META["CONTENT_TYPE"] == 'application/json': HttpResponseBadRequest('Request must be JSON.') bdy = json.loads(request.body) try: obj = obj_from_get(request.GET) except ObjectDoesNotExist: return HttpResponseBadRequest("Did not find models corresponding to given pks.") return HttpResponse(f"received {obj}") elif request.method == "PATCH": if not request.META["CONTENT_TYPE"] == 'application/json': HttpResponseBadRequest('Request must be JSON.') return HttpResponse(f"received {obj}") elif request.method == "GET": return rsp_for_json_from_get(request.GET) return HttpResponseNotAllowed(['PUT', 'PATCH', 'GET'])