julio/carinha-control-panel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2933407b9e095cd99fde175520ff3dfed27dc06e
carinha-control-panel/historico.py
T
julioandClaude Opus 5.5 2933407b9e feat(vida): jeito de gente e console com o porquê de cada decisão 
Console (console.py + tela Console no painel):
- cada parte registra o que fez e o motivo: modo vida (com necessidades
  e humor), sentidos, celular (ordens e o que o rosto fez), IA local,
  aprendizado e ideias do dia; últimas 400 linhas, ao vivo pelo SSE;
- tela com o "agora" (humor, necessidades, o que está fazendo, próxima
  decisão, última decisão, jeito), filtros, detalhes, pausa e busca.

Jeito de gente:
- saudade: a festa cresce com o tempo fora (aceno, "Oi! Você voltou!",
  "Senti sua falta!"); bom dia lembrando de ontem;
- pedidos com motivo ("Brinca comigo?", "Me faz um carinho?", mostrar a
  figurinha nova), no máximo 1 a cada 20 min;
- larga atividade no meio quando fica chato; personalidade (curiosidade,
  brincadeira, timidez) que pesa nos gestos e amadurece com o convívio;
- no rosto: acordar grogue, sonhar com algo do dia (balão no motor),
  hesitar antes de gestos, errar na atividade e ficar sem graça, olhar
  nos olhos e desviar, sorrir de volta (FaceLandmarker no rastreador do
  painel) e reagir à luz (sensor de luz no app 0.10.0);
- ordem "dizer" com legenda e voz de robô.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-11 03:37:07 -03:00

412 lines
20 KiB
Python
Raw Blame History

