test backend for Zhang kexin

padam_django/api/serializers.py

@@ -0,0 +1,91 @@
+from rest_framework import serializers
+from ..apps.transportation.models import BusStop, BusShift
+from ..apps.fleet.models import Bus, Driver
+from ..apps.geography.models import Place
+from django.contrib.auth import get_user_model
+
+User = get_user_model()
+
+class UserSerializer(serializers.ModelSerializer):
+    is_driver = serializers.ReadOnlyField()
+
+    class Meta:
+        model = User
+        fields = ['id', 'username', 'email', 'is_driver']
+
+class BusSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Bus
+        fields = '__all__'
+
+class DriverSerializer(serializers.ModelSerializer):
+    user = UserSerializer(read_only=True)
+
+    class Meta:
+        model = Driver
+        fields = ['id', 'user']
+
+class PlaceSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Place
+        fields = '__all__'
+
+class BusStopSerializer(serializers.ModelSerializer):
+    place = PlaceSerializer(read_only=True)
+    class Meta:
+        model = BusStop
+        fields = '__all__'
+
+class BusShiftSerializer(serializers.ModelSerializer):
+    stops = serializers.PrimaryKeyRelatedField(
+        many=True,
+        queryset=BusStop.objects.all()
+    )
+    departure_time = serializers.SerializerMethodField()
+    arrival_time = serializers.SerializerMethodField()
+    shift_duration = serializers.SerializerMethodField()
+
+    class Meta:
+        model = BusShift
+        fields = ['id', 'bus', 'driver', 'stops', 'departure_time', 'arrival_time', 'shift_duration']
+
+    def get_departure_time(self, obj):
+        return obj.departure_time
+
+    def get_arrival_time(self, obj):
+        return obj.arrival_time
+
+    def get_shift_duration(self, obj):
+        duration = obj.shift_duration
+        if duration:
+            return duration.total_seconds() / 60  # minute
+        return 0
+
+    def validate_stops(self, stops):
+        if len(stops) < 2:
+            raise serializers.ValidationError("At least two stops are required.")
+        return stops
+
+    def validate(self, data):
+        bus = data.get('bus')
+        stops = data.get('stops', [])
+        driver = data.get('driver')
+
+        if stops and len(stops) >= 2:
+            stops_instances = BusStop.objects.filter(id__in=[s.id for s in stops])
+            departure_time = stops_instances.order_by('arrival_time').first().arrival_time
+            arrival_time = stops_instances.order_by('arrival_time').last().arrival_time
+
+            if self._has_time_conflict(BusShift.objects.filter(bus=bus), departure_time, arrival_time):
+                raise serializers.ValidationError({"bus": "The bus is already booked during this period"})
+
+            if self._has_time_conflict(BusShift.objects.filter(driver=driver), departure_time, arrival_time):
+                raise serializers.ValidationError({"driver": "The driver is already assigned to another shift during this period"})
+
+        return data
+
+    def _has_time_conflict(self, shifts, start, end):
+        for shift in shifts:
+            if shift.departure_time <= end and shift.arrival_time >= start:
+                return True
+        return False

padam_django/apps/fleet/urls.py

@@ -0,0 +1,8 @@
+from rest_framework.routers import DefaultRouter
+from .views import BusViewSet, DriverViewSet
+
+router = DefaultRouter()
+router.register(r'buses', BusViewSet)
+router.register(r'drivers', DriverViewSet)
+
+urlpatterns = router.urls

padam_django/apps/fleet/views.py

@@ -0,0 +1,12 @@
+from django.shortcuts import render
+from rest_framework import viewsets
+from .models import Bus, Driver
+from ...api.serializers import BusSerializer, DriverSerializer
+
+class BusViewSet(viewsets.ReadOnlyModelViewSet):
+    queryset = Bus.objects.all()
+    serializer_class = BusSerializer
+
+class DriverViewSet(viewsets.ModelViewSet):
+    queryset = Driver.objects.all()
+    serializer_class = DriverSerializer

padam_django/apps/transportation/admin.py

@@ -0,0 +1,35 @@
+from django.contrib import admin
+from .models import BusShift, BusStop
+from .forms import BusShiftForm
+
+@admin.register(BusShift)
+class BusShiftAdmin(admin.ModelAdmin):
+    form = BusShiftForm
+    list_display = ('bus', 'driver', 'departure_time_display', 'arrival_time_display', 'shift_duration_display', 'stops_list')
+    list_filter = ('bus', 'driver')
+
+    def departure_time_display(self, obj):
+        return obj.departure_time
+    departure_time_display.short_description = 'Departure Time'
+
+    def arrival_time_display(self, obj):
+        return obj.arrival_time
+    arrival_time_display.short_description = 'Arrival Time'
+
+    def shift_duration_display(self, obj):
+        duration = obj.shift_duration
+        if duration:
+            hours, remainder = divmod(duration.seconds, 3600)
+            minutes, _ = divmod(remainder, 60)
+            return f"{hours}h {minutes}m"
+        return "Duration not available"
+    shift_duration_display.short_description = "Shift Duration"
+
+    def stops_list(self, obj):
+        stops = obj.stops.all().order_by('arrival_time')
+        return ','.join(stop.place.name for stop in stops)
+    stops_list.short_description = "Stops"
+
+@admin.register(BusStop)
+class BusStopAdmin(admin.ModelAdmin):
+    list_display = ('place', 'arrival_time')

padam_django/apps/transportation/apps.py

@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+
+class TransportationConfig(AppConfig):
+    name = 'padam_django.apps.transportation'

padam_django/apps/transportation/factories.py

