wifi_scanner.py - Examples

  1#!/usr/bin/env python3
  2"""
  3WiFi 네트워크 스캐너 및 소켓 프로그래밍 예제
  4
  5이 스크립트는 다음 기능을 제공합니다:
  61. WiFi 네트워크 스캔 (시뮬레이션 모드 지원)
  72. TCP 클라이언트/서버 예제
  83. HTTP 클라이언트로 센서 데이터 전송
  94. 네트워크 연결 상태 모니터링
 105. 네트워크 오류 처리 및 재시도 로직
 11
 12참고: content/ko/IoT_Embedded/04_WiFi_Networking.md
 13"""
 14
 15import subprocess
 16import re
 17import socket
 18import json
 19import time
 20import random
 21import sys
 22from typing import Optional, List, Dict
 23from datetime import datetime
 24
 25
 26# ============================================================================
 27# WiFi 네트워크 스캔
 28# ============================================================================
 29
 30def is_raspberry_pi() -> bool:
 31    """라즈베리파이 환경인지 확인"""
 32    try:
 33        with open('/proc/cpuinfo', 'r') as f:
 34            cpuinfo = f.read()
 35            return 'BCM' in cpuinfo or 'Raspberry' in cpuinfo
 36    except FileNotFoundError:
 37        return False
 38
 39
 40def get_wifi_info() -> dict:
 41    """
 42    현재 WiFi 연결 정보 조회
 43
 44    Returns:
 45        dict: WiFi 정보 (ssid, ip_addresses, signal_dbm, mac_address)
 46    """
 47    info = {}
 48
 49    if not is_raspberry_pi():
 50        # 시뮬레이션 모드
 51        print("[시뮬레이션] 실제 라즈베리파이가 아닙니다. 가상 데이터 사용")
 52        info = {
 53            'ssid': 'SimulatedWiFi',
 54            'ip_addresses': ['192.168.1.100'],
 55            'signal_dbm': -45,
 56            'mac_address': 'AA:BB:CC:DD:EE:FF'
 57        }
 58        return info
 59
 60    try:
 61        # SSID 조회
 62        result = subprocess.run(
 63            ['iwgetid', '-r'],
 64            capture_output=True,
 65            text=True,
 66            timeout=5
 67        )
 68        info['ssid'] = result.stdout.strip()
 69
 70        # IP 주소 조회
 71        result = subprocess.run(
 72            ['hostname', '-I'],
 73            capture_output=True,
 74            text=True,
 75            timeout=5
 76        )
 77        ips = result.stdout.strip().split()
 78        info['ip_addresses'] = ips
 79
 80        # 신호 강도 조회
 81        result = subprocess.run(
 82            ['iwconfig', 'wlan0'],
 83            capture_output=True,
 84            text=True,
 85            timeout=5
 86        )
 87        match = re.search(r'Signal level=(-?\d+)', result.stdout)
 88        if match:
 89            info['signal_dbm'] = int(match.group(1))
 90
 91        # MAC 주소
 92        result = subprocess.run(
 93            ['cat', '/sys/class/net/wlan0/address'],
 94            capture_output=True,
 95            text=True,
 96            timeout=5
 97        )
 98        info['mac_address'] = result.stdout.strip()
 99
