#628502899

Committed 27 Nov 2024 04:01AM UTC coverage: 87.7%. First build

Build # #628502899

Build Type

Pull #203

travis-ci

Pull Request Pull Request #203: Filter MTA real time alerts with subway lines

Run Details

0 of 7 new or added lines in 1 file covered. (0.0%)

1041 of 1187 relevant lines covered (87.7%)

0.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.64

/app/views.py

from django.shortcuts import render, redirect, get_object_or_404
from django.contrib.auth.forms import AuthenticationForm, UserCreationForm
from django.contrib.auth import login, logout
from django.contrib import messages
from django.views import generic
from django.conf import settings
from django.db.models import F
from .models import Station, Profile, Review
from django.contrib.auth.decorators import login_required
from .forms import ProfileUpdateForm, RatingForm
from django.utils import timezone
from django.http import JsonResponse
from google.transit import gtfs_realtime_pb2
import json
import requests
from django.core.exceptions import PermissionDenied
import datetime


# User Registration View
def register_view(request):
    if request.user.is_authenticated:
        return redirect("maps:map_view")

    if request.method == "POST":
        form = UserCreationForm(request.POST)
        if form.is_valid():
            user = form.save()

            # Create a profile for the user
            Profile.objects.create(user=user)

            login(request, user)
            messages.success(
                request, f"Account created successfully! Welcome, {user.username}!"
            )
            return redirect("maps:map_view")
        else:
            messages.error(request, "Registration failed. Please try again.")
    else:
        form = UserCreationForm()

    return render(request, "app/register.html", {"form": form})


# Login View
def login_view(request):
    if request.user.is_authenticated:
        return redirect("maps:map_view")

    if request.method == "POST":
        form = AuthenticationForm(request, data=request.POST)
        if form.is_valid():
            user = form.get_user()
            login(request, user)
            messages.success(
                request, f"Welcome, {user.username}! You are now logged in."
            )
            return redirect("maps:map_view")
        else:
            messages.error(request, "Invalid username or password.")
    else:
        form = AuthenticationForm()

    return render(request, "app/login.html", {"form": form})


# Logout View
@login_required
def logout_view(request):
    if request.method == "POST":
        logout(request)
        return redirect("maps:map_view")
    else:
        raise PermissionDenied


# Stations View
class StationsView(generic.ListView):
    paginate_by = 10
    model = Station
    template_name = "app/stations.html"
    context_object_name = "station_list"

    def get_queryset(self):
        # Get the search query from the request
        query = self.request.GET.get("q")
        ada_filter = self.request.GET.get("ada_filter")

        # Start with all stations
        queryset = Station.objects.all().order_by(F("stop_name").asc())

        # Apply search filter if applicable
        if query:
            queryset = queryset.filter(stop_name__icontains=query)

        # Apply ADA filter based on the selected option
        if ada_filter == "fully":
            queryset = queryset.filter(ada=True)
        elif ada_filter == "partially":
            queryset = queryset.filter(ada_southbound=True, ada_northbound=False)
        elif ada_filter == "not":
            queryset = queryset.filter(ada=False)

        return queryset


# Station Detail View
class StationDetailView(generic.DetailView):
    model = Station
    template_name = "app/station_detail.html"


def station_detail(request, station_id):
    station = get_object_or_404(Station, id=station_id)
    form = RatingForm(request.POST)

    already_reviewed = None

    if request.user.is_authenticated:
        user_reviewed = Review.objects.filter(
            station=station, user=request.user
        ).first()
        if user_reviewed:
            already_reviewed = True
        if request.method == "POST":
            if form.is_valid():
                if user_reviewed:
                    # Update the existing review
                    rating = user_reviewed
                    rating.rating = form.cleaned_data["rating"]
                    prev_comment = rating.comment
                    rating.comment = form.cleaned_data["comment"]
                    if rating.comment == "":
                        rating.comment = prev_comment
                    else:
                        rating.created_at = timezone.now()
                    rating.save()
                else:
                    rating = form.save(commit=False)
                    rating.station = station
                    rating.user = request.user
                    rating.created_at = timezone.now()
                    rating.save()
                messages.success(request, "Your review has been added!")
                return redirect("app:station_detail", station_id=station.id)
            else:
                print("Form errors: ", form.errors)
                messages.error(request, "Something went wrong")
        else:
            form = RatingForm(
                instance=user_reviewed
            )  # Pre-populate the form with the user's existing review (if any)

    ratings = station.rating.all()  # Get all ratings for this station
    # Calculate the average rating for this station

    avg_rating = Review.get_average_rating(station_name=station)

    # Find last 5 comments
    last_five_comments = (
        Review.objects.filter(comment__isnull=False, station=station)
        .exclude(comment="")
        .order_by("-created_at")[:5]
    )

    return render(
        request,
        "app/station_detail.html",
        {
            "station": station,
            "ratings": ratings,
            "comments": last_five_comments,
            "form": form,
            "avg_rating": avg_rating,
            "reviewed": already_reviewed,
        },
    )


