"""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 MAX_DECISOES = 800 # decisões guardadas (com o motivo), umas duas semanas AVISO_SONO = 10 * 60 # no máximo uma nota de "não acordou com o sensor" a cada 10 min 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 self.decisoes = [] # [{horario, tipo, texto, motivo}] mais nova primeiro self.sono_avisado = 0.0 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:] self.decisoes = [d for d in salvo.get('decisoes', []) if isinstance(d, dict)][:MAX_DECISOES] 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, self.decisoes = [], 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'], decisoes=self.decisoes) 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.""" return self.esperado_e_fonte(agora)[0] def esperado_e_fonte(self, agora): t = time.localtime(agora) if self.plano: fonte = 'pelo plano que você contou' if self.plano['origem'] == 'voce' else 'pelo plano da IA' for p in self.plano['dias'][t.tm_wday]: if p['ini'] <= t.tm_hour < p['fim']: return p['tipo'], fonte return None, fonte if self.semana_n[t.tm_wday][t.tm_hour] >= 2: v, fonte = self.semana[t.tm_wday][t.tm_hour], 'pelo que vi nas %ss' % DIAS_SEMANA[t.tm_wday].replace('-feira', '') if t.tm_wday >= 5: fonte = 'pelo que vi nos %ss' % DIAS_SEMANA[t.tm_wday] else: v, fonte = self.rotina[t.tm_hour], 'pela média de todos os dias' if v is None: return None, fonte return ('casa' if v >= ROTINA_CHEIA else 'fora' if v < ROTINA_VAZIA else None), fonte # ------------------------------------------------------------------ decisões (com o porquê) def decidir(self, tipo, texto, motivo, agora=None): """Guarda uma decisão tomada com base no que aprendeu, para o painel mostrar dia a dia.""" with self.lock: self.decisoes.insert(0, dict(horario=agora or time.time(), tipo=tipo, texto=texto, motivo=motivo)) del self.decisoes[MAX_DECISOES:] def porque_chegada(self, agora): t = time.localtime(agora + 3600) _, fonte = self.esperado_e_fonte(agora + 3600) return 'às %dh de %s costuma ter gente, %s' % (t.tm_hour, DIAS_SEMANA[t.tm_wday], fonte) def porque_vazia(self, agora): t = time.localtime(agora) _, fonte = self.esperado_e_fonte(agora) return 'às %dh de %s a casa costuma ficar vazia, %s' % (t.tm_hour, DIAS_SEMANA[t.tm_wday], fonte) def segurou_sono(self, agora, precisa, extra): """O PIR viu algo, mas nesta hora ele pede mais movimento: anota (sem repetir toda hora).""" if agora - self.sono_avisado < AVISO_SONO: return self.sono_avisado = agora self.decidir('sono', 'Não acordou com o sensor', 'por volta das %dh já acordou à toa outras vezes: pede %d %s a mais' % (_hora(agora), extra, 'movimento' if extra == 1 else 'movimentos'), agora) def escolheu(self, nome, agora=None): """Escolheu um ato que costuma agradar: anota quando o gosto pesou de verdade.""" peso = self.peso(nome) if peso >= 1.3: self.decidir('gosto', 'Escolheu ' + self.nome(nome), 'costuma agradar: chance ×%s' % ('%.1f' % peso).replace('.', ','), agora) 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] horas = sum(1 for v in fracs if v is not None and v >= ROTINA_VAZIA) nota = self.conferir_dia(sem, fracs) fim_do_dia = time.mktime(time.strptime(hoje['dia'] + ' 23:59:59', '%Y-%m-%d %H:%M:%S')) self.decisoes.insert(0, dict( horario=fim_do_dia, tipo='dia', texto='Fechou o dia: atualizou o quadro de ' + DIAS_SEMANA[sem], motivo='%d %s com gente, %d %s%s' % ( horas, 'hora' if horas == 1 else 'horas', sum(hoje['respostas'].values()), 'carinho ou conversa' if sum(hoje['respostas'].values()) == 1 else 'carinhos ou conversas', '' if nota is None else '; o dia conferiu %d%% com o plano' % round(nota * 100)))) # 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 conferir_dia(self, sem, fracs, plano=None): """Quanto um dia medido bateu com o plano (0 a 1), ou None sem plano ou com poucas horas.""" plano = plano or self.plano if not plano: return None certas = total = 0 for p in plano['dias'][sem]: for h in range(p['ini'], p['fim']): if fracs[h] is None: continue total += 1 certas += (fracs[h] >= ROTINA_VAZIA) if p['tipo'] == 'casa' else (fracs[h] < ROTINA_CHEIA) return round(certas / total, 2) if total >= 6 else None def dia_a_dia(self, dia=None): """Um dia para o painel: o que viu, o que esperava pelo plano, quanto conferiu e as decisões.""" hoje = self.hoje['dia'] dia = dia or hoje try: d = date.fromisoformat(dia) except (TypeError, ValueError): raise ValueError('Dia inválido (use AAAA-MM-DD)') from None sem = d.weekday() with self.lock: if dia == hoje: h = self.hoje fracs = [round(h['presenca'][i] / h['observado'][i], 2) if h['observado'][i] >= 5 else None for i in range(24)] resumo = dict(atos=h['atos'], respostas=sum(h['respostas'].values()), a_toa=sum(h['falsos'])) else: registro = next((x for x in self.historico if x['dia'] == dia), None) fracs = registro['presenca'] if registro else [None] * 24 resumo = dict(atos=registro['atos'], respostas=registro['respostas'], a_toa=registro['a_toa']) if registro else None decisoes = [x for x in self.decisoes if _dia(x['horario']) == dia] dias = sorted({x['dia'] for x in self.historico} | {_dia(x['horario']) for x in self.decisoes} | {hoje}) plano = self.plano horas = sum(1 for v in fracs if v is not None and v >= ROTINA_VAZIA) return dict(dia=dia, sem=sem, nome_dia=DIAS_SEMANA[sem], hoje=dia == hoje, medido=fracs, plano=plano['dias'][sem] if plano else None, origem_plano=plano['origem'] if plano else None, confere=self.conferir_dia(sem, fracs), horas_com_gente=horas, resumo=resumo, decisoes=decisoes, dias=dias[-60:]) 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 usar: motivo = 'plano adotado: confere %d%% com o que mediu' % round(nota * 100) elif nota is None: motivo = 'ficou como sugestão: ainda poucos dados para conferir' elif self.plano and self.plano['origem'] == 'voce' and nota >= CONCORDA: motivo = 'ficou como sugestão: a rotina que você contou continua valendo' else: motivo = 'ficou como sugestão: confere só %d%% (precisa de %d%%)' % (round(nota * 100), round(CONCORDA * 100)) self.decidir('ia', 'IA analisou %d dias' % len(self.historico), motivo, agora) 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) self.decidir('voce', 'Passou a seguir a rotina que você contou', '"%s"' % texto[:140], 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), decisoes_hoje=sum(1 for x in self.decisoes if _dia(x['horario']) == self.hoje['dia']), 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 == 'GET' and rota == '/aprendizado/dia': from urllib.parse import parse_qs, urlsplit resultado = self.dia_a_dia((parse_qs(urlsplit(h.path).query).get('dia') or [None])[0]) 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