Phase 1 : pipeline d'ingestion Sentinel-2
- SentinelHubClient : auth OAuth2, Catalog API (cloud coverage), Process API avec evalscript AFAI binaire (UINT8, cloud-side) - IngestionService : orchestration complète — job tracking, rejet nuages, téléchargement GeoTIFF, polygonize GDAL, simplification PostGIS, persistance SargassumObservation - IngestSentinelCommand : 5 zones Antilles configurées, options --date et --zone, appelable via cron docker exec - Dockerfile : ajout gdal-bin + python3-gdal - Correction entity : paramètre immutable retiré de #[ORM\Column] Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Gwadaking
189
backend/src/Service/Ingestion/IngestionService.php
Normal file
189
backend/src/Service/Ingestion/IngestionService.php
Normal file
@@ -0,0 +1,189 @@
|
||||
<?php
|
||||
|
||||
namespace App\Service\Ingestion;
|
||||
|
||||
use App\Entity\DataIngestionJob;
|
||||
use App\Entity\SargassumObservation;
|
||||
use App\Service\SentinelHub\SentinelHubClient;
|
||||
use Doctrine\DBAL\Connection;
|
||||
use Doctrine\ORM\EntityManagerInterface;
|
||||
use Psr\Log\LoggerInterface;
|
||||
use Symfony\Component\DependencyInjection\Attribute\Autowire;
|
||||
use Symfony\Component\Process\Exception\ProcessFailedException;
|
||||
use Symfony\Component\Process\Process;
|
||||
|
||||
class IngestionService
|
||||
{
|
||||
private const CLOUD_THRESHOLD = 60.0;
|
||||
private const SARGASSUM_DN = 1;
|
||||
private const SIMPLIFY_TOLERANCE = 0.001; // degrés (~100m à l'équateur)
|
||||
private const PROCESSING_VERSION = '1.0.0';
|
||||
|
||||
public function __construct(
|
||||
private SentinelHubClient $sentinelHub,
|
||||
private EntityManagerInterface $em,
|
||||
private LoggerInterface $logger,
|
||||
#[Autowire('%kernel.project_dir%')] private string $projectDir,
|
||||
) {}
|
||||
|
||||
public function ingest(array $bbox, \DateTimeImmutable $date, string $source = 'Sentinel-2'): void
|
||||
{
|
||||
$job = new DataIngestionJob();
|
||||
$job->setSource($source);
|
||||
$job->setTileId(sprintf('[%s]', implode(',', $bbox)));
|
||||
$this->em->persist($job);
|
||||
$this->em->flush();
|
||||
|
||||
$tmpDir = null;
|
||||
|
||||
try {
|
||||
// Étape 1 : vérification couverture nuageuse via Catalog API
|
||||
$cloudCoverage = $this->sentinelHub->getCloudCoverage($bbox, $date);
|
||||
$this->logger->info('Cloud coverage check', ['bbox' => $bbox, 'coverage' => $cloudCoverage]);
|
||||
|
||||
if ($cloudCoverage > self::CLOUD_THRESHOLD) {
|
||||
$job->setStatus('failed');
|
||||
$job->setErrorMessage("Couverture nuageuse {$cloudCoverage}% > seuil " . self::CLOUD_THRESHOLD . '%');
|
||||
$this->em->flush();
|
||||
return;
|
||||
}
|
||||
|
||||
// Étape 2 : téléchargement du raster AFAI binaire
|
||||
$tmpDir = $this->projectDir . '/var/ingestion/' . uniqid('ingest_', true);
|
||||
mkdir($tmpDir, 0755, true);
|
||||
$tifPath = $tmpDir . '/afai.tif';
|
||||
$geojsonPath = $tmpDir . '/polygons.geojson';
|
||||
|
||||
$metadata = $this->sentinelHub->downloadAfaiBinary($bbox, $date, $tifPath);
|
||||
$this->logger->info('GeoTIFF downloaded', ['path' => $tifPath, 'tile' => $metadata['tileId']]);
|
||||
|
||||
// Étape 3 : vectorisation via GDAL
|
||||
$this->polygonize($tifPath, $geojsonPath);
|
||||
|
||||
// Étape 4 : construction du MULTIPOLYGON depuis le GeoJSON
|
||||
$geometry = $this->buildMultiPolygon($geojsonPath);
|
||||
|
||||
if ($geometry === null) {
|
||||
$job->setStatus('success');
|
||||
$job->setProcessedAt(new \DateTimeImmutable());
|
||||
$job->setErrorMessage('Aucune sargasse détectée dans cette scène');
|
||||
$this->em->flush();
|
||||
$this->logger->info('No sargassum detected', ['bbox' => $bbox]);
|
||||
return;
|
||||
}
|
||||
|
||||
// Étape 5 : persistance
|
||||
$observation = new SargassumObservation();
|
||||
$observation->setGeometry($geometry);
|
||||
$observation->setDetectedAt($date);
|
||||
$observation->setSource($source);
|
||||
$observation->setTileId($metadata['tileId']);
|
||||
$observation->setCloudCoverage($cloudCoverage);
|
||||
$observation->setConfidence(0.8);
|
||||
$observation->setAfaiMean($metadata['afaiMean']);
|
||||
$observation->setAfaiStd($metadata['afaiStd']);
|
||||
$observation->setProcessingVersion(self::PROCESSING_VERSION);
|
||||
|
||||
$this->em->persist($observation);
|
||||
$this->em->flush();
|
||||
|
||||
// Étape 6 : simplification + bbox + aire via PostGIS
|
||||
$this->updateSpatialMeta((string) $observation->getId());
|
||||
|
||||
$job->setStatus('success');
|
||||
$job->setProcessedAt(new \DateTimeImmutable());
|
||||
$this->em->flush();
|
||||
|
||||
$this->logger->info('Ingestion OK', ['observation_id' => (string) $observation->getId()]);
|
||||
|
||||
} catch (\Throwable $e) {
|
||||
$job->setStatus('failed');
|
||||
$job->incrementRetry();
|
||||
$job->setErrorMessage($e->getMessage());
|
||||
$this->em->flush();
|
||||
$this->logger->error('Ingestion failed', ['error' => $e->getMessage()]);
|
||||
throw $e;
|
||||
|
||||
} finally {
|
||||
if ($tmpDir !== null && is_dir($tmpDir)) {
|
||||
$this->cleanup($tmpDir);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private function polygonize(string $tifPath, string $geojsonPath): void
|
||||
{
|
||||
$process = new Process([
|
||||
'gdal_polygonize.py',
|
||||
$tifPath,
|
||||
'-f', 'GeoJSON',
|
||||
'-q',
|
||||
$geojsonPath,
|
||||
]);
|
||||
$process->setTimeout(120);
|
||||
$process->run();
|
||||
|
||||
if (!$process->isSuccessful()) {
|
||||
throw new ProcessFailedException($process);
|
||||
}
|
||||
}
|
||||
|
||||
private function buildMultiPolygon(string $geojsonPath): ?array
|
||||
{
|
||||
if (!file_exists($geojsonPath)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
$data = json_decode(file_get_contents($geojsonPath), true);
|
||||
|
||||
if (empty($data['features'])) {
|
||||
return null;
|
||||
}
|
||||
|
||||
$polygons = [];
|
||||
foreach ($data['features'] as $feature) {
|
||||
if (($feature['properties']['DN'] ?? 0) !== self::SARGASSUM_DN) {
|
||||
continue;
|
||||
}
|
||||
|
||||
$geom = $feature['geometry'];
|
||||
|
||||
if ($geom['type'] === 'Polygon') {
|
||||
$polygons[] = $geom['coordinates'];
|
||||
} elseif ($geom['type'] === 'MultiPolygon') {
|
||||
foreach ($geom['coordinates'] as $poly) {
|
||||
$polygons[] = $poly;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (empty($polygons)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return ['type' => 'MultiPolygon', 'coordinates' => $polygons];
|
||||
}
|
||||
|
||||
private function updateSpatialMeta(string $id): void
|
||||
{
|
||||
$this->em->getConnection()->executeStatement(
|
||||
'UPDATE sargassum_observation
|
||||
SET
|
||||
geometry = ST_SimplifyPreserveTopology(geometry, :tolerance),
|
||||
bbox = ST_Envelope(geometry),
|
||||
coverage_area = ROUND(CAST(ST_Area(geometry::geography) / 1000000 AS numeric), 2)
|
||||
WHERE id = :id',
|
||||
['id' => $id, 'tolerance' => self::SIMPLIFY_TOLERANCE]
|
||||
);
|
||||
}
|
||||
|
||||
private function cleanup(string $dir): void
|
||||
{
|
||||
foreach (glob($dir . '/*') ?: [] as $file) {
|
||||
if (is_file($file)) {
|
||||
unlink($file);
|
||||
}
|
||||
}
|
||||
rmdir($dir);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user