class ProfileView(generic.DetailView):
    model = Profile
    template_name = "app/profile.html"
    context_object_name = "profile"

    def get_object(self):
        return self.request.user.profile


def alerts_view(request):
    # Get selected lines from the request
    selected_lines = request.GET.get("lines", "")
    lines_list = selected_lines.split(",") if selected_lines else []

    search_patterns = [f"[{line}]" for line in lines_list]

    url = (
        "https://api-endpoint.mta.info/Dataservice/mtagtfsfeeds/camsys%2Fsubway-alerts"
    )
    try:
        response = requests.get(url)

        # Decode the Protobuf data
        feed = gtfs_realtime_pb2.FeedMessage()
        feed.ParseFromString(response.content)

        # Extract and format alert data
        alerts_data = []
        for entity in feed.entity:
            if entity.HasField("alert"):
                alert = entity.alert

                header = (
                    alert.header_text.translation[0].text
                    if alert.header_text.translation
                    else ""
                )
                description = (
                    alert.description_text.translation[0].text
                    if alert.description_text.translation
                    else ""
                )

                # If lines are selected, filter based on the content
                if lines_list:
                    # Check if any of the search patterns are in the header
                    if any(pattern in header for pattern in search_patterns):
                        # Include this alert
                        alerts_data.append(
                            {
                                "header": header,
                                "description": description,
                            }
                        )
                else:
                    # No lines selected, include all alerts
                    alerts_data.append(
                        {
                            "header": header,
                            "description": description,
                        }
                    )

    except Exception as e:
        alerts_data = None
        print("Error fetching alerts data:", e)

    if request.headers.get("X-Requested-With") == "XMLHttpRequest":
        return JsonResponse({"alerts_data": alerts_data})
    return render(request, "app/alerts.html", {"alerts_data": alerts_data})


@login_required
def edit_profile(request):
    profile = request.user.profile
    if request.method == "POST":
        form = ProfileUpdateForm(request.POST, instance=profile)
        date = request.POST.get("birth_date")
        if date > datetime.date.today().isoformat():
            messages.error(request, "Invalid date")
            return redirect("app:edit_profile")
        if form.is_valid():
            form.save()
            messages.success(request, "Your profile has been updated.")
            return redirect("app:profile")
    else:
        form = ProfileUpdateForm(instance=profile)

    context = {
        "google_maps_api_key": settings.GOOGLE_MAPS_API_KEY,
        "form": form,
    }

    return render(request, "app/edit_profile.html", context)


@login_required
def save_favorite_route(request):
    if request.method == "POST":
        data = json.loads(request.body)
        start = data.get("start")
        end = data.get("end")

        print(start)
        profile = request.user.profile
        profile.fav_source_latitude = start["lat"]
        profile.fav_source_longitude = start["lng"]
        profile.fav_dest_latitude = end["lat"]
        profile.fav_dest_longitude = end["lng"]
        profile.save()

        return JsonResponse({"success": True})

    return JsonResponse({"success": False})

