FIC 2026 - Résolution Scénario 2 - Exercice 4 : Livraison Express

Vue d'ensemble

Dans cet exercice, nous devons retracer le trajet exact d’un camion autonome transportant un prototype volé du moteur Pulsar, dont la localisation GPS a été coupée volontairement par un attaquant. Pour ce faire, nous utilisons uniquement des données Inertial Measurement Unit (IMU) récupérées indépendamment du GPS. Celles-ci comprennent des mesures d’accélération, de vitesse angulaire (gyroscope), d’orientation (compas), et d’odométrie. L’objectif est d’estimer la position du véhicule au fil du temps et de déterminer la destination finale, qui correspond à la ville où le camion s’est rendu.

La partie 1 concerne cette reconstruction de trajectoire à partir des données IMU. La partie 2, est axée sur l’analyse des traces AIS de navires pour la suite de l’enquête.

Structure des fichiers fournis

├── imu_telemetry.dat        # Données binaires IMU brutes
├── AIS_2025_11_05.csv       # Logs AIS complets du jour de l'incident
├── vessels_list.zip         # Liste des navires avec MMSI, noms, etc.

Étape 1 : Identifier la structure binaire et comprendre le protocole

La première étape pour le participant est de comprendre la structure du fichier binaire imu_telemetry.dat afin d’en extraire les données IMU exploitables.

Méthodologie possible

• Exploration hexadécimale : Ouvrir le fichier dans un éditeur hexadécimal pour détecter le motif répétitif de taille fixe, ce qui indique la taille d’un enregistrement (38 octets ici).

• Calculs empiriques : Considérer la durée de l’échantillonnage, la fréquence (~10 Hz) et la taille du fichier pour estimer le nombre d’échantillons. Cela permet d’inférer la taille unitaire d’un enregistrement.

• Hypothèses sur la structure :
  Le fichier contient à chaque itération :

• Timestamp 64-bit (uint64, 8 octets)
• Accéléromètre 3 axes (float32 x3, 12 octets)
• Gyroscope 3 axes (float32 x3, 12 octets)
• Cap ou heading (uint16, 2 octets)

• Odomètre (uint32, 4 octets)
  Total 8 + 12 + 12 + 2 + 4 = 38 octets.

• Reverse engineering par script Python :
  Tester l’extraction des données avec la fonction struct.unpack :
  python import struct with open('imu_telemetry.dat', 'rb') as f: while True: record = f.read(38) if len(record) < 38: break ts, ax, ay, az, gx, gy, gz, heading, odo = struct.unpack('<Qfff fff H I', record) print(f"Timestamp: {ts}, Accel: ({ax:.2f},{ay:.2f},{az:.2f}), Gyro: ({gx:.2f},{gy:.2f},{gz:.2f}), Heading: {heading}, Odometer: {odo}")

• Validation :
  Les timestamps doivent être croissants.
  L’odomètre augmente régulièrement.
  Les accélérations et gyro sont dans des plages réalistes.

La compréhension de ces étapes est cruciale pour parser les logs bruts, première étape indispensable avant la reconstruction.

Étape 2 : Reconstruire la trajectoire à partir des données IMU

Principe technique

• La reconstruction s’appuie sur une méthode de “dead reckoning” : intégration temporelle des accélérations et changements d’orientation pour déduire le déplacement.
• L’odomètre apporte des données cumulées de distance, utilisées pour calculer la vitesse instantanée et valider les déplacements.
• Le gyroscope et compas fournissent la direction du mouvement, avec filtrage par moyenne mobile pour lisser les variations.
• La position initiale connue Austin, Texas (30.226893°N, −97.618270°E), donnée dans le fichier statement.md, sert de point de départ.
• À chaque pas temporel, la nouvelle position est calculée à partir de la dernière, en avançant selon la vitesse et la direction estimées.

Script de résolution

La logique implémentée dans le script solver.py suit ce modèle :

#!/usr/bin/env python3
import argparse
import json
import math
import struct
import sys
from datetime import datetime
from collections import deque

import numpy as np


