Créer un algorithme d’itinéraires pour les transports publics genevois

Bonjour !

Aujourd’hui était l’annonce des résultats pour le baccalauréat du Ministère de l’Éducation nationale et de la Jeunesse en France.

Ayant fait partie des (nombreux) candidats cette année, j’ai passée l’épreuve d’Informatique et Sciences du Numérique, et je souhaitais publier le compte-rendu de mon projet, qui n’étaient rien de moins qu’un algorithme d’itinéraires pour les transports publics genevois !

À noter que le code qui est actuellement en usage dans tpg offline API est un peu différent de ce que j’ai présenté pour le baccalauréat, mais les principes de base restent les mêmes. Bon, ça m’a pas empêché d’avoir 20/20 finalement 😊.

Le compte-rendu original, sous forme de PDF: https://レミー.みんな/ISN.pdf

Introduction

Le compte-rendu que vous allez lire est l’aboutissement d’un projet d’algorithme qui fut construit sur une année scolaire, et c’est avec grand plaisir que je vous le présente aujourd’hui.

L’idée de créer ce projet vient du fait que je sois le développeur d’une application pour les transports public genevois, qui propose une fonction hors-ligne. Je voulais donc permettre à l’utilisateur de chercher des itinéraires dans cette application, même sans accès à internet.

Étant le seul inscrit dans mon école pour cette spécialité, ce projet ne comportera pas de sections consacrées à une analyse de la collaboration entre élèves.

Cahier des charges du projet

Voici les exigences que je me suis fixées pour mon projet :

  • L’algorithme doit être fiable et donner des itinéraires cohérents
  • L’algorithme doit calculer au moins un itinéraire par seconde.
  • L’algorithme doit pouvoir être intégré facilement à une plate-forme web (typiquement, par le biais d’une API, une interface de programmation applicative)
  • L’algorithme doit consommer le moins ressources possible, que ce soit de la puissance de calcul ou de la mémoire temporaire (mémoire RAM).
  • Le format d’entrée des données doit être standardisé par :
    • Les spécifications GTFS (General Transit Feed Specification)
    • Ou une base de données SQL

Le langage de programmation

Pour ce projet, le choix du langage de programmation s’est porté sur Python. Ce langage de programmation, conçu par Guido van Rossum en 1991, est un des langages les plus populaires au monde. Il est notamment adapté pour être utilisé pour gérer des interfaces de programmations d’applications sur Internet, et il peut être intégré aisément à un projet iOS ou Android.

L’algorithme

J’ai décidé de prendre comme base le Connection Scan Algorithm, que l’on peut abréger par l’acronyme CSA. Le principe de cet algorithme est simple à mettre en œuvre, et est économique en termes de ressources.

Initialisation

Pour chaque station s, nous allons garder le trajet le plus rapide arrivant au plus proche de l’horaire désiré sans le dépasser. Le nombre de connexions au sein d’un même trajet n’est pas un facteur pris en compte, seule la durée la plus courte pour arriver à l’heure est considérée. Pour cela, nous allons utiliser les tableaux arrival_timestamp[s] et in_connection[s]. Pour chaque arrêt, ces deux variables seront initialisées avec une valeur impossible à atteindre normalement, tels que par exemple 2″# − 1, la valeur maximum autorisée pour les entiers dans Python, afin de permettre de comparer les meilleures connexions.

# Initialisation

MAX_STATIONS = 100_000
MAX_INT = 2 ** 32 - 1

Boucle principale

La boucle principale de l’algorithme va consister à parcourir les connexions disponibles, avec comme contrainte d’être postérieures à l’heure de départ, et antérieures à l’heure d’arrivée la plus petite, qui est calculée en temps-réel.
Une fois toute les connexions explorées, l’itinéraire sera construit.

for connection in connections:
    if (
        connection.departure_time > earlist_arrival[connection.departure_stop].time
        connection.arrival_time < earlist_arrival[connection.arrival_stop].time
    ):
        earlist_arrival[connection.arrival_stop].time = connection.arrival_time
        in_connection[connection.arrival_stop] = connection.id

Reconstruction de l’itinéraire