1	from django.shortcuts import render, redirect, get_object_or_404	1✔
2	from django.contrib.auth.forms import AuthenticationForm, UserCreationForm	1✔
3	from django.contrib.auth import login, logout	1✔
4	from django.contrib import messages	1✔
5	from django.views import generic	1✔
6	from django.conf import settings	1✔
7	from django.db.models import F	1✔
8	from .models import Station, Profile, Review	1✔
9	from django.contrib.auth.decorators import login_required	1✔
10	from .forms import ProfileUpdateForm, RatingForm	1✔
11	from django.utils import timezone	1✔
12	from django.http import JsonResponse	1✔
13	from google.transit import gtfs_realtime_pb2	1✔
14	import json	1✔
15	import requests	1✔
16	from django.core.exceptions import PermissionDenied	1✔
17	import datetime	1✔
18
19
20	# User Registration View
21	def register_view(request):	1✔
22	if request.user.is_authenticated:	1✔
23	return redirect("maps:map_view")	1✔
24
25	if request.method == "POST":	1✔
26	form = UserCreationForm(request.POST)	1✔
27	if form.is_valid():	1✔
28	user = form.save()	1✔
29
30	# Create a profile for the user
31	Profile.objects.create(user=user)	1✔
32
33	login(request, user)	1✔
34	messages.success(	1✔
35	request, f"Account created successfully! Welcome, {user.username}!"
36	)
37	return redirect("maps:map_view")	1✔
38	else:
39	messages.error(request, "Registration failed. Please try again.")	1✔
40	else:
41	form = UserCreationForm()	1✔
42
43	return render(request, "app/register.html", {"form": form})	1✔
44
45
46	# Login View
47	def login_view(request):	1✔
48	if request.user.is_authenticated:	1✔
49	return redirect("maps:map_view")	1✔
50
51	if request.method == "POST":	1✔
52	form = AuthenticationForm(request, data=request.POST)	1✔
53	if form.is_valid():	1✔
54	user = form.get_user()	1✔
55	login(request, user)	1✔
56	messages.success(	1✔
57	request, f"Welcome, {user.username}! You are now logged in."
58	)
59	return redirect("maps:map_view")	1✔
60	else:
61	messages.error(request, "Invalid username or password.")	1✔
62	else:
63	form = AuthenticationForm()	1✔
64
65	return render(request, "app/login.html", {"form": form})	1✔
66
67
68	# Logout View
69	@login_required	1✔
70	def logout_view(request):	1✔
71	if request.method == "POST":	1✔
72	logout(request)	1✔
73	return redirect("maps:map_view")	1✔
74	else:
75	raise PermissionDenied	1✔
76
77
78	# Stations View
79	class StationsView(generic.ListView):	1✔
80	paginate_by = 10	1✔
81	model = Station	1✔
82	template_name = "app/stations.html"	1✔
83	context_object_name = "station_list"	1✔
84
85	def get_queryset(self):	1✔
86	# Get the search query from the request
87	query = self.request.GET.get("q")	×
88	ada_filter = self.request.GET.get("ada_filter")	×
89
90	# Start with all stations
91	queryset = Station.objects.all().order_by(F("stop_name").asc())	×
92
93	# Apply search filter if applicable
94	if query:	×
95	queryset = queryset.filter(stop_name__icontains=query)	×
96
97	# Apply ADA filter based on the selected option
98	if ada_filter == "fully":	×
99	queryset = queryset.filter(ada=True)	×
100	elif ada_filter == "partially":	×
101	queryset = queryset.filter(ada_southbound=True, ada_northbound=False)	×
102	elif ada_filter == "not":	×
103	queryset = queryset.filter(ada=False)	×
104
105	return queryset	×
106
107
108	# Station Detail View
109	class StationDetailView(generic.DetailView):	1✔
110	model = Station	1✔
111	template_name = "app/station_detail.html"	1✔
112
113
114	def station_detail(request, station_id):	1✔
115	station = get_object_or_404(Station, id=station_id)	1✔
116	form = RatingForm(request.POST)	1✔
117
118	already_reviewed = None	1✔
119
120	if request.user.is_authenticated:	1✔
121	user_reviewed = Review.objects.filter(	1✔
122	station=station, user=request.user
123	).first()
124	if user_reviewed:	1✔
125	already_reviewed = True	1✔
126	if request.method == "POST":	1✔
127	if form.is_valid():	1✔
128	if user_reviewed:	1✔
129	# Update the existing review
130	rating = user_reviewed	1✔
131	rating.rating = form.cleaned_data["rating"]	1✔
132	prev_comment = rating.comment	1✔
133	rating.comment = form.cleaned_data["comment"]	1✔
134	if rating.comment == "":	1✔
135	rating.comment = prev_comment	1✔
136	else:	1✔
137	rating.created_at = timezone.now()	1✔
138	rating.save()
139	else:	1✔
140	rating = form.save(commit=False)	1✔
141	rating.station = station	1✔
142	rating.user = request.user	1✔
143	rating.created_at = timezone.now()	1✔
144	rating.save()	1✔
145	messages.success(request, "Your review has been added!")	1✔
146	return redirect("app:station_detail", station_id=station.id)
147	else:	×
148	print("Form errors: ", form.errors)	×
149	messages.error(request, "Something went wrong")
150	else:	×
151	form = RatingForm(
152	instance=user_reviewed
153	) # Pre-populate the form with the user's existing review (if any)
154		1✔
155	ratings = station.rating.all() # Get all ratings for this station
156	# Calculate the average rating for this station
157		1✔
158	avg_rating = Review.get_average_rating(station_name=station)
159
160	# Find last 5 comments	1✔
161	last_five_comments = (
162	Review.objects.filter(comment__isnull=False, station=station)
163	.exclude(comment="")
164	.order_by("-created_at")[:5]
165	)
166		1✔
167	return render(
168	request,
169	"app/station_detail.html",
170	{
171	"station": station,
172	"ratings": ratings,
173	"comments": last_five_comments,
174	"form": form,
175	"avg_rating": avg_rating,
176	"reviewed": already_reviewed,
177	},
178	)
179
180		1✔
181	class ProfileView(generic.DetailView):	1✔
182	model = Profile	1✔
183	template_name = "app/profile.html"	1✔
184	context_object_name = "profile"
185		1✔
186	def get_object(self):	×
187	return self.request.user.profile
188
189		1✔
190	def alerts_view(request):
NEW 191	# Get selected lines from the request	×
NEW 192	selected_lines = request.GET.get("lines", "")	×
193	lines_list = selected_lines.split(",") if selected_lines else []
NEW 194		×
195	search_patterns = [f"[{line}]" for line in lines_list]
196		×
197	url = (
198	"https://api-endpoint.mta.info/Dataservice/mtagtfsfeeds/camsys%2Fsubway-alerts"
199	)	×
200	try:	×
201	response = requests.get(url)
202
203	# Decode the Protobuf data	×
204	feed = gtfs_realtime_pb2.FeedMessage()	×
205	feed.ParseFromString(response.content)
206
207	# Extract and format alert data	×
208	alerts_data = []	×
209	for entity in feed.entity:	×
210	if entity.HasField("alert"):	×
211	alert = entity.alert
212		×
213	header = (
214	alert.header_text.translation[0].text
215	if alert.header_text.translation
216	else ""
217	)	×
218	description = (
219	alert.description_text.translation[0].text
220	if alert.description_text.translation
221	else ""
222	)
223
NEW 224	# If lines are selected, filter based on the content	×
225	if lines_list:
NEW 226	# Check if any of the search patterns are in the header	×
227	if any(pattern in header for pattern in search_patterns):
NEW 228	# Include this alert	×
229	alerts_data.append(
230	{
231	"header": header,
232	"description": description,
233	}
234	)
235	else:
NEW 236	# No lines selected, include all alerts	×
237	alerts_data.append(
238	{
239	"header": header,
240	"description": description,
241	}
242	)
243		×
244	except Exception as e:	×
245	alerts_data = None	×
246	print("Error fetching alerts data:", e)
247		×
248	if request.headers.get("X-Requested-With") == "XMLHttpRequest":	×
249	return JsonResponse({"alerts_data": alerts_data})	×
250	return render(request, "app/alerts.html", {"alerts_data": alerts_data})
251
252		1✔
253	@login_required	1✔
254	def edit_profile(request):	×
255	profile = request.user.profile	×
256	if request.method == "POST":	×
257	form = ProfileUpdateForm(request.POST, instance=profile)	×
258	date = request.POST.get("birth_date")	×
259	if date > datetime.date.today().isoformat():	×
260	messages.error(request, "Invalid date")	×
261	return redirect("app:edit_profile")	×
262	if form.is_valid():	×
263	form.save()	×
264	messages.success(request, "Your profile has been updated.")	×
265	return redirect("app:profile")
266	else:	×
267	form = ProfileUpdateForm(instance=profile)
268		×
269	context = {
270	"google_maps_api_key": settings.GOOGLE_MAPS_API_KEY,
271	"form": form,
272	}
273		×
274	return render(request, "app/edit_profile.html", context)
275
276		1✔
277	@login_required	1✔
278	def save_favorite_route(request):	×
279	if request.method == "POST":	×
280	data = json.loads(request.body)	×
281	start = data.get("start")	×
282	end = data.get("end")
283		×
284	print(start)	×
285	profile = request.user.profile	×
286	profile.fav_source_latitude = start["lat"]	×
287	profile.fav_source_longitude = start["lng"]	×
288	profile.fav_dest_latitude = end["lat"]	×
289	profile.fav_dest_longitude = end["lng"]	×
290	profile.save()
291		×
292	return JsonResponse({"success": True})
293		×
294	return JsonResponse({"success": False})

gcivil-nyu-org / wed-fall24-team4 / #628502899

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous