diff --git a/deploy.ps1 b/deploy.ps1 index 7392a56..18fe15c 100644 --- a/deploy.ps1 +++ b/deploy.ps1 @@ -27,7 +27,7 @@ Set-Location $PSScriptRoot -ErrorAction Stop # chaves, vendor/ (modelos já instalados na placa) e o fonte do painel (só web/dist vai). $arquivos = @( 'kiosk-server.py', 'comum.py', 'eventos.py', 'controle_pc.py', 'sinalizacao.py', 'servos.py', 'braco.py', - 'cerebro.py', 'conversa.py', 'gemini.py', 'treinamento.py', 'figurinhas.py', 'api_config.py', 'historico.py', 'vida.py', 'criacao.py', 'inspiracao.py', 'console.py', 'aprendizado.py', + 'cerebro.py', 'conversa.py', 'gemini.py', 'treinamento.py', 'figurinhas.py', 'api_config.py', 'historico.py', 'vida.py', 'criacao.py', 'inspiracao.py', 'console.py', 'reconhecimento.py', 'aprendizado.py', 'carinha.html', 'camera.html', 'reacoes-render.js', 'rastreador-worker.js', 'rosto-motor.js', 'voz-robo.js', 'version.json', 'obter-rastreamento.py', 'android/app/assets/qualidade-auto.js' ) diff --git a/historico.py b/historico.py index 651d17f..c434a6c 100644 --- a/historico.py +++ b/historico.py @@ -40,6 +40,7 @@ class Historico: 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 + self.ao_foto = lambda ident, caminho: None # foto nova de um evento (reconhecimento de pessoas) # 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 @@ -298,6 +299,16 @@ class Historico: self.apagar_arquivo(foto) db.execute('UPDATE eventos SET foto=NULL WHERE id=?', (vid,)) self.publicar('historico', dict(foto=ident)) + try: + self.ao_foto(ident, self.fotos / nome) + except Exception: + pass + + def fotos_com_evento(self): + """[(id do evento, arquivo da foto)] das fotos guardadas, mais novas primeiro.""" + with self.db() as db: + linhas = db.execute('SELECT id, foto FROM eventos WHERE foto IS NOT NULL ORDER BY id DESC').fetchall() + return [(i, self.fotos / f) for i, f in linhas if (self.fotos / f).is_file()] # ------------------------------------------------------------------ PIR def vigiar_pir(self): diff --git a/kiosk-server.py b/kiosk-server.py index 4630ff6..9ee5a6c 100644 --- a/kiosk-server.py +++ b/kiosk-server.py @@ -46,6 +46,7 @@ from criacao import Criacao from inspiracao import Inspiracao from aprendizado import Aprendizado from console import Console +from reconhecimento import Reconhecimento TRAVA = threading.Lock() @@ -80,7 +81,7 @@ ESTATICOS = {"/carinha.html", "/camera.html", "/reacoes-render.js", "/rastreador # Rotas que não exigem token: só o necessário para a página do rosto abrir e o app se achar. PUBLICAS_GET = {"/app/health", "/faces", "/gestos", "/config", "/ia/modelo", "/app.apk"} -PREFIXOS_API = ("/eventos", "/console", "/estado", "/fila", "/rtc/", "/conversa/", "/treinamento/", "/historico/", "/vida/", "/aprendizado/", "/criacao/", +PREFIXOS_API = ("/eventos", "/console", "/pessoas", "/estado", "/fila", "/rtc/", "/conversa/", "/treinamento/", "/historico/", "/vida/", "/aprendizado/", "/criacao/", "/app/", "/braco/", "/servos/", "/ia/", "/api/", "/gesto/", "/face/", "/diag/") @@ -282,7 +283,7 @@ class Handler(SimpleHTTPRequestHandler): def modulos(self): s = self.server - return [getattr(s, nome) for nome in ("console", "api_config", "historico", "vida", "aprendizado", "inspiracao", "criacao", "treinamento", "conversa", "cerebro", + return [getattr(s, nome) for nome in ("console", "reconhecimento", "api_config", "historico", "vida", "aprendizado", "inspiracao", "criacao", "treinamento", "conversa", "cerebro", "braco", "servos", "controle") if hasattr(s, nome)] def estado(self): @@ -505,7 +506,7 @@ class Servidor(ThreadingHTTPServer): def ligar_eventos(servidor): for nome in ("controle", "servos", "cerebro", "conversa", "treinamento", "historico", "vida", "criacao", "inspiracao", "aprendizado", - "console"): + "console", "reconhecimento"): if hasattr(servidor, nome): getattr(servidor, nome).publicar = EVENTOS.publicar @@ -581,7 +582,14 @@ def main(): servidor.aprendizado = aprendizado = Aprendizado(Path.cwd() / 'aprendizado.json', nome_criada) # Console: tudo o que ele faz e por quê (cada parte registra o motivo da decisão) servidor.console = CONSOLE - for parte in (vida, aprendizado, inspiracao, servidor.cerebro, servidor.historico, servidor.controle): + # Reconhecimento de pessoas: as fotos dos eventos dizem quem chegou (tudo na placa) + servidor.reconhecimento = rec = Reconhecimento(Path.cwd()) + servidor.historico.ao_foto = rec.nova_foto + rec.fotos_guardadas = servidor.historico.fotos_com_evento + rec.ao_reconhecer = vida.reconhecido + vida.reconhece = lambda: rec.config['ligado'] and rec.visao.pronta() and bool(rec.amostras()) + vida.gostou_de = rec.gostou + for parte in (vida, aprendizado, inspiracao, servidor.cerebro, servidor.historico, servidor.controle, rec): parte.console = CONSOLE.registrar vida.dia_anterior = lambda: aprendizado.historico[-1] if aprendizado.historico else None CONSOLE.registrar("sistema", "Servidor iniciado", "versão " + versao()) diff --git a/reconhecimento.py b/reconhecimento.py new file mode 100644 index 0000000..2516146 --- /dev/null +++ b/reconhecimento.py @@ -0,0 +1,406 @@ +"""Reconhecimento de pessoas: quem é quem, só na placa. + +Cada foto de evento (o celular manda uma quando o sensor vê alguém) passa pelo YuNet, que acha os +rostos, e pelo SFace, que vira cada rosto numa "impressão" de 128 números. A impressão é comparada +com as das pessoas cadastradas (semelhança de cosseno); acima do limiar, é aquela pessoa. Rostos sem +nome ficam numa lista para alguém marcar no painel ("esta é a Ana"), e cada marcação vira amostra. + +Privacidade: nada sai da placa. Fotos, recortes e impressões ficam em /opt/carinha-pc; apagar uma +pessoa apaga as amostras e os recortes dela; rostos sem nome somem depois de DIAS_SEM_NOME dias. +""" +import json +import os +import queue +import sqlite3 +import threading +import time +from contextlib import contextmanager +from pathlib import Path + +from comum import gravar_atomico, ler_json + +LIMIAR = 0.40 # cosseno mínimo para dizer "é essa pessoa" (o SFace recomenda 0,363) +LIMIAR_GRUPO = 0.45 # rostos sem nome tão parecidos assim vão para o mesmo grupo (marcar de uma vez) +SCORE_ROSTO = 0.80 # confiança mínima do detector para considerar um rosto +TAMANHO_MIN = 48 # px: rosto menor que isso é longe demais para reconhecer bem +DIAS_SEM_NOME = 7 # rostos sem nome guardados para marcar no painel +MAX_AMOSTRAS = 30 # por pessoa (as mais novas) +PADRAO = dict(ligado=True) +MODELOS = ('face_detection_yunet_2023mar.onnx', 'face_recognition_sface_2021dec.onnx') + + +def cosseno(a, b): + na = sum(x * x for x in a) ** 0.5 + nb = sum(x * x for x in b) ** 0.5 + if not na or not nb: + return 0.0 + return sum(x * y for x, y in zip(a, b)) / (na * nb) + + +class Visao: + """YuNet + SFace (OpenCV). Carrega na primeira foto; sem OpenCV ou modelo, fica indisponível.""" + + def __init__(self, pasta_modelos): + self.pasta = Path(pasta_modelos) + self.det = self.rec = self.cv2 = self.np = None + self.motivo = '' + + def pronta(self): + if self.rec: + return True + try: + import cv2 + import numpy as np + except ImportError: + self.motivo = 'OpenCV não instalado na placa (apt install python3-opencv)' + return False + faltando = [m for m in MODELOS if not (self.pasta / m).exists()] + if faltando: + self.motivo = 'Faltam os modelos em %s: %s' % (self.pasta, ', '.join(faltando)) + return False + self.cv2, self.np = cv2, np + self.det = cv2.FaceDetectorYN.create(str(self.pasta / MODELOS[0]), '', (320, 320), SCORE_ROSTO, 0.3, 5000) + self.rec = cv2.FaceRecognizerSF.create(str(self.pasta / MODELOS[1]), '') + self.motivo = '' + return True + + def rostos(self, jpeg): + """bytes JPEG -> [(vetor[128], recorte_jpeg, score, largura)]""" + cv2, np = self.cv2, self.np + if not jpeg: # foto que não chegou inteira + return [] + img = cv2.imdecode(np.frombuffer(jpeg, np.uint8), cv2.IMREAD_COLOR) + if img is None: + return [] + h, w = img.shape[:2] + escala = min(1.0, 640 / max(h, w)) # a placa é pequena: analisa em até 640 px + if escala < 1: + img = cv2.resize(img, (int(w * escala), int(h * escala))) + h, w = img.shape[:2] + self.det.setInputSize((w, h)) + _, achados = self.det.detect(img) + saida = [] + for f in (achados if achados is not None else []): + if f[14] < SCORE_ROSTO or min(f[2], f[3]) < TAMANHO_MIN * escala: + continue + alinhado = self.rec.alignCrop(img, f) + vetor = self.rec.feature(alinhado).flatten().tolist() + ok, recorte = cv2.imencode('.jpg', alinhado, [cv2.IMWRITE_JPEG_QUALITY, 85]) + saida.append((vetor, recorte.tobytes() if ok else b'', float(f[14]), float(f[2]))) + return saida + + +class Reconhecimento: + def __init__(self, pasta, visao=None, iniciar=True): + self.pasta = Path(pasta) + self.recortes = self.pasta / 'fotos' / 'rostos' + self.recortes.mkdir(parents=True, exist_ok=True) + self.caminho_db = str(self.pasta / 'pessoas.db') + self.arquivo_config = self.pasta / 'pessoas.json' + self.visao = visao or Visao(self.pasta / 'modelos') + self.lock = threading.RLock() + self.fila = queue.Queue(maxsize=500) + self.publicar = lambda tipo, dados: None + self.console = lambda origem, texto, motivo='', nivel='info', agora=None: None + self.ao_reconhecer = lambda pessoa, confianca, agora: None # o modo vida cumprimenta pelo nome + self.varrendo = None # {feitas, total} ao procurar nas fotos guardadas + self.config = dict(PADRAO) + try: + self.config.update({k: v for k, v in json.loads(self.arquivo_config.read_text(encoding='utf-8')).items() + if k in PADRAO and isinstance(v, bool)}) + except (OSError, ValueError, AttributeError): + pass + with self.db() as db: + db.execute('CREATE TABLE IF NOT EXISTS pessoas(id INTEGER PRIMARY KEY, nome TEXT UNIQUE, criado REAL, ' + 'visto REAL, vezes INTEGER DEFAULT 0, afinidade REAL DEFAULT 0)') + db.execute('CREATE TABLE IF NOT EXISTS rostos(id INTEGER PRIMARY KEY, horario REAL, evento INTEGER, ' + 'pessoa INTEGER, amostra INTEGER DEFAULT 0, confianca REAL, score REAL, vetor TEXT, recorte TEXT)') + if iniciar: + threading.Thread(target=self.trabalhar, daemon=True).start() + threading.Thread(target=self.faxina, daemon=True).start() + + def faxina(self): + while True: + time.sleep(3600) + try: + self.limpar_antigos() + except Exception: + pass + + @contextmanager + def db(self): + con = sqlite3.connect(self.caminho_db) + try: + with con: + yield con + finally: + con.close() + + # ------------------------------------------------------------ pessoas + def amostras(self): + """{pessoa: [vetores]} das faces marcadas por alguém.""" + with self.db() as db: + linhas = db.execute('SELECT pessoa, vetor FROM rostos WHERE amostra=1 AND pessoa IS NOT NULL').fetchall() + saida = {} + for pessoa, vetor in linhas: + saida.setdefault(pessoa, []).append(json.loads(vetor)) + return saida + + def quem(self, vetor, amostras=None): + """(pessoa, confiança) da amostra mais parecida, ou (None, melhor) abaixo do limiar.""" + melhor, quem = 0.0, None + for pessoa, vetores in (amostras if amostras is not None else self.amostras()).items(): + s = max(cosseno(vetor, v) for v in vetores) + if s > melhor: + melhor, quem = s, pessoa + return (quem, melhor) if melhor >= LIMIAR else (None, melhor) + + def pessoa(self, ident): + with self.db() as db: + linha = db.execute('SELECT id, nome, visto, vezes, afinidade FROM pessoas WHERE id=?', (ident,)).fetchone() + return dict(id=linha[0], nome=linha[1], visto=linha[2], vezes=linha[3], afinidade=round(linha[4] or 0, 1)) if linha else None + + def pessoa_por_nome(self, nome): + with self.db() as db: + linha = db.execute('SELECT id FROM pessoas WHERE lower(nome)=lower(?)', (nome,)).fetchone() + return self.pessoa(linha[0]) if linha else None + + # ------------------------------------------------------------ fotos + def nova_foto(self, evento, caminho): + """Chamado pelo histórico quando chega a foto de um evento: analisa em segundo plano.""" + if self.config['ligado']: + try: + self.fila.put_nowait((evento, str(caminho), False)) + except queue.Full: + pass + + def analisar(self, evento, jpeg, agora=None, avisar=True): + """Acha e reconhece os rostos de uma foto. Devolve [(pessoa ou None, confiança)].""" + agora = agora or time.time() + if not self.visao.pronta(): + return [] + achados = self.visao.rostos(jpeg) + amostras = self.amostras() + resultado = [] + for vetor, recorte, score, _ in achados: + pessoa, conf = self.quem(vetor, amostras) + antes = self.pessoa(pessoa)['visto'] if pessoa else None # para a saudade de cada um + with self.lock, self.db() as db: + ident = db.execute('INSERT INTO rostos(horario, evento, pessoa, confianca, score, vetor) VALUES (?,?,?,?,?,?)', + (agora, evento, pessoa, round(conf, 3), round(score, 3), json.dumps([round(x, 5) for x in vetor]))).lastrowid + if recorte: + nome = 'rosto-%d.jpg' % ident + (self.recortes / nome).write_bytes(recorte) + db.execute('UPDATE rostos SET recorte=? WHERE id=?', (nome, ident)) + if pessoa: + db.execute('UPDATE pessoas SET visto=?, vezes=vezes+1 WHERE id=?', (agora, pessoa)) + resultado.append((pessoa, conf)) + if avisar: + if pessoa: + p = dict(self.pessoa(pessoa), visto_antes=antes) + self.console('sentidos', 'Reconheceu %s' % p['nome'], 'semelhança %d%% (evento %s)' % (round(conf * 100), evento), + agora=agora) + try: + self.ao_reconhecer(p, conf, agora) + except Exception: + pass + else: + self.console('sentidos', 'Viu um rosto que ainda não conhece', 'parecido %d%% com alguém (precisa %d%%)' % ( + round(conf * 100), round(LIMIAR * 100)), 'detalhe', agora) + if achados and avisar: + self.avisar() + return resultado + + def trabalhar(self): + while True: + evento, caminho, varredura = self.fila.get() + try: + jpeg = Path(caminho).read_bytes() + self.analisar(evento, jpeg, avisar=not varredura) + except Exception as e: + print('reconhecimento: erro:', e, flush=True) + finally: + if varredura and self.varrendo: + self.varrendo['feitas'] += 1 + if self.varrendo['feitas'] >= self.varrendo['total']: + self.varrendo = None + self.console('sentidos', 'Terminou de procurar rostos nas fotos guardadas') + self.avisar() + + def varrer(self, eventos): + """Procura rostos nas fotos já guardadas (para cadastrar pessoas logo de cara).""" + if not self.visao.pronta(): + raise ValueError(self.visao.motivo) + if self.varrendo: + raise ValueError('Já estou procurando nas fotos guardadas') + with self.db() as db: + feitos = {r[0] for r in db.execute('SELECT DISTINCT evento FROM rostos')} + alvos = [(e, c) for e, c in eventos if e not in feitos and c] + if not alvos: + raise ValueError('Nenhuma foto nova para analisar') + self.varrendo = dict(feitas=0, total=len(alvos)) + for evento, caminho in alvos: + self.fila.put((evento, str(caminho), True)) + self.avisar() + return self.status() + + # ------------------------------------------------------------ marcar no painel + def marcar(self, rostos, pessoa=None, nome=None): + """Diz de quem são estes rostos (vira amostra). nome cria a pessoa; sem pessoa e nome, descarta.""" + if not isinstance(rostos, list) or not rostos or not all(type(r) is int for r in rostos): + raise ValueError('Escolha pelo menos um rosto') + with self.lock, self.db() as db: + if nome is not None: + nome = ' '.join(str(nome).split())[:30] + if not nome: + raise ValueError('Digite o nome') + existente = db.execute('SELECT id FROM pessoas WHERE lower(nome)=lower(?)', (nome,)).fetchone() + pessoa = existente[0] if existente else db.execute( + 'INSERT INTO pessoas(nome, criado) VALUES (?,?)', (nome, time.time())).lastrowid + elif pessoa is not None and not db.execute('SELECT 1 FROM pessoas WHERE id=?', (pessoa,)).fetchone(): + raise ValueError('Pessoa não encontrada') + marcas = ','.join('?' * len(rostos)) + if pessoa is None: # não é ninguém (ou foto ruim): some da lista + for (arquivo,) in db.execute('SELECT recorte FROM rostos WHERE id IN (%s)' % marcas, rostos).fetchall(): + self.apagar_recorte(arquivo) + db.execute('DELETE FROM rostos WHERE id IN (%s)' % marcas, rostos) + else: + db.execute('UPDATE rostos SET pessoa=?, amostra=1 WHERE id IN (%s)' % marcas, [pessoa] + rostos) + # guarda só as amostras mais novas de cada pessoa + velhas = db.execute('SELECT id, recorte FROM rostos WHERE pessoa=? AND amostra=1 ORDER BY id DESC LIMIT -1 OFFSET ?', + (pessoa, MAX_AMOSTRAS)).fetchall() + for vid, arquivo in velhas: + self.apagar_recorte(arquivo) + db.execute('DELETE FROM rostos WHERE id=?', (vid,)) + if pessoa is not None: + self.reconferir() + p = self.pessoa(pessoa) + self.console('sentidos', 'Aprendeu o rosto de %s' % p['nome'], '%d foto(s) marcada(s) no painel' % len(rostos)) + self.avisar() + return self.status() + + def reconferir(self): + """Com amostras novas, os rostos sem nome são comparados de novo (sem reabrir as fotos).""" + amostras = self.amostras() + if not amostras: + return + with self.lock, self.db() as db: + for ident, vetor in db.execute('SELECT id, vetor FROM rostos WHERE pessoa IS NULL').fetchall(): + pessoa, conf = self.quem(json.loads(vetor), amostras) + if pessoa: + db.execute('UPDATE rostos SET pessoa=?, confianca=? WHERE id=?', (pessoa, round(conf, 3), ident)) + + def renomear(self, pessoa, nome): + nome = ' '.join(str(nome or '').split())[:30] + if not nome: + raise ValueError('Digite o nome') + with self.db() as db: + try: + db.execute('UPDATE pessoas SET nome=? WHERE id=?', (nome, pessoa)) + except sqlite3.IntegrityError: + raise ValueError('Já existe alguém com esse nome') from None + self.avisar() + return self.status() + + def apagar_pessoa(self, pessoa): + with self.lock, self.db() as db: + for (arquivo,) in db.execute('SELECT recorte FROM rostos WHERE pessoa=?', (pessoa,)).fetchall(): + self.apagar_recorte(arquivo) + db.execute('DELETE FROM rostos WHERE pessoa=?', (pessoa,)) + db.execute('DELETE FROM pessoas WHERE id=?', (pessoa,)) + self.avisar() + return self.status() + + def gostou(self, pessoa, quanto=1.0): + """Carinho ou conversa de quem foi reconhecido há pouco: a afinidade com essa pessoa cresce.""" + with self.db() as db: + db.execute('UPDATE pessoas SET afinidade=min(100, afinidade+?) WHERE id=?', (quanto, pessoa)) + + def apagar_recorte(self, arquivo): + if arquivo: + (self.recortes / arquivo).unlink(missing_ok=True) + + def limpar_antigos(self, agora=None): + limite = (agora or time.time()) - DIAS_SEM_NOME * 86400 + with self.lock, self.db() as db: + velhos = db.execute('SELECT id, recorte FROM rostos WHERE amostra=0 AND horario < ?', (limite,)).fetchall() + for vid, arquivo in velhos: + self.apagar_recorte(arquivo) + db.execute('DELETE FROM rostos WHERE amostra=0 AND horario < ?', (limite,)) + return len(velhos) + + def configurar(self, dados): + if not isinstance(dados, dict) or set(dados) - set(PADRAO) or not all(isinstance(v, bool) for v in dados.values()): + raise ValueError('Ajuste inválido') + self.config.update(dados) + gravar_atomico(self.arquivo_config, self.config) + self.avisar() + return self.status() + + # ------------------------------------------------------------ painel + def status(self): + pronta = self.visao.pronta() + with self.db() as db: + pessoas = [] + for pid, nome, visto, vezes, afinidade in db.execute('SELECT id, nome, visto, vezes, afinidade FROM pessoas ORDER BY nome'): + fotos = [dict(id=i, recorte=r) for i, r in db.execute( + 'SELECT id, recorte FROM rostos WHERE pessoa=? AND recorte IS NOT NULL ORDER BY amostra DESC, id DESC LIMIT 6', (pid,))] + amostras = db.execute('SELECT count(*) FROM rostos WHERE pessoa=? AND amostra=1', (pid,)).fetchone()[0] + pessoas.append(dict(id=pid, nome=nome, visto=visto, vezes=vezes, afinidade=round(afinidade or 0, 1), + amostras=amostras, fotos=fotos)) + linhas = db.execute('SELECT id, horario, evento, recorte, confianca, vetor FROM rostos WHERE pessoa IS NULL ' + 'AND recorte IS NOT NULL ORDER BY id DESC LIMIT 120').fetchall() + total = db.execute('SELECT count(*) FROM rostos').fetchone()[0] + sem_nome, grupos = [], [] # grupos: rostos parecidos entre si (provável mesma pessoa) + for i, hr, e, r, c, vetor in linhas: + v = json.loads(vetor) + g = next((g for g in grupos if cosseno(v, g['v']) >= LIMIAR_GRUPO), None) + if g is None: + g = dict(v=v, ids=[]) + grupos.append(g) + g['ids'].append(i) + sem_nome.append(dict(id=i, horario=hr, evento=e, recorte=r, parecido=c, grupo=grupos.index(g))) + grupos = [g['ids'] for g in grupos] + return dict(config=dict(self.config), disponivel=pronta, motivo='' if pronta else self.visao.motivo, + pessoas=pessoas, sem_nome=sem_nome, grupos=grupos, rostos=total, varrendo=self.varrendo, limiar=LIMIAR, + dias_sem_nome=DIAS_SEM_NOME) + + def avisar(self): + self.publicar('pessoas', self.status()) + + def rotas(self, h, metodo): + rota = h.path.split('?', 1)[0] + if not rota.startswith('/pessoas'): + return False + try: + if metodo == 'GET' and rota == '/pessoas': + resultado = self.status() + elif metodo == 'GET' and rota.startswith('/pessoas/rosto/'): + nome = os.path.basename(rota) + arquivo = self.recortes / nome + if not nome.startswith('rosto-') or not nome.endswith('.jpg') or not arquivo.is_file(): + raise ValueError('Rosto não encontrado') + h.enviar(arquivo.read_bytes(), 'image/jpeg') + return True + elif metodo == 'POST': + d = ler_json(h, 8192) + if rota == '/pessoas/marcar': + resultado = self.marcar(d.get('rostos'), d.get('pessoa'), d.get('nome')) + elif rota == '/pessoas/renomear': + resultado = self.renomear(d.get('pessoa'), d.get('nome')) + elif rota == '/pessoas/apagar': + resultado = self.apagar_pessoa(d.get('pessoa')) + elif rota == '/pessoas/config': + resultado = self.configurar(d) + elif rota == '/pessoas/varrer': + resultado = self.varrer(self.fotos_guardadas()) + else: + raise ValueError('Rota inválida') + else: + raise ValueError('Método inválido') + h.enviar(json.dumps(resultado, ensure_ascii=False)) + except (ValueError, TypeError) as e: + h.enviar(json.dumps(dict(erro=str(e)), ensure_ascii=False), codigo=400) + return True + + fotos_guardadas = staticmethod(lambda: []) # o main liga ao histórico diff --git a/sistema/reconhecimento/instalar.sh b/sistema/reconhecimento/instalar.sh new file mode 100644 index 0000000..71ae912 --- /dev/null +++ b/sistema/reconhecimento/instalar.sh @@ -0,0 +1,13 @@ +#!/bin/sh +# Reconhecimento de pessoas na placa: OpenCV + modelos YuNet (acha rostos) e SFace (impressão do rosto). +# Rodar como root. Tudo fica na placa; nada vai para a internet depois de baixar os modelos. +set -e +DESTINO=${1:-/opt/carinha-pc} +apt-get install -y --no-install-recommends python3-opencv +mkdir -p "$DESTINO/modelos" +cd "$DESTINO/modelos" +for m in face_detection_yunet/face_detection_yunet_2023mar.onnx face_recognition_sface/face_recognition_sface_2021dec.onnx; do + [ -s "$(basename "$m")" ] || curl -fsSL -o "$(basename "$m")" "https://github.com/opencv/opencv_zoo/raw/main/models/$m" +done +python3 -c "import cv2; assert hasattr(cv2, 'FaceRecognizerSF'); print('reconhecimento pronto: OpenCV', cv2.__version__)" +systemctl restart carinha-pc diff --git a/tests/test_reconhecimento.py b/tests/test_reconhecimento.py new file mode 100644 index 0000000..010908a --- /dev/null +++ b/tests/test_reconhecimento.py @@ -0,0 +1,177 @@ +"""Reconhecimento de pessoas: marcar rostos, reconhecer, saudade por pessoa, afinidade e privacidade.""" +import random +import sys +import tempfile +import time +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from reconhecimento import DIAS_SEM_NOME, Reconhecimento, cosseno # noqa: E402 +from vida import Vida # noqa: E402 + +DIA = time.mktime((2026, 10, 12, 15, 0, 0, 0, 0, -1)) + + +def rosto_de(semente, ruido=0.25): + """Impressão de 128 números: a mesma pessoa varia pouco, pessoas diferentes são bem diferentes.""" + base = random.Random(semente) + r = random.Random() + return [base.gauss(0, 1) + r.gauss(0, ruido) for _ in range(128)] + + +class VisaoFalsa: + motivo = '' + + def __init__(self): + self.fotos = {} + + def pronta(self): + return True + + def rostos(self, jpeg): + return [(v, b'\xff\xd8recorte\xff\xd9', 0.95, 120) for v in self.fotos.get(jpeg, [])] + + +class ReconhecimentoTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.visao = VisaoFalsa() + self.r = Reconhecimento(self.tmp.name, visao=self.visao, iniciar=False) + self.vistos, self.linhas = [], [] + self.r.ao_reconhecer = lambda p, conf, agora: self.vistos.append((p['nome'], round(conf, 2), p.get('visto_antes'))) + self.r.console = lambda origem, texto, motivo='', nivel='info', agora=None: self.linhas.append(texto) + + def tearDown(self): + self.tmp.cleanup() + + def foto(self, evento, *pessoas, agora=DIA): + jpeg = ('foto-%d' % evento).encode() + self.visao.fotos[jpeg] = [rosto_de(p) for p in pessoas] + return self.r.analisar(evento, jpeg, agora) + + def test_impressoes_da_mesma_pessoa_se_parecem(self): + self.assertGreater(cosseno(rosto_de('ana'), rosto_de('ana')), 0.8) + self.assertLess(abs(cosseno(rosto_de('ana'), rosto_de('bia'))), 0.3) + + def test_marcar_e_depois_reconhecer_pelo_nome(self): + (pessoa, _), = self.foto(1, 'ana') # ninguém cadastrado: rosto sem nome + self.assertIsNone(pessoa) + self.foto(2, 'ana'); self.foto(3, 'ana') + sem_nome = [s['id'] for s in self.r.status()['sem_nome']] + self.assertEqual(len(sem_nome), 3) + self.r.marcar(sem_nome, nome='Ana') + st = self.r.status() + self.assertEqual([p['nome'] for p in st['pessoas']], ['Ana']) + self.assertEqual(st['pessoas'][0]['amostras'], 3) + self.assertEqual(st['sem_nome'], []) + # foto nova: reconhece a Ana e não a Bia + (pessoa, conf), = self.foto(4, 'ana', agora=DIA + 60) + self.assertIsNotNone(pessoa) + self.assertEqual(self.vistos[-1][0], 'Ana') + self.assertIn('Reconheceu Ana', self.linhas) + (pessoa, _), = self.foto(5, 'bia', agora=DIA + 120) + self.assertIsNone(pessoa) + self.assertEqual(len(self.r.status()['sem_nome']), 1) + + def test_marcar_a_pessoa_reconhece_os_rostos_antigos_sem_nome(self): + for i in range(4): + self.foto(10 + i, 'bia') + ids = [s['id'] for s in self.r.status()['sem_nome']] + self.r.marcar(ids[:1], nome='Bia') # marcou um: os outros parecidos já passam a ser dela + st = self.r.status() + self.assertEqual(st['sem_nome'], []) + self.assertEqual(st['pessoas'][0]['amostras'], 1) + + def test_saudade_conta_quando_cada_um_foi_visto(self): + self.foto(1, 'ana') + self.r.marcar([s['id'] for s in self.r.status()['sem_nome']], nome='Ana') + self.foto(2, 'ana', agora=DIA + 100) + self.foto(3, 'ana', agora=DIA + 5000) + self.assertEqual(self.vistos[-1][2], DIA + 100) # visto antes desta foto + + def test_rostos_parecidos_ficam_no_mesmo_grupo(self): + self.foto(1, 'ana'); self.foto(2, 'bia'); self.foto(3, 'ana'); self.foto(4, 'ana') + grupos = sorted(len(g) for g in self.r.status()['grupos']) + self.assertEqual(grupos, [1, 3]) + + def test_descartar_renomear_apagar(self): + self.foto(1, 'ana'); self.foto(2, 'gato') + ana, gato = sorted(s['id'] for s in self.r.status()['sem_nome']) + self.r.marcar([gato]) # não é ninguém + self.r.marcar([ana], nome='ana') + self.r.marcar([ana], nome='ANA') # mesmo nome: mesma pessoa + pid = self.r.status()['pessoas'][0]['id'] + self.r.renomear(pid, 'Ana Paula') + self.assertEqual(self.r.status()['pessoas'][0]['nome'], 'Ana Paula') + self.assertEqual(len(list((Path(self.tmp.name) / 'fotos' / 'rostos').glob('*.jpg'))), 1) + self.r.apagar_pessoa(pid) # apaga a pessoa e tudo dela + self.assertEqual(self.r.status()['pessoas'], []) + self.assertEqual(self.r.status()['rostos'], 0) + self.assertEqual(list((Path(self.tmp.name) / 'fotos' / 'rostos').glob('*.jpg')), []) + with self.assertRaises(ValueError): + self.r.marcar([], nome='x') + + def test_rostos_sem_nome_somem_depois_de_uns_dias(self): + self.foto(1, 'ana', agora=DIA) + self.foto(2, 'bia', agora=DIA) + self.r.marcar([self.r.status()['sem_nome'][0]['id']], nome='Bia') + self.assertEqual(self.r.limpar_antigos(DIA + DIAS_SEM_NOME * 86400 + 1), 1) # a amostra da Bia fica + self.assertEqual(self.r.status()['rostos'], 1) + + def test_desligado_nao_analisa_fotos_novas(self): + self.r.configurar(dict(ligado=False)) + self.r.nova_foto(1, 'x.jpg') + self.assertTrue(self.r.fila.empty()) + with self.assertRaises(ValueError): + self.r.configurar(dict(outra=True)) + + +class VidaComNomesTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.v = Vida(Path(self.tmp.name) / 'vida.json', iniciar=False, rng=random.Random(1)) + self.ordens, self.afinidade = [], [] + self.v.enfileirar = self.ordens.append + self.v.celular_online = lambda: True + self.v.celular_dormindo = lambda: False + self.v.madrugada = lambda agora: False + self.v.reconhece = lambda: True + self.v.gostou_de = lambda pessoa, quanto=1.0: self.afinidade.append((pessoa, quanto)) + self.v.saudou_dia = '2026-10-12' + self.perto = False + self.v.alguem_perto = lambda: self.perto + + def tearDown(self): + self.tmp.cleanup() + + def falas(self): + return [o[6:] for o in self.ordens if o.startswith('dizer ')] + + def test_espera_saber_quem_chegou_para_cumprimentar_pelo_nome(self): + self.v.ultimo_perto, self.perto = DIA - 2 * 3600, True + self.v.tick(DIA) + self.assertEqual(self.falas(), []) # esperando a foto + self.v.reconhecido(dict(id=1, nome='Ana', afinidade=25, visto_antes=DIA - 6 * 3600), 0.8, DIA + 3) + self.assertEqual(self.falas(), ['Ana! Senti sua falta!']) # 6 h sem a Ana, mesmo a casa tendo gente há 2 h + self.assertIsNone(self.v.reencontro_pendente) + + def test_sem_reconhecer_cumprimenta_sem_nome_depois_de_esperar(self): + self.v.ultimo_perto, self.perto = DIA - 2 * 3600, True + self.v.tick(DIA) + self.v.tick(DIA + 5) + self.assertEqual(self.falas(), []) + self.v.tick(DIA + 9) + self.assertEqual(self.falas(), ['Oi! Você voltou!']) + + def test_outra_pessoa_chega_com_alguem_ja_por_perto(self): + self.v.reconhecido(dict(id=2, nome='Bia', afinidade=0, visto_antes=DIA - 3 * 3600), 0.7, DIA) + self.assertEqual(self.falas(), ['Oi, Bia! Você voltou!']) + self.v.reconhecido(dict(id=2, nome='Bia', afinidade=0, visto_antes=DIA), 0.7, DIA + 60) # acabou de ver + self.assertEqual(len(self.falas()), 1) + self.v.estimulo('carinho', DIA + 90) # carinho logo depois: afinidade com a Bia + self.assertEqual(self.afinidade, [(2, 1.0)]) + + +if __name__ == '__main__': + unittest.main() diff --git a/vida.py b/vida.py index 8d32e6d..ea6d7d7 100644 --- a/vida.py +++ b/vida.py @@ -124,6 +124,8 @@ GESTOS_TRACO = {'brincadeira': {'rir', 'comemorar', 'alegre', 'musica', 'piscadi AUSENCIA_ACENO, AUSENCIA_OI, AUSENCIA_SAUDADE = 10 * 60, 60 * 60, 4 * 3600 INTERVALO_PEDIDO = 20 * 60 # no mínimo 20 min entre um pedido ("brinca comigo?") e outro DESISTE = 0.12 # chance de largar uma atividade no meio porque ficou chato +ESPERA_NOME = 8 # s esperando o reconhecimento antes de cumprimentar sem o nome +OI_DE_NOVO = 10 * 60 # reconheceu alguém que não via há esse tempo: "Oi, Ana!" GESTOS_VIDA = {g for atos in ATOS.values() for g, _, _ in atos} @@ -172,6 +174,11 @@ class Vida: self.pediu_em = 0.0 self.desistencia = None # (quando, o que) se largou a atividade no meio self.figurinha_mostrada = 0.0 + # reconhecimento de pessoas (reconhecimento.py): o main liga + self.reconhece = lambda: False # True: vale esperar a foto para cumprimentar pelo nome + self.gostou_de = lambda pessoa, quanto=1.0: None + self.reencontro_pendente = None # (desde, ausência) esperando saber quem chegou + self.ultimo_reconhecido = None # (id, nome, quando) agora = time.time() self.config = dict(PADRAO) @@ -359,6 +366,9 @@ class Vida: if self.aprendizado: self.aprendizado.resposta(tipo, agora) self.amadurecer(tipo) + rec = self.ultimo_reconhecido + if rec and agora - rec[2] < 120: + self.gostou_de(rec[0], 1.0 if tipo in ('carinho', 'brincar', 'conversa') else 0.5) self.console('sentidos', CUIDADOS.get(tipo) or NOME_ESTIMULO.get(tipo, tipo), ', '.join('%s %+d' % (k, v) for k, v in efeito.items()), 'detalhe' if tipo in ('som', 'movimento') else 'info', agora) @@ -453,10 +463,12 @@ class Vida: def tempo(seg): return '%d min' % round(seg / 60) if seg < 3600 else '%dh%02d' % (seg // 3600, seg % 3600 // 60) - def saudacao(self, agora): + def saudacao(self, agora, nome=None): """Bom dia com uma lembrança de ontem (do que ele viveu, sem IA).""" hora = time.localtime(agora).tm_hour prefixo = 'Bom dia' if 5 <= hora < 12 else 'Boa tarde' if hora < 18 else 'Boa noite' + if nome: + prefixo += ', ' + nome fatos = [] ontem = None try: @@ -478,29 +490,35 @@ class Vida: fatos.append('Acordei cheio de energia!' if self.necessidades['energia'] >= 80 else 'Que bom te ver!') return '%s! %s' % (prefixo, self.acaso.choice(fatos)) - def reencontro(self, agora, ausencia): + def reencontro(self, agora, ausencia, nome=None, afinidade=0.0): """Alguém apareceu: a reação depende de quanto tempo ficou fora (e de quem ele é).""" if self.dormindo() or self.madrugada(agora) or self.bloqueado(): return None n, tim = self.necessidades, self.personalidade['timidez'] dia = time.strftime('%Y-%m-%d', time.localtime(agora)) - motivo = 'ninguém por perto havia ' + self.tempo(ausencia) + motivo = ('%s não aparecia havia ' % nome if nome else 'ninguém por perto havia ') + self.tempo(ausencia) + querido = afinidade >= 20 # muito carinho dessa pessoa: festa com coração if self.saudou_dia != dia and time.localtime(agora).tm_hour >= 5 and ausencia >= AUSENCIA_ACENO: self.saudou_dia = dia - frase = self.saudacao(agora) + frase = self.saudacao(agora, nome) self.dizer(frase, 'timido' if tim > 0.65 else 'acenar') ganho, texto, motivo = 10, 'Deu bom dia: "%s"' % frase, 'primeiro encontro do dia; ' + motivo elif ausencia >= AUSENCIA_SAUDADE: - self.dizer('Senti sua falta!', 'comemorar') + self.dizer('%s! Senti sua falta!' % nome if nome else 'Senti sua falta!', 'comemorar') self.enfileirar('coracao') - ganho, texto = 20, 'Fez festa: estava com saudade' + ganho, texto = 20, 'Fez festa: estava com saudade' + (' de ' + nome if nome else '') n['alegria'] = _limitar(n['alegria'] + 10) elif ausencia >= AUSENCIA_OI: - self.dizer('Oi! Você voltou!', 'timido' if tim > 0.65 else 'comemorar') - ganho, texto = 10, 'Comemorou que alguém voltou' + self.dizer('Oi, %s! Você voltou!' % nome if nome else 'Oi! Você voltou!', 'timido' if tim > 0.65 else 'comemorar') + if querido: + self.enfileirar('coracao') + ganho, texto = 10, 'Comemorou que %s voltou' % (nome or 'alguém') elif ausencia >= AUSENCIA_ACENO: - self.enfileirar('acenar') - ganho, texto = 4, 'Acenou para quem voltou' + if nome: + self.dizer('Oi, %s!' % nome, 'acenar') + else: + self.enfileirar('acenar') + ganho, texto = 4, 'Acenou para %s' % (nome or 'quem voltou') else: return None with self.lock: @@ -510,6 +528,21 @@ class Vida: self.avisar() return texto + def reconhecido(self, pessoa, confianca, agora=None): + """O reconhecimento disse quem está aí (vem da thread do reconhecimento).""" + agora = agora or time.time() + nome = pessoa['nome'] + self.ultimo_reconhecido = (pessoa['id'], nome, agora) + antes = pessoa.get('visto_antes') + pendente = self.reencontro_pendente + if pendente: # chegou alguém e agora sabe quem é + self.reencontro_pendente = None + ausencia = agora - antes if antes else pendente[1] + return self.reencontro(agora, ausencia, nome, pessoa.get('afinidade', 0)) + if antes and agora - antes > OI_DE_NOVO: # chegou outra pessoa com alguém já por perto + return self.reencontro(agora, agora - antes, nome, pessoa.get('afinidade', 0)) + return None + def pedir_algo(self, agora): """Iniciativa com motivo: pede o que está faltando (ou mostra o que fez).""" n = self.necessidades @@ -769,9 +802,16 @@ class Vida: perto = self.alguem_perto() if perto: if not self.perto_antes: - self.reencontro(agora, agora - self.ultimo_perto) + if self.reconhece(): + self.reencontro_pendente = (agora, agora - self.ultimo_perto) # espera a foto dizer quem é + else: + self.reencontro(agora, agora - self.ultimo_perto) self.ultimo_perto = agora self.perto_antes = perto + pendente = self.reencontro_pendente + if pendente and agora - pendente[0] >= ESPERA_NOME: + self.reencontro_pendente = None + self.reencontro(agora, pendente[1]) if self.desistencia and agora >= self.desistencia[0]: _, oque = self.desistencia self.desistencia = None diff --git a/web/src/App.svelte b/web/src/App.svelte index d53f1d5..22b4fc7 100644 --- a/web/src/App.svelte +++ b/web/src/App.svelte @@ -1,5 +1,5 @@ + +
+ + {#snippet acoes()} + + {/snippet} + {#if !st} +

Carregando…

+ {:else} +
+ pessoas.ligar(v)} /> + {#if !st.disponivel} + {st.motivo} + {:else if st.varrendo} + Procurando rostos: {st.varrendo.feitas} de {st.varrendo.total} fotos + {/if} +

Tudo fica na placa: fotos, rostos e nomes não vão para a internet. Rostos sem nome somem em {st.dias_sem_nome} dias; apagar uma pessoa apaga tudo dela.

+
+ {/if} + + + {#if st} + + {#if st.pessoas.length} + + {/if} + + + + {#if st.sem_nome.length} + {#each st.grupos as ids, g (ids[0])} + {@const todos = ids.every((i) => escolhidos.includes(i))} +
+
+ {st.grupos.length > 1 ? `Desconhecido ${g + 1}` : 'Desconhecido'} · {ids.length} {ids.length === 1 ? 'foto' : 'fotos'} + +
+
+ {#each st.sem_nome.filter((r) => r.grupo === g) as r (r.id)} + + {/each} +
+
+ {/each} + {#if escolhidos.length} +
+ {escolhidos.length} escolhido(s): é… + {#each st.pessoas as p (p.id)} + + {/each} +
+ + +
+ +
+ {/if} + {:else} +

Nenhum rosto esperando nome. Eles aparecem aqui quando o sensor vê alguém e a câmera pega o rosto (com luz).

+ {/if} + + {/if} +
+ +