@@ -0,0 +1,38 @@
+import factory
+from faker import Faker
+from transportation import models as transportation_models
+from geography.factories import PlaceFactory
+from fleet.factories import BusFactory, DriverFactory
+from datetime import datetime, timedelta
+
+fake = Faker(['fr'])
+
+def generate_random_time_today():
+    today = datetime.today().date()
+    random_time = fake.time_object()
+    return datetime.combine(today, random_time)
+
+class BusStopFactory(factory.django.DjangoModelFactory):
+    place = factory.SubFactory(PlaceFactory)
+    arrival_time = factory.LazyFunction(generate_random_time_today)
+
+    class Meta:
+        model = transportation_models.BusStop
+
+class BusShiftFactory(factory.django.DjangoModelFactory):
+    bus = factory.SubFactory(BusFactory)
+    driver = factory.SubFactory(DriverFactory)
+
+    @factory.post_generation
+    def stops(self, create, extracted, **kwargs):
+        if not create:
+            return
+
+        if extracted:
+            for stop in extracted:
+                self.stops.add(stop)
+        else:
+            self.stops.add(BusStopFactory(), BusStopFactory())
+
+    class Meta:
+        model = transportation_models.BusShift

padam_django/apps/transportation/forms.py

@@ -0,0 +1,45 @@
+from datetime import datetime
+from django import forms
+from django.core.exceptions import ValidationError
+from .models import BusShift, BusStop
+
+class BusShiftForm(forms.ModelForm):
+
+    class Meta:
+        model = BusShift
+        fields = ('bus', 'driver', 'stops')
+
+    def clean(self):
+        cleaned_data = super().clean()
+        bus = cleaned_data.get('bus')
+        stops = cleaned_data.get('stops')
+        driver = cleaned_data.get('driver')
+
+        # Check if there are at least two stops
+        if stops and stops.count() >= 2:
+            departure_time = stops.order_by('arrival_time').first().arrival_time
+            arrival_time = stops.order_by('arrival_time').last().arrival_time
+        else:
+            self.add_error('stops', 'At least two stops are required')
+            raise ValidationError('At least two stops are required')
+
+        # Validate the bus and driver availability
+        self._validate_bus_availability(bus, departure_time, arrival_time)
+        self._validate_driver_availability(driver, departure_time, arrival_time)
+
+
+    def _validate_bus_availability(self, bus, departure_time, arrival_time):
+        existing_shifts = BusShift.objects.filter(bus=bus)
+        if self._has_time_conflict(existing_shifts, departure_time, arrival_time):
+            raise ValidationError(f'The bus {bus} is already booked during this time period.')
+
+    def _validate_driver_availability(self, driver, departure_time, arrival_time):
+        existing_shifts = BusShift.objects.filter(driver=driver)
+        if self._has_time_conflict(existing_shifts, departure_time, arrival_time):
+            raise ValidationError(f'The driver {driver} is already assigned to another shift during this time period.')
+
+    def _has_time_conflict(self, shifts, start, end):
+        for shift in shifts:
+            if shift.departure_time <= end and shift.arrival_time >= start:
+                return True
+        return False

padam_django/apps/transportation/migrations/0001_initial.py

@@ -0,0 +1,36 @@
+# Generated by Django 3.2.5 on 2024-06-07 17:54
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('fleet', '0002_auto_20211109_1456'),
+        ('geography', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='BusStop',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('arrival_time', models.DateTimeField()),
+                ('place', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='geography.place')),
+            ],
+        ),
+        migrations.CreateModel(
+            name='BusShift',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('start_time', models.DateTimeField()),
+                ('end_time', models.DateTimeField()),
+                ('bus', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='fleet.bus')),
+                ('driver', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='fleet.driver')),
+                ('stops', models.ManyToManyField(to='transportation.BusStop')),
+            ],
+        ),
+    ]

padam_django/apps/transportation/migrations/0002_auto_20240607_2038.py

@@ -0,0 +1,21 @@
+# Generated by Django 3.2.5 on 2024-06-07 18:38
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('transportation', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name='busshift',
+            name='end_time',
+        ),
+        migrations.RemoveField(
+            model_name='busshift',
+            name='start_time',
+        ),
+    ]

padam_django/apps/transportation/models.py

@@ -0,0 +1,40 @@
+from django.db import models
+from datetime import datetime, timedelta
+from django.core.exceptions import ValidationError
+
+class BusStop(models.Model):
+    place = models.ForeignKey('geography.Place', on_delete=models.CASCADE)
+    arrival_time = models.DateTimeField(auto_now=False, auto_now_add=False)
+
+    def __str__(self) -> str:
+        arrival_time_str = self.arrival_time.strftime('%Y-%m-%d %H:%M')
+        return f"BusStop: {self.place.name} arrive at: {arrival_time_str} (ID: {self.id})"
+
+class BusShift(models.Model):
+    bus = models.ForeignKey('fleet.Bus', on_delete=models.CASCADE)
+    driver = models.ForeignKey('fleet.Driver', on_delete=models.CASCADE)
+    stops = models.ManyToManyField(BusStop)
+
+    def __str__(self) -> str:
+        return f'Bus Shift: {self.bus.licence_plate} driven by {self.driver.user.username} (ID: {self.id})'
+
+    @property
+    def departure_time(self):
+        """Returns the departure time of the bus shift, based on the earliest stop time."""
+        if self.stops.exists():
+            return self.stops.all().order_by('arrival_time').first().arrival_time
+        return None
+
+    @property
+    def arrival_time(self):
+        """Returns the arrival time of the bus shift, based on the latest stop time."""
+        if self.stops.exists():
+            return self.stops.all().order_by('arrival_time').last().arrival_time
+        return None
+
+    @property
+    def shift_duration(self):
+        """Calculates the total duration of the bus shift."""
+        if self.departure_time and self.arrival_time:
+            return self.arrival_time - self.departure_time
+        return timedelta(0)