#!/usr/bin/env python3 """ SeismicDetect Gateway Client Open-source IoT Gateway untuk menghubungkan hardware ke platform SeismicDetect """ import json import time import threading import logging import serial import requests import yaml import socket import sys import os from datetime import datetime from typing import Dict, List, Any, Optional # Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger('SeismicGateway') class SeismicGateway: """Gateway utama untuk menghubungkan hardware ke server""" def __init__(self, config_path: str = 'config.yaml'): self.load_config(config_path) self.device_id = None self.api_key = None self.is_registered = False self.data_buffer = [] self.running = True self.serial_connection = None def load_config(self, config_path: str): """Load konfigurasi dari file YAML""" try: with open(config_path, 'r') as f: self.config = yaml.safe_load(f) logger.info("Konfigurasi loaded") except FileNotFoundError: logger.warning("Config file not found, using defaults") self.config = self.get_default_config() def get_default_config(self) -> Dict: """Default configuration""" return { 'gateway': { 'name': 'Seismic Gateway', 'device_id': f'GATEWAY_{os.getenv("DEVICE_ID", "001")}', 'type': 'gateway' }, 'server': { 'url': os.getenv('SERVER_URL', 'http://localhost:8000'), 'api_endpoint': '/api/gateway/data', 'register_endpoint': '/api/gateway/register', 'heartbeat_endpoint': '/api/gateway/heartbeat' }, 'hardware': { 'serial': {'enabled': True, 'baud_rate': 115200, 'timeout': 1}, 'network': {'enabled': True, 'port': 5000} }, 'data': { 'batch_size': 10, 'send_interval': 2, 'heartbeat_interval': 30 } } def register_device(self) -> bool: """Registrasi device ke server""" try: register_data = { 'device_id': self.config['gateway']['device_id'], 'name': self.config['gateway']['name'], 'type': self.config['gateway']['type'], 'capabilities': ['gempa', 'getaran', 'suhu'], 'metadata': { 'version': '1.0.0', 'platform': sys.platform, 'python_version': sys.version } } response = requests.post( f"{self.config['server']['url']}{self.config['server']['register_endpoint']}", json=register_data, headers={'Authorization': f'Bearer {os.getenv("USER_TOKEN", "")}'} ) if response.status_code == 201: data = response.json() self.api_key = data['api_key'] self.device_id = data['device']['device_id'] self.is_registered = True logger.info(f"✅ Device registered! API Key: {self.api_key[:20]}...") self.save_api_key() return True else: logger.error(f"Registration failed: {response.status_code}") return False except Exception as e: logger.error(f"Registration error: {e}") return False def save_api_key(self): """Simpan API Key ke file""" with open('.api_key', 'w') as f: f.write(self.api_key) os.chmod('.api_key', 0o600) def load_api_key(self) -> Optional[str]: """Load API Key dari file""" try: with open('.api_key', 'r') as f: return f.read().strip() except FileNotFoundError: return None def send_data(self, data: List[Dict]): """Kirim data ke server""" if not self.api_key: self.api_key = self.load_api_key() if not self.api_key: logger.error("No API Key found") return False try: payload = { 'data': data, 'gateway_id': self.device_id, 'timestamp': datetime.now().isoformat() } response = requests.post( f"{self.config['server']['url']}{self.config['server']['api_endpoint']}", json=payload, headers={ 'X-API-Key': self.api_key, 'Content-Type': 'application/json' }, timeout=10 ) if response.status_code == 200: logger.info(f"✅ Data sent: {len(data)} records") return True elif response.status_code == 401: logger.error("Invalid API Key, need re-registration") self.is_registered = False return False else: logger.error(f"Send failed: {response.status_code}") return False except Exception as e: logger.error(f"Send error: {e}") return False def send_heartbeat(self): """Kirim heartbeat ke server""" if not self.api_key: return try: response = requests.post( f"{self.config['server']['url']}{self.config['server']['heartbeat_endpoint']}", headers={'X-API-Key': self.api_key}, json={'status': 'online', 'timestamp': datetime.now().isoformat()}, timeout=5 ) if response.status_code == 200: logger.debug("💓 Heartbeat sent") except Exception as e: logger.debug(f"Heartbeat error: {e}") def read_serial_data(self): """Baca data dari serial port (USB)""" if not self.config['hardware']['serial']['enabled']: return try: # Auto-detect serial ports import glob ports = glob.glob('/dev/ttyUSB*') + glob.glob('/dev/ttyACM*') if not ports: logger.warning("No serial ports found") return for port in ports: try: ser = serial.Serial( port, self.config['hardware']['serial']['baud_rate'], timeout=self.config['hardware']['serial']['timeout'] ) while self.running and ser.in_waiting: line = ser.readline().decode('utf-8', errors='ignore').strip() if line: self.process_raw_data(line, source='serial') ser.close() except serial.SerialException as e: logger.debug(f"Cannot open {port}: {e}") except Exception as e: logger.error(f"Serial read error: {e}") def read_network_data(self): """Baca data dari network (UDP/TCP) untuk ESP32 via WiFi""" if not self.config['hardware']['network']['enabled']: return try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(0.1) sock.bind(('0.0.0.0', self.config['hardware']['network']['port'])) while self.running: try: data, addr = sock.recvfrom(4096) message = data.decode('utf-8') self.process_raw_data(message, source=f'network:{addr[0]}') except socket.timeout: continue except Exception as e: logger.error(f"Network read error: {e}") def process_raw_data(self, raw_data: str, source: str = 'unknown'): """Proses data mentah dari hardware""" try: # Coba parse sebagai JSON data = json.loads(raw_data) # Standarisasi format standardized = { 'type': data.get('type', 'unknown'), 'value': data.get('value', data.get('intensitas', 0)), 'source': source, 'timestamp': datetime.now().isoformat(), 'raw': data } self.data_buffer.append(standardized) # Kirim jika buffer sudah penuh if len(self.data_buffer) >= self.config['data']['batch_size']: self.flush_buffer() logger.info(f"📊 Data received: {standardized['type']}={standardized['value']}") except json.JSONDecodeError: # Bukan JSON, coba parse sebagai format sederhana logger.warning(f"Non-JSON data: {raw_data[:100]}") # Contoh: "SENSOR:1234" atau "GETARAN:567" if ':' in raw_data: parts = raw_data.split(':') if len(parts) == 2: standardized = { 'type': parts[0].lower(), 'value': int(parts[1]), 'source': source, 'timestamp': datetime.now().isoformat(), 'raw': raw_data } self.data_buffer.append(standardized) def flush_buffer(self): """Kirim semua data dalam buffer""" if self.data_buffer: if self.send_data(self.data_buffer): self.data_buffer = [] def get_command(self) -> Optional[Dict]: """Ambil command dari server""" if not self.api_key: return None try: response = requests.get( f"{self.config['server']['url']}{self.config['server']['command_endpoint']}", headers={'X-API-Key': self.api_key}, timeout=5 ) if response.status_code == 200: command = response.json().get('command') if command: logger.info(f"📟 Command received: {command}") return command except Exception as e: logger.debug(f"Command check error: {e}") return None def execute_command(self, command: Dict): """Eksekusi command yang diterima""" cmd_type = command.get('command') params = command.get('params', {}) if cmd_type == 'led_on': # Contoh: kirim ke serial untuk menyalakan LED self.send_to_serial("LED_ON\n") logger.info("💡 LED turned ON") elif cmd_type == 'led_off': self.send_to_serial("LED_OFF\n") logger.info("💡 LED turned OFF") elif cmd_type == 'calibrate': logger.info("🔧 Calibration requested") self.send_to_serial("CALIBRATE\n") def send_to_serial(self, message: str): """Kirim perintah ke hardware via serial""" try: ports = glob.glob('/dev/ttyUSB*') + glob.glob('/dev/ttyACM*') for port in ports: try: ser = serial.Serial(port, 115200, timeout=1) ser.write(message.encode()) ser.close() logger.debug(f"Sent to {port}: {message.strip()}") except: pass except Exception as e: logger.error(f"Serial send error: {e}") def run(self): """Main loop gateway""" logger.info("🚀 Starting SeismicDetect Gateway...") # Load atau register device self.api_key = self.load_api_key() if not self.api_key: if not self.register_device(): logger.error("Failed to register device") return # Start threads threads = [] # Serial reader thread serial_thread = threading.Thread(target=self.read_serial_data, daemon=True) serial_thread.start() threads.append(serial_thread) # Network reader thread network_thread = threading.Thread(target=self.read_network_data, daemon=True) network_thread.start() threads.append(network_thread) # Heartbeat thread last_heartbeat = 0 logger.info("✅ Gateway running! Waiting for data...") try: while self.running: # Send heartbeat if time.time() - last_heartbeat > self.config['data']['heartbeat_interval']: self.send_heartbeat() last_heartbeat = time.time() # Check for commands command = self.get_command() if command: self.execute_command(command) # Flush buffer periodically if len(self.data_buffer) > 0: self.flush_buffer() time.sleep(1) except KeyboardInterrupt: logger.info("Shutting down...") finally: self.running = False self.flush_buffer() logger.info("Gateway stopped") def main(): """Entry point""" gateway = SeismicGateway() gateway.run() if __name__ == '__main__': main()