Sonde de presence video : la camera muette devient un evenement capteur

Le domaine connaissait deja CAMERA_OFFLINE, classe CRITICAL, mais rien ne
l'emettait. CameraLivenessProbe interroge l'API du serveur media et comble
ce trou.

Elle passe par le meme port IngestSensorEventUseCase que l'endpoint REST
et le simulateur : trois producteurs de natures tres differentes, un seul
contrat d'entree, et le meme trajet Kafka -> agregat ensuite. Si l'alarme
est desarmee, une coupure video ne reveille personne - c'est l'agregat qui
tranche, pas la sonde.

Le seuil est la partie qui compte. 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 CRITICAL par jour, et on apprend vite a ignorer des
alertes qui crient pour rien - y compris le jour ou elles ont raison.
Trois sondes consecutives, soit une minute d'absence reelle. Six tests
couvrent ce comportement, dont la remise a zero du compteur au retour du
flux.

Utilisateur d'API dedie cote MediaMTX, restreint au reseau Docker interne.

41 tests au vert.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Adam Belkacemi 2026-09-12 18:24:45 +02:00
parent 5e4dfee665
commit d017ba4958
12 changed files with 446 additions and 4 deletions

View file

@ -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

View file

@ -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://<serveur>:8890?streamid=publish:biblio:iphone:<mot-de-passe>
RTMP rtmp://<serveur>:1935/biblio?user=iphone&pass=<mot-de-passe>
SRT srt://<serveur>:8890 · streamid = publish:biblio:iphone:<mot-de-passe>
RTMP rtmp://<serveur>:1935/biblio?user=iphone&pass=<mot-de-passe>
```
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.

View file

@ -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"

View file

@ -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: []

View file

@ -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,

View file

@ -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

View file

@ -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.
*
* <p>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.</p>
*
* <h2>Pourquoi un seuil et pas une detection immediate</h2>
*
* <p>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.</p>
*
* <p>Il faut donc {@code offlineThreshold} sondes consecutives ratees, soit une
* minute d'absence reelle avec les valeurs par defaut.</p>
*
* <p>Point d'architecture : la sonde passe par le <em>meme port</em>
* {@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.</p>
*/
@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())));
}
}

View file

@ -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) {
}
}

View file

@ -0,0 +1,15 @@
package me.belkacemi.iotalarm.infrastructure.media;
/**
* Interrogation du serveur media : le flux video est-il vivant ?
*
* <p>Interface volontairement minuscule. Elle n'expose pas MediaMTX mais la
* seule question que le systeme d'alarme se pose reellement.</p>
*/
public interface MediaServerClient {
/**
* @return {@code true} si un publieur alimente ce flux en ce moment
*/
boolean isStreamLive(String streamPath);
}

View file

@ -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

View file

@ -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}

View file

@ -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.
*
* <p>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.</p>
*/
@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<IngestSensorEventCommand> 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());
}
}