diff --git a/historico.py b/historico.py index c86e336..9ff8362 100644 --- a/historico.py +++ b/historico.py @@ -48,7 +48,8 @@ class Historico: pass self.ultimo = {} # tipo -> horário do último evento (intervalo mínimo) self.pedido = None # (id do evento, monotonic) aguardando foto do app - self.pir = dict(disponivel=False, erro='', linha=os.environ.get('CARINHA_PIR', 'gpiochip1:70'), ultimo=None) + self.pir = dict(disponivel=False, erro='', linha=os.environ.get('CARINHA_PIR', 'gpiochip1:70'), ultimo=None, + ativo=False) # ativo = pino em nível alto agora (o HC-SR505 segura ~8 s e rearma) self.processo_pir = None with self.db() as db: db.execute('CREATE TABLE IF NOT EXISTS eventos(id INTEGER PRIMARY KEY, horario REAL, tipo TEXT, evento TEXT, nivel REAL, foto TEXT)') @@ -119,13 +120,30 @@ class Historico: if not (self.config['presenca'] and self.config['pir'] and self.pir['disponivel']): return False with self.lock: + if self.pir['ativo']: # sensor ainda segurando: a pessoa continua ali + self.presenca['ultimo'] = agora + return False if self.presenca['estado'] != 'presente' or agora - self.presenca['ultimo'] < self.config['ausencia'] * 60: return False self.presenca.update(estado='ausente', desde=agora) self.ao_ausencia() - self.publicar('pir', dict(ultimo=self.pir['ultimo'], presenca=dict(self.presenca))) + self.avisar_pir() return True + def avisar_pir(self): + self.publicar('pir', dict(ultimo=self.pir['ultimo'], ativo=self.pir['ativo'], presenca=dict(self.presenca))) + + def nivel_pir(self, alto, agora=None, registrar=True): + """Mudança no pino do PIR. Subida = presença (evento + foto); enquanto alto, a bolinha fica acesa.""" + agora = agora if agora is not None else time.time() + self.pir['ativo'] = alto + if alto: + self.pir['ultimo'] = agora + self.marcar_presenca(agora) + self.avisar_pir() + if alto and registrar: + self.registrar('pir', 'presença') + def vigiar_ausencia(self): while True: time.sleep(10) @@ -240,7 +258,7 @@ class Historico: # ------------------------------------------------------------------ PIR def vigiar_pir(self): - """Mantém um `gpiomon` rodando enquanto o PIR estiver ligado; cada borda de subida = presença.""" + """Mantém um `gpiomon` rodando enquanto o PIR estiver ligado e acompanha o nível do pino.""" while True: if not self.config['pir'] or os.name != 'posix' or not shutil.which('gpiomon'): self.pir.update(disponivel=False, erro='' if self.config['pir'] else 'desligado') @@ -251,25 +269,28 @@ class Historico: chip, _, linha = self.pir['linha'].partition(':') try: # pull-down: com o sensor desligado o pino não fica "flutuando" (disparos falsos) - self.processo_pir = subprocess.Popen(['gpiomon', '-c', chip, '-b', 'pull-down', '-e', 'rising', linha], + lido = subprocess.run(['gpioget', '-c', chip, '-b', 'pull-down', '--numeric', linha], + capture_output=True, text=True, timeout=5) + self.processo_pir = subprocess.Popen(['gpiomon', '-c', chip, '-b', 'pull-down', '-e', 'both', '-F', '%E', linha], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) self.pir.update(disponivel=True, erro='') + # nível de partida: alguém já na frente não gera borda de subida, mas acende a bolinha + self.nivel_pir(lido.stdout.strip() == '1', registrar=False) inicio = time.monotonic() - for _ in self.processo_pir.stdout: + for linha_evento in self.processo_pir.stdout: if not self.config['pir']: break - if time.monotonic() - inicio < 2: # o pino "assenta" ao ligar o pull-down - continue - self.pir['ultimo'] = time.time() - self.marcar_presenca(self.pir['ultimo']) - self.publicar('pir', dict(ultimo=self.pir['ultimo'], presenca=dict(self.presenca))) # todo disparo, ao vivo - self.registrar('pir', 'presença') + alto = linha_evento.strip() == 'rising' + # o pino "assenta" ao ligar o pull-down: acompanha o nível, mas não registra evento + self.nivel_pir(alto, registrar=time.monotonic() - inicio >= 2) erro = (self.processo_pir.stderr.read() or '').strip()[:200] except OSError as e: erro = str(e) + except subprocess.TimeoutExpired: + erro = 'gpioget não respondeu' finally: self.parar_pir() - self.pir.update(disponivel=False, erro=erro if self.config['pir'] else 'desligado') + self.pir.update(disponivel=False, ativo=False, erro=erro if self.config['pir'] else 'desligado') time.sleep(3) def parar_pir(self): diff --git a/tests/test_historico.py b/tests/test_historico.py index 8bd2366..8bd36ee 100644 --- a/tests/test_historico.py +++ b/tests/test_historico.py @@ -93,6 +93,20 @@ class HistoricoTest(unittest.TestCase): self.h.marcar_presenca(3000) self.assertFalse(self.h.checar_ausencia(9999)) + def test_pir_segurando_conta_como_presenca(self): + avisos = [] + self.h.publicar = lambda tipo, dados: avisos.append((tipo, dados)) + self.h.pir['disponivel'] = True + self.h.nivel_pir(True, agora=1000) # subida: evento + bolinha acesa + self.assertTrue(self.h.pir['ativo']) + self.assertEqual(self.h.status()['eventos'][0]['tipo'], 'pir') + self.assertFalse(self.h.checar_ausencia(1000 + 10 * 60)) # pino ainda alto: não dorme + self.h.nivel_pir(False, agora=1000 + 10 * 60) # descida: só apaga a bolinha + self.assertFalse(self.h.pir['ativo']) + self.assertEqual(self.h.status()['total'], 1) + self.assertTrue(self.h.checar_ausencia(1000 + 15 * 60 + 1)) + self.assertIn(('pir', True), [(t, d.get('ativo')) for t, d in avisos]) + def test_apagar_e_limpar(self): a = self.h.registrar('pir', agora=1); self.h.guardar_foto(a, JPEG) b = self.h.registrar('som', agora=1) diff --git a/web/src/lib/estado.svelte.ts b/web/src/lib/estado.svelte.ts index cc50e8e..531a09a 100644 --- a/web/src/lib/estado.svelte.ts +++ b/web/src/lib/estado.svelte.ts @@ -35,7 +35,7 @@ class Estado { api = $state(null); feed = $state([]); /** Sensor PIR da placa: último disparo (s, epoch) e se está ligado. */ - pir = $state<{ disponivel: boolean; ultimo: number | null; presenca?: { estado: 'presente' | 'ausente'; desde: number; ultimo: number } } | null>(null); + pir = $state<{ disponivel: boolean; ultimo: number | null; ativo?: boolean; presenca?: { estado: 'presente' | 'ausente'; desde: number; ultimo: number } } | null>(null); /** Relógio de 1 s para derivar "online" sem esperar evento novo. */ agora = $state(Date.now()); @@ -108,7 +108,7 @@ function aplicarFoto(f: EstadoServidor) { estado.reacao = f.reacao ?? null; if (f.treinamento) estado.treinamento = f.treinamento; if (f.api) estado.api = f.api; - if (f.pir) estado.pir = { disponivel: f.pir.disponivel, ultimo: f.pir.ultimo, presenca: (f.pir as any).presenca }; + if (f.pir) estado.pir = { disponivel: f.pir.disponivel, ultimo: f.pir.ultimo, ativo: (f.pir as any).ativo, presenca: (f.pir as any).presenca }; } const tratadores: Record void> = { @@ -142,7 +142,7 @@ const tratadores: Record void> = { if (d?.receita?.nome) noFeed('reacao', 'Reação: ' + d.receita.nome); }, treinamento: (d) => (estado.treinamento = d), - pir: (d: { ultimo: number; presenca?: any }) => (estado.pir = { disponivel: true, ultimo: d.ultimo, presenca: d.presenca }), + pir: (d: { ultimo: number; ativo?: boolean; presenca?: any }) => (estado.pir = { disponivel: true, ultimo: d.ultimo, ativo: d.ativo, presenca: d.presenca }), historico: (d) => { const n = d?.novo; if (n) noFeed('historico', { pir: '🚶 Presença no PIR', movimento: '👀 Movimento na câmera', som: '🔊 Som alto' }[n.tipo as string] ?? 'Detecção'); diff --git a/web/src/telas/AoVivo.svelte b/web/src/telas/AoVivo.svelte index 28f557a..ae384d2 100644 --- a/web/src/telas/AoVivo.svelte +++ b/web/src/telas/AoVivo.svelte @@ -49,7 +49,10 @@ const cfg = $derived(estado.appConfig); // Pontinho do PIR sobre a imagem: verde pulsando nos 8 s depois de cada disparo. - const pirAtivo = $derived(!!estado.pir?.ultimo && estado.agora - estado.pir.ultimo * 1000 < 8000); + // aceso enquanto o pino do sensor está alto; servidor antigo (sem `ativo`) cai nos 8 s após o disparo +const pirAtivo = $derived( + estado.pir?.ativo ?? (!!estado.pir?.ultimo && estado.agora - estado.pir.ultimo * 1000 < 8000), +); const transmitindo = $derived(!!cfg?.preview); const hora = (t: number) => new Date(t).toLocaleTimeString('pt-BR', { hour: '2-digit', minute: '2-digit', second: '2-digit' });