diff --git a/.gitignore b/.gitignore index e43b0f9..783b798 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,3 @@ .DS_Store +/**/*.shp +/**/*.shx diff --git a/bats_ai/core/models/__init__.py b/bats_ai/core/models/__init__.py index 3dfd89d..cc8bf05 100644 --- a/bats_ai/core/models/__init__.py +++ b/bats_ai/core/models/__init__.py @@ -4,6 +4,7 @@ from .recording_annotation_status import RecordingAnnotationStatus from .species import Species from .spectrogram import Spectrogram +from .temporal_annotations import TemporalAnnotations __all__ = [ 'Annotations', @@ -12,4 +13,5 @@ 'RecordingAnnotationStatus', 'Species', 'Spectrogram', + 'TemporalAnnotations', ] diff --git a/bats_ai/core/models/temporal_annotations.py b/bats_ai/core/models/temporal_annotations.py new file mode 100644 index 0000000..1f5207e --- /dev/null +++ b/bats_ai/core/models/temporal_annotations.py @@ -0,0 +1,13 @@ +from django.contrib.auth.models import User +from django.db import models + +from .recording import Recording + + +class TemporalAnnotations(models.Model): + recording = models.ForeignKey(Recording, on_delete=models.CASCADE) + owner = models.ForeignKey(User, on_delete=models.CASCADE) + start_time = models.IntegerField(blank=True, null=True) + end_time = models.IntegerField(blank=True, null=True) + type = models.TextField(blank=True, null=True) + comments = models.TextField(blank=True, null=True) diff --git a/bats_ai/core/views/__init__.py b/bats_ai/core/views/__init__.py index 6a0e84a..ddf5efb 100644 --- a/bats_ai/core/views/__init__.py +++ b/bats_ai/core/views/__init__.py @@ -1,7 +1,11 @@ +from .annotations import router as AnnotationRouter from .recording import router as RecordingRouter from .species import router as SpeciesRouter +from .temporal_annotations import router as TemporalAnnotationRouter __all__ = [ 'RecordingRouter', 'SpeciesRouter', + 'AnnotationRouter', + 'TemporalAnnotationRouter', ] diff --git a/bats_ai/core/views/annotations.py b/bats_ai/core/views/annotations.py index 68c9445..e83fd25 100644 --- a/bats_ai/core/views/annotations.py +++ b/bats_ai/core/views/annotations.py @@ -2,9 +2,9 @@ from django.http import HttpRequest from ninja import Schema -from ninja.errors import HttpError from ninja.pagination import RouterPaginated -from oauth2_provider.models import AccessToken + +from bats_ai.core.models import Annotations, Recording logger = logging.getLogger(__name__) @@ -23,10 +23,28 @@ class AnnotationSchema(Schema): comments: str -def get_owner_id(request: HttpRequest): - token = request.headers.get('Authorization').replace('Bearer ', '') - token_found = AccessToken.objects.get(token=token) - if not token_found: - raise HttpError(401, 'Authentication credentials were not provided.') +@router.get('/{id}') +def get_annotation(request: HttpRequest, id: int): + try: + annotation = Annotations.objects.get(pk=id) + recording = annotation.recording + + # Check if the user owns the recording or if the recording is public + if recording.owner == request.user or recording.public: + # Query annotations associated with the recording that are owned by the current user + annotations_qs = Annotations.objects.filter(recording=recording, owner=request.user) + + # Serialize the annotations using AnnotationSchema + annotations_data = [ + AnnotationSchema.from_orm(annotation, owner_email=request.user.email).dict() + for annotation in annotations_qs + ] + + return annotations_data + else: + return { + 'error': 'Permission denied. You do not own this annotation, or the associated recording is not public.' + } - return token_found.user.pk + except Recording.DoesNotExist: + return {'error': 'Recording not found'} diff --git a/bats_ai/core/views/recording.py b/bats_ai/core/views/recording.py index 951c042..5b14cce 100644 --- a/bats_ai/core/views/recording.py +++ b/bats_ai/core/views/recording.py @@ -8,8 +8,9 @@ from ninja.files import UploadedFile from ninja.pagination import RouterPaginated -from bats_ai.core.models import Annotations, Recording, Species +from bats_ai.core.models import Annotations, Recording, Species, TemporalAnnotations from bats_ai.core.views.species import SpeciesSchema +from bats_ai.core.views.temporal_annotations import TemporalAnnotationSchema logger = logging.getLogger(__name__) @@ -461,3 +462,89 @@ def delete_annotation(request, recording_id: int, id: int): return {'error': 'Recording not found'} except Annotations.DoesNotExist: return {'error': 'Annotation not found'} + + +# TEMPORAL ANNOTATIONS + + +@router.get('recording/{id}/temporal-annotations') +def get_temporal_annotations(request: HttpRequest, id: int): + try: + recording = Recording.objects.get(pk=id) + + # Check if the user owns the recording or if the recording is public + if recording.owner == request.user or recording.public: + # Query annotations associated with the recording that are owned by the current user + annotations_qs = TemporalAnnotations.objects.filter( + recording=recording, owner=request.user + ) + + # Serialize the annotations using AnnotationSchema + annotations_data = [ + TemporalAnnotationSchema.from_orm(annotation, owner_email=request.user.email).dict() + for annotation in annotations_qs + ] + + return annotations_data + else: + return { + 'error': 'Permission denied. You do not own this recording, and it is not public.' + } + + except Recording.DoesNotExist: + return {'error': 'Recording not found'} + + +@router.put('recording/{id}/temporal-annotations') +def put_temporal_annotation( + request, + id: int, + annotation: TemporalAnnotationSchema, +): + try: + recording = Recording.objects.get(pk=id) + if recording.owner == request.user or recording.public: + # Create a new annotation + new_annotation = TemporalAnnotations.objects.create( + recording=recording, + owner=request.user, + start_time=annotation.start_time, + end_time=annotation.end_time, + type=annotation.type, + comments=annotation.comments, + ) + + return {'message': 'Annotation added successfully', 'id': new_annotation.pk} + else: + return { + 'error': 'Permission denied. You do not own this recording, and it is not public.' + } + + except Recording.DoesNotExist: + return {'error': 'Recording not found'} + + +@router.delete('/{recording_id}/temporal-annotations/{id}') +def delete_temporal_annotation(request, recording_id: int, id: int): + try: + recording = Recording.objects.get(pk=recording_id) + + # Check if the user owns the recording or if the recording is public + if recording.owner == request.user or recording.public: + annotation_instance = TemporalAnnotations.objects.get( + pk=id, recording=recording, owner=request.user + ) + + # Delete the annotation + annotation_instance.delete() + + return {'message': 'Annotation deleted successfully'} + else: + return { + 'error': 'Permission denied. You do not own this recording, and it is not public.' + } + + except Recording.DoesNotExist: + return {'error': 'Recording not found'} + except Annotations.DoesNotExist: + return {'error': 'Annotation not found'} diff --git a/bats_ai/core/views/temporal_annotations.py b/bats_ai/core/views/temporal_annotations.py new file mode 100644 index 0000000..e961363 --- /dev/null +++ b/bats_ai/core/views/temporal_annotations.py @@ -0,0 +1,45 @@ +from django.http import HttpRequest +from ninja import Schema +from ninja.pagination import RouterPaginated + +from bats_ai.core.models import Annotations, Recording, TemporalAnnotations + +router = RouterPaginated() + + +class TemporalAnnotationSchema(Schema): + recording: int # Foreign Key to index + owner_username: str + start_time: int + end_time: int + type: str + comments: str + + +@router.get('/{id}') +def get_temporal_annotation(request: HttpRequest, id: int): + try: + annotation = Annotations.objects.get(pk=id) + recording = annotation.recording + + # Check if the user owns the recording or if the recording is public + if recording.owner == request.user or recording.public: + # Query annotations associated with the recording that are owned by the current user + annotations_qs = TemporalAnnotations.objects.filter( + recording=recording, owner=request.user + ) + + # Serialize the annotations using AnnotationSchema + annotations_data = [ + TemporalAnnotationSchema.from_orm(annotation, owner_email=request.user.email).dict() + for annotation in annotations_qs + ] + + return annotations_data + else: + return { + 'error': 'Permission denied. You do not own this annotation, or the associated recording is not public.' + } + + except Recording.DoesNotExist: + return {'error': 'Recording not found'}