La reconstruction de l’itinéraire se fait à partir de l’arrêt d’arrivée et de regarder dans in_connection les connexions jusqu’à atteindre l’arrêt de départ, et nous aurons l’itinéraire avec les horaires.

Inconvénients de la forme actuelle de l’algorithme

L’algorithme a plusieurs inconvénients sous cette forme, par exemple, il ne prend pas en compte les temps de connexion, ni même les stations multi modes de transport. Enfin, le programme dans l’état tel qu’il est ne peut être intégré sous forme de module dans un programme Python, car il n’offre aucune option de personnalisation de l’algorithme, que ce soit par rapport aux arrêts ou aux horaires.

Mon but cette année a donc été de corriger ces inconvénients, et d’améliorer l’algorithme.

Écriture du programme

Note : Pour respecter la norme PEP8, les noms de variables / fonctions / classes et les sorties du programme seront en anglais, à l’exception des commentaires qui, pour une meilleure compréhension des lecteurs francophones, seront en français. Il est à noter que le code et les commentaires seront totalement en anglais sur les versions en ligne de l’algorithme.

Programmation Orienté Objet (POO)

La première étape de l’écriture du programme est la définition de sa structure. Nous allons ici utiliser le principe de programmation orienté objet, qui est une approche solide et utilisé de manière universelle en programmation.

Pour expliquer de manière très succincte la POO, cette structure consiste en l’utilisation de objets, définis par des classes, qui seront elles-mêmes composés de méthodes et d’attributs, pouvant être appelés ou non selon si elles sont définies comme abstraites, et les classes peuvent être hérités pour créer des classes enfants qui reprendront les attributs et méthodes de leurs parents.
Donc, pour commencer, nous initialisons une classe qui servira de base à notre programme

class TpgRoutes:
    def __init__(self, logger=None):
        super(TpgRoutes, self).__init__()

if __name__ == "__main__":
    # Si le fichier est appelé directement, 
    # nous pouvons lancer un test de 
    # fonctionnement de l'algorithme.
    tpgRoutes = TpgRoutes()
    tpgRoutes.runTest()

Les sorties du programme

Pour comprendre comment fonctionne un programme, ou pour tout simplement voir les résultats du programme, nous avons besoin de sorties. Python intègre un moyen simple d’afficher des sorties, avec la fonction print. Cependant, il est assezdifficile de récupérer ces sorties si le programme a été importé comme module, ou s’il est exécuté par un démon (programme lancé de manière autonome, généralement par le système d’exploitation). C’est pourquoi nous allons utiliser la librairie logging pour afficher nos sorties. Cette librairie a aussi l’avantage de permettre une catégorisation des sorties, allant de la catégorie debug, pour les sorties permettant aux développeurs de comprendre comment fonctionne le programme, à la catégorie critical, pour indiquer des erreurs de fonctionnement critiques, en passant par les modes info, warning, et error.

class TpgRoutes:
    def __init__(self, logger=None):
        super(TpgRoutes, self).__init__()
        if logger == None:
            self.configure_logs()
        else:
            self.logger = logger
            # Si un logger existe déjà,
            # par exemple en cas d'utilisation de l'algorithme en tant que module,
            # on évite d'initialiser un nouveau logger, et on utilise celui existant

    def configure_logs(self):
        """
        Avoir les sorties du programme est crucial, et la fonction print est quasiment inutilisable si le programme est exécuté par un démon.
        C'est pourquoi nous initialisons ici la librairie logging, qui fait partie de la bibliothèque de librairie standard à Python.
        """

        self.logger = logging.getLogger()
        self.logger.setLevel(logging.DEBUG)

        formatter = logging.Formatter("%(asctime)s :: %(levelname)s :: %(message)s")
        file_handler = RotatingFileHandler("activity.log", "a", 1_000_000, 1)

        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)

        stream_handler = logging.StreamHandler()
        stream_handler.setLevel(logging.DEBUG)
        self.logger.addHandler(stream_handler)

Les données du programme

