diff --git a/.env.example b/.env.example index bafc097..914d61a 100644 --- a/.env.example +++ b/.env.example @@ -27,3 +27,8 @@ LOG_LEVEL=INFO # Genere au deploiement, comme DB_PASSWORD : # printf 'STREAM_PUBLISH_PASSWORD=%s\n' "$(openssl rand -hex 16)" >> .env STREAM_PUBLISH_PASSWORD=CHANGEME + +# Mot de passe de l'API MediaMTX, utilise par la sonde de presence video. +# Genere au deploiement : +# printf 'MEDIA_API_PASSWORD=%s\n' "$(openssl rand -hex 16)" >> .env +MEDIA_API_PASSWORD=CHANGEME diff --git a/README.md b/README.md index b4eed5a..82d89b7 100644 --- a/README.md +++ b/README.md @@ -234,15 +234,45 @@ d'entreprise. | **Lecture** | publique — le flux est une vitrine, c'est assumé | | **Publication** | authentifiée — sans ça, n'importe qui sur Internet pousserait sa propre vidéo sur le flux `biblio` | +### La caméra muette est un événement comme un autre + +Le domaine connaissait déjà `CAMERA_OFFLINE`, classé `CRITICAL` — un capteur +qui se tait alors que l'alarme est armée, c'est soit une batterie morte, soit +quelqu'un qui l'a trouvé derrière les livres. Rien ne l'émettait jamais. +`CameraLivenessProbe` comble ce trou : elle interroge l'API du serveur média et +traduit une absence prolongée en événement capteur. + +Elle passe par le **même port** `IngestSensorEventUseCase` que l'endpoint REST +et le simulateur. Trois producteurs d'événements de natures totalement +différentes — un téléphone, un simulateur, une sonde de supervision — un seul +contrat d'entrée, et le même trajet Kafka → agrégat ensuite. Si l'alarme est +désarmée, une coupure vidéo ne réveille personne : c'est l'agrégat qui tranche, +pas la sonde. + +**Le seuil est la partie qui compte.** Sur un lien 4G, le flux tombe +régulièrement sans que personne n'ait touché au téléphone : tunnel, bascule +d'antenne, congestion. Alerter dès la première sonde ratée produirait plusieurs +`CRITICAL` par jour — et la seule chose qu'on apprend à vivre avec des alertes +qui crient pour rien, c'est à les ignorer, y compris le jour où elles ont +raison. Il faut donc trois sondes consécutives, soit une minute d'absence +réelle. + ### Brancher un vrai téléphone -Application : **IP Camera Lite** (iOS), menu *Live Push*. +**Larix Broadcaster** (iOS/Android, gratuite, sans filigrane) : elle gère SRT, +la reconnexion automatique et le débit adaptatif. C'est ce qu'il faut sur un +lien mobile. IP Camera Lite sait aussi pousser, mais elle est conçue pour le +mode *pull* — lequel ne peut pas fonctionner depuis l'extérieur. ``` -SRT srt://:8890?streamid=publish:biblio:iphone: -RTMP rtmp://:1935/biblio?user=iphone&pass= +SRT srt://:8890 · streamid = publish:biblio:iphone: +RTMP rtmp://:1935/biblio?user=iphone&pass= ``` +Réglages pour la 4G : **H.264, 640×360, 15 fps, 400–600 kbps, latence SRT +2000 ms, audio coupé**. À 500 kbps, compter **~225 Mo par heure** — un flux +permanent dépasse 5 Go par jour, ce qui n'a de sens que sur le WiFi du studio. + Le mot de passe est généré au déploiement dans le `.env` du serveur (`STREAM_PUBLISH_PASSWORD`) et n'existe nulle part dans ce dépôt. diff --git a/deploy/deploy.sh b/deploy/deploy.sh index 942bdf6..441ffa9 100755 --- a/deploy/deploy.sh +++ b/deploy/deploy.sh @@ -29,8 +29,10 @@ echo "==> Rendu de la configuration MediaMTX" # part dans le depot. Le fichier rendu est gitignore. ssh "${HOST}" "cd ${REMOTE_DIR} && \ PASS=\$(grep '^STREAM_PUBLISH_PASSWORD=' .env | cut -d= -f2-) && \ + APIPASS=\$(grep '^MEDIA_API_PASSWORD=' .env | cut -d= -f2-) && \ PUBLIC_IP=\$(curl -s --max-time 10 https://api.ipify.org) && \ sed -e \"s|__PUBLISH_PASSWORD__|\${PASS}|\" \ + -e \"s|__API_PASSWORD__|\${APIPASS}|\" \ -e \"s|__PUBLIC_IP__|\${PUBLIC_IP}|\" \ deploy/mediamtx/mediamtx.yml.template > deploy/mediamtx/mediamtx.yml && \ chmod 600 deploy/mediamtx/mediamtx.yml" diff --git a/deploy/mediamtx/mediamtx.yml.template b/deploy/mediamtx/mediamtx.yml.template index 5b748c5..f4edbdc 100644 --- a/deploy/mediamtx/mediamtx.yml.template +++ b/deploy/mediamtx/mediamtx.yml.template @@ -43,6 +43,15 @@ authInternalUsers: - action: read - action: playback + # Lecture seule de l'API, pour la sonde de presence de l'application. + # Restreinte au reseau Docker interne : l'API n'est de toute facon pas + # publiee sur l'hote, mais deux verrous valent mieux qu'un. + - user: probe + pass: __API_PASSWORD__ + ips: ['172.16.0.0/12'] + permissions: + - action: api + - user: iphone pass: __PUBLISH_PASSWORD__ ips: [] diff --git a/docker-compose.yml b/docker-compose.yml index c156c76..207020d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -89,6 +89,10 @@ services: condition: service_healthy postgres: condition: service_healthy + # Demarrage seulement : la sonde tolere un serveur media absent, elle + # ne doit pas empecher l'alarme de fonctionner. + mediamtx: + condition: service_started environment: SPRING_PROFILES_ACTIVE: prod KAFKA_BOOTSTRAP_SERVERS: kafka:9092 @@ -99,6 +103,9 @@ services: SIMULATOR_INTERVAL: ${SIMULATOR_INTERVAL:-8s} ALARM_EXIT_DELAY: ${ALARM_EXIT_DELAY:-30s} ALARM_MIN_CONFIDENCE: ${ALARM_MIN_CONFIDENCE:-60} + MEDIA_API_URL: http://mediamtx:9997 + MEDIA_API_PASSWORD: ${MEDIA_API_PASSWORD} + CAMERA_PROBE_ENABLED: ${CAMERA_PROBE_ENABLED:-true} LOG_LEVEL: ${LOG_LEVEL:-INFO} mem_limit: 512m # Joint les deux reseaux : "default" pour parler a Kafka et PostgreSQL, diff --git a/src/main/java/me/belkacemi/iotalarm/infrastructure/config/AlarmProperties.java b/src/main/java/me/belkacemi/iotalarm/infrastructure/config/AlarmProperties.java index 89ada6e..f2c4a98 100644 --- a/src/main/java/me/belkacemi/iotalarm/infrastructure/config/AlarmProperties.java +++ b/src/main/java/me/belkacemi/iotalarm/infrastructure/config/AlarmProperties.java @@ -19,11 +19,35 @@ public record AlarmProperties( Duration alertCooldown, int minimumConfidence, Kafka kafka, - Simulator simulator) { + Simulator simulator, + Camera camera) { public record Kafka(String topic, String dlqTopic, int partitions, short replicationFactor) { } + /** + * Sonde de presence du flux video. + * + * @param livenessProbeEnabled active la sonde + * @param probeInterval periode d'interrogation du serveur media + * @param offlineThreshold sondes ratees consecutives avant de crier au loup. + * Sur un lien 4G, une coupure de quelques secondes + * (tunnel, bascule d'antenne) est normale : alerter + * des la premiere raterait la cible et apprendrait + * surtout a ignorer les alertes. + * @param streamPath nom du flux cote serveur media + * @param apiUrl base de l'API MediaMTX, sur le reseau interne + */ + public record Camera( + boolean livenessProbeEnabled, + Duration probeInterval, + int offlineThreshold, + String streamPath, + String apiUrl, + String apiUser, + String apiPassword) { + } + /** * Simulateur de capteur. Sur le deploiement public, l'iPhone n'est pas * joignable : ce simulateur rejoue un trafic realiste pour que la diff --git a/src/main/java/me/belkacemi/iotalarm/infrastructure/media/CameraLivenessProbe.java b/src/main/java/me/belkacemi/iotalarm/infrastructure/media/CameraLivenessProbe.java new file mode 100644 index 0000000..70d25ad --- /dev/null +++ b/src/main/java/me/belkacemi/iotalarm/infrastructure/media/CameraLivenessProbe.java @@ -0,0 +1,114 @@ +package me.belkacemi.iotalarm.infrastructure.media; + +import me.belkacemi.iotalarm.domain.model.DeviceId; +import me.belkacemi.iotalarm.domain.model.SensorEvent; +import me.belkacemi.iotalarm.domain.model.SensorEventType; +import me.belkacemi.iotalarm.domain.port.in.IngestSensorEventUseCase; +import me.belkacemi.iotalarm.domain.port.in.command.IngestSensorEventCommand; +import me.belkacemi.iotalarm.infrastructure.config.AlarmProperties; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.time.Clock; + +/** + * ADAPTER ENTRANT : transforme la perte du flux video en evenement capteur. + * + *

