feat: ingestion automatique toutes les 6h + indicateur fraîcheur UI

- Symfony Scheduler (MainSchedule) : cron 0,6,12,18h UTC
- IngestionScheduleMessage + IngestionScheduleHandler : pipeline complet
  ingestion → forecasts → impact scores pour toutes les zones Antilles
- messenger.yaml : transport scheduler_main + async Redis (rs971_async)
- entrypoint.sh : 2 workers en background (scheduler_main + async)
- GET /api/status : date/heure de la dernière observation
- SargassesMap : badge "Données du JJ/MM à HH:MM" en bas à gauche

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Gwadaking
2026-04-03 22:31:58 -04:00
parent 55caca001c
commit 48e6e5060c
9 changed files with 223 additions and 0 deletions

View File

@@ -0,0 +1,20 @@
framework:
messenger:
transports:
# Transport du scheduler — traité en synchrone dans le worker scheduler_main
scheduler_main:
dsn: 'scheduler://main'
# Transport async Redis pour les messages lourds (ingestion)
async:
dsn: '%env(REDIS_URL)%'
options:
stream: rs971_async
retry_strategy:
max_retries: 3
delay: 60000 # 1 min entre chaque retry
multiplier: 2
routing:
# L'ingestion planifiée tourne dans le worker async (pas dans le worker scheduler)
App\Message\IngestionScheduleMessage: async

View File

@@ -0,0 +1,34 @@
<?php
namespace App\Controller;
use Doctrine\DBAL\Connection;
use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\Routing\Attribute\Route;
/**
* GET /api/status — fraîcheur des données pour l'indicateur UI
*/
#[Route('/api/status', name: 'api_status', methods: ['GET'])]
class StatusController extends AbstractController
{
public function __construct(private Connection $connection) {}
public function __invoke(): JsonResponse
{
$row = $this->connection->fetchAssociative(
'SELECT detected_at, source, coverage_area
FROM sargassum_observation
ORDER BY detected_at DESC LIMIT 1'
);
return $this->json([
'lastObservation' => $row !== false ? [
'detectedAt' => $row['detected_at'],
'source' => $row['source'],
'coverageKm2' => $row['coverage_area'] !== null ? (float) $row['coverage_area'] : null,
] : null,
]);
}
}

View File

@@ -0,0 +1,9 @@
<?php
namespace App\Message;
/**
* Message déclenché automatiquement par le Scheduler toutes les 6h.
* Lance le pipeline complet : ingestion Sentinel → forecasts → scores.
*/
final class IngestionScheduleMessage {}

View File

@@ -0,0 +1,80 @@
<?php
namespace App\MessageHandler;
use App\Entity\SargassumObservation;
use App\Message\IngestionScheduleMessage;
use App\Service\Forecast\DriftSimulationService;
use App\Service\Forecast\ImpactScoreService;
use App\Service\Ingestion\IngestionService;
use Doctrine\ORM\EntityManagerInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
/**
* Exécute le pipeline complet à chaque tick du scheduler :
* 1. Ingestion Sentinel-2/3 pour chaque zone Antilles
* 2. Calcul des forecasts de dérive pour les nouvelles observations
* 3. Calcul des ImpactScores pour tous les spots côtiers
*/
#[AsMessageHandler]
final class IngestionScheduleHandler
{
// Zones d'intérêt — même liste que IngestSentinelCommand
private const ZONES = [
'martinique' => [-61.30, 14.30, -60.70, 14.90],
'guadeloupe' => [-61.90, 15.80, -61.00, 16.60],
'sainte-lucie' => [-61.10, 13.70, -60.80, 14.20],
'barbade' => [-59.80, 13.00, -59.30, 13.40],
'saint-martin' => [-63.20, 17.80, -62.90, 18.20],
];
public function __construct(
private IngestionService $ingestion,
private DriftSimulationService $drift,
private ImpactScoreService $scores,
private EntityManagerInterface $em,
private LoggerInterface $logger,
) {}
public function __invoke(IngestionScheduleMessage $message): void
{
$date = new \DateTimeImmutable('yesterday');
$this->logger->info('Scheduled ingestion started', ['date' => $date->format('Y-m-d')]);
// Étape 1 : ingestion de toutes les zones
foreach (self::ZONES as $name => $bbox) {
try {
$this->ingestion->ingestWithFallback($bbox, $date);
$this->logger->info('Zone ingested', ['zone' => $name]);
} catch (\Throwable $e) {
$this->logger->error('Ingestion failed', ['zone' => $name, 'error' => $e->getMessage()]);
}
}
// Étape 2 : forecasts pour les observations sans prévision
$observations = $this->em->getRepository(SargassumObservation::class)
->createQueryBuilder('o')
->leftJoin('o.forecasts', 'f')
->where('f.id IS NULL')
->orderBy('o.detectedAt', 'DESC')
->setMaxResults(10)
->getQuery()
->getResult();
foreach ($observations as $observation) {
try {
$forecasts = $this->drift->computeForecasts($observation);
$this->scores->computeForObservation($observation, $forecasts);
$this->logger->info('Forecasts computed', ['observation' => (string) $observation->getId()]);
} catch (\Throwable $e) {
$this->logger->error('Forecast failed', [
'observation' => (string) $observation->getId(),
'error' => $e->getMessage(),
]);
}
}
$this->logger->info('Scheduled ingestion completed');
}
}

View File

@@ -0,0 +1,32 @@
<?php
namespace App\Scheduler;
use App\Message\IngestionScheduleMessage;
use Symfony\Component\Scheduler\Attribute\AsSchedule;
use Symfony\Component\Scheduler\RecurringMessage;
use Symfony\Component\Scheduler\Schedule;
use Symfony\Component\Scheduler\ScheduleProviderInterface;
use Symfony\Contracts\Cache\CacheInterface;
/**
* Planifie l'ingestion satellite toutes les 6h.
* Le Schedule est mis en cache pour éviter les recréations inutiles.
*/
#[AsSchedule('main')]
final class MainSchedule implements ScheduleProviderInterface
{
private ?Schedule $schedule = null;
public function __construct(private CacheInterface $cache) {}
public function getSchedule(): Schedule
{
return $this->schedule ??= (new Schedule())
->add(
// 06:00, 12:00, 18:00, 00:00 UTC — données satellite disponibles ~2h après le passage
RecurringMessage::cron('0 0,6,12,18 * * *', new IngestionScheduleMessage())
)
->stateful($this->cache);
}
}