Skip to content

Use get_object_or_404 where possible #442

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
brianhelba wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Use get_object_or_404 where possible #442

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2c06984
Use `get_object_or_404` where possible
brianhelba Mar 17, 2026
1b60916
Use `get_list_or_404` where possible
brianhelba Mar 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 40 additions & 48 deletions bats_ai/core/tasks/nabat/nabat_data_retrieval.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,53 +172,45 @@ def nabat_recording_initialize(self, recording_id: int, survey_event_id: int, ap


def create_nabat_recording_from_response(response_data, recording_id, survey_event_id):
try:
# Extract the batch data from the response
nabat_recording_data = response_data["data"]

# Optional fields
recording_location_data = nabat_recording_data["surveyEventById"][
"eventGeometryByEventGeometryId"
]["geom"]["geojson"]
file_name = nabat_recording_data["acousticFileById"]["fileName"]

# Create geometry for the recording location if available
if recording_location_data:
coordinates = recording_location_data.get("coordinates", [])
recording_location = (
Point(coordinates[0], coordinates[1]) if len(coordinates) == 2 else None
)
else:
recording_location = None

# Create the NABatRecording instance
nabat_recording = NABatRecording.objects.create(
recording_id=recording_id,
survey_event_id=survey_event_id,
name=file_name,
recording_location=recording_location,
# Extract the batch data from the response
nabat_recording_data = response_data["data"]

# Optional fields
recording_location_data = nabat_recording_data["surveyEventById"][
"eventGeometryByEventGeometryId"
]["geom"]["geojson"]
file_name = nabat_recording_data["acousticFileById"]["fileName"]

# Create geometry for the recording location if available
if recording_location_data:
coordinates = recording_location_data.get("coordinates", [])
recording_location = (
Point(coordinates[0], coordinates[1]) if len(coordinates) == 2 else None
)
else:
recording_location = None

# Create the NABatRecording instance
nabat_recording = NABatRecording.objects.create(
recording_id=recording_id,
survey_event_id=survey_event_id,
name=file_name,
recording_location=recording_location,
)

acoustic_batches_nodes = nabat_recording_data["surveyEventById"][
"acousticBatchesBySurveyEventId"
]["nodes"]
if len(acoustic_batches_nodes) > 0:
batch_data = acoustic_batches_nodes[0]["acousticFileBatchesByBatchId"]["nodes"]
for node in batch_data:
species_id = node.get("manualId", False)
if species_id is not False:
annotation = NABatRecordingAnnotation.objects.create(
nabat_recording=nabat_recording,
user_email=node["vetter"],
)
species = Species.objects.get(pk=species_id)
annotation.species.add(species)

return nabat_recording

except KeyError:
logger.exception("Missing key")
raise
except Exception:
logger.exception("Error creating NABatRecording")
raise
acoustic_batches_nodes = nabat_recording_data["surveyEventById"][
"acousticBatchesBySurveyEventId"
]["nodes"]
if len(acoustic_batches_nodes) > 0:
batch_data = acoustic_batches_nodes[0]["acousticFileBatchesByBatchId"]["nodes"]
for node in batch_data:
species_id = node.get("manualId", False)
if species_id is not False:
annotation = NABatRecordingAnnotation.objects.create(
nabat_recording=nabat_recording,
user_email=node["vetter"],
)
species = Species.objects.get(pk=species_id)
annotation.species.add(species)

return nabat_recording
135 changes: 65 additions & 70 deletions bats_ai/core/views/grts_cells.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from django.contrib.gis.geos import Point, Polygon
from django.http import HttpRequest, JsonResponse
from django.shortcuts import get_list_or_404
from ninja import Query
from ninja.pagination import RouterPaginated

Expand Down Expand Up @@ -34,88 +35,82 @@ def get_grid_cell_id(

@router.get("/{pk}")
def get_cell_center(request: HttpRequest, pk: int, quadrant: str | None = None):
try:
cells = GRTSCells.objects.filter(grts_cell_id=pk)
cells = get_list_or_404(GRTSCells, grts_cell_id=pk)

# Define a custom order for sample_frame_id
custom_order = GRTSCells.sort_order() # Define your custom order here
# Define a custom order for sample_frame_id
custom_order = GRTSCells.sort_order() # Define your custom order here

# Define a custom key function to sort cells based on the custom order
def custom_sort_key(cell):
return custom_order.index(cell.sample_frame_id)
# Define a custom key function to sort cells based on the custom order
def custom_sort_key(cell):
return custom_order.index(cell.sample_frame_id)

# Sort the cells queryset based on the custom order
sorted_cells = sorted(cells, key=custom_sort_key)
cell = sorted_cells[0]
geom_4326 = cell.geom_4326
# Sort the cells queryset based on the custom order
sorted_cells = sorted(cells, key=custom_sort_key)
cell = sorted_cells[0]
geom_4326 = cell.geom_4326

# Get the centroid of the entire cell polygon
center = geom_4326.centroid
# Get the centroid of the entire cell polygon
center = geom_4326.centroid

if quadrant:
# If quadrant is specified, divide the cell polygon into quadrants
min_x, min_y, max_x, max_y = geom_4326.extent
mid_x = (min_x + max_x) / 2
mid_y = (min_y + max_y) / 2
if quadrant:
# If quadrant is specified, divide the cell polygon into quadrants
min_x, min_y, max_x, max_y = geom_4326.extent
mid_x = (min_x + max_x) / 2
mid_y = (min_y + max_y) / 2

# Determine the bounding box coordinates of the specified quadrant
if quadrant.upper() == "NW":
bbox = (min_x, mid_y, mid_x, max_y)
elif quadrant.upper() == "SE":
bbox = (mid_x, min_y, max_x, mid_y)
elif quadrant.upper() == "SW":
bbox = (min_x, min_y, mid_x, mid_y)
elif quadrant.upper() == "NE":
bbox = (mid_x, mid_y, max_x, max_y)
# Determine the bounding box coordinates of the specified quadrant
if quadrant.upper() == "NW":
bbox = (min_x, mid_y, mid_x, max_y)
elif quadrant.upper() == "SE":
bbox = (mid_x, min_y, max_x, mid_y)
elif quadrant.upper() == "SW":
bbox = (min_x, min_y, mid_x, mid_y)
elif quadrant.upper() == "NE":
bbox = (mid_x, mid_y, max_x, max_y)

quadrant_polygon = Polygon.from_bbox(bbox)
quadrant_polygon = Polygon.from_bbox(bbox)

# Intersect the cell polygon with the specified quadrant's polygon
quadrant_polygon = geom_4326.intersection(quadrant_polygon)
# Intersect the cell polygon with the specified quadrant's polygon
quadrant_polygon = geom_4326.intersection(quadrant_polygon)

# Get the centroid of the intersected polygon
center = quadrant_polygon.centroid
# Get the centroid of the intersected polygon
center = quadrant_polygon.centroid

# Get the latitude and longitude of the centroid
center_latitude = center.y
center_longitude = center.x
# Get the latitude and longitude of the centroid
center_latitude = center.y
center_longitude = center.x

return JsonResponse({"latitude": center_latitude, "longitude": center_longitude})
except GRTSCells.DoesNotExist:
return JsonResponse({"error": f"Cell with cellId={pk} does not exist"}, status=200)
return JsonResponse({"latitude": center_latitude, "longitude": center_longitude})


@router.get("/{pk}/bbox")
def get_grts_cell_bbox(request: HttpRequest, pk: int):
try:
cells = GRTSCells.objects.filter(grts_cell_id=pk)
custom_order = GRTSCells.sort_order()

def custom_sort_key(cell):
return custom_order.index(cell.sample_frame_id)

sorted_cells = sorted(cells, key=custom_sort_key)
cell = sorted_cells[0]
geom = cell.geom_4326

min_x, min_y, max_x, max_y = geom.extent

geojson = {
"type": "Feature",
"geometry": {
"type": "Polygon",
"coordinates": [
[min_x, min_y],
[min_x, max_y],
[max_x, max_y],
[max_x, min_y],
],
},
"properties": {
"grts_cell_id": pk,
"annotationType": "rectangle",
},
}
return JsonResponse(geojson)
except (GRTSCells.DoesNotExist, IndexError):
return JsonResponse({"error": f"Cell with id {pk} does not exist"}, status=200)
cells = get_list_or_404(GRTSCells, grts_cell_id=pk)
custom_order = GRTSCells.sort_order()

def custom_sort_key(cell):
return custom_order.index(cell.sample_frame_id)

sorted_cells = sorted(cells, key=custom_sort_key)
cell = sorted_cells[0]
geom = cell.geom_4326

min_x, min_y, max_x, max_y = geom.extent

geojson = {
"type": "Feature",
"geometry": {
"type": "Polygon",
"coordinates": [
[min_x, min_y],
[min_x, max_y],
[max_x, max_y],
[max_x, min_y],
],
},
"properties": {
"grts_cell_id": pk,
"annotationType": "rectangle",
},
}
return JsonResponse(geojson)
6 changes: 2 additions & 4 deletions bats_ai/core/views/nabat/nabat_configuration.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from django.db import transaction
from django.db.models import Count
from django.http import HttpRequest, JsonResponse
from django.shortcuts import get_object_or_404
from django.utils.timezone import now
from ninja import Query, Router, Schema
from ninja.pagination import paginate
Expand Down Expand Up @@ -161,10 +162,7 @@ def recording_annotations(
if not request.user.is_authenticated or not request.user.is_superuser:
return JsonResponse({"error": "Permission denied"}, status=403)

try:
recording = NABatRecording.objects.get(pk=recording_id)
except NABatRecording.DoesNotExist:
return JsonResponse({"error": "Recording not found"}, status=404)
recording = get_object_or_404(NABatRecording, pk=recording_id)

annotations = NABatRecordingAnnotation.objects.filter(nabat_recording=recording)

Expand Down
Loading
Loading