[BACK] feat: create pathing apps that allow to make and see bus pathing

padam_django/apps/common/management/commands/create_data.py

@@ -12,3 +12,5 @@ def handle(self, *args, **options):
         management.call_command('create_drivers', number=5)
         management.call_command('create_buses', number=10)
         management.call_command('create_places', number=30)
+        management.call_command('create_bus_stops', number=30)
+        management.call_command('create_bus_shifts', number=5)

padam_django/apps/pathing/admin.py

@@ -0,0 +1,26 @@
+
+from django.contrib import admin
+from django.core.exceptions import ValidationError
+from padam_django.apps.pathing.forms.bus_stop_form import BusStopForm
+from padam_django.apps.pathing.models.bus_shift import BusShift
+from padam_django.apps.pathing.models.bus_stop import BusStop
+from padam_django.apps.pathing.forms.bus_shift_form import BusShiftForm
+from django.utils.timezone import now
+
+
+@admin.register(BusShift)
+class BusShiftAdmin(admin.ModelAdmin):
+    form = BusShiftForm
+    list_display = ['bus', 'driver']
+
+    def save_model(self, request, obj, form, change):
+        bus_stops = form.cleaned_data.get('bus_stops')
+        obj.save()
+        for bus_stop in bus_stops:
+            bus_stop.bus_shift = obj
+            bus_stop.save()
+
+
+@admin.register(BusStop)
+class BusStopAdmin(admin.ModelAdmin):
+    form = BusStopForm

padam_django/apps/pathing/apps.py

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

padam_django/apps/pathing/factories/bus_shift_factory.py

@@ -0,0 +1,18 @@
+import factory
+from factory import django, List, SubFactory, LazyAttribute
+from padam_django.apps.fleet.factories import BusFactory, DriverFactory
+from padam_django.apps.pathing.factories.bus_stop_factory import BusStopFactory
+from padam_django.apps.pathing.models import BusShift
+
+
+class BusShiftFactory(django.DjangoModelFactory):
+    bus = factory.SubFactory(BusFactory)
+    driver = factory.SubFactory(DriverFactory)
+    bus_stops = factory.RelatedFactoryList(
+        BusStopFactory,
+        factory_related_name='bus_shift',
+        size=2
+    )
+
+    class Meta:
+        model = BusShift

padam_django/apps/pathing/factories/bus_stop_factory.py

@@ -0,0 +1,17 @@
+import factory
+from padam_django.apps.geography.factories import PlaceFactory
+from datetime import datetime, timedelta
+from random import randint
+
+from padam_django.apps.pathing.models import BusStop
+from django.utils.timezone import make_aware
+
+def futur_datetime():
+    return make_aware(datetime.now() + timedelta(days=randint(1, 30)))
+
+class BusStopFactory(factory.django.DjangoModelFactory):
+    visit_date_time = factory.LazyFunction(futur_datetime)
+    place = factory.SubFactory(PlaceFactory)
+
+    class Meta:
+        model = BusStop

padam_django/apps/pathing/forms/bus_shift_form.py

@@ -0,0 +1,46 @@
+from django import forms
+from padam_django.apps.pathing.models.bus_shift import BusShift
+from padam_django.apps.pathing.models.bus_stop import BusStop
+from django.core.exceptions import ValidationError
+
+class BusShiftForm(forms.ModelForm):
+    bus_stops = forms.ModelMultipleChoiceField(
+        queryset=BusStop.objects.filter(bus_shift__isnull=True),
+        widget=forms.CheckboxSelectMultiple,
+        required=True,
+        label="Select Bus Stops"
+    )
+
+    class Meta:
+        model = BusShift
+        fields = ['bus', 'driver']
+
+    def clean_bus_stops(self):
+        bus_stops = self.cleaned_data.get('bus_stops')
+        if len(bus_stops) < 2:
+            raise ValidationError("You must select at least 2 bus stops.")
+
+        bus = self.cleaned_data.get('bus')
+        driver = self.cleaned_data.get('driver')
+        bus_stops = self.cleaned_data.get('bus_stops')
+
+        ordered_stops = sorted(bus_stops, key=lambda stop: stop.visit_date_time)
+        departure_time = ordered_stops[0].visit_date_time
+        arrival_time = ordered_stops[-1].visit_date_time
+
+        overlapping_bus_shifts = BusShift.objects.filter(
+            bus=bus,
+            bus_stops__visit_date_time__lt=arrival_time,
+            bus_stops__visit_date_time__gt=departure_time,
+        ).exclude(pk=self.instance.pk)
+
+        overlapping_driver_shifts = BusShift.objects.filter(
+            driver=driver,
+            bus_stops__visit_date_time__lt=arrival_time,
+            bus_stops__visit_date_time__gt=departure_time,
+        )
+
+        if overlapping_bus_shifts.exists() or overlapping_driver_shifts.exists():
+            raise ValidationError("This bus or driver is already assigned to a shift during this time.")
+
+        return bus_stops