class IMUTrajectoryReconstructor:
    METERS_PER_LAT_DEGREE = 111000

    def __init__(self, start_lat, start_lon):
        self.start_lat = start_lat
        self.start_lon = start_lon
        self.current_lat = start_lat
        self.current_lon = start_lon
        self.current_heading = 0.0

        self.trajectory = []
        self.start_time_ms = None

        self.last_odometer = 0
        self.last_timestamp_ms = None

    def add_position(self, lat, lon, timestamp_ms, source='DEAD_RECKONING', speed_kmh=0.0):
        self.trajectory.append({
            'timestamp_ms': timestamp_ms,
            'latitude': lat,
            'longitude': lon,
            'source': source,
            'speed_kmh': speed_kmh
        })

    def lat_lon_from_distance_heading(self, lat, lon, distance_m, heading_deg):
        heading_rad = math.radians(heading_deg)
        delta_lat = (distance_m * math.cos(heading_rad)) / self.METERS_PER_LAT_DEGREE

        lat_rad = math.radians(lat)
        meters_per_lon = self.METERS_PER_LAT_DEGREE * math.cos(lat_rad)
        delta_lon = (distance_m * math.sin(heading_rad)) / meters_per_lon if meters_per_lon > 0 else 0

        return lat + delta_lat, lon + delta_lon

    def parse_binary_imu(self, filepath):
        print(f"Reading IMU data from {filepath}...")

        imu_data = []
        RECORD_SIZE = 38

        try:
            with open(filepath, 'rb') as f:
                data = f.read()
                num_records = len(data) // RECORD_SIZE

                for i in range(num_records):
                    offset = i * RECORD_SIZE
                    record = data[offset:offset + RECORD_SIZE]

                    if len(record) < RECORD_SIZE:
                        break

                    timestamp_ms, = struct.unpack('<Q', record[0:8])
                    accel_x, accel_y, accel_z = struct.unpack('<fff', record[8:20])
                    gyro_x, gyro_y, gyro_z = struct.unpack('<fff', record[20:32])
                    heading_raw, = struct.unpack('<H', record[32:34])
                    odometer, = struct.unpack('<I', record[34:38])

                    heading_deg = (heading_raw / 10.0) % 360.0

                    imu_data.append({
                        'timestamp_ms': timestamp_ms,
                        'accel_x': accel_x,
                        'accel_y': accel_y,
                        'accel_z': accel_z,
                        'gyro_x': gyro_x,
                        'gyro_y': gyro_y,
                        'gyro_z': gyro_z,
                        'heading': heading_deg,
                        'odometer': odometer
                    })

                print(f"Loaded {len(imu_data)} data points.\n")
                return imu_data

        except FileNotFoundError:
            print(f"File not found: {filepath}", file=sys.stderr)
            return []
        except Exception as e:
            print(f"Error reading file: {e}", file=sys.stderr)
            return []

    def reconstruct_trajectory(self, imu_data):
        print("Reconstructing trajectory...\n")

        if not imu_data:
            print("No IMU data available", file=sys.stderr)
            return

        self.start_time_ms = imu_data[0]['timestamp_ms']
        self.last_timestamp_ms = self.start_time_ms
        self.last_odometer = imu_data[0]['odometer']

        heading_buffer = deque(maxlen=5)

        for i, point in enumerate(imu_data):
            compass_heading = point['heading']
            heading_buffer.append(compass_heading)
            smoothed_heading = np.median(list(heading_buffer))

            gyro_contribution = math.degrees(point['gyro_z']) * 0.1 if i > 0 else 0
            self.current_heading = (smoothed_heading + gyro_contribution) % 360.0

            if i > 0:
                delta_odom = point['odometer'] - self.last_odometer
                delta_time_s = (point['timestamp_ms'] - self.last_timestamp_ms) / 1000.0
                speed_kmh = (delta_odom / delta_time_s) * 3.6 if delta_time_s > 0 else 0.0
            else:
                speed_kmh = 0.0

            if i > 0:
                distance_m = point['odometer'] - self.last_odometer
                if distance_m > 0.05:
                    self.current_lat, self.current_lon = self.lat_lon_from_distance_heading(
                        self.current_lat, self.current_lon, distance_m, self.current_heading
                    )

            self.add_position(self.current_lat, self.current_lon, point['timestamp_ms'],
                              'DEAD_RECKONING+COMPASS', speed_kmh)

            self.last_odometer = point['odometer']
            self.last_timestamp_ms = point['timestamp_ms']

            if (i + 1) % 5000 == 0:
                pct = (i + 1) / len(imu_data) * 100
                print(f"  {pct:.0f}% processed: {self.current_lat:.6f}°N, {self.current_lon:.6f}°E, speed {speed_kmh:.1f} km/h")

        print(f"\nTrajectory reconstructed with {len(self.trajectory)} points.\n")

    def save_trajectory_csv(self, filename='trajectory_reconstruction.csv'):
        with open(filename, 'w') as f:
            f.write("timestamp_ms,timestamp_iso,latitude,longitude,source,speed_kmh\n")
            for point in self.trajectory:
                ts_iso = datetime.fromtimestamp(point['timestamp_ms'] / 1000).isoformat() + 'Z'
                f.write(f"{point['timestamp_ms']},{ts_iso},"
                       f"{point['latitude']:.8f},{point['longitude']:.8f},"
                       f"{point['source']},{point['speed_kmh']:.2f}\n")
        print(f"CSV saved: {filename}")

    def visualize_trajectory(self, output_html='trajectory_map.html', reference_geojson=None):
        print(f"Generating interactive map...\n")

        lats = [p['latitude'] for p in self.trajectory]
        lons = [p['longitude'] for p in self.trajectory]
        min_lat, max_lat = min(lats), max(lats)
        min_lon, max_lon = min(lons), max(lons)
        center_lat = (min_lat + max_lat) / 2
        center_lon = (min_lon + max_lon) / 2

        trajectory_coords = [[p['latitude'], p['longitude']] for p in self.trajectory]

        traj_data_with_speed = [
            {'lat': p['latitude'], 'lon': p['longitude'], 'speed': p['speed_kmh']}
            for p in self.trajectory
        ]
        traj_data_json = json.dumps(traj_data_with_speed)

        html = f"""<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8"/>
    <title>Trajectory Reconstruction</title>
    <link rel="stylesheet" href="https://unpkg.com/leaflet@1.9.4/dist/leaflet.css" />
    <script src="https://unpkg.com/leaflet@1.9.4/dist/leaflet.js"></script>
    <style>
        body {{ margin:0; padding:0; font-family: Arial, sans-serif; }}
        #map {{ position:absolute; top:0; bottom:0; width:100%; }}
        .info {{ position: absolute; top: 10px; left: 10px; z-index: 1000;
                 background: white; padding: 15px; border-radius: 5px;
                 box-shadow: 0 0 15px rgba(0,0,0,0.2); max-width: 300px; }}
        .info h3 {{ margin: 0 0 10px; }}
        .stat {{ margin: 5px 0; }}
        .label {{ font-weight: bold; }}
        .legend {{ position: absolute; bottom: 30px; right: 10px; z-index: 1000;
                   background: white; padding: 10px; border-radius: 5px;
                   box-shadow: 0 0 15px rgba(0,0,0,0.2); }}
        .legend h4 {{ margin: 0 0 10px; }}
        .legend-item {{ margin: 5px 0; display: flex; align-items: center; }}
        .color-box {{ width: 30px; height: 15px; margin-right: 8px; border: 1px solid #ccc; }}
    </style>
</head>
<body>
    <div id="map"></div>
    <div class="info">
        <h3>Reconstruction</h3>
        <div class="stat"><span class="label">Points:</span> {len(self.trajectory)}</div>
        <div class="stat"><span class="label">Average speed:</span> {np.mean([p['speed_kmh'] for p in self.trajectory]):.1f} km/h</div>
        <div class="stat"><span class="label">Max speed:</span> {max([p['speed_kmh'] for p in self.trajectory]):.1f} km/h</div>
        <div class="stat"><span class="label">Duration:</span> {(self.trajectory[-1]['timestamp_ms'] - self.trajectory[0]['timestamp_ms']) / 1000 / 60:.1f} min</div>
    </div>

    <script>
        const map = L.map('map').setView([{center_lat}, {center_lon}], 12);
        L.tileLayer('https://{{s}}.tile.openstreetmap.org/{{z}}/{{x}}/{{y}}.png').addTo(map);

        const trajectoryData = {traj_data_json};

        function getSpeedColor(speed) {{
            if (speed < 20) return '#00C853';
            if (speed < 50) return '#FFC107';
            if (speed < 80) return '#FF9800';
            return '#F44336';
        }}

        for (let i = 0; i < trajectoryData.length - 1; i++) {{
            const p1 = trajectoryData[i];
            const p2 = trajectoryData[i + 1];
            const speed = p1.speed;

            L.polyline([[p1.lat, p1.lon], [p2.lat, p2.lon]], {{
                color: getSpeedColor(speed),
                weight: 3,
                opacity: 0.8
            }}).addTo(map);
        }}

        L.marker([{self.trajectory[0]['latitude']}, {self.trajectory[0]['longitude']}])
            .addTo(map).bindPopup('Start');
        L.marker([{self.trajectory[-1]['latitude']}, {self.trajectory[-1]['longitude']}])
            .addTo(map).bindPopup('End');

        const legend = L.control({{position: 'bottomright'}});
        legend.onAdd = function() {{
            const div = L.DomUtil.create('div', 'legend');
            div.innerHTML = `
                <h4>Speed (km/h)</h4>
                <div class="legend-item"><div class="color-box" style="background:#00C853"></div> 0-20</div>
                <div class="legend-item"><div class="color-box" style="background:#FFC107"></div> 20-50</div>
                <div class="legend-item"><div class="color-box" style="background:#FF9800"></div> 50-80</div>
                <div class="legend-item"><div class="color-box" style="background:#F44336"></div> 80+</div>
            `;
            return div;
        }};
        legend.addTo(map);
    </script>
</body>
</html>"""

        with open(output_html, 'w', encoding='utf-8') as f:
            f.write(html)
        print(f"Map generated: {output_html}\n")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="IMU Trajectory Solver")
    parser.add_argument("--start-lat", type=float, default=48.8566)
    parser.add_argument("--start-lon", type=float, default=2.3522)
    parser.add_argument("--imu-file", type=str, default="imu_telemetry.dat")
    parser.add_argument("--output-csv", type=str, default="trajectory_reconstruction.csv")
    parser.add_argument("--output-html", type=str, default="trajectory_map.html")

    args = parser.parse_args()

    reconstructor = IMUTrajectoryReconstructor(start_lat=args.start_lat, start_lon=args.start_lon)
    imu_data = reconstructor.parse_binary_imu(args.imu_file)

    if imu_data:
        reconstructor.add_position(args.start_lat, args.start_lon, imu_data[0]['timestamp_ms'], 'GPS', 0.0)
        reconstructor.reconstruct_trajectory(imu_data)
        reconstructor.save_trajectory_csv(args.output_csv)
        reconstructor.visualize_trajectory(args.output_html)
    else:
        print("Failed to load IMU data.", file=sys.stderr)
        sys.exit(1)

