rewrote calculer_frequentation, added basic test for it.

maraudes/tests.py

@@ -8,9 +8,7 @@ from .models import (
     Maraude, Maraudeur, Planning,
     WEEKDAYS, HORAIRES_SOIREE,
     )
-# Create your tests here.
 
-from maraudes_project.base_data import MARAUDEURS
 
 MARAUDE_DAYS = [
     True, True, False, True, True, False, False

statistiques/charts.py

@@ -1,4 +1,5 @@
 import datetime
+import collections
 
 from django.db.models.functions.datetime import ExtractMonth
 from django.db.models import (Field, NullBooleanField,
@@ -28,6 +29,7 @@ LABELS = {
         NullBooleanField: {True: "Oui", False: "Non", None: "Ne sait pas"},
     }
 
+
 # TODO: Retrieve charts data from cache to avoid recalculating on each request
 class CachedDataSource:
     pass
@@ -136,7 +138,7 @@ class DonneeGeneraleChart(gchart.BarChart):
 
     def __init__(self, maraudes=None, rencontres=None, sujets=None):
 
-        data = [("...", "Soirée", "Journée", { 'role': 'annotation'})]
+        data = [("...", "Soirée", "Journée", {'role': 'annotation'})]
 
         data += [("Maraudes", maraudes[2].count(), maraudes[1].count(), maraudes[0].count())]
         data += [("Rencontres", rencontres[2].count(), rencontres[1].count(), rencontres[0].count())]
@@ -145,6 +147,7 @@ class DonneeGeneraleChart(gchart.BarChart):
 
         super().__init__(SimpleDataSource(data), options={'title': 'Données générales', 'isStacked': 'percent'})
 
+
 def generate_piechart_for_field(field):
     """ Returns a PieWrapper subclass working with a fixed field """
     class FieldChart(PieWrapper):
@@ -217,7 +220,7 @@ class RencontreParMoisChart(gchart.ColumnChart):
                             ).order_by()
             data += [(NOM_MOIS[item['mois']], item['nbr']) for item in par_mois]
         else:
-            data += [("Mois",0)]
+            data += [("Mois", 0)]
         super().__init__(SimpleDataSource(data),
                          options={
                             "title": "Nombre de rencontres par mois"
@@ -229,8 +232,8 @@ class RencontreParHeureChart(gchart.AreaChart):
     def __init__(self, queryset):
         data = [("Heure", "Rencontres démarrées", "Au total (démarré + en cours)")]
         if queryset:
-            par_heure = self.calculer_frequentation_par_quart_heure(queryset, continu=False)
+            par_heure = self.calculer_frequentation(queryset, continu=False)
-            en_continu = self.calculer_frequentation_par_quart_heure(queryset, continu=True)
+            en_continu = self.calculer_frequentation(queryset, continu=True)
             data += [(heure, par_heure[heure], en_continu[heure]) for heure in sorted(par_heure.keys())]
         else:
             data += [("Heure", 0, 0)]
@@ -241,56 +244,33 @@ class RencontreParHeureChart(gchart.AreaChart):
                          )
 
     @staticmethod
-    def calculer_frequentation_par_quart_heure(observations, continu=False):
+    def calculer_frequentation(observations, step=15, continu=False):
-        """ Calcule le nombre d'observations, de 16h à 24h, par tranche de 15min.
-            L'algorithme est *très peu* efficace mais simple à comprendre : on calcule pour
-            chaque tranche les observations qui y sont contenues.
-            On peut calculer seulement les observations démarrées (continu = False) ou considérer
-            que l'observation est contenue dans un intervalle sur toute sa durée (continu = True).
 
-        """
+        def find_intervalle(temps):
-        data = dict()
+            return datetime.time(temps.hour, temps.minute // step * step)
 
-        def genere_filtre_pour(heure, indice):
+        def range_over_intervalles(rencontre):
-            """ Renvoie une fonction qui renvoie True si l'intervalle donné contient l'observation, c'est-à-dire :
+            """ Génère tous les intervalles contenus entre le début et la fin de la rencontre """
-            1. Elle démarre/finit dans l'intervalle.
+            curseur = find_intervalle(rencontre.heure_debut)
-            2. Elle démarre avant et fini après l'intervalle.
+            # Convertir en objet datetime
-            """
+            curseur = datetime.datetime.now().replace(hour=curseur.hour,
-            debut_intervalle = indice * 15
+                                                      minute=curseur.minute)
-            fin_intervalle = debut_intervalle + 15
+            debut = datetime.datetime.now().replace(hour=rencontre.heure_debut.hour,
-            rng = range(debut_intervalle, fin_intervalle)
+                                                    minute=rencontre.heure_debut.minute)
+            fin = debut + datetime.timedelta(0, rencontre.duree * 60)
+            delta = datetime.timedelta(0, step * 60)
 
-            def est_contenue(observation):
+            while curseur < fin:
-                """ Vérifie l'observation est contenue dans l'intervalle """
+                yield datetime.time(curseur.hour, curseur.minute)
-                debut = datetime.datetime.strptime(
+                curseur += delta
-                    "%s" % observation.rencontre.heure_debut,
-                    "%H:%M:%S"
-                )
-                fin = debut + datetime.timedelta(0, observation.rencontre.duree * 60)
 
-                # L'observation démarre dans l'intervalle
+        data = collections.defaultdict(lambda: 0)
-                if debut.hour == heure and debut.minute in rng:
-                    return True
-                # L'observation finit dans l'intervalle, seulement si continu est True
-                elif continu and (fin.hour == heure and fin.minute in rng):
-                    return True
-                # L'observation démarre avant ET finit après l'intervalle,
-                # seulement si continu est True
-                elif (continu
-                      and (debut.hour <= heure and debut.minute <= debut_intervalle)
-                      and (fin.hour >= heure and fin.minute >= fin_intervalle)):
-                    return True
-                else:
-                    return False
 
-            return est_contenue
+        for rencontre in map(lambda o: o.rencontre, observations):
-
+            if not continu:
-        for h in range(16, 24):
+                data[find_intervalle(rencontre.heure_debut)] += 1
-            for i in range(4):
+            else:
-                filtre = genere_filtre_pour(heure=h, indice=i)
+                for intervalle in range_over_intervalles(rencontre):
-                contenus = list(filter(filtre, observations))
+                    data[intervalle] += 1
-
-                key = datetime.time(h, i * 15)
-                data[key] = len(contenus)
 
         return data

statistiques/tests.py

@@ -1,7 +1,67 @@
 from django.test import TestCase
+from datetime import date, time
+
+from maraudes.models import Maraude, Rencontre, Lieu
+from notes.models import Sujet
+from utilisateurs.models import Maraudeur
+from maraudes.notes import Observation
+from .charts import RencontreParHeureChart
 
 # Create your tests here.
 
 # MANDATORY FEATURES
 
 # FicheStatistique primary key IS it's foreign sujet pk
+class TestCalculerFrequentationParHeure(TestCase):
+
+    def setUp(self):
+        maraude = Maraude.objects.create(date=date(2017, 1, 1),
+                                         heure_debut=time(20, 00),
+                                         binome=Maraudeur.objects.create(first_name="Asterix", last_name="Gaulois"),
+                                         referent=Maraudeur.objects.create(first_name="Referent", last_name="R", is_superuser=True))
+        self._generate_test_data(maraude, [
+            ("Gare", time(20, 30), 45, ("S1",)),  #
+            ("Gare", time(20, 40), 10, ("S2", "S3",)),
+            ("Centre-ville", time(21, 30), 30, ("S4",)), # Borne finale fermée
+            ("Urgences", time(22, 20), 26, ("S5", "S3")),  # Déborde sur un nouvel intervalle
+        ])
+
+    def _generate_test_data(self, maraude, data):
+        for lieu, heure, duree, sujets in data:
+            rencontre = Rencontre.objects.create(maraude=maraude,
+                                                 lieu=Lieu.objects.get_or_create(nom=lieu)[0],
+                                                 heure_debut=heure,
+                                                 duree=duree)
+            for s in sujets:
+                observation = Observation.objects.create(rencontre=rencontre,
+                                                         sujet=Sujet.objects.get_or_create(surnom=s)[0],
+                                                         text="RAS")
+
+    def differences(self, first, second):
+        diff = dict()
+        for key, val in first.items():
+            try:
+                if not second[key] == val:
+                    diff[key] = (val, second[key])
+            except KeyError:
+                diff[key] = (val, None)
+        return diff
+
+    def test_with_test_data(self):
+
+        # Retrieve observations queryset
+        queryset = Observation.objects.filter(created_date=date(2017,1,1))
+
+        non_continu = RencontreParHeureChart.calculer_frequentation(queryset, continu=False)
+        continu = RencontreParHeureChart.calculer_frequentation(queryset, continu=True)
+        test_non_continu = {time(20, 30): 3,
+                            time(21, 30): 1,
+                            time(22, 15): 2,
+                            }
+        self.assertEqual(non_continu, test_non_continu, "\nDifferences :\n{}".format(self.differences(non_continu, test_non_continu)))
+
+        test_continu = {time(20, 30): 3, time(20, 45): 3,
+                        time(21, 0): 1, time(21, 30): 1, time(21, 45): 1,
+                        time(22, 15): 2, time(22, 30): 2, time(22, 45): 2,
+                        }
+        self.assertEqual(continu, test_continu,  "\nDifferences :\n{}".format(self.differences(continu, test_continu)))