#614980386

Committed 13 Dec 2023 12:50AM UTC coverage: 86.973%. First build

Build # #614980386

Build Type

Pull #257

travis-ci

Pull Request Pull Request #257: Final sprint

Run Details

702 of 804 new or added lines in 23 files covered. (87.31%)

1382 of 1589 relevant lines covered (86.97%)

0.87 hits per line

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

67.52

/vibematch/views.py

from django.shortcuts import redirect, render
from django.utils import timezone
from user_profile.models import Vibe, User, UserTop, UserFriendRelation
import numpy as np
from vibematch.models import UserLocation
import re
from dashboard.models import EmotionVector, Track, Artist
from django.db.models import OuterRef, Subquery, F
from django.contrib import messages
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
import json
from django.contrib.auth.decorators import login_required
from math import radians, cos, sin, asin, sqrt
import math
from django.db.models import Q


def vibe_match(request):
    if request.user.is_authenticated:
        user_info = request.user
        user_id = user_info.user_id
        # Pass username to navbar
        username = user_info.username

        nearest_neighbors_ids, all_users, physical_distances = k_nearest_neighbors(
            5, user_id
        )

        matches = []
        for uid, _ in nearest_neighbors_ids:
            user_top = UserTop.objects.filter(user_id=uid).order_by("-time").first()
            this_user = User.objects.get(user_id=uid)

            matches.append(
                {
                    "user_id": uid,
                    "username": this_user,
                    "vibe": all_users.filter(user_id=uid)
                    .values_list("user_lyrics_vibe", "user_audio_vibe", flat=False)
                    .first(),
                    "fav_track": Track.objects.filter(id=this_user.track_id).first()
                    if this_user.track_id
                    else None,
                    "distance": math.ceil(physical_distances.get(uid, None))
                    if physical_distances.get(uid) is not None
                    else None,
                    "similarity": distance_to_similarity(_),
                    "top_artist": Artist.objects.filter(id__in=user_top.top_artist[:5])
                    if user_top and len(user_top.top_artist) > 0
                    else None,
                    "top_tracks": Track.objects.filter(id__in=user_top.top_track[:3])
                    if user_top and len(user_top.top_track) > 0
                    else None,
                }
            )

        context = {
            "neighbors": matches,
            "username": username,
        }

        return render(request, "match.html", context)
    else:
        # No token, redirect to login again
        messages.error(request, "Vibe_match failed, please try again later.")
        return redirect("login:index")


def k_nearest_neighbors(k, target_user_id):
    # Get the most recent vibe for each user
    # Subquery to get the latest vibe_time for each user
    latest_vibe_times = (
        Vibe.objects.filter(user_id=OuterRef("user_id"))
        .order_by("-vibe_time")
        .values("vibe_time")[:1]
    )

    # Filter Vibe objects to only get those matching the latest vibe_time for each user
    latest_vibes = Vibe.objects.annotate(
        latest_vibe_time=Subquery(latest_vibe_times)
    ).filter(vibe_time=F("latest_vibe_time"))

    all_users, physical_distances = get_users(target_user_id, latest_vibes)

    # Fetch Emotion Vectors
    lyrics_vibe_values = list(set(user[1] for user in all_users))
    audio_vibe_values = list(set(user[2] for user in all_users))
    all_vibe_values = list(set(lyrics_vibe_values + audio_vibe_values))

    need_emotion_vectors = EmotionVector.objects.filter(
        emotion__in=[value for value in all_vibe_values]
    )

    emotion_vectors = {
        str(emotion.emotion): vector_to_array(emotion.vector)
        for emotion in need_emotion_vectors
    }

    all_users_array = []
    target_user_features = None

    for user in all_users:
        user_id, lyrics_vibe, audio_vibe, *features = user
        lyrics_vector = emotion_vectors.get(
            lyrics_vibe, np.zeros_like(next(iter(emotion_vectors.values())))
        )
        audio_vector = emotion_vectors.get(
            audio_vibe, np.zeros_like(next(iter(emotion_vectors.values())))
        )
        features = [float(feature) for feature in features]

        if user_id != target_user_id:
            # Check friend relation status here.
            # Do not want to recommend people you are already friends
            # or have a pending friend request with

            friend_request = UserFriendRelation.objects.filter(
                (Q(user1_id=user_id) & Q(user2_id=target_user_id))
                | (Q(user2_id=user_id) & Q(user1_id=target_user_id))
            ).first()

            if not friend_request or friend_request.status == "not_friend":
                # Friend request does not exist or there's a row in table but not friends anymore
                all_users_array.append(
                    (user_id, [*lyrics_vector, *audio_vector, *features])
                )

        else:
            target_user_features = [*lyrics_vector, *audio_vector, *features]

    if target_user_features is None:
        return []

    # Calculate distances, excluding the target user
    distances = [
        (user_id, euclidean_distance(target_user_features, features))
        for user_id, features in all_users_array
    ]

    # Sort by distance and select top k
    nearest_neighbors_ids = sorted(distances, key=lambda x: x[1])[:k]

    return nearest_neighbors_ids, all_users, physical_distances