padam_django/apps/pathing/forms/bus_stop_form.py

@@ -0,0 +1,11 @@
+from django import forms
+from padam_django.apps.geography.models import Place
+from padam_django.apps.pathing.models.bus_stop import BusStop
+from django.core.exceptions import ValidationError
+
+class BusStopForm(forms.ModelForm):
+    class Meta:
+        model = BusStop
+        fields = ['visit_date_time', 'place']
+
+

padam_django/apps/pathing/management/commands/create_bus_shifts.py

@@ -0,0 +1,12 @@
+from padam_django.apps.common.management.base import CreateDataBaseCommand
+
+from padam_django.apps.pathing.factories.bus_shift_factory import BusShiftFactory
+
+class Command(CreateDataBaseCommand):
+
+    help = 'Create few bus shifts'
+
+    def handle(self, *args, **options):
+        super().handle(*args, **options)
+        self.stdout.write(f'Creating {self.number} bus shifts ...')
+        BusShiftFactory.create_batch(size=self.number)

padam_django/apps/pathing/management/commands/create_bus_stops.py

@@ -0,0 +1,12 @@
+from padam_django.apps.common.management.base import CreateDataBaseCommand
+
+from padam_django.apps.pathing.factories.bus_stop_factory import BusStopFactory
+
+class Command(CreateDataBaseCommand):
+
+    help = 'Create few bus stops'
+
+    def handle(self, *args, **options):
+        super().handle(*args, **options)
+        self.stdout.write(f'Creating {self.number} bus stops ...')
+        BusStopFactory.create_batch(size=self.number)

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

@@ -0,0 +1,42 @@
+# Generated by Django 4.2.16 on 2024-11-22 20:12
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('geography', '0001_initial'),
+        ('fleet', '0002_auto_20211109_1456'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='BusShift',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('bus', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='bus_shifts', to='fleet.bus')),
+                ('driver', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='bus_shifts', to='fleet.driver')),
+            ],
+            options={
+                'verbose_name': 'Bus Shift',
+                'verbose_name_plural': 'Bus Shifts',
+            },
+        ),
+        migrations.CreateModel(
+            name='BusStop',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('visit_date_time', models.DateTimeField()),
+                ('bus_shift', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='bus_stops', to='pathing.busshift')),
+                ('place', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='bus_stops', to='geography.place')),
+            ],
+            options={
+                'verbose_name': 'Bus Stop',
+                'verbose_name_plural': 'Bus Stops',
+            },
+        ),
+    ]

padam_django/apps/pathing/models/__init__.py

@@ -0,0 +1,2 @@
+from .bus_stop import BusStop
+from .bus_shift import BusShift

padam_django/apps/pathing/models/bus_shift.py

@@ -0,0 +1,26 @@
+from django.db import models
+from padam_django.apps.fleet.models import Bus, Driver
+
+
+class BusShift(models.Model):
+    bus = models.ForeignKey(Bus, on_delete=models.CASCADE, related_name="bus_shifts")
+    driver = models.ForeignKey(Driver, on_delete=models.CASCADE, related_name="bus_shifts")
+
+    class Meta:
+        verbose_name = "Bus Shift"
+        verbose_name_plural = "Bus Shifts"
+
+    def get_ordered_bus_stops(self):
+        return sorted(self.bus_stops.all(), key=lambda stop: stop.visit_date_time)
+
+    def get_departure(self):
+        return self.get_ordered_bus_stops()[0]
+
+    def get_arrival(self):
+        return self.get_ordered_bus_stops()[-1]
+
+    def get_travel_time(self):
+        return self.get_arrival().visit_date_time - self.get_departure().visit_date_time
+
+    def __str__(self):
+        return f"{self.id} - {self.bus} - {self.driver}"

padam_django/apps/pathing/models/bus_stop.py