"""Histórico de detecções (PIR da placa, movimento e som do celular) com uma foto por evento.
- PIR: lê o pino com `gpiomon` (libgpiod). Padrão: gpiochip1 linha 70 = pino 11 (PC6) do
Orange Pi Zero 3; outro pino com CARINHA_PIR=gpiochip1:79.
- Foto: ao registrar um evento, o servidor deixa um "pedido de foto"; o app do S10 consulta
/historico/pedido, tira a foto com a câmera frontal e envia em /historico/foto?id=N.
- Guarda só os últimos 3 dias (eventos e fotos mais antigos são apagados sozinhos), com teto de
segurança de 2000 eventos e 500 fotos. Cada sensor e a foto podem ser desligados.
"""
import json
import os
from pathlib import Path
import shutil
import sqlite3
import subprocess
import threading
import time
from contextlib import contextmanager
from urllib.parse import parse_qs, urlsplit
from comum import gravar_atomico, ler_json
TIPOS = ('pir', 'movimento', 'som')
PADRAO = dict(pir=True, movimento=True, som=True, foto=True, intervalo=15, dias=3,
presenca=True, ausencia=5, # modo presença: o PIR liga a câmera; sem ninguém, ela desliga
sensibilidade_pir='media')
# O HC-SR505 não tem ajuste de sensibilidade e segura ~8 s a cada disparo. Para separar alguém de
# verdade de um disparo à toa: (disparos que bastam, dentro de quantos s, ou ligado direto por quantos s).
REGRAS_PIR = {'alta': (1, 0, 0), 'media': (2, 60, 12), 'baixa': (3, 90, 20)}
MAX_EVENTOS, MAX_FOTOS = 2000, 500 # teto de segurança; quem manda é a retenção em dias
VALIDADE_PEDIDO = 12 # s que o app tem para mandar a foto
class Historico:
def __init__(self, pasta, iniciar_pir=True):
self.pasta = Path(pasta)
self.fotos = self.pasta / 'fotos'
self.fotos.mkdir(exist_ok=True)
self.caminho_db = str(self.pasta / 'historico.db')
self.arquivo_config = self.pasta / 'historico.json'
self.lock = threading.RLock()
self.publicar = lambda tipo, dados: None
self.ao_detectar = lambda tipo: None # o servidor usa para o PIR acordar o rosto
# Modo presença: o servidor liga câmera/acorda (ao_presenca) e desliga/dorme (ao_ausencia).
self.ao_presenca = lambda: None
self.ao_ausencia = lambda: None
self.presenca = dict(estado='presente', desde=time.time(), ultimo=time.time())
self.config = dict(PADRAO)
try:
salvo = json.loads(self.arquivo_config.read_text(encoding='utf-8'))
self.config.update(self.validar(salvo))
except (OSError, ValueError, TypeError):
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,
ativo=False) # ativo = pino em nível alto agora (o HC-SR505 segura ~8 s e rearma)
self.processo_pir = None
self.subidas_pir = [] # disparos ainda não confirmados (horários)
self.console = lambda origem, texto, motivo='', nivel='info', agora=None: None
self.subiu_em = None # quando o pino subiu da última vez (ligado direto desde então)
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)')
self.limpar_antigos()
if iniciar_pir:
threading.Thread(target=self.vigiar_pir, daemon=True).start()
threading.Thread(target=self.faxina, daemon=True).start()
threading.Thread(target=self.vigiar_ausencia, daemon=True).start()
@contextmanager
def db(self):
con = sqlite3.connect(self.caminho_db)
try:
with con:
yield con
finally:
con.close()
@staticmethod
def validar(dados):
if not isinstance(dados, dict):
raise ValueError('Esperava um objeto')
limpo = {}
for chave, valor in dados.items():
if chave not in PADRAO:
raise ValueError('Ajuste desconhecido: ' + chave)
if chave == 'ausencia':
if isinstance(valor, bool) or not isinstance(valor, int) or not 1 <= valor <= 120:
raise ValueError('Ausência de 1 a 120 minutos')
elif chave == 'dias':
if isinstance(valor, bool) or not isinstance(valor, int) or not 1 <= valor <= 30:
raise ValueError('Guardar de 1 a 30 dias')
elif chave == 'sensibilidade_pir':
if valor not in REGRAS_PIR:
raise ValueError('Sensibilidade do sensor: alta, media ou baixa')
elif chave == 'intervalo':
if isinstance(valor, bool) or not isinstance(valor, (int, float)) or not 0 <= valor <= 600:
raise ValueError('Intervalo entre 0 e 600 s')
elif not isinstance(valor, bool):
raise ValueError(chave + ' precisa ser verdadeiro ou falso')
limpo[chave] = valor
return limpo
# ------------------------------------------------------------------ retenção
def limpar_antigos(self, agora=None):
"""Apaga eventos (e fotos) mais velhos que a retenção em dias."""
limite = (agora if agora is not None else time.time()) - self.config['dias'] * 86400
with self.lock, self.db() as db:
velhos = db.execute('SELECT id, foto FROM eventos WHERE horario < ?', (limite,)).fetchall()
for vid, foto in velhos:
self.apagar_arquivo(foto)
db.execute('DELETE FROM eventos WHERE horario < ?', (limite,))
if velhos:
self.publicar('historico', dict(expirados=len(velhos)))
return len(velhos)
# ------------------------------------------------------------------ modo presença
def marcar_presenca(self, agora=None):
agora = agora if agora is not None else time.time()
with self.lock:
self.presenca['ultimo'] = agora
voltou = self.presenca['estado'] != 'presente'
if voltou:
self.presenca.update(estado='presente', desde=agora)
if voltou and self.config['presenca']:
self.ao_presenca()
def checar_ausencia(self, agora=None):
"""Sem presença no PIR por `ausencia` minutos: desliga câmera, dorme e fica quieto."""
agora = agora if agora is not None else time.time()
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.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. A bolinha acompanha o pino; presença (evento, foto, acordar) só
quando o disparo se confirma pela sensibilidade escolhida."""
agora = agora if agora is not None else time.time()
self.pir['ativo'] = alto
if alto:
self.pir['ultimo'] = agora
self.avisar_pir()
if not alto:
self.checar_pir(agora) # ligado direto até agora pode já ter confirmado
self.subiu_em = None
return
if not registrar: # pino assentando ao ligar: só a bolinha
return
quantos, janela, _ = REGRAS_PIR[self.config['sensibilidade_pir']]
with self.lock:
self.subidas_pir = [t for t in self.subidas_pir if agora - t < janela] + [agora]
self.subiu_em = agora
confirmou = len(self.subidas_pir) >= quantos
if confirmou:
self.confirmar_pir(agora)
else:
self.console('sentidos', 'Disparo solto do sensor de presença', 'sensibilidade %s: ainda não confirmou (%d de %d)' % (
self.config['sensibilidade_pir'], len(self.subidas_pir), quantos), 'detalhe', agora)
def checar_pir(self, agora=None):
"""Ligado direto por bastante tempo = alguém se mexendo na frente (o sensor foi redisparando)."""
agora = agora if agora is not None else time.time()
_, _, segurar = REGRAS_PIR[self.config['sensibilidade_pir']]
with self.lock:
pronto = segurar and self.subiu_em is not None and self.subidas_pir and agora - self.subiu_em >= segurar
if pronto:
self.confirmar_pir(agora)
def confirmar_pir(self, agora):
with self.lock:
self.subidas_pir = []
self.marcar_presenca(agora)
self.registrar('pir', 'presença', agora=agora)
def vigiar_ausencia(self):
passo = 0
while True:
time.sleep(1)
passo += 1
try:
self.checar_pir()
if passo % 10 == 0:
self.checar_ausencia()
except Exception:
pass
def faxina(self):
while True:
time.sleep(600)
try:
self.limpar_antigos()
except Exception:
pass
# ------------------------------------------------------------------ eventos
def registrar(self, tipo, evento='', nivel=None, agora=None):
"""Grava o evento se o sensor estiver ligado e fora do intervalo mínimo; devolve o id ou None."""
if tipo not in TIPOS:
raise ValueError('Tipo de evento inválido')
agora = agora if agora is not None else time.time()
with self.lock:
if not self.config.get(tipo, False):
return None
if tipo in self.ultimo and agora - self.ultimo[tipo] < self.config['intervalo']:
return None
self.ultimo[tipo] = agora
evento = str(evento or '')[:40]
nivel = float(nivel) if isinstance(nivel, (int, float)) and not isinstance(nivel, bool) else None
with self.db() as db:
ident = db.execute('INSERT INTO eventos(horario, tipo, evento, nivel) VALUES (?,?,?,?)',
(agora, tipo, evento, nivel)).lastrowid
velhos = db.execute('SELECT id, foto FROM eventos ORDER BY id DESC LIMIT -1 OFFSET ?', (MAX_EVENTOS,)).fetchall()
for vid, foto in velhos:
self.apagar_arquivo(foto)
db.execute('DELETE FROM eventos WHERE id=?', (vid,))
if self.config['foto']:
self.pedido = (ident, time.monotonic())
self.limpar_antigos(agora)
self.publicar('historico', dict(novo=self.evento(ident)))
try:
self.ao_detectar(tipo)
except Exception:
pass
return ident
def evento(self, ident):
with self.db() as db:
linha = db.execute('SELECT id, horario, tipo, evento, nivel, foto FROM eventos WHERE id=?', (ident,)).fetchone()
return self.como_dict(linha) if linha else None
@staticmethod
def como_dict(linha):
i, h, t, e, n, f = linha
return dict(id=i, horario=h, tipo=t, evento=e, nivel=n, foto=bool(f))
def listar(self, limite=500):
with self.db() as db:
linhas = db.execute('SELECT id, horario, tipo, evento, nivel, foto FROM eventos ORDER BY id DESC LIMIT ?', (limite,)).fetchall()
total = db.execute('SELECT count(*) FROM eventos').fetchone()[0]
return [self.como_dict(l) for l in linhas], total
def apagar(self, ident):
with self.lock, self.db() as db:
linha = db.execute('SELECT foto FROM eventos WHERE id=?', (ident,)).fetchone()
if linha:
self.apagar_arquivo(linha[0])
db.execute('DELETE FROM eventos WHERE id=?', (ident,))
self.publicar('historico', dict(apagado=ident))
def limpar(self):
with self.lock, self.db() as db:
db.execute('DELETE FROM eventos')
for f in self.fotos.glob('evento-*.jpg'):
f.unlink(missing_ok=True)
self.pedido = None
self.publicar('historico', dict(limpo=True))
def apagar_arquivo(self, nome):
if nome:
(self.fotos / nome).unlink(missing_ok=True)
# ------------------------------------------------------------------ foto
def pedido_atual(self):
with self.lock:
if self.pedido and time.monotonic() - self.pedido[1] < VALIDADE_PEDIDO:
return self.pedido[0]
self.pedido = None
return None
def guardar_foto(self, ident, jpeg):
if not (jpeg.startswith(b'\xff\xd8') and jpeg.endswith(b'\xff\xd9')) or len(jpeg) > 3_000_000:
raise ValueError('Esperava uma foto JPEG de até 3 MB')
with self.lock:
if not self.config['foto']:
raise ValueError('Fotos desligadas')
if self.pedido and self.pedido[0] == ident:
self.pedido = None
nome = f'evento-{ident}.jpg'
with self.db() as db:
if not db.execute('SELECT 1 FROM eventos WHERE id=?', (ident,)).fetchone():
raise ValueError('Evento não existe mais')
tmp = self.fotos / (nome + '.tmp')
tmp.write_bytes(jpeg); tmp.replace(self.fotos / nome)
db.execute('UPDATE eventos SET foto=? WHERE id=?', (nome, ident))
antigas = db.execute('SELECT id, foto FROM eventos WHERE foto IS NOT NULL ORDER BY id DESC LIMIT -1 OFFSET ?', (MAX_FOTOS,)).fetchall()
for vid, foto in antigas:
self.apagar_arquivo(foto)
db.execute('UPDATE eventos SET foto=NULL WHERE id=?', (vid,))
self.publicar('historico', dict(foto=ident))
# ------------------------------------------------------------------ PIR
def vigiar_pir(self):
"""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')
if os.name == 'posix' and not shutil.which('gpiomon'):
self.pir['erro'] = 'gpiomon não instalado (apt install gpiod)'
time.sleep(2)
continue
chip, _, linha = self.pir['linha'].partition(':')
try:
# pull-down: com o sensor desligado o pino não fica "flutuando" (disparos falsos)
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 linha_evento in self.processo_pir.stdout:
if not self.config['pir']:
break
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, ativo=False, erro=erro if self.config['pir'] else 'desligado')
time.sleep(3)
def parar_pir(self):
p, self.processo_pir = self.processo_pir, None
if p and p.poll() is None:
p.terminate()
try:
p.wait(timeout=2)
except subprocess.TimeoutExpired:
p.kill()
def status(self):
eventos, total = self.listar()
with self.lock:
return dict(config=dict(self.config), pir=dict(self.pir), presenca=dict(self.presenca), eventos=eventos, total=total,
max_eventos=MAX_EVENTOS, max_fotos=MAX_FOTOS)
# ------------------------------------------------------------------ HTTP
def rotas(self, h, metodo):
url = urlsplit(h.path)
rota = url.path
if not rota.startswith('/historico/'):
return False
try:
if metodo == 'GET' and rota == '/historico/status':
return h.json(self.status()) or True
if metodo == 'GET' and rota == '/historico/pedido':
return h.json(dict(id=self.pedido_atual())) or True
if metodo == 'GET' and rota.startswith('/historico/foto/'):
nome = rota.rsplit('/', 1)[-1]
if not nome.startswith('evento-') or not nome.endswith('.jpg') or '/' in nome or '..' in nome:
raise ValueError('Foto inválida')
arquivo = self.fotos / nome
if not arquivo.is_file():
h.json(dict(erro='Foto não encontrada'), 404)
return True
h.enviar(arquivo.read_bytes(), 'image/jpeg')
return True
if metodo == 'POST' and rota == '/historico/foto':
ident = int(parse_qs(url.query).get('id', ['0'])[0])
if h.headers.get('Content-Type', '').split(';')[0] != 'image/jpeg':
raise ValueError('Use image/jpeg')
tamanho = int(h.headers.get('Content-Length', '0'))
if not 0 < tamanho <= 3_000_000:
raise ValueError('Foto vazia ou grande demais')
self.guardar_foto(ident, h.rfile.read(tamanho))
return h.json(dict(ok=True)) or True
if metodo == 'POST':
dados = ler_json(h, 4096)
if rota == '/historico/evento':
ident = self.registrar(dados.get('tipo'), dados.get('evento'), dados.get('nivel'))
return h.json(dict(ok=True, id=ident)) or True
if rota == '/historico/config':
limpo = self.validar(dados)
with self.lock:
self.config.update(limpo)
gravar_atomico(self.arquivo_config, self.config)
if 'pir' in limpo and not limpo['pir']:
self.parar_pir()
if 'dias' in limpo:
self.limpar_antigos()
self.publicar('historico', dict(config=dict(self.config)))
return h.json(self.status()) or True
if rota == '/historico/apagar':
if type(dados.get('id')) is not int:
raise ValueError('ID inválido')
self.apagar(dados['id'])
return h.json(dict(ok=True)) or True
if rota == '/historico/limpar':
self.limpar()
return h.json(dict(ok=True)) or True
h.json(dict(erro='Rota do histórico desconhecida'), 404)
except (ValueError, TypeError) as e:
h.json(dict(erro=str(e)), 400)
return True
Reference in New Issue View Git Blame Copy Permalink