• Le script lit le fichier binaire imu_telemetry.dat en respectant la structure (timestamp 8B, accel 3x float, gyro 3x float, heading uint16, odomètre uint32).
• Il calcule la vitesse à partir de la différence d’odomètre et la variation de temps.
• Par intégration avec la direction issue du compas et gyroscope, il mise à jour la position GPS estimée.
• Le résultat est exporté en CSV et en carte Leaflet interactive, où la vitesse est visualisée par un dégradé de couleur sur la trajectoire, facilitant la compréhension des phases d’accélération ou arrêt.

Le meilleur moyen de travailler sur cet exercice est de visualiser les résultats sur une carte interactive pour itérer sur les paramètres de calcul.

Sur ces captures d'écran, on voit la trajectoire reconstruite (en vert/jaune/rouge selon la vitesse) superposée à la carte OpenStreetMap avec la trajectoire réelle (en bleu) pour comparaison. Même si des écarts subsistent dus aux imprécisions des capteurs IMU, la ville de destination finale peut clairement être identifiée si on se fie aux routes importantes proches de la trajectoire (ou qu'on a le temps d'implémenter des filtres plus avancés comme un Kalman, etc.).

On identifie le premier flag.

Étape 3 : Analyse des traces AIS et détection du navire suspect

Le camion arrive au port de cette ville identifié précédemment Il faut maintenant identifier quel navire a récupéré la cargaison volée parmi les traces AIS du jour.