@@ -0,0 +1,29 @@
+from django.core.exceptions import ValidationError
+from django.db import models
+
+from padam_django.apps.geography.models import Place
+from padam_django.apps.pathing.models.bus_shift import BusShift
+from django.utils.timezone import now
+
+
+class BusStop(models.Model):
+    visit_date_time = models.DateTimeField()
+    place = models.ForeignKey(Place, on_delete=models.CASCADE, related_name="bus_stops")
+    bus_shift = models.ForeignKey(BusShift, on_delete=models.CASCADE, related_name="bus_stops", null=True, blank=True)
+
+    def clean(self):
+        super().clean()
+
+        if self.visit_date_time and self.visit_date_time <= now():
+            raise ValidationError("The visit date and time can't be in the past.")
+
+    def save(self, *args, **kwargs):
+        self.full_clean()
+        super().save(*args, **kwargs)
+
+    class Meta:
+        verbose_name = "Bus Stop"
+        verbose_name_plural = "Bus Stops"
+
+    def __str__(self):
+        return f"{self.visit_date_time} - {self.place}"

padam_django/apps/pathing/tests/conftest.py

@@ -0,0 +1,17 @@
+import pytest
+from django.contrib.auth import get_user_model
+
+@pytest.fixture
+def create_admin_user():
+    user_model = get_user_model()
+    admin_user = user_model.objects.create_superuser(
+        username='admin',
+        email='admin@example.com',
+        password='adminpassword'
+    )
+    return admin_user
+
+@pytest.fixture
+def authenticated_client(create_admin_user, client):
+    client.login(username='admin', password='adminpassword')
+    return client

padam_django/apps/pathing/tests/test_bus_shift.py

@@ -0,0 +1,69 @@
+import datetime
+from django.utils import timezone
+
+import pytest
+from django.urls import reverse
+from padam_django.apps.fleet.factories import DriverFactory, BusFactory
+from padam_django.apps.pathing.factories.bus_shift_factory import BusShiftFactory
+from padam_django.apps.pathing.factories.bus_stop_factory import BusStopFactory
+from padam_django.apps.pathing.models import BusShift
+
+
+@pytest.mark.django_db
+class TestAdminBusShift:
+    def test_add_bus_shift(self, authenticated_client):
+        bus = BusFactory()
+        driver = DriverFactory()
+        bus_stop_1 = BusStopFactory()
+        bus_stop_2 = BusStopFactory()
+        data = {
+            'bus': bus.id,
+            'driver': driver.id,
+            'bus_stops': [bus_stop_1.id, bus_stop_2.id],
+        }
+        url = reverse('admin:pathing_busshift_add')
+        response = authenticated_client.post(url, data)
+        assert response.status_code == 302
+        assert BusShift.objects.count() == 1
+        bus_shift = BusShift.objects.first()
+        assert bus_shift.bus.id == bus.id
+
+    def test_add_bus_shift_only_one_stop(self, authenticated_client):
+        bus = BusFactory()
+        driver = DriverFactory()
+        bus_stop_1 = BusStopFactory()
+        data = {
+            'bus': bus.id,
+            'driver': driver.id,
+            'bus_stops': [bus_stop_1.id],
+        }
+        url = reverse('admin:pathing_busshift_add')
+        response = authenticated_client.post(url, data)
+        assert response.status_code == 200
+        assert BusShift.objects.count() == 0
+        assert "You must select at least 2 bus stops." in response.content.decode()
+
+
+    def test_add_bus_shift_overlap_bus(self, authenticated_client):
+        bus = BusFactory()
+        driver = DriverFactory()
+        now = timezone.now()
+        bus_stop_1 = BusStopFactory(visit_date_time=now + datetime.timedelta(hours=1))
+        bus_stop_2 = BusStopFactory(visit_date_time=now + datetime.timedelta(hours=2))
+        bus_stop_3 = BusStopFactory(visit_date_time=now + datetime.timedelta(hours=3))
+        bus_stop_4 = BusStopFactory(visit_date_time=now + datetime.timedelta(hours=4))
+        shift = BusShiftFactory(bus=bus, bus_stops=[bus_stop_2, bus_stop_4])
+        shift.bus_stops.add(bus_stop_2, bus_stop_4)
+        data = {
+            'bus': bus.id,
+            'driver': driver.id,
+            'bus_stops': [bus_stop_1.id, bus_stop_3.id],
+        }
+        url = reverse('admin:pathing_busshift_add')
+        response = authenticated_client.post(url, data)
+        assert response.status_code == 200
+        assert "This bus or driver is already assigned to a shift during this time." in response.content.decode()
+
+
+
+