def distance_to_similarity(distance):
    return math.ceil((1 / (1 + distance)) * 100)


def get_users(target_user_id, latest_vibes):
    today = timezone.localdate()
    phys_distances = {}

    # Check if a location for today already exists
    if UserLocation.objects.filter(
        user=User.objects.get(user_id=target_user_id), created_at__date=today
    ).exists():
        # Filter for users within 60 miles of the target user
        # Getting latest location for each user
        all_user_locations = UserLocation.objects.filter(
            created_at=Subquery(
                UserLocation.objects.filter(user=OuterRef("user"))
                .order_by("-created_at")
                .values("created_at")[:1]
            )
        )

        nearby_users, phys_distances = get_nearby_users(
            all_user_locations, target_user_id
        )

        all_users = latest_vibes.filter(user_id__in=nearby_users).values_list(
            "user_id",
            "user_lyrics_vibe",
            "user_audio_vibe",
            "user_acousticness",
            "user_danceability",
            "user_energy",
            "user_valence",
            flat=False,
        )
    else:
        # If no location for the target user, use the existing method
        all_users = latest_vibes.values_list(
            "user_id",
            "user_lyrics_vibe",
            "user_audio_vibe",
            "user_acousticness",
            "user_danceability",
            "user_energy",
            "user_valence",
            flat=False,
        )

    return all_users, phys_distances


def get_nearby_users(all_user_locations, target_user_id):
    # Get target user's location
    target_user_location = (
        UserLocation.objects.filter(user_id=target_user_id)
        .order_by("-created_at")
        .first()
    )

    nearby_users = []
    user_distances = {}
    for location in all_user_locations:
        distance = haversine(
            target_user_location.longitude,
            target_user_location.latitude,
            location.longitude,
            location.latitude,
        )
        if distance <= 60:
            nearby_users.append(location.user_id)
            user_distances[location.user_id] = distance

    return nearby_users, user_distances


def euclidean_distance(user_1, user_2):
    user_1 = np.array(user_1)
    user_2 = np.array(user_2)
    return np.sqrt(np.sum((user_1 - user_2) ** 2))


def vector_to_array(vector_str):
    clean = re.sub(r"[\[\]\n\t]", "", vector_str)
    clean = clean.split()
    clean = [float(e) for e in clean]
    return np.array(clean)


# Haversine formula to calculate distance between two lat/long points
def haversine(lon1, lat1, lon2, lat2):
    # Convert decimal degrees to radians
    lon1, lat1, lon2, lat2 = map(radians, [lon1, lat1, lon2, lat2])

    # Haversine formula
    dlon = lon2 - lon1
    dlat = lat2 - lat1
    a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2
    c = 2 * asin(sqrt(a))
    r = 3956  # Radius of Earth in miles
    return c * r