Fichiers fournis

AIS_2025_11_05.csv          # Logs AIS complets du jour de l'incident

Le fichier contient des messages AIS pour l'ensemble des navires dans les côtes américaines le 5 novembre 2025 (ce ne sont pas des données de ce jour-là, pour éviter que les participants puissent trouver facilement les modifications que nous avons apportées aux données réelles).

Résolution

Avant même de s'intéresser aux données AIS, le participant doit comprendre ce qu'elles sont, car il est probable qu'il n'ait jamais travaillé avec ce type de données. Une fois cette compréhension acquise, l'objectif est de détecter un navire suspect qui aurait pu usurper l'identité d'un navire local pour récupérer la cargaison volée.

Voici les axes d’investigation recommandés :

• Chercher un MMSI présentant un blackout AIS anormal : un intervalle temporel important sans messages, par exemple un gap de 60 à 90 minutes ou plus, au cours de la journée.

• Observer que ce même MMSI réapparaît soudainement très loin géographiquement, à plusieurs centaines de kilomètres, ce qui ne correspond pas aux habitudes ou à la vitesse plausible d’un cargo.

• Identifier que ce navire suspect est lié à un mélange de trajectoires issues de deux navires réels différents, avec l’identité (MMSI) d’un navire opérant autour de la Nouvelle-Orléans et le trajet spatial d’un navire local.

• Le joueur doit donc combiner analyse temporelle des gaps, analyse spatiale des positions après gap, et connaissance des vitesses plausibles d’un navire (typiquement 15 à 25 nœuds, hors cas exceptionnel).

Ici on peut voir dans les logs AIS le fameux comportement suspect :

• Le suspect final présente une trajectoire nettement incohérente, avec un gap temporaire suivi d’une réapparition impossiblement lointaine pour un même MMSI. (Comme on peut le voir sur cette visualisation ci-dessous).

Cette détection demande un travail d’investigation croisé sur les données temporelles et géographiques.