Le domaine connait deja {@link SensorEventType#CAMERA_OFFLINE}, classe + * {@code CRITICAL} : un capteur qui se tait alors que l'alarme est armee, c'est + * soit une batterie morte, soit quelqu'un qui l'a trouve derriere les livres. + * Jusqu'ici, rien n'emettait cet evenement. C'est cette sonde.

+ * + *

Pourquoi un seuil et pas une detection immediate

+ * + *

Sur un lien 4G, le flux tombe regulierement sans que personne n'ait touche + * au telephone : tunnel, bascule d'antenne, congestion. Alerter des la premiere + * sonde ratee produirait plusieurs alertes CRITICAL par jour, et la seule chose + * qu'on apprend a vivre avec des alertes qui crient pour rien, c'est a les + * ignorer - y compris le jour ou elles ont raison.

+ * + *

Il faut donc {@code offlineThreshold} sondes consecutives ratees, soit une + * minute d'absence reelle avec les valeurs par defaut.

+ * + *

Point d'architecture : la sonde passe par le meme port + * {@link IngestSensorEventUseCase} que l'endpoint REST et le simulateur. Trois + * producteurs d'evenements tres differents, un seul contrat d'entree - et + * l'evenement traverse ensuite Kafka et l'agregat comme n'importe quel autre. + * Si l'alarme est desarmee, une coupure video ne reveille personne : c'est + * l'agregat qui en decide, pas cette classe.

+ */ +@Component +@ConditionalOnProperty( + prefix = "alarm.camera", + name = "liveness-probe-enabled", + havingValue = "true") +public class CameraLivenessProbe { + + private static final Logger log = LoggerFactory.getLogger(CameraLivenessProbe.class); + + private final MediaServerClient mediaServer; + private final IngestSensorEventUseCase ingestSensorEvent; + private final AlarmProperties.Camera config; + private final DeviceId deviceId; + private final Clock clock; + + /** Sondes ratees d'affilee. Remis a zero des que l'image revient. */ + private int consecutiveFailures; + + /** Evite de re-emettre l'evenement a chaque sonde tant que l'absence dure. */ + private boolean offlineAlreadyReported; + + public CameraLivenessProbe(MediaServerClient mediaServer, + IngestSensorEventUseCase ingestSensorEvent, + AlarmProperties properties, + Clock clock) { + this.mediaServer = mediaServer; + this.ingestSensorEvent = ingestSensorEvent; + this.config = properties.camera(); + this.deviceId = DeviceId.of(properties.simulator().deviceId()); + this.clock = clock; + log.info("Sonde de presence video active sur '{}' - {} sondes ratees " + + "consecutives declenchent un CAMERA_OFFLINE", + config.streamPath(), config.offlineThreshold()); + } + + @Scheduled(fixedDelayString = "${alarm.camera.probe-interval}") + public void probe() { + if (mediaServer.isStreamLive(config.streamPath())) { + onStreamAlive(); + return; + } + onStreamMissing(); + } + + private void onStreamAlive() { + if (offlineAlreadyReported) { + log.info("Flux video retabli sur '{}'", config.streamPath()); + } + consecutiveFailures = 0; + offlineAlreadyReported = false; + } + + private void onStreamMissing() { + consecutiveFailures++; + + if (offlineAlreadyReported) { + return; + } + if (consecutiveFailures < config.offlineThreshold()) { + log.debug("Flux '{}' absent ({}/{} avant signalement)", + config.streamPath(), consecutiveFailures, config.offlineThreshold()); + return; + } + + log.warn("Flux video '{}' absent depuis {} sondes - emission d'un CAMERA_OFFLINE", + config.streamPath(), consecutiveFailures); + offlineAlreadyReported = true; + + ingestSensorEvent.ingest(new IngestSensorEventCommand( + SensorEvent.of(deviceId, SensorEventType.CAMERA_OFFLINE, clock.instant()))); + } +} diff --git a/src/main/java/me/belkacemi/iotalarm/infrastructure/media/MediaMtxClient.java b/src/main/java/me/belkacemi/iotalarm/infrastructure/media/MediaMtxClient.java new file mode 100644 index 0000000..2bf96a7 --- /dev/null +++ b/src/main/java/me/belkacemi/iotalarm/infrastructure/media/MediaMtxClient.java @@ -0,0 +1,65 @@ +package me.belkacemi.iotalarm.infrastructure.media; + +import me.belkacemi.iotalarm.infrastructure.config.AlarmProperties; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.http.client.SimpleClientHttpRequestFactory; +import org.springframework.stereotype.Component; +import org.springframework.web.client.RestClient; + +import java.time.Duration; + +/** + * ADAPTER : interroge l'API de MediaMTX pour savoir si le flux est alimente. + */ +@Component +public class MediaMtxClient implements MediaServerClient { + + private static final Logger log = LoggerFactory.getLogger(MediaMtxClient.class); + + private final RestClient restClient; + + public MediaMtxClient(RestClient.Builder builder, AlarmProperties properties) { + AlarmProperties.Camera camera = properties.camera(); + + // Timeouts serres, poses ici plutot que par un RestClientCustomizer : + // ce dernier s'appliquerait a tous les RestClient de l'application. + // Les valeurs par defaut de Spring sont infinies - une sonde qui ne + // revient jamais immobilise un thread du scheduler, et l'alarme cesse + // silencieusement de detecter la perte de camera. + SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory(); + requestFactory.setConnectTimeout(Duration.ofSeconds(2)); + requestFactory.setReadTimeout(Duration.ofSeconds(3)); + + this.restClient = builder + .baseUrl(camera.apiUrl()) + .requestFactory(requestFactory) + .defaultHeaders(headers -> + headers.setBasicAuth(camera.apiUser(), camera.apiPassword())) + .build(); + } + + @Override + public boolean isStreamLive(String streamPath) { + try { + PathInfo info = restClient.get() + .uri("/v3/paths/get/{path}", streamPath) + .retrieve() + .body(PathInfo.class); + + return info != null && info.ready(); + + } catch (Exception exception) { + // API injoignable : on ne sait pas si la camera filme, donc on + // considere qu'elle ne filme pas. Du point de vue du systeme + // d'alarme, "pas d'image" et "serveur media muet" ont exactement + // la meme consequence : il n'y a plus de surveillance video. + log.debug("API du serveur media injoignable : {}", exception.getMessage()); + return false; + } + } + + /** Reponse partielle de l'API MediaMTX : on n'extrait que ce qui sert. */ + record PathInfo(String name, boolean ready) { + } +} diff --git a/src/main/java/me/belkacemi/iotalarm/infrastructure/media/MediaServerClient.java b/src/main/java/me/belkacemi/iotalarm/infrastructure/media/MediaServerClient.java new file mode 100644 index 0000000..44d8f0f --- /dev/null +++ b/src/main/java/me/belkacemi/iotalarm/infrastructure/media/MediaServerClient.java @@ -0,0 +1,15 @@ +package me.belkacemi.iotalarm.infrastructure.media; + +/** + * Interrogation du serveur media : le flux video est-il vivant ? + * + *

Interface volontairement minuscule. Elle n'expose pas MediaMTX mais la + * seule question que le systeme d'alarme se pose reellement.

+ */ +public interface MediaServerClient { + + /** + * @return {@code true} si un publieur alimente ce flux en ce moment + */ + boolean isStreamLive(String streamPath); +} diff --git a/src/main/resources/application-local.yml b/src/main/resources/application-local.yml index 44e6c94..31adf9e 100644 --- a/src/main/resources/application-local.yml +++ b/src/main/resources/application-local.yml @@ -18,3 +18,9 @@ spring: logging: level: me.belkacemi.iotalarm: DEBUG + +alarm: + # Pas de MediaMTX en developpement local : la sonde signalerait une camera + # hors ligne toutes les 20 secondes. + camera: + liveness-probe-enabled: false diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 733a13c..bf3f7b1 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -81,6 +81,21 @@ alarm: partitions: 3 replication-factor: 1 + # Sonde de presence du flux video. Traduit une camera muette en evenement + # capteur CAMERA_OFFLINE, que le domaine classe CRITICAL - un capteur qui se + # tait alors que l'alarme est armee, c'est soit une batterie morte, soit + # quelqu'un qui l'a trouve derriere les livres. + camera: + liveness-probe-enabled: ${CAMERA_PROBE_ENABLED:true} + probe-interval: ${CAMERA_PROBE_INTERVAL:20s} + # 3 sondes ratees = 60 s reelles sans image. En 4G, le flux tombe + # regulierement sans que personne n'ait touche au telephone. + offline-threshold: ${CAMERA_OFFLINE_THRESHOLD:3} + stream-path: ${CAMERA_STREAM_PATH:biblio} + api-url: ${MEDIA_API_URL:http://mediamtx:9997} + api-user: probe + api-password: ${MEDIA_API_PASSWORD:} + simulator: enabled: ${SIMULATOR_ENABLED:true} interval: ${SIMULATOR_INTERVAL:8s} diff --git a/src/test/java/me/belkacemi/iotalarm/infrastructure/media/CameraLivenessProbeTest.java b/src/test/java/me/belkacemi/iotalarm/infrastructure/media/CameraLivenessProbeTest.java new file mode 100644 index 0000000..049fab0 --- /dev/null +++ b/src/test/java/me/belkacemi/iotalarm/infrastructure/media/CameraLivenessProbeTest.java @@ -0,0 +1,150 @@ +package me.belkacemi.iotalarm.infrastructure.media; + +import me.belkacemi.iotalarm.domain.model.SensorEventType; +import me.belkacemi.iotalarm.domain.port.in.IngestSensorEventUseCase; +import me.belkacemi.iotalarm.domain.port.in.command.IngestSensorEventCommand; +import me.belkacemi.iotalarm.infrastructure.config.AlarmProperties; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * La sonde doit distinguer une coupure 4G d'un capteur reellement neutralise. + * + *

C'est la seule chose qui compte ici : une alerte CRITICAL qui se declenche + * a chaque tunnel devient une alerte qu'on apprend a ignorer - y compris le + * jour ou elle a raison.

+ */ +@DisplayName("CameraLivenessProbe - tolerance aux coupures reseau") +class CameraLivenessProbeTest { + + private static final int THRESHOLD = 3; + private static final String STREAM = "biblio"; + + private MediaServerClient mediaServer; + private IngestSensorEventUseCase ingest; + private CameraLivenessProbe probe; + + @BeforeEach + void setUp() { + mediaServer = mock(MediaServerClient.class); + ingest = mock(IngestSensorEventUseCase.class); + + AlarmProperties properties = new AlarmProperties( + "studio", + Duration.ofSeconds(30), + Duration.ofSeconds(15), + 60, + new AlarmProperties.Kafka("sensor-events", "sensor-events-dlq", 3, (short) 1), + new AlarmProperties.Simulator(false, Duration.ofSeconds(8), "iphone-biblio", 25), + new AlarmProperties.Camera(true, Duration.ofSeconds(20), THRESHOLD, + STREAM, "http://mediamtx:9997", "probe", "secret")); + + probe = new CameraLivenessProbe(mediaServer, ingest, properties, + Clock.fixed(Instant.parse("2026-09-12T22:00:00Z"), ZoneOffset.UTC)); + } + + private void streamIs(boolean live) { + when(mediaServer.isStreamLive(STREAM)).thenReturn(live); + } + + @Test + @DisplayName("flux present : aucun evenement") + void aliveStreamEmitsNothing() { + streamIs(true); + + probe.probe(); + probe.probe(); + + verify(ingest, never()).ingest(org.mockito.ArgumentMatchers.any()); + } + + @Test + @DisplayName("coupure breve sous le seuil : aucune alerte") + void shortOutageStaysSilent() { + streamIs(false); + + probe.probe(); + probe.probe(); // 2 sondes ratees, seuil a 3 + + verify(ingest, never()).ingest(org.mockito.ArgumentMatchers.any()); + } + + @Test + @DisplayName("seuil atteint : un CAMERA_OFFLINE est emis") + void sustainedOutageRaisesCameraOffline() { + streamIs(false); + + probe.probe(); + probe.probe(); + probe.probe(); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(IngestSensorEventCommand.class); + verify(ingest).ingest(captor.capture()); + + assertThat(captor.getValue().event().type()).isEqualTo(SensorEventType.CAMERA_OFFLINE); + assertThat(captor.getValue().event().deviceId().value()).isEqualTo("iphone-biblio"); + } + + @Test + @DisplayName("coupure prolongee : un seul evenement, pas un par sonde") + void outageIsReportedOnlyOnce() { + streamIs(false); + + for (int i = 0; i < 10; i++) { + probe.probe(); + } + + verify(ingest, times(1)).ingest(org.mockito.ArgumentMatchers.any()); + } + + @Test + @DisplayName("apres retour puis nouvelle coupure, une nouvelle alerte est possible") + void counterResetsWhenStreamComesBack() { + streamIs(false); + probe.probe(); + probe.probe(); + probe.probe(); // 1re alerte + + streamIs(true); + probe.probe(); // retour du flux : compteur remis a zero + + streamIs(false); + probe.probe(); + probe.probe(); + probe.probe(); // 2e alerte + + verify(ingest, times(2)).ingest(org.mockito.ArgumentMatchers.any()); + } + + @Test + @DisplayName("un retour avant le seuil annule le comptage") + void recoveryBeforeThresholdCancelsEverything() { + streamIs(false); + probe.probe(); + probe.probe(); // 2 ratees + + streamIs(true); + probe.probe(); // le flux revient + + streamIs(false); + probe.probe(); + probe.probe(); // 2 ratees a nouveau, pas 4 + + verify(ingest, never()).ingest(org.mockito.ArgumentMatchers.any()); + } +}