Concernant l’alimentation du programme en données, nous allons utiliser le jeu de données fournis par le Secrétariat des tâches du système d’information à la clientèle (CFF SA), et disponible sur le portail des données ouvertes de l’administration publique suisse opendata.swiss (https://opendata.swiss/fr/dataset/timetable-2019-gtfs). Ce jeu de données est disponible au format GTFS, mais pour des raisons pratiques et de performances, nous allons convertir les données GTFS en base de données SQLite avec le format suivant pour chaque jour :

CREATE TABLE monday (
id integer PRIMARY KEY,
departure_stop integer NOT NULL,
arrival_stop integer NOT NULL,
departure_time integer NOT NULL,
arrival_time integer NOT NULL,
line string NOT NULL,
trip_id integer NOT NULL,
destination_stop integer NOT NULL)

Il est à noter que lors de la création de la base de données, un filtrage des départs est effectué car le jeu de données concernait la totalité des réseaux de transports urbain et interurbains de Suisse, alors que nous n’avons besoin que d’une partie restreinte.

def load_database(self, path):
        """
        Initialiser les données d'itinéraires avec une base SQLite
        :param path: Chemin vers la base de données SQLite
        """
        self.logger.info("Loading database...")
        try:
            self.database = sqlite3.connect(path)
            self.logger.info("Database loaded")
        except:
            self.logger.exception("Loading database failed")

Une fois la base de données créée, il est nécessaire de la charger dans le programme. Pour cela, nous allons utiliser la librairie sqlite3, qui est fournie dans la bibliothèque de librairie standard à Python.

L’algorithme

Pour commencer l’algorithme, nous allons d’abord avoir besoin de recueillir les données et de les vérifier, à savoir l’accès à la base de données SQLite, la station de départ, la station d’arrivée, l’heure de départ et l’heure d’arrivée.

class TpgRoutes:
    def compute_route(
        self, departure_station: int, arrival_stop: int, departure_time: int, day: int
    ):
        """
        Calcul un itinéraire depuis un arrêt A vers un arrêt B avec le temps de trajet le plus court possible, et au plus proche de l'heure de départ
        :param departure_station: Identifiant de l'arrêt de départ dans la base de données, habituellement l'identifiant CFF
        :param arrival_stop: Identifiant de l'arrêt d'arrivée dans la base de données, habituellement l'identifiant CFF
        :param departure_time: Heure minimum de départ, en secondes depuis minuit
        :param day: Jour de la semaine, peut-être un entier en 0 et 6, de dimanche à samedi
        """

        try:
            self.cursor = self.database.cursor()
        except:
            self.logger.exception(
                "Access to database failed. Please, check you had loaded the database with load_database."
            )
            return

        assert departure_time > 0, "departure_time is less than 0 seconds"
        assert (
            departure_time < 86400
        ), "departure_time is greater than 86400 seconds, also known as the number of seconds in a day"

        if day == 0:
            dayString = "sunday"
        elif day == 6:
            dayString = "saturday"
        elif day == 5:
            dayString = (
                "friday"
            )  # Les horaires de vendredi sont différents de ceux des jours entre lundi et vendredi, en raison de la présence des horaires des lignes Noctambus.
        elif day > 0 and day < 5:
            dayString = "monday"
        else:
            self.logger.exception(
                "Day is incorrect, it should be an integer between 0 and 6 included"
            )
            return

Une fois ces vérifications effectuées, nous pouvons commencer à coder l’algorithme. Nous allons reprendre les principes indiqués précédemment, en commençant par initialiser le programme. Nous initialisons l’algorithme avec les tableaux arrival_timestamp[s] et in_connection[s], et nous attribuons comme heure d’arrivée à la station de départ, l’heure de départ.

Nous créons aussi une classe EarliestArrival pour rendre le code plus lisible.

class EarliestArrival:
    time: int
    trip_id: str
    connection: int

    def __init__(self, time, trip_id, connection):
        self.time = time
        self.trip_id = trip_id
        self.connection = connection

class TpgRoutes:
    def compute_route(
        self, departure_station: int, arrival_stop: int, departure_time: int, day: int
    ):
        ...
        in_connection = array("I", [MAX_INT for _ in range(MAX_STATIONS)])
        earliest_arrival = [
            EarliestArrival(MAX_INT, "", 0) for _ in range(MAX_STATIONS)
        ]

        earliest_arrival[departure_station].time = departure_time

Pour continuer, nous allons charger les données depuis la base de données. Pour cela, nous utilisons la variable cursor créée précédemment, et nous allons sélectionner les connexions qui nous intéressent avec une requête SQL.

class ConnectionRow:
    id: int
    departure_stop: int
    arrival_stop: int
    departure_time: int
    arrival_time: int
    line: str
    trip_id: int
    destination_stop: int

    def __init__(self, c):
        self.id = c[0]
        self.departure_stop = c[1]
        self.arrival_stop = c[2]
        self.departure_time = c[3]
        self.arrival_time = c[4]
        self.line = c[5]
        self.trip_id = c[6]
        self.destination_stop = c[7]

class TpgRoutes:
    def compute_route(
        self, departure_station: int, arrival_stop: int, departure_time: int, day: int
    ):
        ...
        # Juste au cas où, nous vérifions que les arrêts de départs et les arrêts d'arrivée sont dans l'écart des identifiants autorisés
        if departure_station <= MAX_STATIONS and arrival_stop <= MAX_STATIONS:
            earliest = MAX_INT

            # Nous prenons ici dans la base de données tous les départs étant postérieur à l'heure minimum de départ, ordonné par l'heure de départ
            self.cursor.execute(
                f"SELECT * FROM {dayString} WHERE departure_time >= {departure_time} and departure_time < arrival_time ORDER BY departure_time"
            )
            for c in self.cursor.fetchall():
                connection = ConnectionRow(
                    c
                )  # Initialiser avec une classe permet de rendre le code plus lisible, et de rendre le débogage plus simple

ISN

À noter que nous créons ici une nouvelle classe ConnectionRow, qui va représenter chaque résultat de la requête SQL, de manière à rendre le code plus lisible et rendre le débogage beaucoup plus simple. Ainsi, au lieu d’appeler la valeur pour la ligne par connection[5], on utilisera connection.line.

Maintenant que nous sommes dans la boucle pour explorer chaque connexion possible, nous pouvons implémenter la suite de l’algorithme. À noter que, contrairement à l’algorithme initial, nous introduisons ici la notion de temps de correspondances. Ce temps de correspondance est personnalisable, et n’est pas appliqué lorsqu’on ne change pas de véhicule. Nous allons aussi calculer le nombre de lignes empruntées, ce qui pourra être utilisé à titre informatif ou pour de futures améliorations.

class TpgRoutes:
    def compute_route(
        self, departure_station: int, arrival_stop: int, departure_time: int, day: int
    ):
        ...
            for c in self.cursor.fetchall():
                ...
                minimum_connection_time = 120  # en secondes
                if not earliest_arrival[connection.departure_stop].trip_id:
                    # Si nous sommes à l'arrêt de départ, alors il n'y a pas besoin d'un délai de connexion
                    minimum_connection_time = 0
                elif c[6] == earliest_arrival[c[1]].trip_id:
                    # Si nous continuons notre trajet avec le même véhicule, alors il n'y a pas besoin d'un délai de connexion
                    minimum_connection_time = 0

                # Pour de futures améliorations, nous calculons le nombres de lignes différentes
                differents_lines = 0
                if (
                    connection.trip_id
                    != earliest_arrival[connection.departure_stop].trip_id
                ):
                    differents_lines = 1

                if (
                    connection.departure_time
                    > (
                        earliest_arrival[connection.departure_stop].time
                        + minimum_connection_time
                    )
                    and connection.arrival_time
                    < earliest_arrival[connection.arrival_stop].time
                ):
                    # Si cette connexion est un meilleur choix que celui utilisé précédamment, alors nous allons enregistrer celui-ci
                    earliest_arrival[
                        connection.arrival_stop
                    ].time = connection.arrival_time
                    earliest_arrival[connection.arrival_stop].connection = (
                        earliest_arrival[connection.departure_stop].connection
                        + differents_lines
                    )
                    earliest_arrival[
                        connection.arrival_stop
                    ].trip_id = connection.trip_id
                    in_connection[connection.arrival_stop] = connection.id

                    if connection.arrival_stop == arrival_stop:
                        # Nous sommes à l'arrêt d'arrivée, il faut enregistrer le parcours le plus court
                        earliest = min(earliest, connection.arrival_time)
                elif connection.departure_time > earliest:
                    # Parce que les départs sont déjà ordonnés par heure de départ,
                    # si le présent départ à une heure de départ postérieur à l'heure d'arrivée minimum,
                    # nous pouvons arrêter la boucle, nous avons trouvé le meilleur itinéraire
                    continue

Une fois l’itinéraire calculé, il ne manque plus qu’à reconstruire l’itinéraire depuis l’arrêt d’arrivée, et à retourner les résultats

class TpgRoutes:
    def compute_route(
        self, departure_station: int, arrival_stop: int, departure_time: int, day: int
    ):
        ...
        # Maintenant que l'itinéraire est calculé dans deux tableaux, nous pouvons donc le reconstruire

            if in_connection[arrival_stop] == MAX_INT:
                # Si l'arrêt d'arrivée n'a pas de connexion, alors aucun itinéraire n'a été trouvé
                # Cela peut arriver parfois, notamment lorsque l'heure de départ est proche de l'heure de fin de service
                self.logger.info("No solution was found")
            else:
                route = []

                # Nous devons reconstruire l'itinéraire depuis l'arrêt d'arrivée
                last_connection_index = in_connection[arrival_stop]

                while last_connection_index != MAX_INT:
                    self.cursor.execute(
                        f"SELECT * FROM {dayString} WHERE id = {last_connection_index}"
                    )
                    connection = self.cursor.fetchall()[0]
                    route.append(connection)
                    last_connection_index = in_connection[connection[1]]

                # Et nous pouvons retourner le résultat dans le bon ordre
                return reversed(route)

Une fois les résultats retournés, l’itinéraire a été calculé.

Le code complet

Pour des raisons d’espace dans ce compte-rendu, vous trouverez le code complet du programme dans l’annexe, ainsi que sur Internet (https://github.com/tpgoffline/TpgRoutes/blob/master/tpgroutes.py)

Performances

Afin d’avoir une moyenne de performance correcte conformément au cahier des charges, l’algorithme a été exécuté plus d’un milliard de fois, avec les arrêts de départs, d’arrivée, et les heures de départs choisis aléatoirement.
Sur un processeur avec une fréquence moyenne de 2.8 GHz, le chargement des 11 Mo de base de données au format SQLite a été ce qu’il y a de plus lent, prenant en moyenne 2.48 secondes, et l’algorithme a, par la suite, été capable de retourner les résultats en moyenne en 0.43 seconde.

Respect du cahier des charges

Le cahier des charges défini en début du compte-rendu a été respecté intégralement. L’algorithme donne des itinéraires cohérents, n’utilise que très peu de ressources systèmes, et donne des résultats dans les temps impartis.

Publication du projet

Le code en lui-même de ce projet a été rendu public sur GitHub, sous la licence MIT (licence simple et permissive concernant les droits du projet). Vous pourrez ainsi retrouver le projet sur https://github.com/tpgoffline/TpgRoutes/.
Le projet a aussi été publié sur PyPi, et peut désormais être installé avec pip, ce qui permet une installation et une exécution simplifiée de ce module.

Ce code est actuellement exploité à usage interne dans les futures mises à jour de mon application tpg offline, ainsi que sur l’API que j’ai réalisée, que vous pourrez retrouver et tester sur https://api.tpgoffline.com.

De possibles améliorations

En amélioration futures, il serait théoriquement possible de rendre l’algorithme plus rapide. La version actuelle de cet algorithme ne cherche pas non plus l’itinéraire avec le moins de connexions possibles, ce qui serait intéressant à faire afin de donner le choix de l’itinéraire à un usager.

Pour une version serveur, il serait envisageable d’utiliser une base de données MySQL à la place de SQLite, afin d’avoir de meilleures performances, et déléguer les tâches concernant la base de données à un autre processus, voir délocaliser la base de données auprès d’un autre poste informatique ou auprès d’un prestataire d’hébergement.

Expérience personnelle

En ce qui me concerne, je suis très satisfait de ce que j’ai réalisé. Créer un algorithme de cette envergure m’a permis d’enrichir mes connaissances en matière de logique, et la réalisation d’un projet de bout en bout est toujours un défi en soit.