fix(pir): acompanha o nível do pino em vez de só a borda de subida

O HC-SR505 segura a saída em alto ~8 s e rearma a cada movimento; com
alguém na frente o pino ficava alto direto e passar a mão não gerava
borda nova, então o painel parecia não detectar nada. Agora o gpiomon
escuta as duas bordas, o nível inicial vem do gpioget e o status/SSE
levam `ativo`. A bolinha do Ao vivo fica acesa enquanto o pino está
alto, e o modo presença não manda dormir com o sensor ainda segurando.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
julio
2026-10-06 04:04:41 -03:00
co-authored by Claude Opus 5.5
parent 784042335c
commit fd6a14054d
4 changed files with 54 additions and 16 deletions
+33 -12
View File
@@ -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):