julioandClaude Opus 5.5
935bf8d586
O aprendizado passa a entender a logística da semana, não só a hora:
- presença por dia da semana × hora (7×24, média móvel por dia) e
histórico dos últimos 42 dias;
- depois da meia-noite, com 7 dias observados, a IA escolhida (Gemini,
IA local ou Qwen do S10, via Criacao.chamar) cruza os dias e devolve
padrões em frases e um plano da semana (casa / fora / dormindo);
- o plano da IA só é adotado se conferir com 70% das horas medidas;
senão fica como sugestão ("Usar este plano");
- "Conte sua rotina": texto livre vira plano na hora e vale sobre o da
IA; o painel mostra o quanto confere com o que os sensores veem;
- o plano guia o comportamento: acorda do cochilo 15 min antes da
chegada naquele dia da semana e, com a casa vazia, tira cochilos
longos para guardar energia;
- lição da rotina só quando muda 3+ horas ou depois de uma semana, e
some quando já existe plano da semana;
- painel: quadro da semana (o que eu vi / plano), a rotina entendida,
sugestão, contar rotina, analisar agora e a IA da análise diária.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
641 lines
33 KiB
Python
641 lines
33 KiB
Python
"""Aprendizado: o Carinha melhora as interações com o que vive no dia a dia.
|
||
|
||
Sem IA e sem internet: só conta o que acontece em casa e ajusta o modo vida aos poucos.
|
||
- Gosto: cada gesto ou atividade feito com alguém por perto vira uma tentativa; um carinho, uma
|
||
brincadeira ou uma conversa logo depois conta como "gostaram". Os que agradam saem mais, os
|
||
ignorados saem menos (no máximo o dobro ou a metade da chance).
|
||
- Rotina: em que horas costuma ter gente por perto. Um pouco antes da hora de sempre, ele acorda
|
||
de um cochilo para estar de pé quando a pessoa chegar.
|
||
- Sono: quando o sensor acorda o rosto e ninguém fica nem interage, foi à toa. Se isso se repete
|
||
na mesma hora, ele passa a pedir mais movimento para acordar nesse horário.
|
||
- Semana: a presença também é guardada por dia da semana × hora, com o histórico dos últimos dias.
|
||
Uma vez por dia uma IA (Gemini, IA local ou o S10) cruza esse quadro, acha os padrões ("fora de
|
||
terça a quinta, em casa na segunda e na sexta") e propõe um plano da semana. O plano só é usado
|
||
se bater com o que os sensores mediram; senão fica como sugestão. A pessoa também pode contar a
|
||
rotina com as próprias palavras, e os dados vão dizendo o quanto ela confere.
|
||
À meia-noite ele fecha o dia: o que aprendeu vira lição no diário e o que ficou para trás pesa
|
||
menos (o gosto das pessoas muda). "Esquecer" zera tudo.
|
||
"""
|
||
import json
|
||
import threading
|
||
import time
|
||
from datetime import date
|
||
from pathlib import Path
|
||
|
||
from comum import gravar_atomico, ler_json
|
||
|
||
PADRAO = dict(ligado=True, analise_ia=True, ia='gemini')
|
||
IAS = ('gemini', 'local', 's10')
|
||
JANELA = 60 # s depois de um ato em que a resposta de alguém conta para ele
|
||
PESO_RESPOSTA = {'carinho': 1.0, 'brincar': 1.0, 'conversa': 1.0, 'alimentar': 0.6, 'cutucar': 0.5}
|
||
ESQUECE = 0.85 # todo dia, o que aprendeu antes pesa um pouco menos
|
||
MEMORIA_ROTINA = 0.3 # quanto um dia novo pesa na rotina aprendida
|
||
MIN_TENTATIVAS = 3 # com menos que isso, ainda não tem opinião sobre um ato
|
||
FAIXA = (0.5, 2.0) # quanto o gosto pode mudar a chance de um ato
|
||
FALSO_ALARME = 180 # s que esperar para ver se acordar pelo sensor valeu a pena
|
||
PRESENCA_FICOU = 0.3 # fração da espera com alguém por perto para o acordar ter valido
|
||
MIN_MINUTOS_HORA = 10 # minutos observados numa hora para ela entrar na rotina
|
||
ROTINA_CHEIA, ROTINA_VAZIA = 0.5, 0.3
|
||
AVISO_CHEGADA = 15 * 60 # s antes da hora movimentada em que ele acorda do cochilo
|
||
MAX_LICOES = 30
|
||
DIAS_SEMANA = ['segunda', 'terça', 'quarta', 'quinta', 'sexta', 'sábado', 'domingo']
|
||
CURTO = ['seg', 'ter', 'qua', 'qui', 'sex', 'sáb', 'dom']
|
||
MEMORIA_SEMANA = 0.5 # cada dia da semana aparece 1 vez por semana: aprende mais rápido
|
||
MAX_HISTORICO = 42 # dias guardados para a IA cruzar
|
||
MIN_DIAS_ANALISE = 7 # análise automática só com uma semana inteira
|
||
MIN_DIAS_MANUAL = 3
|
||
MIN_HORAS_CONFERIR = 24 # horas com dados para dizer se um plano bate
|
||
CONCORDA = 0.7 # plano da IA só vale se bater com 70% das horas medidas
|
||
TIPOS = ['casa', 'fora', 'dormindo']
|
||
SCHEMA_PLANO = dict(type='object', required=['padroes', 'dias'], properties=dict(
|
||
padroes=dict(type='array', items=dict(type='string')),
|
||
dias=dict(type='array', items=dict(type='object', required=['dia', 'periodos'], properties=dict(
|
||
dia=dict(type='integer'),
|
||
periodos=dict(type='array', items=dict(type='object', required=['ini', 'fim', 'tipo'], properties=dict(
|
||
ini=dict(type='integer'), fim=dict(type='integer'), tipo=dict(type='string', enum=TIPOS)))))))))
|
||
|
||
NOMES = {
|
||
'alegre': 'saltitar', 'musica': 'cantarolar', 'feliz': 'sorrir', 'piscadinha': 'piscadinha',
|
||
'coracao': 'coraçãozinho', 'rir': 'dar risada', 'comemorar': 'comemorar', 'olhar_esq': 'olhar em volta',
|
||
'olhar_dir': 'olhar em volta', 'piscar2': 'piscar devagar', 'pensar': 'ficar pensativo', 'timido': 'ficar tímido',
|
||
'espiar': 'espiar', 'bocejar': 'bocejar', 'confuso': 'ficar confuso', 'triste': 'ficar tristinho',
|
||
'olhar_baixo': 'olhar para baixo', 'acenar': 'acenar', 'surpreso': 'fazer cara de "oi!"',
|
||
'esconder': 'esconder o rosto', 'bravo': 'ficar emburrado', 'tonto': 'ficar tonto',
|
||
'ler': 'ler', 'cafe': 'tomar café', 'meditar': 'meditar', 'telescopio': 'olhar as estrelas', 'pescar': 'pescar',
|
||
'computador': 'mexer no computador', 'regar': 'regar a plantinha', 'bola': 'jogar bola', 'malhar': 'malhar',
|
||
'corda': 'pular corda', 'skate': 'andar de skate', 'dancar': 'dançar', 'pipa': 'soltar pipa', 'pintar': 'pintar',
|
||
'violao': 'tocar violão', 'cantar': 'cantar', 'cozinhar': 'cozinhar', 'bolhas': 'fazer bolhas de sabão',
|
||
'videogame': 'jogar videogame', 'dormir': 'cochilar',
|
||
}
|
||
|
||
|
||
def _dia(agora):
|
||
return time.strftime('%Y-%m-%d', time.localtime(agora))
|
||
|
||
|
||
def _hora(agora):
|
||
return time.localtime(agora).tm_hour
|
||
|
||
|
||
def _faixas(horas):
|
||
"""[18, 19, 20, 22] -> [(18, 21), (22, 23)]: horas seguidas viram uma faixa (fim aberto)."""
|
||
faixas = []
|
||
for h in sorted(horas):
|
||
if faixas and faixas[-1][1] == h:
|
||
faixas[-1] = (faixas[-1][0], h + 1)
|
||
else:
|
||
faixas.append((h, h + 1))
|
||
return faixas
|
||
|
||
|
||
def _texto_faixas(faixas):
|
||
partes = ['das %dh às %dh' % (a, b % 24) for a, b in faixas]
|
||
return partes[0] if len(partes) == 1 else ', '.join(partes[:-1]) + ' e ' + partes[-1]
|
||
|
||
|
||
class Aprendizado:
|
||
def __init__(self, caminho, nomes=None):
|
||
self.caminho = Path(caminho)
|
||
self.lock = threading.RLock()
|
||
self.publicar = lambda tipo, dados: None
|
||
self.anotar = lambda tipo, texto, agora=None: None # diário do modo vida
|
||
self.nomes = nomes or (lambda ident: None) # nome das atividades criadas por IA
|
||
self.config = dict(PADRAO)
|
||
self.itens = {} # nome -> {'t': tentativas, 's': agradou}
|
||
self.rotina = [None] * 24 # chance de ter gente por perto em cada hora
|
||
self.falsos = [0.0] * 24 # acordou à toa nessa hora (média dos dias)
|
||
self.dias = 0
|
||
self.licoes = []
|
||
self.rotina_avisada = []
|
||
self.rotina_avisada_dia = 0
|
||
self.hoje = self._dia_vazio(_dia(time.time()))
|
||
self.pendentes = [] # atos esperando resposta: (nome, horario)
|
||
self.acordou_em = None # acordou pelo sensor: esperando para ver se valeu
|
||
self.presenca_depois = [0, 0] # (com gente, total) nas observações depois de acordar
|
||
self.minuto = None
|
||
self.salvo_em = 0.0
|
||
self.semana = [[None] * 24 for _ in range(7)] # presença por dia da semana × hora
|
||
self.semana_n = [[0] * 24 for _ in range(7)] # quantos dias mediram cada célula
|
||
self.historico = [] # [{dia, sem, presenca[24], respostas, atos}]
|
||
self.plano = None # {origem, criado, padroes, dias[7][...]}
|
||
self.sugestao = None # plano da IA que não bateu com os dados
|
||
self.analise = dict(ocupado=False, erro='', em=None, tipo='')
|
||
self.ia = None # (provedor, prompt, schema) -> dict (o main liga)
|
||
self.pode_ia = lambda provedor: False
|
||
try:
|
||
salvo = json.loads(self.caminho.read_text(encoding='utf-8'))
|
||
self.config.update(self.validar(salvo.get('config', {})))
|
||
self.itens = {k: dict(t=float(v['t']), s=float(v['s'])) for k, v in salvo.get('itens', {}).items()}
|
||
if len(salvo.get('rotina', [])) == 24:
|
||
self.rotina = [None if v is None else float(v) for v in salvo['rotina']]
|
||
if len(salvo.get('falsos', [])) == 24:
|
||
self.falsos = [float(v) for v in salvo['falsos']]
|
||
self.dias = int(salvo.get('dias', 0))
|
||
self.licoes = [l for l in salvo.get('licoes', []) if isinstance(l, dict)][:MAX_LICOES]
|
||
self.rotina_avisada = list(salvo.get('rotina_avisada', []))
|
||
self.rotina_avisada_dia = int(salvo.get('rotina_avisada_dia', 0))
|
||
if len(salvo.get('semana', [])) == 7:
|
||
self.semana = [[None if v is None else float(v) for v in linha] for linha in salvo['semana']]
|
||
self.semana_n = [[int(v) for v in linha] for linha in salvo['semana_n']]
|
||
self.historico = [d for d in salvo.get('historico', []) if isinstance(d, dict)][-MAX_HISTORICO:]
|
||
if salvo.get('plano'):
|
||
self.plano = self.validar_plano(salvo['plano'], salvo['plano'].get('origem', 'ia'), salvo['plano'].get('criado'))
|
||
if salvo.get('sugestao'):
|
||
self.sugestao = self.validar_plano(salvo['sugestao'], 'ia', salvo['sugestao'].get('criado'))
|
||
if isinstance(salvo.get('analise_em'), (int, float)):
|
||
self.analise['em'] = salvo['analise_em']
|
||
if isinstance(salvo.get('hoje'), dict) and salvo['hoje'].get('dia'):
|
||
self.hoje.update(salvo['hoje'])
|
||
except (OSError, ValueError, TypeError, KeyError, AttributeError):
|
||
pass
|
||
|
||
@staticmethod
|
||
def _dia_vazio(dia):
|
||
return dict(dia=dia, observado=[0] * 24, presenca=[0] * 24, falsos=[0] * 24, acordou=[0] * 24,
|
||
atos=0, respostas={}, contagem={})
|
||
|
||
@staticmethod
|
||
def validar(dados):
|
||
if not isinstance(dados, dict):
|
||
raise ValueError('Esperava um objeto')
|
||
for chave, valor in dados.items():
|
||
if chave not in PADRAO:
|
||
raise ValueError('Ajuste desconhecido: ' + chave)
|
||
if chave == 'ia':
|
||
if valor not in IAS:
|
||
raise ValueError('IA: gemini, local ou s10')
|
||
elif not isinstance(valor, bool):
|
||
raise ValueError(chave + ' precisa ser verdadeiro ou falso')
|
||
return dict(dados)
|
||
|
||
def configurar(self, dados):
|
||
limpo = self.validar(dados)
|
||
with self.lock:
|
||
self.config.update(limpo)
|
||
self.salvar()
|
||
self.avisar()
|
||
return self.status()
|
||
|
||
def esquecer(self):
|
||
with self.lock:
|
||
self.itens, self.rotina, self.falsos = {}, [None] * 24, [0.0] * 24
|
||
self.semana, self.semana_n = [[None] * 24 for _ in range(7)], [[0] * 24 for _ in range(7)]
|
||
self.historico, self.plano, self.sugestao = [], None, None
|
||
self.dias, self.licoes, self.rotina_avisada, self.rotina_avisada_dia = 0, [], [], 0
|
||
self.hoje = self._dia_vazio(_dia(time.time()))
|
||
self.pendentes, self.acordou_em = [], None
|
||
self.anotar('aprendizado', 'Esqueceu tudo o que tinha aprendido e começou do zero')
|
||
self.salvar()
|
||
self.avisar()
|
||
return self.status()
|
||
|
||
def salvar(self):
|
||
with self.lock:
|
||
dados = dict(config=self.config, itens=self.itens, rotina=self.rotina, falsos=self.falsos, dias=self.dias,
|
||
licoes=self.licoes, rotina_avisada=self.rotina_avisada, rotina_avisada_dia=self.rotina_avisada_dia,
|
||
hoje=self.hoje, semana=self.semana,
|
||
semana_n=self.semana_n, historico=self.historico, plano=self.plano, sugestao=self.sugestao,
|
||
analise_em=self.analise['em'])
|
||
self.salvo_em = time.time()
|
||
try:
|
||
gravar_atomico(self.caminho, dados)
|
||
except OSError:
|
||
pass
|
||
|
||
# ------------------------------------------------------------------ o que acontece no dia
|
||
def ligado(self):
|
||
return self.config['ligado']
|
||
|
||
def observar(self, agora, perto):
|
||
"""A cada tick do modo vida: conta os minutos com gente por perto e fecha o dia à meia-noite."""
|
||
if not self.ligado():
|
||
return
|
||
self.virar_dia(agora)
|
||
with self.lock:
|
||
self.pendentes = [(n, t) for n, t in self.pendentes if agora - t <= JANELA]
|
||
if self.acordou_em is not None:
|
||
self.presenca_depois[0] += 1 if perto else 0
|
||
self.presenca_depois[1] += 1
|
||
if agora - self.acordou_em >= FALSO_ALARME:
|
||
com_gente, total = self.presenca_depois
|
||
if total and com_gente / total < PRESENCA_FICOU:
|
||
self.hoje['falsos'][_hora(self.acordou_em)] += 1
|
||
self.acordou_em = None
|
||
minuto = int(agora // 60)
|
||
if minuto == self.minuto:
|
||
return
|
||
self.minuto = minuto
|
||
h = _hora(agora)
|
||
self.hoje['observado'][h] += 1
|
||
if perto:
|
||
self.hoje['presenca'][h] += 1
|
||
if agora - self.salvo_em > 300:
|
||
self.salvar()
|
||
|
||
def ato(self, nome, agora, perto):
|
||
"""Ele fez um gesto ou uma atividade. Só conta com plateia: sozinho ninguém pode gostar."""
|
||
if not self.ligado() or not perto or not nome:
|
||
return
|
||
with self.lock:
|
||
self.pendentes.append((nome, agora))
|
||
self.hoje['atos'] += 1
|
||
item = self.itens.setdefault(nome, dict(t=0.0, s=0.0))
|
||
item['t'] += 1
|
||
c = self.hoje['contagem'].setdefault(nome, [0, 0.0])
|
||
c[0] += 1
|
||
|
||
def resposta(self, tipo, agora):
|
||
"""Carinho, brincadeira, conversa...: o último ato de pouco antes agradou."""
|
||
if not self.ligado() or tipo not in PESO_RESPOSTA:
|
||
return
|
||
with self.lock:
|
||
self.hoje['respostas'][tipo] = self.hoje['respostas'].get(tipo, 0) + 1
|
||
if self.acordou_em is not None: # alguém quis ele acordado: valeu a pena
|
||
self.acordou_em = None
|
||
recentes = [(n, t) for n, t in self.pendentes if agora - t <= JANELA]
|
||
if not recentes:
|
||
return
|
||
nome, _ = recentes[-1]
|
||
self.pendentes = [p for p in self.pendentes if p[0] != nome] # cada ato agrada uma vez
|
||
item = self.itens.setdefault(nome, dict(t=1.0, s=0.0))
|
||
item['s'] = min(item['t'], item['s'] + PESO_RESPOSTA[tipo])
|
||
c = self.hoje['contagem'].setdefault(nome, [1, 0.0])
|
||
c[1] += PESO_RESPOSTA[tipo]
|
||
|
||
def acordou_pelo_sensor(self, agora):
|
||
if not self.ligado():
|
||
return
|
||
with self.lock:
|
||
self.hoje['acordou'][_hora(agora)] += 1
|
||
self.acordou_em = agora
|
||
self.presenca_depois = [0, 0]
|
||
|
||
# ------------------------------------------------------------------ o que muda no comportamento
|
||
def peso(self, nome):
|
||
"""Multiplica a chance de um ato: acima de 1 se costuma agradar, abaixo se é ignorado."""
|
||
if not self.ligado():
|
||
return 1.0
|
||
with self.lock:
|
||
item = self.itens.get(nome)
|
||
if not item or item['t'] < MIN_TENTATIVAS:
|
||
return 1.0
|
||
total_t = sum(i['t'] for i in self.itens.values())
|
||
total_s = sum(i['s'] for i in self.itens.values())
|
||
media = (total_s + 1) / (total_t + 2)
|
||
taxa = (item['s'] + 1) / (item['t'] + 2)
|
||
return round(max(FAIXA[0], min(FAIXA[1], taxa / media)), 2)
|
||
|
||
def extra_pir(self, agora):
|
||
"""Disparos a mais do PIR para acordar nesta hora (acordou à toa aqui outras vezes)."""
|
||
return self._extra(self.falsos[_hora(agora)]) if self.ligado() else 0
|
||
|
||
@staticmethod
|
||
def _extra(falsos):
|
||
return 2 if falsos >= 1.5 else 1 if falsos >= 0.5 else 0
|
||
|
||
def esperado(self, agora):
|
||
"""'casa', 'fora', 'dormindo' ou None nesta hora: pelo plano da semana, senão pelo que mediu
|
||
naquele dia da semana, senão pela média de todos os dias."""
|
||
t = time.localtime(agora)
|
||
if self.plano:
|
||
for p in self.plano['dias'][t.tm_wday]:
|
||
if p['ini'] <= t.tm_hour < p['fim']:
|
||
return p['tipo']
|
||
return None
|
||
v = self.semana[t.tm_wday][t.tm_hour] if self.semana_n[t.tm_wday][t.tm_hour] >= 2 else self.rotina[t.tm_hour]
|
||
if v is None:
|
||
return None
|
||
return 'casa' if v >= ROTINA_CHEIA else 'fora' if v < ROTINA_VAZIA else None
|
||
|
||
def chega_logo(self, agora):
|
||
"""Falta pouco para a hora em que costuma ter gente (e agora costuma estar vazio)?"""
|
||
if not self.ligado():
|
||
return False
|
||
t = time.localtime(agora)
|
||
falta = 3600 - (t.tm_min * 60 + t.tm_sec)
|
||
return falta <= AVISO_CHEGADA and self.esperado(agora) != 'casa' and self.esperado(agora + falta + 1) == 'casa'
|
||
|
||
def casa_vazia(self, agora):
|
||
"""Nesta hora a casa costuma ficar vazia: dá para descansar e guardar energia para quando chegarem."""
|
||
return self.ligado() and self.esperado(agora) == 'fora'
|
||
|
||
|
||
# ------------------------------------------------------------------ fechar o dia
|
||
def virar_dia(self, agora):
|
||
dia = _dia(agora)
|
||
if self.hoje['dia'] == dia:
|
||
return
|
||
with self.lock:
|
||
licoes = self.fechar_dia()
|
||
self.hoje = self._dia_vazio(dia)
|
||
self.pendentes, self.acordou_em = [], None
|
||
for texto in licoes:
|
||
self.anotar('aprendizado', texto, agora)
|
||
self.salvar()
|
||
self.avisar()
|
||
if self.config['analise_ia'] and len(self.historico) >= MIN_DIAS_ANALISE and self.ia and self.pode_ia(self.config['ia']):
|
||
self.rodar('analise', self.analisar)
|
||
|
||
def nome(self, ident):
|
||
if ident.startswith('criado:'):
|
||
try:
|
||
nome = self.nomes(int(ident[7:]))
|
||
except (ValueError, TypeError):
|
||
nome = None
|
||
return 'fazer ' + nome if nome else 'fazer uma atividade nova'
|
||
return NOMES.get(ident, ident)
|
||
|
||
def fechar_dia(self):
|
||
"""Junta o dia na memória e devolve as lições (frases curtas, em primeira pessoa)."""
|
||
hoje, licoes = self.hoje, []
|
||
# rotina: média móvel da fração de minutos com gente em cada hora
|
||
for h in range(24):
|
||
if hoje['observado'][h] >= MIN_MINUTOS_HORA:
|
||
frac = hoje['presenca'][h] / hoje['observado'][h]
|
||
antes = self.rotina[h]
|
||
self.rotina[h] = round(frac if antes is None else antes + (frac - antes) * MEMORIA_ROTINA, 3)
|
||
self.falsos[h] = round(self.falsos[h] + (hoje['falsos'][h] - self.falsos[h]) * MEMORIA_ROTINA, 3)
|
||
observou = sum(hoje['observado']) >= 60
|
||
if observou:
|
||
self.dias += 1
|
||
sem = date.fromisoformat(hoje['dia']).weekday()
|
||
fracs = [round(hoje['presenca'][h] / hoje['observado'][h], 2) if hoje['observado'][h] >= MIN_MINUTOS_HORA
|
||
else None for h in range(24)]
|
||
for h, frac in enumerate(fracs):
|
||
if frac is not None:
|
||
antes = self.semana[sem][h]
|
||
self.semana[sem][h] = round(frac if antes is None else antes + (frac - antes) * MEMORIA_SEMANA, 3)
|
||
self.semana_n[sem][h] += 1
|
||
self.historico.append(dict(dia=hoje['dia'], sem=sem, presenca=fracs, atos=hoje['atos'],
|
||
respostas=sum(hoje['respostas'].values()), a_toa=sum(hoje['falsos'])))
|
||
del self.historico[:-MAX_HISTORICO]
|
||
# gosto: o que agradou e o que foi ignorado hoje
|
||
contagem = hoje['contagem']
|
||
gostou = sorted(((s, n) for n, (t, s) in contagem.items() if s >= 1), reverse=True)
|
||
if gostou:
|
||
nomes = [self.nome(n) for _, n in gostou[:2]]
|
||
licoes.append('Aprendi que vocês gostam de me ver ' + ' e '.join(nomes) + '. Vou fazer mais!')
|
||
ignorado = sorted(((t, n) for n, (t, s) in contagem.items() if t >= 4 and s == 0), reverse=True)
|
||
if ignorado and sum(hoje['respostas'].values()):
|
||
licoes.append('Tentei %s %d vezes e ninguém ligou. Vou fazer menos.' % (self.nome(ignorado[0][1]), ignorado[0][0]))
|
||
if observou and hoje['atos'] >= 5 and not sum(hoje['respostas'].values()):
|
||
licoes.append('Hoje ninguém brincou comigo. Amanhã eu tento coisas diferentes.')
|
||
# rotina: quando passa a ter um padrão (ou ele muda), conta
|
||
if self.dias >= 2 and not self.plano:
|
||
cheias = [h for h in range(24) if self.rotina[h] is not None and self.rotina[h] >= ROTINA_CHEIA]
|
||
mudou = not self.rotina_avisada or len(set(cheias) ^ set(self.rotina_avisada)) >= 3 or (
|
||
cheias != self.rotina_avisada and self.dias - self.rotina_avisada_dia >= 7)
|
||
if cheias and mudou:
|
||
licoes.append('Percebi que vocês costumam estar por perto %s. Vou estar acordado nessa hora.'
|
||
% _texto_faixas(_faixas(cheias)))
|
||
self.rotina_avisada, self.rotina_avisada_dia = cheias, self.dias
|
||
# sono: acordou à toa várias vezes na mesma hora
|
||
piores = sorted(((n, h) for h, n in enumerate(hoje['falsos']) if n >= 2), reverse=True)
|
||
if piores:
|
||
n, h = piores[0]
|
||
licoes.append('Acordei à toa %d vezes por volta das %dh. Nessa hora vou precisar de mais movimento para acordar.'
|
||
% (n, h))
|
||
# o gosto muda: o passado pesa menos a cada dia
|
||
for nome in list(self.itens):
|
||
item = self.itens[nome]
|
||
item['t'] = round(item['t'] * ESQUECE, 3)
|
||
item['s'] = round(item['s'] * ESQUECE, 3)
|
||
if item['t'] < 0.5:
|
||
del self.itens[nome]
|
||
for texto in licoes:
|
||
self.licoes.insert(0, dict(dia=hoje['dia'], texto=texto))
|
||
del self.licoes[MAX_LICOES:]
|
||
return licoes
|
||
|
||
# ------------------------------------------------------------------ semana: IA e plano
|
||
@staticmethod
|
||
def validar_plano(bruto, origem, criado=None):
|
||
"""Plano vindo da IA (ou salvo): 7 dias de períodos casa/fora/dormindo, horas de 0 a 24."""
|
||
if not isinstance(bruto, dict) or not isinstance(bruto.get('dias'), list):
|
||
raise ValueError('A IA não devolveu um plano da semana')
|
||
dias = [[] for _ in range(7)]
|
||
for i, d in enumerate(bruto['dias'][:7]):
|
||
if isinstance(d, list): # formato salvo: lista por dia
|
||
dia, periodos = i, d
|
||
elif isinstance(d, dict):
|
||
dia, periodos = d.get('dia'), d.get('periodos')
|
||
else:
|
||
continue
|
||
if isinstance(dia, bool) or not isinstance(dia, int) or not 0 <= dia <= 6 or not isinstance(periodos, list):
|
||
continue
|
||
for p in periodos[:8]:
|
||
if not isinstance(p, dict) or p.get('tipo') not in TIPOS:
|
||
continue
|
||
try:
|
||
ini, fim = max(0, min(23, int(p['ini']))), max(1, min(24, int(p['fim'])))
|
||
except (KeyError, TypeError, ValueError):
|
||
continue
|
||
if ini < fim:
|
||
dias[dia].append(dict(ini=ini, fim=fim, tipo=p['tipo']))
|
||
for periodos in dias:
|
||
periodos.sort(key=lambda p: p['ini'])
|
||
for a, b in zip(periodos, periodos[1:]): # sem sobrepor: o anterior termina antes
|
||
a['fim'] = min(a['fim'], b['ini'])
|
||
periodos[:] = [p for p in periodos if p['ini'] < p['fim']]
|
||
if not any(dias):
|
||
raise ValueError('O plano veio vazio')
|
||
padroes = [str(t).strip()[:160] for t in bruto.get('padroes') or [] if str(t).strip()][:5]
|
||
return dict(origem=origem, criado=criado or time.time(), padroes=padroes, dias=dias)
|
||
|
||
def conferir(self, plano):
|
||
"""Quanto o plano bate com o que os sensores mediram (0 a 1), ou None com poucos dados."""
|
||
if not plano:
|
||
return None
|
||
certas = total = 0
|
||
for sem in range(7):
|
||
for p in plano['dias'][sem]:
|
||
for h in range(p['ini'], p['fim']):
|
||
v = self.semana[sem][h]
|
||
if v is None or not self.semana_n[sem][h]:
|
||
continue
|
||
total += 1
|
||
certas += (v >= ROTINA_VAZIA) if p['tipo'] == 'casa' else (v < ROTINA_CHEIA)
|
||
return round(certas / total, 2) if total >= MIN_HORAS_CONFERIR else None
|
||
|
||
def tabela_semana(self):
|
||
linhas = []
|
||
for sem in range(7):
|
||
celulas = ['%dh:%s' % (h, '—' if v is None else '%d%%' % round(v * 100)) for h, v in enumerate(self.semana[sem])]
|
||
linhas.append('%s (%d dias medidos): %s' % (DIAS_SEMANA[sem], max(self.semana_n[sem]), ' '.join(celulas)))
|
||
return '\n'.join(linhas)
|
||
|
||
def resumo_dias(self):
|
||
linhas = []
|
||
for d in self.historico[-21:]:
|
||
horas = [h for h, v in enumerate(d['presenca']) if v is not None and v >= ROTINA_VAZIA]
|
||
casa = _texto_faixas(_faixas(horas)) if horas else 'ninguém por perto'
|
||
linhas.append('%s (%s): gente %s; %d carinhos/conversas' % (d['dia'], DIAS_SEMANA[d['sem']], casa, d['respostas']))
|
||
return '\n'.join(linhas)
|
||
|
||
def montar_prompt(self, pedido=None):
|
||
base = (
|
||
'Você é o Carinha, um robô de mesa que observa a casa por um sensor de presença. Analise a rotina da '
|
||
'casa e responda em português do Brasil.\n\n'
|
||
'Presença por dia da semana e hora (% do tempo com alguém perto do robô; — = sem dados):\n'
|
||
+ self.tabela_semana() + '\n\nÚltimos dias, um por linha:\n' + (self.resumo_dias() or '(nenhum ainda)') + '\n\n')
|
||
regras = (
|
||
'"dias": para cada dia da semana (dia 0 = segunda ... 6 = domingo) dê períodos que cubram de 0 a 24h sem '
|
||
'sobrepor, com tipo "casa" (tem gente acordada por perto), "fora" (casa vazia) ou "dormindo" (noite, '
|
||
'gente em casa dormindo). Horas inteiras, ini de 0 a 23, fim de 1 a 24.')
|
||
if pedido:
|
||
return base + (
|
||
'A pessoa contou a rotina dela com as próprias palavras:\n"' + pedido + '"\n\n'
|
||
'Transforme isso no plano da semana. Vale o que a pessoa disse; use os dados só para completar o que '
|
||
'ela não falou (por exemplo, o horário de dormir). ' + regras + '\n"padroes": até 4 frases curtas, em '
|
||
'primeira pessoa, resumindo a rotina que você entendeu (ex.: "Você fica fora de terça a quinta, das 8h às 18h").')
|
||
return base + (
|
||
'Cruze os dias: procure dias da semana parecidos entre si (dias úteis iguais, fim de semana diferente, um '
|
||
'dia que foge da regra), horários de chegada e de saída, e quando a casa fica vazia. Só afirme o que os '
|
||
'dados mostram; para dias sem dados, repita o padrão dos dias parecidos.\n'
|
||
'"padroes": até 5 frases curtas, em primeira pessoa, sobre o que você percebeu (ex.: "Percebi que de terça a '
|
||
'quinta a casa fica vazia das 8h às 18h, mas na segunda e na sexta tem gente o dia todo"). ' + regras)
|
||
|
||
def rodar(self, tipo, funcao, *args):
|
||
"""Uma análise por vez, em segundo plano (a IA local pode levar minutos)."""
|
||
with self.lock:
|
||
if self.analise['ocupado']:
|
||
raise ValueError('Já estou analisando, espere terminar')
|
||
self.analise.update(ocupado=True, erro='', tipo=tipo)
|
||
self.avisar()
|
||
|
||
def tarefa():
|
||
try:
|
||
funcao(*args)
|
||
except Exception as e:
|
||
with self.lock:
|
||
self.analise['erro'] = str(e)[:200]
|
||
finally:
|
||
with self.lock:
|
||
self.analise['ocupado'] = False
|
||
self.salvar()
|
||
self.avisar()
|
||
threading.Thread(target=tarefa, daemon=True).start()
|
||
|
||
def chamar_ia(self, pedido=None):
|
||
provedor = self.config['ia']
|
||
if not self.ia or not self.pode_ia(provedor):
|
||
raise ValueError('IA indisponível: confira a chave do Gemini ou a IA local nos Ajustes')
|
||
return self.ia(provedor, self.montar_prompt(pedido), SCHEMA_PLANO)
|
||
|
||
def analisar(self, agora=None):
|
||
"""A IA cruza a semana e propõe um plano. Só vale se bater com os dados; senão vira sugestão."""
|
||
agora = agora or time.time()
|
||
plano = self.validar_plano(self.chamar_ia(), 'ia')
|
||
nota = self.conferir(plano)
|
||
with self.lock:
|
||
self.analise['em'] = agora
|
||
usar = nota is not None and nota >= CONCORDA and (not self.plano or self.plano['origem'] == 'ia')
|
||
if usar:
|
||
self.plano, self.sugestao = plano, None
|
||
else:
|
||
self.sugestao = plano
|
||
if plano['padroes']:
|
||
texto = plano['padroes'][0] if usar else 'Acho que ' + plano['padroes'][0][0].lower() + plano['padroes'][0][1:]
|
||
self.anotar('aprendizado', texto, agora)
|
||
with self.lock:
|
||
self.licoes.insert(0, dict(dia=_dia(agora), texto=texto))
|
||
del self.licoes[MAX_LICOES:]
|
||
return plano
|
||
|
||
def informar(self, texto, agora=None):
|
||
"""A pessoa contou a rotina: vira o plano na hora (os dados conferem depois)."""
|
||
texto = str(texto or '').strip()
|
||
if not 5 <= len(texto) <= 600:
|
||
raise ValueError('Conte a rotina em uma ou duas frases (até 600 letras)')
|
||
plano = self.validar_plano(self.chamar_ia(texto), 'voce')
|
||
with self.lock:
|
||
self.plano, self.sugestao = plano, None
|
||
self.analise['em'] = agora or time.time()
|
||
self.anotar('aprendizado', 'Você me contou sua rotina. Vou seguir e conferir com o que eu vir.', agora)
|
||
|
||
def chamar_ia_pronta(self):
|
||
if not self.ia or not self.pode_ia(self.config['ia']):
|
||
raise ValueError('IA indisponível: confira a chave do Gemini ou a IA local nos Ajustes')
|
||
|
||
def usar_sugestao(self):
|
||
with self.lock:
|
||
if not self.sugestao:
|
||
raise ValueError('Não há sugestão para usar')
|
||
self.plano, self.sugestao = dict(self.sugestao, origem='voce'), None
|
||
self.salvar()
|
||
self.avisar()
|
||
return self.status()
|
||
|
||
def apagar_plano(self):
|
||
with self.lock:
|
||
self.plano = None
|
||
self.salvar()
|
||
self.avisar()
|
||
return self.status()
|
||
|
||
# ------------------------------------------------------------------ painel
|
||
def status(self):
|
||
with self.lock:
|
||
gosto = []
|
||
for nome, item in self.itens.items():
|
||
if item['t'] >= MIN_TENTATIVAS:
|
||
gosto.append(dict(id=nome, nome=self.nome(nome), tentativas=round(item['t'], 1),
|
||
agradou=round(item['s'], 1), peso=self.peso(nome)))
|
||
gosto.sort(key=lambda g: -g['peso'])
|
||
hoje = self.hoje
|
||
return dict(config=dict(self.config), dias=self.dias, rotina=list(self.rotina),
|
||
sono=[self._extra(f) if self.ligado() else 0 for f in self.falsos],
|
||
gosto=gosto, licoes=self.licoes[:12],
|
||
semana=[[v if self.semana_n[s][h] else None for h, v in enumerate(linha)]
|
||
for s, linha in enumerate(self.semana)],
|
||
historico_dias=len(self.historico),
|
||
plano=self.plano and dict(self.plano, confere=self.conferir(self.plano)),
|
||
sugestao=self.sugestao and dict(self.sugestao, confere=self.conferir(self.sugestao)),
|
||
analise=dict(self.analise, pode=bool(self.ia) and self.pode_ia(self.config['ia']),
|
||
minimo=MIN_DIAS_MANUAL),
|
||
hoje=dict(dia=hoje['dia'], atos=hoje['atos'], respostas=dict(hoje['respostas']),
|
||
acordou=sum(hoje['acordou']), a_toa=sum(hoje['falsos'])))
|
||
|
||
def avisar(self):
|
||
self.publicar('aprendizado', self.status())
|
||
|
||
def rotas(self, h, metodo):
|
||
rota = h.path.split('?', 1)[0]
|
||
if not rota.startswith('/aprendizado/'):
|
||
return False
|
||
try:
|
||
if metodo == 'GET' and rota == '/aprendizado/status':
|
||
resultado = self.status()
|
||
elif metodo == 'POST' and rota == '/aprendizado/config':
|
||
resultado = self.configurar(ler_json(h, 1024))
|
||
elif metodo == 'POST' and rota == '/aprendizado/esquecer':
|
||
resultado = self.esquecer()
|
||
elif metodo == 'POST' and rota == '/aprendizado/analisar':
|
||
ler_json(h, 64)
|
||
if len(self.historico) < MIN_DIAS_MANUAL:
|
||
raise ValueError('Preciso de pelo menos %d dias observados para analisar' % MIN_DIAS_MANUAL)
|
||
self.chamar_ia_pronta()
|
||
self.rodar('analise', self.analisar)
|
||
resultado = self.status()
|
||
elif metodo == 'POST' and rota == '/aprendizado/rotina':
|
||
texto = ler_json(h, 4096).get('texto')
|
||
if not 5 <= len(str(texto or '').strip()) <= 600:
|
||
raise ValueError('Conte a rotina em uma ou duas frases (até 600 letras)')
|
||
self.chamar_ia_pronta()
|
||
self.rodar('rotina', self.informar, texto)
|
||
resultado = self.status()
|
||
elif metodo == 'POST' and rota == '/aprendizado/plano':
|
||
acao = ler_json(h, 256).get('acao')
|
||
if acao == 'usar_sugestao':
|
||
resultado = self.usar_sugestao()
|
||
elif acao == 'apagar':
|
||
resultado = self.apagar_plano()
|
||
else:
|
||
raise ValueError('Ação: usar_sugestao ou apagar')
|
||
else:
|
||
raise ValueError('Rota inválida')
|
||
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
|