100    except Exception as e:
101        info['error'] = str(e)
102
103    return info
104
105
106def get_wifi_networks() -> List[dict]:
107    """
108    주변 WiFi 네트워크 스캔
109
110    Returns:
111        list: 네트워크 정보 리스트 (ssid, quality, signal_dbm)
112    """
113    networks = []
114
115    if not is_raspberry_pi():
116        # 시뮬레이션 모드
117        print("[시뮬레이션] WiFi 스캔 결과 생성 중...")
118        networks = [
119            {'ssid': 'HomeNetwork', 'quality': '70/70', 'signal_dbm': -35},
120            {'ssid': 'OfficeWiFi', 'quality': '65/70', 'signal_dbm': -42},
121            {'ssid': 'GuestNetwork', 'quality': '50/70', 'signal_dbm': -58},
122            {'ssid': 'Neighbor_2.4G', 'quality': '40/70', 'signal_dbm': -65},
123            {'ssid': 'CafeWiFi', 'quality': '30/70', 'signal_dbm': -75},
124        ]
125        return networks
126
127    try:
128        result = subprocess.run(
129            ['sudo', 'iwlist', 'wlan0', 'scan'],
130            capture_output=True,
131            text=True,
132            timeout=10
133        )
134
135        current_network = {}
136        for line in result.stdout.split('\n'):
137            if 'ESSID:' in line:
138                ssid = re.search(r'ESSID:"(.+)"', line)
139                if ssid and current_network:
140                    networks.append(current_network)
141                current_network = {'ssid': ssid.group(1) if ssid else ''}
142
143            elif 'Quality=' in line:
144                quality = re.search(r'Quality=(\d+)/(\d+)', line)
145                if quality:
146                    current_network['quality'] = f"{quality.group(1)}/{quality.group(2)}"
147
148                signal = re.search(r'Signal level=(-?\d+)', line)
149                if signal:
150                    current_network['signal_dbm'] = int(signal.group(1))
151
152        if current_network:
153            networks.append(current_network)
154
155    except Exception as e:
156        print(f"스캔 실패: {e}")
157        print("시뮬레이션 모드로 전환합니다...")
158        return get_wifi_networks()  # 재귀 호출로 시뮬레이션 모드 사용
159
160    return networks
161
162
163def monitor_wifi_signal(threshold_dbm: int = -70, interval: int = 5, duration: int = 30):
164    """
165    WiFi 신호 강도 모니터링
166
167    Args:
168        threshold_dbm: 경고 임계값 (dBm)
169        interval: 확인 간격 (초)
170        duration: 모니터링 기간 (초)
171    """
172    print(f"=== WiFi 신호 모니터링 시작 ===")
173    print(f"경고 임계값: {threshold_dbm} dBm")
174    print(f"확인 간격: {interval}초, 총 {duration}초")
175    print()
176
177    start_time = time.time()
178
179    while time.time() - start_time < duration:
180        info = get_wifi_info()
181
182        if 'error' in info:
183            print(f"[{datetime.now().strftime('%H:%M:%S')}] 오류: {info['error']}")
184        else:
185            signal = info.get('signal_dbm', 0)
186            ssid = info.get('ssid', 'Unknown')
187
188            status = "양호" if signal > threshold_dbm else "⚠️  경고"
189            print(f"[{datetime.now().strftime('%H:%M:%S')}] {ssid}: {signal} dBm ({status})")
190
191            if signal <= threshold_dbm:
192                print(f"  ⚠️  신호가 약합니다! ({signal} dBm <= {threshold_dbm} dBm)")
193
194        time.sleep(interval)
195
196    print("\n모니터링 종료")
197
198
199# ============================================================================
200# TCP 소켓 프로그래밍
201# ============================================================================
202
203def tcp_server(host: str = '0.0.0.0', port: int = 9999):
204    """
205    TCP 서버 - 센서 데이터 수신
206
207    Args:
208        host: 바인드할 IP 주소
209        port: 포트 번호
210    """
211    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
212        # 주소 재사용 허용
213        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
214
215        server.bind((host, port))
216        server.listen(5)
217
218        print(f"TCP 서버 시작: {host}:{port}")
219        print("클라이언트 연결 대기 중... (Ctrl+C로 종료)")
220
221        try:
222            while True:
223                client, address = server.accept()
224                print(f"\n클라이언트 연결: {address}")
225
226                with client:
227                    while True:
228                        data = client.recv(1024)
229                        if not data:
230                            break
231
232                        try:
233                            # JSON 데이터 파싱
234                            message = json.loads(data.decode('utf-8'))
235                            timestamp = datetime.now().strftime("%H:%M:%S")
236                            print(f"[{timestamp}] 수신: {message}")
237
238                            # 응답 전송
239                            response = {
240                                "status": "ok",
241                                "received": message.get("sensor_id"),
242                                "timestamp": time.time()
243                            }
244                            client.sendall(json.dumps(response).encode('utf-8'))
245
246                        except json.JSONDecodeError:
247                            print(f"잘못된 JSON: {data}")
248
249                print(f"클라이언트 연결 종료: {address}")
250
251        except KeyboardInterrupt:
252            print("\n서버 종료")
253
254
255def tcp_client(server_host: str, server_port: int = 9999, count: int = 10):
256    """
257    TCP 클라이언트 - 센서 데이터 전송
258
259    Args:
260        server_host: 서버 IP 주소
261        server_port: 서버 포트
262        count: 전송 횟수
263    """
264    try:
265        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
266            print(f"서버 연결 중: {server_host}:{server_port}")
267            client.connect((server_host, server_port))
268            print(f"서버 연결 성공!")
269
270            sensor_id = "temp_sensor_01"
271
272            for i in range(count):
273                # 센서 데이터 생성 (시뮬레이션)
274                data = {
275                    "sensor_id": sensor_id,
276                    "temperature": round(random.uniform(20, 30), 1),
277                    "humidity": round(random.uniform(40, 70), 1),
278                    "timestamp": time.time(),
279                    "sequence": i + 1
280                }
281
282                # 전송
283                message = json.dumps(data).encode('utf-8')
284                client.sendall(message)
285                print(f"[{i+1}/{count}] 전송: 온도={data['temperature']}°C, 습도={data['humidity']}%")
286
287                # 응답 수신
288                try:
289                    client.settimeout(5.0)
290                    response = client.recv(1024)
291                    if response:
292                        resp_data = json.loads(response.decode('utf-8'))
293                        print(f"         응답: {resp_data['status']}")
294                except socket.timeout:
295                    print(f"         응답 타임아웃")
296
297                if i < count - 1:
298                    time.sleep(2)
299
300            print("\n전송 완료")
301
302    except ConnectionRefusedError:
303        print(f"오류: 서버에 연결할 수 없습니다 ({server_host}:{server_port})")
304        print("서버가 실행 중인지 확인하세요.")
305    except Exception as e:
306        print(f"오류: {e}")
307
308
309# ============================================================================
310# HTTP 클라이언트
311# ============================================================================
312
313def send_sensor_data_http(api_base: str, sensor_id: str, data: dict) -> bool:
314    """
315    HTTP POST로 센서 데이터 전송
316
317    Args:
318        api_base: API 베이스 URL
319        sensor_id: 센서 ID
320        data: 전송할 데이터
321
322    Returns:
323        bool: 성공 여부
324    """
325    try:
326        import requests
327    except ImportError:
328        print("requests 라이브러리가 필요합니다: pip install requests")
329        return False
330
331    url = f"{api_base}/sensors/{sensor_id}/data"
332
333    try:
334        response = requests.post(
335            url,
336            json=data,
337            headers={"Content-Type": "application/json"},
338            timeout=5
339        )
340
341        if response.status_code in (200, 201):
342            print(f"데이터 전송 성공: {data}")
343            return True
344        else:
345            print(f"전송 실패: {response.status_code} - {response.text}")
346            return False
347
348    except requests.exceptions.RequestException as e:
349        print(f"네트워크 오류: {e}")
350        return False
351
352
353def http_client_with_retry(api_base: str, sensor_id: str, retries: int = 3):
354    """
355    재시도 로직이 있는 HTTP 클라이언트
356
357    Args:
358        api_base: API 베이스 URL
359        sensor_id: 센서 ID
360        retries: 재시도 횟수
361    """
362    print(f"=== HTTP 클라이언트 (재시도: {retries}회) ===")
363    print(f"API: {api_base}")
364    print(f"센서 ID: {sensor_id}\n")
365
366    for i in range(10):
367        data = {
368            "temperature": round(random.uniform(20, 30), 1),
369            "humidity": round(random.uniform(40, 70), 1),
370            "timestamp": int(time.time())
371        }
372
373        success = False
374        for attempt in range(retries):
375            if send_sensor_data_http(api_base, sensor_id, data):
376                success = True
377                break
378            else:
379                if attempt < retries - 1:
380                    wait_time = 2 ** attempt  # Exponential backoff
381                    print(f"  재시도 대기: {wait_time}초...")
382                    time.sleep(wait_time)
383
384        if not success:
385            print(f"  ❌ {retries}번 재시도 후 실패")
386
387        time.sleep(5)
388
389
390# ============================================================================
391# 네트워크 연결 상태 확인
392# ============================================================================
393
394def check_internet_connection(host: str = "8.8.8.8", port: int = 53, timeout: int = 3) -> bool:
395    """
396    인터넷 연결 확인
397
398    Args:
399        host: 테스트할 호스트 (기본: Google DNS)
400        port: 포트
401        timeout: 타임아웃 (초)
402
403    Returns:
404        bool: 연결 가능 여부
405    """
406    try:
407        socket.setdefaulttimeout(timeout)
408        socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect((host, port))
409        return True
410    except socket.error:
411        return False
412
413
414def get_local_ip() -> str:
415    """
416    로컬 IP 주소 조회
417
418    Returns:
419        str: IP 주소
420    """
421    try:
422        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
423        s.connect(('8.8.8.8', 80))
424        ip = s.getsockname()[0]
425        s.close()
426        return ip
427    except Exception:
428        return "127.0.0.1"
429
430
431def network_status():
432    """네트워크 상태 확인"""
433    print("=== 네트워크 상태 ===\n")
434
435    # 로컬 IP
436    local_ip = get_local_ip()
437    print(f"로컬 IP: {local_ip}")
438
439    # 인터넷 연결
440    internet = check_internet_connection()
441    print(f"인터넷 연결: {'✓ 연결됨' if internet else '✗ 연결 안됨'}")
442
443    # WiFi 정보
444    wifi_info = get_wifi_info()
445    if 'error' not in wifi_info:
446        print(f"\nWiFi SSID: {wifi_info.get('ssid', 'N/A')}")
447        print(f"신호 강도: {wifi_info.get('signal_dbm', 'N/A')} dBm")
448        print(f"MAC 주소: {wifi_info.get('mac_address', 'N/A')}")
449
450
451# ============================================================================
452# 메인 함수
453# ============================================================================
454
455def main():
456    """메인 함수"""
457    if len(sys.argv) < 2:
458        print("WiFi 네트워크 스캐너 및 소켓 프로그래밍 예제")
459        print("\n사용법:")
460        print("  python wifi_scanner.py info         - WiFi 정보 조회")
461        print("  python wifi_scanner.py scan         - WiFi 네트워크 스캔")
462        print("  python wifi_scanner.py monitor      - WiFi 신호 모니터링")
463        print("  python wifi_scanner.py status       - 네트워크 상태 확인")
464        print("  python wifi_scanner.py server       - TCP 서버 시작")
465        print("  python wifi_scanner.py client <IP>  - TCP 클라이언트 시작")
466        print("\n예제:")
467        print("  python wifi_scanner.py info")
468        print("  python wifi_scanner.py client 192.168.1.100")
469        return
470
471    command = sys.argv[1].lower()
472
473    if command == 'info':
474        print("=== WiFi 연결 정보 ===")
475        info = get_wifi_info()
476        for key, value in info.items():
477            print(f"  {key}: {value}")
478
479    elif command == 'scan':
480        print("=== 주변 WiFi 네트워크 ===")
481        networks = get_wifi_networks()
482        print(f"발견된 네트워크: {len(networks)}개\n")
483        for net in networks:
484            ssid = net.get('ssid', 'Unknown')
485            signal = net.get('signal_dbm', 'N/A')
486            quality = net.get('quality', 'N/A')
487            print(f"  {ssid:20} - 신호: {signal:4} dBm, 품질: {quality}")
488
489    elif command == 'monitor':
490        monitor_wifi_signal(threshold_dbm=-70, interval=5, duration=30)
491
492    elif command == 'status':
493        network_status()
494
495    elif command == 'server':
496        tcp_server()
497
498    elif command == 'client':
499        if len(sys.argv) < 3:
500            print("오류: 서버 IP 주소를 입력하세요")
501            print("예제: python wifi_scanner.py client 192.168.1.100")
502            return
503        server_ip = sys.argv[2]
504        tcp_client(server_ip, count=10)
505
506    else:
507        print(f"알 수 없는 명령: {command}")
508        print("'python wifi_scanner.py'를 실행하여 도움말을 확인하세요")
509
510
511if __name__ == "__main__":
512    main()