#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Sistema de Atendimento Integral - Script de Integração Raspberry Pi GPIO ------------------------------------------------------------------------ Este script conecta pinos físicos da Raspberry Pi à aplicação web em PHP. Ele executa duas tarefas principais simultaneamente (Multithreading): 1. ENTRADAS (Botões Físicos): - Botão A (Pino 17): Pede ao servidor PHP para gerar/imprimir uma nova senha Normal. - Botão B (Pino 27): Pede ao servidor PHP para gerar/imprimir uma nova senha Prioritária. - Botão C (Pino 22): Acoplado ao guichê do operador para chamar "Próximo Normal". - Botão D (Pino 23): Acoplado ao guichê do operador para chamar "Próximo Prioritário". 2. SAÍDAS (Alerta Visual, Sonoro e Voz TTS): - Monitora o estado da API (Long Polling) de 1 em 1 segundo. - Ao detectar uma nova chamada de senha: * Aciona um Relé/LED (Pino 18) piscando de forma intermitente para chamar atenção. * Toca um sinal sonoro (Buzzer no Pino 24) no ritmo do carrilhão médico. * Executa opcionalmente uma chamada falada (Text-to-Speech) em áudio analógico: Ex: "Atenção: Senha Normal zero doze, dirija-se ao Guichê zero três." Instalação de Dependências na Raspberry Pi: ------------------------------------------- $ sudo apt-get update $ sudo apt-get install python3-pip python3-rpi.gpio mpg123 espeak -y $ pip3 install requests """ import time import sys import threading import subprocess try: import RPi.GPIO as GPIO except ImportError: print("\n[ALERTA] Biblioteca RPi.GPIO não encontrada!") print("Para testes em computadores comuns, simulando o comportamento de botões físicos.") print("Execute diretamente na Raspberry Pi para o controle de hardware nativo.\n") # Mock do GPIO para não quebrar a execução fora da Raspberry Pi class GPIOMock: BCM = 'BCM' IN = 'IN' OUT = 'OUT' PUD_UP = 'PUD_UP' FALLING = 'FALLING' def setmode(self, mode): pass def setup(self, pin, mode, pull_up_down=None, initial=None): pass def output(self, pin, state): pass def add_event_detect(self, pin, edge, callback, bouncetime=0): pass def cleanup(self): pass GPIO = GPIOMock() import requests # ========================================== # CONFIGURAÇÕES DA INTEGRAÇÃO # ========================================== # Altere esta URL com o endereço IP ou domínio real da sua hospedagem do painel em PHP API_BASE_URL = "http://SEU_IP_OU_DOMINIO/php-mysql-code" # MAPEAMENTO DE HARDWARE (GPIO no padrão BCM da Raspberry Pi) PIN_BOTAO_SENHA_NORMAL = 17 # Pino físico 11 -> botão para o totem imprimir Normal PIN_BOTAO_SENHA_PRIORIDADE = 27 # Pino físico 13 -> botão para o totem imprimir Prioritário PIN_BOTAO_PROXIMO_NORMAL = 22 # Pino físico 15 -> botão no guichê para chamar Próximo Normal PIN_BOTAO_PROXIMO_PRIORIDADE = 23 # Pino físico 16 -> botão no guichê para chamar Próximo Prioritário PIN_ALERTA_LED_RELE = 18 # Pino físico 12 -> Saída para LED gigante ou Relé de Giroflex PIN_BUZZER_SONORO = 24 # Pino físico 18 -> Saída para Buzzer ativo de 5v # CONFIGURAÇÕES DO OPERADOR (Para quando chamar do guichê via botão físico) NUMERO_GUICHE_PADRAO = "Guichê 01" # CONFIGURAÇÕES DOS ALERTAS ATIVAR_VOZ_TTS = True # Syntetiza voz ("Senha ... Vá para o ...") via alto-falantes da Pi BOUNCE_TIME_OTO = 400 # Debounce dos botões em milissegundos para evitar disparo duplo # ========================================== # IMPLEMENTAÇÃO DO CLIENTE DA API # ========================================== class RaspberryQueueClient: def __init__(self): self.api_url = f"{API_BASE_URL}/api.php" self.ultimo_ticket_chamado_id = None self.rodando = True def requisitar(self, params, payload=None, metodo="GET"): """Método helper para efetuar requisições HTTP seguras ao servidor PHP""" try: if metodo == "POST": res = requests.post(self.api_url, params=params, json=payload, timeout=5) else: res = requests.get(self.api_url, params=params, timeout=5) if res.status_code == 200: try: return res.json() except ValueError: return res.text else: print(f"[ERRO API] Código de retorno HTTP inválido: {res.status_code}") return None except Exception as e: print(f"[ERRO REDE] Não foi possível comunicar com o servidor: {e}") return None # Ações do Totem de Impressão (Gerar Senhas) def emitir_senha_normal(self): print("\n[TOTEM FÍSICO] Botão Pressionado -> Solicitando Senha Normal...") resposta = self.requisitar(params={"action": "gerar"}, payload={"type": "Normal"}, metodo="POST") if resposta and "ticket" in resposta: ticket = resposta["ticket"] num = ticket.get("numero", "---") print(f"[TOTEM FÍSICO] Senha Gerada com Sucesso: {num}") self.imprimir_fisicamente(num, "NORMAL") else: print("[TOTEM FÍSICO] Falha ao solicitar senha.") def emitir_senha_prioridade(self): print("\n[TOTEM FÍSICO] Botão Pressionado -> Solicitando Senha Prioritária...") resposta = self.requisitar(params={"action": "gerar"}, payload={"type": "Prioritário"}, metodo="POST") if resposta and "ticket" in resposta: ticket = resposta["ticket"] num = ticket.get("numero", "---") print(f"[TOTEM FÍSICO] Senha Gerada com Sucesso: {num}") self.imprimir_fisicamente(num, "PRIORIDADE") else: print("[TOTEM FÍSICO] Falha ao solicitar senha.") def imprimir_fisicamente(self, numero_senha, tipo_senha): """ Em totens reais comerciais, uma impressora térmica de 80mm de ribbon (ESC/POS) pode ser acionada diretamente por comandos na porta serial /dev/ttyUSB0 ou USB raw. """ print(f"----- IMPRESSÃO DE SENHA -----") print(f" {tipo_senha.upper()}") print(f" SENHA: {numero_senha}") print(f" Aguarde sua chamada na TV.") print(f" Data/Hora: {time.strftime('%d/%m/%Y %H:%M:%S')}") print(f"------------------------------") # Exemplo opcional de comando shell para imprimir em impressora padrão CUPS # subprocess.run(["lp", "-d", "thermal_printer", "impressao_temporaria.txt"]) # Ações do painel do guichê (Chamar Senhas) def chamar_proximo_normal(self): print(f"\n[GUICHÊ FÍSICO] Botão Chamar Normal Pressionado por: {NUMERO_GUICHE_PADRAO}") resposta = self.requisitar(params={"action": "chamar"}, payload={"desk": NUMERO_GUICHE_PADRAO, "typeSelection": "normalOnly"}, metodo="POST") if resposta and "ticket" in resposta: ticket = resposta["ticket"] print(f"[GUICHÊ FÍSICO] Chamado com sucesso: {ticket.get('numero')} no {NUMERO_GUICHE_PADRAO}") else: print("[GUICHÊ FÍSICO] Nenhuma senha Normal aguardando na fila.") def chamar_proximo_prioridade(self): print(f"\n[GUICHÊ FÍSICO] Botão Chamar Prioritário Pressionado por: {NUMERO_GUICHE_PADRAO}") resposta = self.requisitar(params={"action": "chamar"}, payload={"desk": NUMERO_GUICHE_PADRAO, "typeSelection": "prioritarioOnly"}, metodo="POST") if resposta and "ticket" in resposta: ticket = resposta["ticket"] print(f"[GUICHÊ FÍSICO] Chamado com sucesso: {ticket.get('numero')} no {NUMERO_GUICHE_PADRAO}") else: print("[GUICHÊ FÍSICO] Nenhuma senha Prioritária aguardando na fila.") # ========================================== # SAÍDA FÍSICA E ALERTAS AO ATENDER SENHA # ========================================== def acionar_alertas_hardware(self, numero_ticket, guiche_chamado): """Dispara buzinas, giroflex e voz TTS no Raspberry Pi quando uma nova senha surge""" print(f"\n>>> ATENÇÃO: SENHA {numero_ticket} CHAMADA PARA O {guiche_chamado}! <<<") # 1. PISCAR LED / ATIVAR RELÊ (Aviso Visual de alta prioridade na TV/Kiosk) def piscar_led(): for _ in range(8): # Pisca 8 vezes GPIO.output(PIN_ALERTA_LED_RELE, True) time.sleep(0.15) GPIO.output(PIN_ALERTA_LED_RELE, False) time.sleep(0.12) # 2. TOCAR BUZZER FÍSICO (Tons no estilo bip-bip duplo) def tocar_buzzer(): for _ in range(3): # Bipa três vezes de forma dupla GPIO.output(PIN_BUZZER_SONORO, True) time.sleep(0.1) GPIO.output(PIN_BUZZER_SONORO, False) time.sleep(0.05) GPIO.output(PIN_BUZZER_SONORO, True) time.sleep(0.1) GPIO.output(PIN_BUZZER_SONORO, False) time.sleep(0.4) # Inicia efeitos em threads separadas para não congelar o polling principal threading.Thread(target=piscar_led, daemon=True).start() threading.Thread(target=tocar_buzzer, daemon=True).start() # 3. TEXT-TO-SPEECH (VOD DE AUDIO EM PORTUGUÊS) if ATIVAR_VOZ_TTS: # Letrista amigável para sintetizador tipo_letra = "Normal" if numero_ticket.startswith("N") else "Prioritária" # Decompor números para fala articulada tranquila (Ex: N025 -> N zero vinte e cinco) num_clean = numero_ticket[1:] # Tira o N ou P try: num_int = int(num_clean) numero_pronunciado = str(num_int) except ValueError: numero_pronunciado = numero_ticket texto_voz = f"Atenção. Senha {tipo_letra}, número {numero_pronunciado}. Compareça ao {guiche_chamado}." # Executa sintetizador de voz offline ultra leve em uma thread secundária threading.Thread(target=self.falar_texto, args=(texto_voz,), daemon=True).start() def falar_texto(self, texto): """Utiliza o sintetizador de fala espeak nativo do Linux""" try: print(f"[SÍNTESE DE VOZ] Falando: '{texto}'") # pt-br: português brasileiro; -s 140: velocidade regulada amigável; -p 50: tom de voz padrão vocal do gongo subprocess.run(["espeak", "-v", "pt-br", "-s", "145", "-p", "45", texto], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except Exception as e: print(f"[ERRO VOZ] Falha ao executar subprocesso do espeak: {e}") # ========================================== # POLLING DE MONITORAMENTO (LOOP DE LEITURA) # ========================================== def loop_monitoramento_api(self): """Consulta o servidor PHP a cada 1 segundo para pescar a última senha chamada no visor""" print("[MONITOR] Serviço de escuta inicializado com sucesso. Monitorando o PHP...") while self.rodando: estado = self.requisitar(params={"action": "estado"}) if estado and isinstance(estado, dict): ticket_ativo = estado.get("activeTicket") if ticket_ativo: ticket_id = ticket_ativo.get("ticketId") numero = ticket_ativo.get("numero") guiche = ticket_ativo.get("desk") # Se um novo ticket diferente foi chamado, dispara evento de som/voz/luz if ticket_id != self.ultimo_ticket_chamado_id: if self.ultimo_ticket_chamado_id is not None: self.acionar_alertas_hardware(numero, guiche) self.ultimo_ticket_chamado_id = ticket_id time.sleep(1.0) # ========================================== # CONFIGURAÇÃO DE EVENTOS DE EVENT-DETECTION GPUS # ========================================== client = RaspberryQueueClient() # Callbacks de interrupção física por borda de descida (botão pressionado aterrado) def callback_botao_normal(channel): client.emitir_senha_normal() def callback_botao_prioritario(channel): client.emitir_senha_prioridade() def callback_proximo_normal(channel): client.chamar_proximo_normal() def callback_proximo_prioritario(channel): client.chamar_proximo_prioridade() def inicializar_hardware(): """Configura o endereçamento físico das portas da Raspberry Pi""" print("[HARDWARE] Inicializando mapeamento de portas da placa...") GPIO.setmode(GPIO.BCM) # Configuração de Saídas (LEDs/Relés/Buzzers) GPIO.setup(PIN_ALERTA_LED_RELE, GPIO.OUT, initial=False) GPIO.setup(PIN_BUZZER_SONORO, GPIO.OUT, initial=False) # Configuração de Entradas com resistores de Pull-Up internos (Ativo em nível lógico Baixo) GPIO.setup(PIN_BOTAO_SENHA_NORMAL, GPIO.IN, pull_up_down=GPIO.PUD_UP) GPIO.setup(PIN_BOTAO_SENHA_PRIORIDADE, GPIO.IN, pull_up_down=GPIO.PUD_UP) GPIO.setup(PIN_BOTAO_PROXIMO_NORMAL, GPIO.IN, pull_up_down=GPIO.PUD_UP) GPIO.setup(PIN_BOTAO_PROXIMO_PRIORIDADE, GPIO.IN, pull_up_down=GPIO.PUD_UP) pinos_entrada = [ (PIN_BOTAO_SENHA_NORMAL, callback_botao_normal), (PIN_BOTAO_SENHA_PRIORIDADE, callback_botao_prioritario), (PIN_BOTAO_PROXIMO_NORMAL, callback_proximo_normal), (PIN_BOTAO_PROXIMO_PRIORIDADE, callback_proximo_prioritario) ] for pin, callback in pinos_entrada: try: # Tentar remover detecção de eventos prévia para evitar conflitos de processos anteriores try: GPIO.remove_event_detect(pin) except Exception: pass # Adicionar evento de descida (borda de descida) com debounce em milissegundos GPIO.add_event_detect(pin, GPIO.FALLING, callback=callback, bouncetime=BOUNCE_TIME_OTO) except RuntimeError as e: print("\n" + "="*80) print("❌ ERRO CRÍTICO RASPBERRY PI: Falha ao adicionar detecção de borda no pino!") print("="*80) print(f"Pino com falha: BCM GPIO {pin} | Erro retornado: {e}") print("\nPROVÁVEIS CAUSAS E SOLUÇÕES:") print("1. FALTA DE PERMISSÃO ROOT (Mais comum):") print(" Execute o script usando privilégios de administrador (sudo):") print(" --> sudo python3 raspberry_pi_gpio.py") print("\n2. PROCESSO EM SEGUNDO PLANO JÁ EM EXECUÇÃO:") print(" Outra cópia deste script pode estar ocupando os recursos do GPIO da Raspberry Pi.") print(" Encontre e encerre outros processos Python rodando:") print(" --> sudo killall python3") print("\n3. COMPATIBILIDADE COM NOVAS VERSÕES DE KERNEL OU RASPBERRY PI 5:") print(" Em sistemas operacionais mais recentes (Debian Bookworm / Pi 5), a biblioteca RPi.GPIO") print(" está obsoleta e não consegue se comunicar diretamente com o hardware.") print(" Nessas versões, recomenda-se fortemente o uso da moderna biblioteca 'gpiozero'.") print(" Gostaria que gerássemos uma versão alternativa utilizando 'gpiozero' para resolver?") print("="*80 + "\n") raise e print("[HARDWARE] GPIO mapeado com sucesso.") if __name__ == "__main__": print("=====================================================================") print(" SERVIÇO RASPBERRY PI GPIO PARA SISTEMA DE FILAS INTEGRAIS ") print("=====================================================================") print(f"Endereço do Servidor Alvo: {API_BASE_URL}\n") try: inicializar_hardware() # Dispara thread secundária responsável pela sincronização do Visor da TV thread_escuta = threading.Thread(target=client.loop_monitoramento_api, daemon=True) thread_escuta.start() print("\n[OK] Tudo pronto! Pressione Ctrl+C para encerrar o script de integração.") print("Pressione fisicamente os botões da Pi ou acione chamadas no Painel Web para certificar.") # Loop infinito mantendo a thread principal ativa while True: time.sleep(10) except KeyboardInterrupt: print("\n[SISTEMA] Sinal de fechamento detectado. Desativando serviços...") finally: client.rodando = False GPIO.cleanup() print("[SISTEMA] Portas limpas e circuito higienizado com segurança. Tchau!")