@csrf_exempt
@require_http_methods(["POST"])
def store_location(request):
    if not request.user.is_authenticated:
        return JsonResponse({"status": "unauthorized"}, status=401)

    # Get today's date
    today = timezone.localdate()

    # Check if a location for today already exists
    if UserLocation.objects.filter(user=request.user, created_at__date=today).exists():
        # If it does, return a success response without creating a new entry
        return JsonResponse({"status": "location already stored for today"}, status=200)

    try:
        data = json.loads(request.body)
        latitude = data["latitude"]
        longitude = data["longitude"]

        # Create a new UserLocation instance and save it to the database
        UserLocation.objects.create(
            user=request.user, latitude=latitude, longitude=longitude
        )

        return JsonResponse({"status": "success"}, status=200)
    except (KeyError, json.JSONDecodeError, TypeError) as e:
        # Return an error message if something goes wrong
        return JsonResponse({"status": "error", "message": str(e)}, status=400)


@login_required
def check_location_stored(request):
    today = timezone.localdate()
    location_exists = UserLocation.objects.filter(
        user=request.user, created_at__date=today
    ).exists()
    return JsonResponse({"locationStored": location_exists})

1	from django.shortcuts import redirect, render	1✔
2	from django.utils import timezone	1✔
3	from user_profile.models import Vibe, User, UserTop, UserFriendRelation	1✔
4	import numpy as np	1✔
5	from vibematch.models import UserLocation	1✔
6	import re	1✔
7	from dashboard.models import EmotionVector, Track, Artist	1✔
8	from django.db.models import OuterRef, Subquery, F	1✔
9	from django.contrib import messages	1✔
10	from django.http import JsonResponse	1✔
11	from django.views.decorators.csrf import csrf_exempt	1✔
12	from django.views.decorators.http import require_http_methods	1✔
13	import json	1✔
14	from django.contrib.auth.decorators import login_required	1✔
15	from math import radians, cos, sin, asin, sqrt	1✔
16	import math	1✔
17	from django.db.models import Q	1✔
18
19
20	def vibe_match(request):	1✔
21	if request.user.is_authenticated:	1✔
22	user_info = request.user	1✔
23	user_id = user_info.user_id	1✔
24	# Pass username to navbar
25	username = user_info.username	1✔
26
27	nearest_neighbors_ids, all_users, physical_distances = k_nearest_neighbors(	1✔
28	5, user_id
29	)
30
31	matches = []	1✔
32	for uid, _ in nearest_neighbors_ids:	1✔
33	user_top = UserTop.objects.filter(user_id=uid).order_by("-time").first()	1✔
34	this_user = User.objects.get(user_id=uid)	1✔
35
36	matches.append(	1✔
37	{
38	"user_id": uid,
39	"username": this_user,
40	"vibe": all_users.filter(user_id=uid)
41	.values_list("user_lyrics_vibe", "user_audio_vibe", flat=False)
42	.first(),
43	"fav_track": Track.objects.filter(id=this_user.track_id).first()
44	if this_user.track_id
45	else None,
46	"distance": math.ceil(physical_distances.get(uid, None))
47	if physical_distances.get(uid) is not None
48	else None,
49	"similarity": distance_to_similarity(_),
50	"top_artist": Artist.objects.filter(id__in=user_top.top_artist[:5])
51	if user_top and len(user_top.top_artist) > 0
52	else None,
53	"top_tracks": Track.objects.filter(id__in=user_top.top_track[:3])
54	if user_top and len(user_top.top_track) > 0
55	else None,
56	}
57	)
58
59	context = {	1✔
60	"neighbors": matches,
61	"username": username,
62	}
63
64	return render(request, "match.html", context)	1✔
65	else:
66	# No token, redirect to login again
67	messages.error(request, "Vibe_match failed, please try again later.")	×
68	return redirect("login:index")	×
69
70
71	def k_nearest_neighbors(k, target_user_id):	1✔
72	# Get the most recent vibe for each user
73	# Subquery to get the latest vibe_time for each user
74	latest_vibe_times = (	1✔
75	Vibe.objects.filter(user_id=OuterRef("user_id"))
76	.order_by("-vibe_time")
77	.values("vibe_time")[:1]
78	)
79
80	# Filter Vibe objects to only get those matching the latest vibe_time for each user
81	latest_vibes = Vibe.objects.annotate(	1✔
82	latest_vibe_time=Subquery(latest_vibe_times)
83	).filter(vibe_time=F("latest_vibe_time"))
84
85	all_users, physical_distances = get_users(target_user_id, latest_vibes)	1✔
86
87	# Fetch Emotion Vectors
88	lyrics_vibe_values = list(set(user[1] for user in all_users))	1✔
89	audio_vibe_values = list(set(user[2] for user in all_users))	1✔
90	all_vibe_values = list(set(lyrics_vibe_values + audio_vibe_values))	1✔
91
92	need_emotion_vectors = EmotionVector.objects.filter(	1✔
93	emotion__in=[value for value in all_vibe_values]
94	)
95
96	emotion_vectors = {	1✔
97	str(emotion.emotion): vector_to_array(emotion.vector)
98	for emotion in need_emotion_vectors
99	}
100
101	all_users_array = []	1✔
102	target_user_features = None	1✔
103
104	for user in all_users:	1✔
105	user_id, lyrics_vibe, audio_vibe, *features = user	1✔
106	lyrics_vector = emotion_vectors.get(	1✔
107	lyrics_vibe, np.zeros_like(next(iter(emotion_vectors.values())))
108	)
109	audio_vector = emotion_vectors.get(	1✔
110	audio_vibe, np.zeros_like(next(iter(emotion_vectors.values())))
111	)
112	features = [float(feature) for feature in features]	1✔
113
114	if user_id != target_user_id:	1✔
115	# Check friend relation status here.
116	# Do not want to recommend people you are already friends
117	# or have a pending friend request with
118
119	friend_request = UserFriendRelation.objects.filter(	1✔
120	(Q(user1_id=user_id) & Q(user2_id=target_user_id))
121	\| (Q(user2_id=user_id) & Q(user1_id=target_user_id))
122	).first()
123
124	if not friend_request or friend_request.status == "not_friend":	1✔
125	# Friend request does not exist or there's a row in table but not friends anymore
126	all_users_array.append(	1✔
127	(user_id, [lyrics_vector, audio_vector, *features])
128	)
129
130	else:
131	target_user_features = [lyrics_vector, audio_vector, *features]	1✔
132
133	if target_user_features is None:	1✔
134	return []	×
135
136	# Calculate distances, excluding the target user
137	distances = [	1✔
138	(user_id, euclidean_distance(target_user_features, features))
139	for user_id, features in all_users_array
140	]
141
142	# Sort by distance and select top k
143	nearest_neighbors_ids = sorted(distances, key=lambda x: x[1])[:k]	1✔
144
145	return nearest_neighbors_ids, all_users, physical_distances	1✔
146
147
148	def distance_to_similarity(distance):	1✔
149	return math.ceil((1 / (1 + distance)) * 100)	1✔
150
151
152	def get_users(target_user_id, latest_vibes):	1✔
153	today = timezone.localdate()	1✔
154	phys_distances = {}	1✔
155
156	# Check if a location for today already exists
157	if UserLocation.objects.filter(	1✔
158	user=User.objects.get(user_id=target_user_id), created_at__date=today
159	).exists():
160	# Filter for users within 60 miles of the target user
161	# Getting latest location for each user
NEW 162	all_user_locations = UserLocation.objects.filter(	×
163	created_at=Subquery(
164	UserLocation.objects.filter(user=OuterRef("user"))
165	.order_by("-created_at")
166	.values("created_at")[:1]
167	)
168	)
169
NEW 170	nearby_users, phys_distances = get_nearby_users(	×
171	all_user_locations, target_user_id
172	)
173
NEW 174	all_users = latest_vibes.filter(user_id__in=nearby_users).values_list(	×
175	"user_id",
176	"user_lyrics_vibe",
177	"user_audio_vibe",
178	"user_acousticness",
179	"user_danceability",
180	"user_energy",
181	"user_valence",
182	flat=False,
183	)
184	else:
185	# If no location for the target user, use the existing method
186	all_users = latest_vibes.values_list(	1✔
187	"user_id",
188	"user_lyrics_vibe",
189	"user_audio_vibe",
190	"user_acousticness",
191	"user_danceability",
192	"user_energy",
193	"user_valence",
194	flat=False,
195	)
196
197	return all_users, phys_distances	1✔
198
199
200	def get_nearby_users(all_user_locations, target_user_id):	1✔
201	# Get target user's location
NEW 202	target_user_location = (	×
203	UserLocation.objects.filter(user_id=target_user_id)
204	.order_by("-created_at")
205	.first()
206	)
207
NEW 208	nearby_users = []	×
NEW 209	user_distances = {}	×
NEW 210	for location in all_user_locations:	×
NEW 211	distance = haversine(	×
212	target_user_location.longitude,
213	target_user_location.latitude,
214	location.longitude,
215	location.latitude,
216	)
NEW 217	if distance <= 60:	×
NEW 218	nearby_users.append(location.user_id)	×
NEW 219	user_distances[location.user_id] = distance	×
220
NEW 221	return nearby_users, user_distances	×
222
223
224	def euclidean_distance(user_1, user_2):	1✔
225	user_1 = np.array(user_1)	1✔
226	user_2 = np.array(user_2)	1✔
227	return np.sqrt(np.sum((user_1 - user_2) ** 2))	1✔
228
229
230	def vector_to_array(vector_str):	1✔
231	clean = re.sub(r"[\[\]\n\t]", "", vector_str)	1✔
232	clean = clean.split()	1✔
233	clean = [float(e) for e in clean]	1✔
234	return np.array(clean)	1✔
235
236
237	# Haversine formula to calculate distance between two lat/long points
238	def haversine(lon1, lat1, lon2, lat2):	1✔
239	# Convert decimal degrees to radians
NEW 240	lon1, lat1, lon2, lat2 = map(radians, [lon1, lat1, lon2, lat2])	×
241
242	# Haversine formula
NEW 243	dlon = lon2 - lon1	×
NEW 244	dlat = lat2 - lat1	×
NEW 245	a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2	×
NEW 246	c = 2 * asin(sqrt(a))	×
NEW 247	r = 3956 # Radius of Earth in miles	×
NEW 248	return c * r	×
249
250
251	@csrf_exempt	1✔
252	@require_http_methods(["POST"])	1✔
253	def store_location(request):	1✔
254	if not request.user.is_authenticated:	×
255	return JsonResponse({"status": "unauthorized"}, status=401)	×
256
257	# Get today's date
258	today = timezone.localdate()	×
259
260	# Check if a location for today already exists
261	if UserLocation.objects.filter(user=request.user, created_at__date=today).exists():	×
262	# If it does, return a success response without creating a new entry
263	return JsonResponse({"status": "location already stored for today"}, status=200)	×
264
265	try:	×
266	data = json.loads(request.body)	×
267	latitude = data["latitude"]	×
268	longitude = data["longitude"]	×
269
270	# Create a new UserLocation instance and save it to the database
271	UserLocation.objects.create(	×
272	user=request.user, latitude=latitude, longitude=longitude
273	)
274
275	return JsonResponse({"status": "success"}, status=200)	×
276	except (KeyError, json.JSONDecodeError, TypeError) as e:	×
277	# Return an error message if something goes wrong
278	return JsonResponse({"status": "error", "message": str(e)}, status=400)	×
279
280
281	@login_required	1✔
282	def check_location_stored(request):	1✔
283	today = timezone.localdate()	×
NEW 284	location_exists = UserLocation.objects.filter(	×
285	user=request.user, created_at__date=today
286	).exists()
NEW 287	return JsonResponse({"locationStored": location_exists})	×

gcivil-nyu-org / Wednesday-Fall2023-Team-1 / #614980386

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