julioandClaude Opus 5.5
673dea62e8
- Controle: opt-in de movimento fisico, botao "segure para mover" (mouse, toque e teclado), sliders das articulacoes tambem so movem enquanto pressionados; velocidade e interpolacao ficam no servidor - sequencias prontas e posturas salvas: previa so na tela ou execucao segurando o botao ate o fim; soltar interrompe na hora - posturas com nome inline (sem prompt), exclusao confirmada inline, exportar/importar; posturas aparecem na paleta (Ctrl+K) - Calibracao: testes de fim de curso por canal, marcar limites e tabela editavel com validacao antes de salvar - Diagnostico (antigo Servos): ao vivo limitado a 10/s, bloqueado quando o braco esta com os motores - desenho esquematico do braco em SVG com previa animada - servidor: evento 'motores' tambem ao ligar canais; renovar no alvo devolve movendo=false para a sequencia avancar - aba escondida ou E-Stop soltam o braco e desativam o movimento Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
235 lines
11 KiB
Python
235 lines
11 KiB
Python
"""Servidor base: token, /eventos (SSE), /estado, /comando, E-Stop, trava de hardware e fila."""
|
|
import http.client
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
spec = importlib.util.spec_from_file_location('kiosk', ROOT / 'kiosk-server.py')
|
|
kiosk = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(kiosk)
|
|
from braco import Braco
|
|
from comum import carregar_token
|
|
from servos import ControleServos
|
|
|
|
TOKEN = 'segredo-de-teste'
|
|
|
|
|
|
class HW:
|
|
def __init__(self): self.calls = []
|
|
def pulse(self, ch, us): self.calls.append((ch, us))
|
|
def off(self, ch): self.calls.append((ch, None))
|
|
|
|
|
|
class ServidorTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
pasta = Path(self.tmp.name)
|
|
self.hw = HW()
|
|
self.server = kiosk.Servidor(('127.0.0.1', 0), kiosk.Handler)
|
|
self.server.token = TOKEN
|
|
self.server.controle = kiosk.ControlePC(pasta / 'app.json')
|
|
self.server.servos = ControleServos(pasta / 'servos.json', self.hw)
|
|
self.server.braco = Braco(pasta / 'braco.json', self.server.servos)
|
|
kiosk.ligar_eventos(self.server)
|
|
kiosk.FILA.limpar()
|
|
kiosk.ESTOP.update(ativo=False, desde=None)
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
self.porta = self.server.server_port
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.server.servos.close()
|
|
self.thread.join()
|
|
self.tmp.cleanup()
|
|
|
|
def pedir(self, metodo, rota, dados=None, token=TOKEN, cabecalhos=None):
|
|
conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5)
|
|
h = {'Content-Type': 'application/json'}
|
|
if token:
|
|
h['Authorization'] = 'Bearer ' + token
|
|
h.update(cabecalhos or {})
|
|
corpo = None if dados is None else json.dumps(dados)
|
|
conn.request(metodo, rota, body=corpo, headers=h)
|
|
r = conn.getresponse()
|
|
texto = r.read()
|
|
conn.close()
|
|
return r.status, json.loads(texto) if texto else None
|
|
|
|
# ------------------------------------------------------------- token
|
|
def test_rotas_de_controle_exigem_token(self):
|
|
for metodo, rota, dados in [('POST', '/braco/mover', {'angulos': [90] * 7}),
|
|
('POST', '/comando', {'tipo': 'gesto', 'valor': 'feliz'}),
|
|
('POST', '/parar-tudo', {}),
|
|
('GET', '/estado', None), ('GET', '/fila?aparelho=1', None),
|
|
('GET', '/app/status', None), ('GET', '/gesto/feliz', None)]:
|
|
for token in (None, 'errado'):
|
|
codigo, corpo = self.pedir(metodo, rota, dados, token=token)
|
|
self.assertEqual(codigo, 401, (metodo, rota, token))
|
|
self.assertEqual(self.hw.calls, [])
|
|
self.assertEqual(kiosk.FILA.ordens(), [])
|
|
|
|
def test_rotas_publicas_e_token_na_query(self):
|
|
self.assertEqual(self.pedir('GET', '/gestos', token=None)[0], 200)
|
|
self.assertEqual(self.pedir('GET', '/app/health', token=None)[0], 200)
|
|
self.assertEqual(self.pedir('GET', '/estado?t=' + TOKEN, token=None)[0], 200)
|
|
|
|
def test_sem_token_configurado_fica_aberto(self):
|
|
self.server.token = None
|
|
self.assertEqual(self.pedir('GET', '/estado', token=None)[0], 200)
|
|
|
|
def test_carregar_token(self):
|
|
pasta = Path(self.tmp.name)
|
|
self.assertIsNone(carregar_token(pasta))
|
|
(pasta / 'seguranca.json').write_text(json.dumps({'token': 'abcdefgh123'}))
|
|
self.assertEqual(carregar_token(pasta), 'abcdefgh123')
|
|
(pasta / 'seguranca.json').write_text(json.dumps({'token': 'curto'}))
|
|
with self.assertRaises(ValueError):
|
|
carregar_token(pasta)
|
|
|
|
def test_origem_estrangeira_recusada(self):
|
|
codigo, _ = self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'feliz'},
|
|
cabecalhos={'Origin': 'http://malicioso.example'})
|
|
self.assertEqual(codigo, 400)
|
|
|
|
# ------------------------------------------------------------- comando e fila
|
|
def test_comando_valida_e_enfileira(self):
|
|
self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'feliz'})[0], 200)
|
|
self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'rosto', 'valor': 'miudo'})[0], 200)
|
|
self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'rm -rf'})[0], 400)
|
|
self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'rosto', 'valor': 'nada'})[0], 400)
|
|
self.assertEqual(kiosk.FILA.ordens(), ['feliz', 'rosto miudo'])
|
|
self.assertEqual(self.pedir('GET', '/fila?aparelho=1')[1], ['feliz', 'rosto miudo'])
|
|
self.assertEqual(self.pedir('GET', '/fila?aparelho=1')[1], [])
|
|
|
|
def test_post_face_usa_prefixo_rosto(self):
|
|
conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5)
|
|
conn.request('POST', '/face', body='miudo', headers={'Authorization': 'Bearer ' + TOKEN})
|
|
self.assertEqual(conn.getresponse().status, 200)
|
|
conn.close()
|
|
self.assertEqual(kiosk.FILA.ordens(), ['rosto miudo'])
|
|
|
|
def test_fila_limitada_e_com_validade(self):
|
|
fila = kiosk.Fila(maximo=3, validade=10)
|
|
for i in range(5):
|
|
fila.adicionar(f'g{i}', agora=0)
|
|
self.assertEqual(fila.ordens(), ['g2', 'g3', 'g4'])
|
|
self.assertEqual(fila.consumir(agora=11), [])
|
|
fila.adicionar('recarregar'); fila.adicionar('recarregar'); fila.adicionar('feliz')
|
|
self.assertEqual(fila.ordens(), ['recarregar', 'feliz'])
|
|
# Rosto no navegador não rouba o 'recarregar' do celular.
|
|
self.assertEqual(fila.consumir(aparelho=False), ['feliz'])
|
|
self.assertEqual(fila.consumir(aparelho=True), ['recarregar'])
|
|
|
|
# ------------------------------------------------------------- SSE
|
|
def test_eventos_entrega_comando(self):
|
|
conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5)
|
|
conn.request('GET', '/eventos?t=' + TOKEN)
|
|
r = conn.getresponse()
|
|
self.assertEqual(r.status, 200)
|
|
self.assertIn('text/event-stream', r.getheader('Content-Type'))
|
|
linhas = []
|
|
while not linhas or linhas[-1] != '':
|
|
linhas.append(r.fp.readline().decode().rstrip('\n'))
|
|
self.assertIn('event: pronto', linhas)
|
|
self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'coracao'})
|
|
bloco = []
|
|
while not bloco or bloco[-1] != '':
|
|
bloco.append(r.fp.readline().decode().rstrip('\n'))
|
|
conn.close()
|
|
self.assertIn('event: expressao', bloco)
|
|
self.assertIn('data: {"ordem": "coracao"}', bloco)
|
|
|
|
def test_estado_agrega(self):
|
|
codigo, estado = self.pedir('GET', '/estado')
|
|
self.assertEqual(codigo, 200)
|
|
for chave in ('versao', 'seq', 'estop', 'aparencia', 'gestos', 'app', 'motores'):
|
|
self.assertIn(chave, estado)
|
|
self.assertTrue(estado['autenticacao'])
|
|
|
|
# ------------------------------------------------------------- E-Stop e trava
|
|
def test_parar_tudo_desliga_e_trava_ate_rearmar(self):
|
|
self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 200)
|
|
self.hw.calls.clear()
|
|
codigo, corpo = self.pedir('POST', '/parar-tudo', {})
|
|
self.assertEqual(codigo, 200)
|
|
self.assertTrue(corpo['estop']['ativo'])
|
|
self.assertEqual({c for c, us in self.hw.calls if us is None}, set(range(16)))
|
|
self.assertEqual(kiosk.FILA.ordens(), ['calma'])
|
|
self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 409)
|
|
self.assertEqual(self.pedir('POST', '/servos/mover', {'indice': 0, 'pulso': 1500})[0], 409)
|
|
self.assertEqual(self.pedir('POST', '/rearmar', {})[0], 200)
|
|
self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 200)
|
|
|
|
def test_braco_e_diagnostico_nao_disputam_motores(self):
|
|
self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 200)
|
|
codigo, corpo = self.pedir('POST', '/servos/mover', {'indice': 0, 'pulso': 1500})
|
|
self.assertEqual(codigo, 409)
|
|
self.assertIn('braco', corpo['erro'])
|
|
self.pedir('POST', '/braco/parar', {})
|
|
self.assertEqual(self.pedir('POST', '/servos/mover', {'indice': 0, 'pulso': 1500})[0], 200)
|
|
|
|
|
|
class BracoMovimentoTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.hw = HW()
|
|
self.servos = ControleServos(Path(self.tmp.name) / 's.json', self.hw)
|
|
self.servos.closed.set() # sem threads: os passos são chamados à mão
|
|
self.arm = Braco(Path(self.tmp.name) / 'b.json', self.servos)
|
|
cal = self.arm.status()
|
|
for row in cal['calibracao']:
|
|
row.update(minimo=500, maximo=2500, centro=1500)
|
|
self.arm.save(dict(calibracao=cal['calibracao'], posturas=[]))
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_interpolacao_no_servidor_respeita_velocidade(self):
|
|
self.arm.move([90] * 7)
|
|
resposta = self.arm.move([90, 90, 90, 90, 90, 90, 130], velocidade=100, ttl_ms=2000)
|
|
self.assertTrue(resposta['movendo'])
|
|
agora = self.arm.lease - 1.5
|
|
self.arm.passo(agora, .1) # 100 °/s * 0,1 s = 10°
|
|
self.assertEqual(self.arm.atual[6], 100)
|
|
for i in range(10):
|
|
self.arm.passo(agora + .1 * (i + 1), .1)
|
|
self.assertEqual(self.arm.atual[6], 130)
|
|
self.assertIsNone(self.arm.alvo)
|
|
|
|
def test_lease_vencido_interrompe_movimento(self):
|
|
self.arm.move([90] * 7)
|
|
self.arm.move([90] * 6 + [170], velocidade=50, ttl_ms=300)
|
|
self.arm.passo(self.arm.lease + .01, .1)
|
|
self.assertIsNone(self.arm.alvo)
|
|
self.assertEqual(self.arm.atual[6], 90)
|
|
# O watchdog desliga os canais no mesmo prazo.
|
|
self.servos.expire(self.arm.lease + .01)
|
|
self.assertEqual(self.servos.active, {})
|
|
self.assertIsNone(self.servos.dono)
|
|
|
|
def test_renovar_no_alvo_informa_que_chegou(self):
|
|
self.arm.move([90] * 7)
|
|
self.hw.calls.clear()
|
|
r = self.arm.move([90] * 7, velocidade=60, ttl_ms=400)
|
|
self.assertFalse(r['movendo'])
|
|
self.assertIsNone(self.arm.alvo)
|
|
self.assertEqual(len(self.hw.calls), 8) # reenergiza os 8 canais com o novo prazo
|
|
|
|
def test_parametros_invalidos(self):
|
|
for kwargs in (dict(velocidade=1000), dict(velocidade=True), dict(ttl_ms=50), dict(ttl_ms=5000)):
|
|
with self.assertRaises(ValueError):
|
|
self.arm.move([90] * 7, **kwargs)
|
|
self.assertEqual(self.hw.calls, [])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|