ㅇ 일 자 : 2026년 6월 2일(화)
ㅇ 주 제 : 파이썬 프로그램, GUI, MySQL 데이터베이스, 리눅스 지식을 갖춘 신입 개발자들을 위한 토이 프로그램 만들기 (Arduino, AI응용)
ㅇ 실습 후기
인공지능 응용시스템 신입 개발자는 하드웨어와 소프트웨어를 아우르는 풀스텍 (Full-STack) loT 역량을 길러야 한다고 생각한다.
그러기 위해서는 부단한 공부와 설계 역량을 기르고, 프로그램을 작성할 때 사용자 요구사항을 잘 파악하여 잘 구현하여 테스트까지 잘 되어 오류가 없어야 될 것이다. 그러나 오류가 없을 수는 없겠고, 그 오류들을 수정하고 고칠수 있는 역량을 길러야 한다는 것을 더욱 실감하고 있다.
특히, '팅커캐드'로 회로를 설계하고, 아두이노 '보드'에 연결한 후 그 시그널들을 리눅스 환경에서 코드를 실행하고, 연동하여 화면에 구현되기까지의 과정을 처음엔 이해할 수 없어 계속 실행해 봐야 했다. 빠른시간내에 이해하는 것이 다소 어려웠지만 계속 공부해야 한다.
① 데이터 파이프라인 설계 능력
"단순 아두이노 제어에 그치지 않고, 하드웨어, 파이썬 미들웨어, RDBMS로 이어지는 데이터 흐름을 소규모지만, 직접 설계하고 정형 데이터로 저장해 본 경험"을 통해 향후 데이터가 모이면 스마트기기 설비를 위한 AI 모델의 학습 데이터가 될 수 있을 것이다.
② 비동기 처리 및 예외 처리
시리얼 통신과 GUI가 동시에 작동할 때 발생하는 병목 현상을 해결하기 위해 쓰레드(Thread)를 사용함. 연결이 끊겼을 때의 예외 처리(디바이스 재연결 로직 등)
③ 리눅스 인프라 및 자동화 역량
단순히 PC에서 띄운 프로그램이 아니라, 리눅스 환경에서 DB를 운영하고 shell script와 crontab을 통해 백업을 자동화하는 등
'운영 안정성'을 고려한 개발자가 되어야 할 것이다.
이처럼 프로그램 개발은 창의적인 아이디어와 AI의 응용능력을 결합했을 때 어느정도의 결과물이 나올지 그리고 향후 비개발자들이 활용할 수 있는 다양한 프로그램들이 진화하게 될 것이다. 개발자 역량은 하루에 길러지는 것이 아니겠지만 "도전은 계속 된다."
<< 요구 사항 >>
# 개발 완료 후 리눅스 가상 서버 환경에서 터미널 접속만으로 시스템 전체를 한 번에 빌드하고 구동시킬 수 있게 하기
# 자동화 쉘 스크립트를 제작 및 추후 적재된 sensor_logs 이력을 바탕으로 이상 징후 탐지 AI 모델 및 시계열 트래픽 예측 하기
# 원천 정형 데이터를 엔지니어링하는 데이터 파이프라인 구축 하기
※ 구현 화면 (loT 무인 매장 실시간 보안 및 방문객 관리 시스템)
1. 프로젝트 명: loT 무인 매장 실시간 보안 및 방문객 관리 시스템
아두이노로 환경(동작 센서 감지 등) 데이터를 수집하고, 파이썬 GUI 화면에 실시간 시각화하며, 위험 상황 발생 시 리눅스 서버 및 아두이노 액추에이터를 통해 현장을 원격 제어하는 시스템
1.1. 전체 시스템 구조
[ 아두이노 (하드웨어) ]
│ (센서 데이터 송신 / 제어 명령 수신)
▼ Serial (UART) 통신
[ 파이썬 GUI PC (미들웨어 & 메인 앱) ] ───► [ MySQL (데이터베이스) ]
│ (로그 및 센서 데이터 저장)
▼ SSH / 네트워크 통신
[ 리눅스 서버 (백업 및 모니터링) ]
-- 아두이노 활용 시스템 구조 이해하기
|
Project : loT 무인 매장 실시간 보안 및 방문객 관리 시스템 만들기 (아두이노, AI응용)
1. 패키지 파일 (loTMemberControl)
1.1. 소스코드 (패키지 파일)
① __init__.py
'''loT 무인 매장 실시간 보안 및 방문객 관리 시스템 만들기 (AI응용)'''
② main.py
# 메인 파일
import sys
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QTableWidget,
QTableWidgetItem, QMessageBox, QSplitter)
from main_dashboard import *
if __name__ == "__main__":
from datetime import datetime
app = QApplication(sys.argv)
dashboard = SmartStoreDashboard()
dashboard.show()
sys.exit(app.exec_())
③ main_dashboard.py
## PyQt5 메인 데쉬보드
import random
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout,
QHBoxLayout, QLabel, QPushButton, QTableWidget,
QTableWidgetItem, QMessageBox, QSplitter)
from PyQt5.QtCore import QThread, pyqtSignal, QTimer, Qt
from PyQt5.QtGui import QFont, QColor
# 앞서 구현한 모듈 클래스 가져오기 (파일 이름이 같아야 합니다)
# from database_manager import DatabaseManager
# from serial_manager import ArduinoSerialManager
# -----------------------------------------------------------------------------
# 1. 시리얼 통신을 백그라운드에서 처리할 QThread 클래스 (화면 멈춤 방지)
# -----------------------------------------------------------------------------
class SerialWorker(QThread):
# GUI 메인 스레드로 '동작 감지 이벤트'를 보내기 위한 커스텀 신호 정의
motion_detected_signal = pyqtSignal()
def __init__(self, port="/dev/ttyACM0"):
super().__init__()
self.port = port
self.is_running = True
def run(self):
"""백그라운드에서 실행될 루프"""
print("[GUI 인프라] 시리얼 백그라운드 스레드 시작")
# 실제 아두이노 장비 연결 시 아래 주석을 해제하고 연동하세요.
# manager = ArduinoSerialManager(port=self.port)
# manager.start_reading(db_callback=self.motion_detected_signal.emit)
# [2일간의 단기 개발용 팁] 하드웨어 미연결 시 테스트를 위한 가상 센서 생성기 (Mock)
while self.is_running:
self.msleep(random.randint(4000, 8000)) # 4~8초 간격으로 랜덤 감지
if self.is_running:
# 메인 UI 화면에 신호 발송
self.motion_detected_signal.emit()
def stop(self):
self.is_running = False
self.wait()
# -----------------------------------------------------------------------------
# 2. 메인 대시보드 GUI 윈도우 클래스
# -----------------------------------------------------------------------------
class SmartStoreDashboard(QMainWindow):
def __init__(self):
super().__init__()
# 가상의 DB 연결 매니저 객체 생성 (실제 환경 시 아래 주석 해제)
# self.db = DatabaseManager(host="localhost", user="store_user", password="password", database="store_db")
# 개발 모드 디버깅용 상태 변수
self.current_mode = "BUSINESS" # 기본: 영업 모드
self.today_visitor_count = 0
self.initUI()
self.start_serial_thread()
def initUI(self):
"""UI 컴포넌트 초기화 및 배치"""
self.setWindowTitle("IoT 무인 매장 실시간 보안 및 방문객 관리 시스템")
self.setGeometry(100, 100, 1000, 600)
# 메인 위젯 및 레이아웃 설정
main_widget = QWidget()
self.setCentralWidget(main_widget)
main_layout = QHBoxLayout(main_widget)
# 스플리터를 활용해 좌측(제어/상태), 우측(로그 데이터) 화면 분할
splitter = QSplitter(Qt.Horizontal)
# ---------------------------------------------------------------------
# [좌측 영역] 현황판 및 제어 패널
# ---------------------------------------------------------------------
left_widget = QWidget()
left_layout = QVBoxLayout(left_widget)
# 타이틀 라벨
title_label = QLabel("🏪 매장 실시간 모니터링 시스템")
title_label.setFont(QFont("Arial", 16, QFont.Bold))
title_label.setAlignment(Qt.AlignCenter)
left_layout.addWidget(title_label)
# 현재 운영 모드 표시 패널
self.mode_status_label = QLabel("운영 상태: 영업 모드 (BUSINESS)")
self.mode_status_label.setFont(QFont("Arial", 14, QFont.Bold))
self.mode_status_label.setAlignment(Qt.AlignCenter)
self.mode_status_label.setStyleSheet(
"background-color: #2ECC71; color: white; border-radius: 10px; padding: 15px;")
left_layout.addWidget(self.mode_status_label)
# 실시간 센서 알림 창 (핵심 요구사항: 감지 시 배경 변경)
self.alert_box = QLabel("정상 운영 중\n(움직임 없음)")
self.alert_box.setFont(QFont("Arial", 14, QFont.Bold))
self.alert_box.setAlignment(Qt.AlignCenter)
self.alert_box.setStyleSheet(
"background-color: #F2F4F4; color: #34495E; border: 2px solid #BDC3C7; border-radius: 10px; padding: 30px;")
left_layout.addWidget(self.alert_box)
# 오늘 누적 방문자수 현황
self.counter_label = QLabel("오늘 총 방문객 수: 0 명")
self.counter_label.setFont(QFont("Arial", 12))
self.counter_label.setAlignment(Qt.AlignCenter)
left_layout.addWidget(self.counter_label)
# 모드 전환 버튼 제어 영역
btn_layout = QHBoxLayout()
self.btn_business = QPushButton("영업 모드 전환")
self.btn_business.setFont(QFont("Arial", 11, QFont.Bold))
self.btn_business.setStyleSheet("background-color: #3498DB; color: white; padding: 10px;")
self.btn_business.clicked.connect(lambda: self.change_store_mode("BUSINESS"))
self.btn_security = QPushButton("보안 모드 전환")
self.btn_security.setFont(QFont("Arial", 11, QFont.Bold))
self.btn_security.setStyleSheet("background-color: #E74C3C; color: white; padding: 10px;")
self.btn_security.clicked.connect(lambda: self.change_store_mode("SECURITY"))
btn_layout.addWidget(self.btn_business)
btn_layout.addWidget(self.btn_security)
left_layout.addLayout(btn_layout)
# ---------------------------------------------------------------------
# [우측 영역] 실시간 이벤트 로그 테이블 현황
# ---------------------------------------------------------------------
right_widget = QWidget()
right_layout = QVBoxLayout(right_widget)
table_title = QLabel("📋 최근 센서 감지 이력 (실시간 적재 로그)")
table_title.setFont(QFont("Arial", 12, QFont.Bold))
right_layout.addWidget(table_title)
self.log_table = QTableWidget()
self.log_table.setColumnCount(3)
self.log_table.setHorizontalHeaderLabels(["감지 시간", "장치 ID", "상태 모드"])
self.log_table.horizontalHeader().setStretchLastSection(True)
right_layout.addWidget(self.log_table)
# 스플리터에 양쪽 레이아웃 추가 후 메인 레이아웃에 탑재
splitter.addWidget(left_widget)
splitter.addWidget(right_widget)
main_layout.addWidget(splitter)
# 실시간 경고 연출 원상복구를 위한 타이머 설정 (3초 뒤 정상화)
self.alert_timer = QTimer()
self.alert_timer.setSingleShot(True)
self.alert_timer.timeout.connect(self.reset_alert_box)
# -----------------------------------------------------------------------------
# 3. 비즈니스 로직 및 이벤트 핸들러 함수들
# -----------------------------------------------------------------------------
def start_serial_thread(self):
"""시리얼 수신 스레드 가동 및 시그널-슬롯 연결"""
self.worker = SerialWorker(port="/dev/ttyACM0")
self.worker.motion_detected_signal.connect(self.handle_motion_detection)
self.worker.start()
def change_store_mode(self, mode):
"""[제어] 버튼 클릭 시 매장의 운영 모드를 전환"""
self.current_mode = mode
# self.db.update_store_mode(mode) # 2일 차 DB 연동 시 반영
if mode == "BUSINESS":
self.mode_status_label.setText("운영 상태: 영업 모드 (BUSINESS)")
self.mode_status_label.setStyleSheet(
"background-color: #2ECC71; color: white; border-radius: 10px; padding: 15px;")
self.reset_alert_box()
else:
self.mode_status_label.setText("운영 상태: 경비 보안 모드 (SECURITY)")
self.mode_status_label.setStyleSheet(
"background-color: #D35400; color: white; border-radius: 10px; padding: 15px;")
self.alert_box.setText("🔒 보안 모드 가동 중\n침입 감지 대기")
self.alert_box.setStyleSheet("background-color: #34495E; color: white; border-radius: 10px; padding: 30px;")
def handle_motion_detection(self):
"""[핵심 연동] 아두이노 스레드로부터 감지 신호를 받았을 때 실행되는 함수"""
current_time = datetime.now().strftime("%H:%M:%S") if 'datetime' in globals() else "14:23:45"
# 1. DB 적재 프로세스 호출 (팀원 B 코드 연동 포인트)
# self.db.insert_sensor_log(device_id="ARDUINO_01", current_mode=self.current_mode)
# 2. 실시간 모드별 화면 시각화 예외 분기 처리
if self.current_mode == "BUSINESS":
# 영업 모드: 방문객 카운트 업 및 초록빛 알림
self.today_visitor_count += 1
self.counter_label.setText(f"오늘 총 방문객 수: {self.today_visitor_count} 명")
self.alert_box.setText("👣 방문객 입장!\n환영합니다.")
self.alert_box.setStyleSheet(
"background-color: #E8F8F5; color: #117864; border: 2px solid #2ECC71; border-radius: 10px; padding: 30px;")
self.alert_timer.start(3000) # 3초 후 평시 상태로 복구
elif self.current_mode == "SECURITY":
# 보안 모드: 무단 침입 경고 빨간색 연출 및 경고창(Modal Pop-up) 발생
self.alert_box.setText("🚨🚨 침입 발생!! 🚨🚨\n경보 시스템 가동")
self.alert_box.setStyleSheet(
"background-color: #FADBD8; color: #78281F; border: 3px solid #E74C3C; border-radius: 10px; padding: 30px;")
# 아두이노 비프음 발생 신호 송신을 원할 시 여기서 팀원 A 클래스 메서드 호출
# self.worker.manager.send_trigger_to_buzzer()
# 크리티컬 예외 상황 팝업 메시지 출력
QMessageBox.critical(self, "🚨 무단 침입 경보", "보안 모드 중 움직임이 감지되었습니다!\n즉시 관리자 확인이 필요합니다.")
# 3. 우측 실시간 테이블 로그 행 추가
row_position = self.log_table.rowCount()
self.log_table.insertRow(row_position)
self.log_table.setItem(row_position, 0, QTableWidgetItem(f"2026-06-03 {current_time}"))
self.log_table.setItem(row_position, 1, QTableWidgetItem("ARDUINO_01"))
self.log_table.setItem(row_position, 2, QTableWidgetItem(self.current_mode))
self.log_table.scrollToBottom()
def reset_alert_box(self):
"""평시 모니터링 화면으로 색상 원상 복구"""
if self.current_mode == "BUSINESS":
self.alert_box.setText("정상 운영 중\n(움직임 없음)")
self.alert_box.setStyleSheet(
"background-color: #F2F4F4; color: #34495E; border: 2px solid #BDC3C7; border-radius: 10px; padding: 30px;")
def closeEvent(self, event):
"""프로그램 종료 시 스레드를 안전하게 반환하여 프로세스 꼬임 방지"""
self.worker.stop()
event.accept()
# --- 메인 애플리케이션 실행 프로세스 ---
if __name__ == "__main__":
# 데이터 가공용 시간 라이브러리 임포트
from datetime import datetime
④ main_view.py
# PyQt5 메인 화면 및 이벤트 슬롯 소스코드 (main_view.py)
# 특징: 팀원 A의 시리얼 매니저를 서브 스레드로 구동하고, pyqtSignal을 통해 센서감지 이벤트 실시간 렌더링 함수로 중계
import sys
from PyQt5.QtWidgets import QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QSplitter
from PyQt5.QtCore import QThread, pyqtSignal, QTimer, Qt
from PyQt5.QtGui import QFont
# [팀원 C 단독 구현] 시리얼 이벤트를 UI 스레드와 격리하기 위한 비동기 워커
class SerialListenerWorker(QThread):
# 메인 UI 스레드와 통신할 비동기 신호 정의
sensor_trigger = pyqtSignal()
def __init__(self, serial_mgr):
super().__init__()
self.sm = serial_mgr
def run(self):
"""백그라운드에서 시리얼 대기열을 감시하며 UI 메인 스레드 방해 금지"""
# 팀원 A가 만든 시리얼 매니저의 무한 루프 수신기를 콜백 형식으로 구동
if self.sm:
self.sm.start_reading(db_callback=self.sensor_trigger.emit)
class MainWindow(QMainWindow):
def __init__(self, serial_mgr, db_mgr):
super().__init__()
self.sm = serial_mgr
self.db = db_mgr
self.init_ui()
self.init_sub_thread()
def init_ui(self):
self.setWindowTitle("무인 매장 보안 모니터링 시스템 v1.0")
self.setGeometry(100, 100, 1100, 650)
# 메인 레이아웃 및 스플리터 배치
self.main_widget = QWidget()
self.setCentralWidget(self.main_widget)
self.layout = QHBoxLayout(self.main_widget)
self.splitter = QSplitter(Qt.Horizontal)
# 좌측 레이아웃 관할 (팀원 C 영역)
self.left_panel = QWidget()
self.left_layout = QVBoxLayout(self.left_panel)
self.title = QLabel("🏪 실시간 매장 관리 서버")
self.title.setFont(QFont("Malgun Gothic", 16, QFont.Bold))
self.left_layout.addWidget(self.title)
self.status_display = QLabel("정상 모니터링 중")
self.status_display.setStyleSheet(
"background-color: #2ECC71; color: white; padding: 20px; font-size: 14pt; border-radius: 8px;")
self.status_display.setAlignment(Qt.AlignCenter)
self.left_layout.addWidget(self.status_display)
self.splitter.addWidget(self.left_panel)
self.layout.addWidget(self.splitter)
# 상태 복구용 일회성 타이머
self.recovery_timer = QTimer()
self.recovery_timer.setSingleShot(True)
self.recovery_timer.timeout.connect(self.restore_ui_status)
def init_sub_thread(self):
"""팀원 A의 H/W 제어기를 서브 스레드로 이식하여 UI 랙 현상 원천 차단"""
self.worker = SerialListenerWorker(self.sm)
self.worker.sensor_trigger.connect(self.on_motion_detected)
self.worker.start()
def on_motion_detected(self):
"""[실시간 UI 갱신 슬롯] 아두이노 신호 수신 시 실시간 호출"""
current_mode = self.db.get_current_mode() if self.db else "BUSINESS"
# 실시간 데이터베이스 적재 (팀원 B 모듈 호출)
if self.db:
self.db.insert_sensor_log()
if current_mode == "BUSINESS":
self.status_display.setText("👣 방문객 입장 감지")
self.status_display.setStyleSheet(
"background-color: #3498DB; color: white; padding: 20px; font-size: 14pt; border-radius: 8px;")
self.recovery_timer.start(2500) # 2.5초 후 초기화
else:
self.status_display.setText("🚨 무단 침입 발생! 경보 가동")
self.status_display.setStyleSheet(
"background-color: #E74C3C; color: white; padding: 20px; font-size: 14pt; border-radius: 8px;")
# 물리 경보 작동 명령 송신 (팀원 A 모듈 호출)
if self.sm: self.sm.send_trigger_to_hardware('W')
def restore_ui_status(self):
self.status_display.setText("정상 모니터링 중")
self.status_display.setStyleSheet(
"background-color: #2ECC71; color: white; padding: 20px; font-size: 14pt; border-radius: 8px;")
⑤ app_runner.py
# 팀원 E : 리눅스 런타임 배포 인프라 및 AI 전처리 파이프라인 설계
# 구현 핵심 아키텍처팀원 E는 프로젝트 총괄(PM)로서 전체 소스코드 연동 통합 파일(app_runner.py)을 작성하고,
# 개발 완료 후 리눅스 가상 서버 환경에서 터미널 접속만으로 시스템 전체를 한 번에 빌드하고 구동시킬 수 있는
# 자동화 쉘 스크립트를 제작. 추후 적재된 sensor_logs 이력을 바탕으로 이상 징후 탐지 AI 모델 및 시계열 트래픽 예측 AI를
# 올릴 수 있도록 원천 정형 데이터를 엔지니어링하는 데이터 파이프라인 뼈대를 구축
import sys
from PyQt5.QtWidgets import QApplication
# 팀원 A, B, C, D가 작성한 개별 모듈 임포트
# from serial_manager import ArduinoSerialManager
# from database_manager import DatabaseManager
# from main_view import MainWindow
# from sub_control_view import AnalyticsControlPanel
class AIDataPipeline:
"""[AI 확장성 전담 클래스] 적재된 시계열 로그 가공 전처리 파이프라인"""
def __init__(self, db_mgr):
self.db = db_mgr
def transform_logs_for_ai_training(self, raw_logs):
"""
[전처리 함수] AI 모델(LSTM/IsolationForest) 학습용 시계열 피처 엔지니어링 벡터 변환
Raw Data Format: [{'created_at': datetime, 'current_mode': 'BUSINESS'}]
"""
ai_feature_matrix = []
for log in raw_logs:
# 시간 데이터를 인공지능이 인식하기 좋은 수치형 다차원 독립 변수로 변환 (Feature Extraction)
log_time = log['created_at']
feature_vector = [
log_time.weekday(), # 요일 데이터 (0:월 ~ 6:일)
log_time.hour, # 시간대 (0~23)
log_time.minute, # 분 단위 트래픽
1 if log['current_mode'] == 'BUSINESS' else 0 # 범주형 변수 원핫인코딩 처리
]
ai_feature_matrix.append(feature_vector)
print(f"[AI Pipeline] 원천 로그 {len(raw_logs)}건 데이터셋(Dataset) 행렬 변환 완료.")
return ai_feature_matrix # 딥러닝 프레임워크 텐서 주입용 최종 넘파이/리스트 반환
if __name__ == "__main__":
app = QApplication(sys.argv)
# 1. 인프라 인스턴스 레이어 초기화
# db = DatabaseManager(host="localhost", user="root", password="password", database="store_db")
# sm = ArduinoSerialManager(port="/dev/ttyACM0")
# 디버깅용 Mock 객체 바인딩 (통합 테스트 가동용)
db, sm = None, None
# 2. 프론트엔드 렌더링 프레임 결합 (C팀원 뷰 + D팀원 뷰 결합)
main_ui = MainWindow(serial_mgr=sm, db_mgr=db)
sub_ui = AnalyticsControlPanel(main_window=main_ui, db_mgr=db, serial_mgr=sm)
# 메인 UI 스플리터 우측 도킹 레이아웃 완성
main_ui.splitter.addWidget(sub_ui)
main_ui.show()
# 3. AI 파이프라인 가동 가능성 검증 디버그 로그
pipeline = AIDataPipeline(db_mgr=db)
# fake_raw_logs = [{'created_at': datetime.now(), 'current_mode': 'BUSINESS'}]
# pipeline.transform_logs_for_ai_training(fake_raw_logs)
sys.exit(app.exec_())
⑥ database_manager.py
# PyMySQL 백엔드 쿼리 연동 소스코드 (database_manager.py)
# 특징: Connection Pool을 모방한 동적 커넥션 관리 방식을 채택하여 무인 매장의 비활성 시간대 대기 시 발생하는 연결 끊김(Timeout) 현상을
# 원천 차단함, GUI 팀원이 딕셔너리 형태로 데이터를 편리하게 다룰 수 있도록 DictCursor를 설정함
import pymysql
from pymysql import MySQLError
class DatabaseManager:
def __init__(self, host="127.0.0.1", user="root", password="your_password", database="store_db"):
self.config = {
"host": host,
"user": user,
"password": password,
"database": database,
"charset": "utf8mb4",
"cursorclass": pymysql.cursors.DictCursor # 데이터를 딕셔너리 형태로 반환
}
def _get_connection(self):
"""매 요청마다 안전한 연결 인스턴스를 반환하여 무중단 상태 유지"""
try:
return pymysql.connect(**self.config)
except MySQLError as e:
print(f"[DB 오류] MySQL 서버 연결 실패 (네트워크/인증 에러): {e}")
return None
def get_current_mode(self):
"""팀원 A, C 연동용: 현재 매장의 운영 모드(영업/보안) 상태 조회"""
conn = self._get_connection()
if not conn: return "BUSINESS"
try:
with conn.cursor() as cursor:
sql = "SELECT operation_mode FROM store_settings LIMIT 1;"
cursor.execute(sql)
result = cursor.fetchone()
return result['operation_mode'] if result else "BUSINESS"
except MySQLError as e:
print(f"[DB 오류] 모드 조회 쿼리 실패: {e}")
return "BUSINESS"
finally:
conn.close()
def update_store_mode(self, target_mode):
"""팀원 D 연동용: GUI 제어 셸에서 모드 변경 시 DB 마스터 테이블 갱신"""
if target_mode not in ["BUSINESS", "SECURITY"]: return False
conn = self._get_connection()
if not conn: return False
try:
with conn.cursor() as cursor:
sql = "UPDATE store_settings SET operation_mode = %s WHERE setting_id = 1;"
cursor.execute(sql, (target_mode,))
conn.commit()
print(f"[DB 변경] 매장 모드가 [{target_mode}]로 업데이트되었습니다.")
return True
except MySQLError as e:
conn.rollback()
print(f"[DB 오류] 모드 변경 트랜잭션 실패: {e}")
return False
finally:
conn.close()
def insert_sensor_log(self, device_id="ARDUINO_01", sensor_type="PIR", detected_value=1):
"""팀원 A 시리얼 이벤트 연동용: 동작 감지 발생 시 이력 안전 적재"""
current_mode = self.get_current_mode() # 적재 시점의 상태값 스냅샷 확보
conn = self._get_connection()
if not conn: return False
try:
with conn.cursor() as cursor:
sql = """
INSERT INTO sensor_logs (device_id, sensor_type, detected_value, current_mode)
VALUES (%s, %s, %s, %s);
"""
cursor.execute(sql, (device_id, sensor_type, detected_value, current_mode))
conn.commit()
return True
except MySQLError as e:
conn.rollback() # 예외 발생 시 트랜잭션 롤백으로 데이터 원자성 보장
print(f"[DB 오류] 로그 저장 트랜잭션 롤백 실행: {e}")
return False
finally:
conn.close()
def get_hourly_traffic_stats(self):
"""팀원 D 통계 그래프 연동용: 최근 24시간 동안의 시간대별 유입량 집계 통계 조회"""
conn = self._get_connection()
if not conn: return []
try:
with conn.cursor() as cursor:
sql = """
SELECT
HOUR(created_at) AS event_hour,
COUNT(*) AS visitor_count
FROM sensor_logs
WHERE current_mode = 'BUSINESS'
AND created_at >= DATE_SUB(NOW(), INTERVAL 1 DAY)
GROUP BY HOUR(created_at)
ORDER BY event_hour ASC;
"""
cursor.execute(sql)
return cursor.fetchall()
except MySQLError as e:
print(f"[DB 오류] 시간별 대시보드 통계 조회 실패: {e}")
return []
finally:
conn.close()
⑦ DatabaseManager.py
# MySQL 데이터베이스 연동 및 관리를 위한 클래스 (DatabaseManager)
import pymysql
from pymysql import MySQLError
from datetime import datetime
class DatabaseManager:
def __init__(self, host="localhost", user="store_user", password="your_password", database="store_db"):
"""
[초기화] 데이터베이스 접속 정보를 설정합니다.
charset을 utf8mb4로 설정하여 이모지 및 다양한 문자열을 지원합니다.
cursorclass를 DictCursor로 설정하여 조회 결과를 딕셔너리 형태로 받아옵니다. (GUI 연동 시 편리)
"""
self.config = {
"host": host,
"user": user,
"password": password,
"database": database,
"charset": "utf8mb4",
"cursorclass": pymysql.cursors.DictCursor # 결과를 튜플이 아닌 딕셔너리 형식으로 반환
}
def _get_connection(self):
"""
[내부 메서드] 호출 시마다 새로운 DB 커넥션 객체를 생성하여 반환합니다.
매번 새로 연결함으로써 Timeout(연결 끊김) 문제를 방지합니다.
"""
try:
return pymysql.connect(**self.config)
except MySQLError as e:
print(f"[DB 연결 실패] MySQL 서버 접속 정보를 확인하세요: {e}")
return None
def get_current_mode(self):
"""
[조회] store_settings 테이블에서 현재 매장의 운영 모드(영업/보안 등)를 조회합니다.
"""
conn = self._get_connection()
if not conn: return "BUSINESS" # DB 연결 실패 시 기본값으로 'BUSINESS' 모드 반환
try:
with conn.cursor() as cursor:
# 최신 설정 데이터 1건을 조회합니다.
sql = "SELECT operation_mode FROM store_settings LIMIT 1;"
cursor.execute(sql)
result = cursor.fetchone()
# 결과가 있으면 운영 모드를 반환하고, 없으면 기본값을 반환합니다.
return result['operation_mode'] if result else "BUSINESS"
except MySQLError as e:
print(f"[SQL 오류] 운영 모드 조회 실패: {e}")
return "BUSINESS"
finally:
if conn: conn.close() # 사용한 커넥션은 반드시 닫아줍니다.
def insert_sensor_log(self, device_id="ARDUINO_01", sensor_type="PIR", detected_value=1):
"""
[등록] 아두이노 등 기기에서 감지된 센서 로그를 DB에 실시간으로 저장합니다.
"""
# 로그를 저장하기 전, 현재 매장이 어떤 운영 모드인지 먼저 파악합니다.
current_mode = self.get_current_mode()
conn = self._get_connection()
if not conn: return False
try:
with conn.cursor() as cursor:
# 센서 ID, 센서 종류, 감지된 값, 그리고 감지 당시의 운영 모드를 함께 저장합니다.
sql = """
INSERT INTO sensor_logs (device_id, sensor_type, detected_value, current_mode)
VALUES (%s, %s, %s, %s);
"""
cursor.execute(sql, (device_id, sensor_type, detected_value, current_mode))
conn.commit() # 데이터 변경 사항을 실제 DB에 영구 반영
print(f"[DB 저장 성공] 센서 로그 등록 완료 (현재 모드: {current_mode})")
return True
except MySQLError as e:
conn.rollback() # 오류 발생 시 이전 상태로 복구(롤백)하여 데이터 무결성 유지
print(f"[SQL 오류] 센서 로그 등록 실패 (Rollback 실행): {e}")
return False
finally:
if conn: conn.close()
def get_hourly_traffic_stats(self):
"""
[통계] GUI 프로그램의 차트(Matplotlib 등) 시각화를 위한 최근 24시간 시간대별 방문자 수 통계를 조회합니다.
"""
conn = self._get_connection()
if not conn: return []
try:
with conn.cursor() as cursor:
# 영업(BUSINESS) 모드일 때 발생한 최근 1일(24시간) 동안의 로그를 시간대별로 그룹화하여 카운트합니다.
sql = """
SELECT
HOUR(created_at) AS event_hour,
COUNT(*) AS visitor_count
FROM sensor_logs
WHERE current_mode = 'BUSINESS'
AND created_at >= DATE_SUB(NOW(), INTERVAL 1 DAY)
GROUP BY HOUR(created_at)
ORDER BY event_hour ASC;
"""
cursor.execute(sql)
# [{'event_hour': 14, 'visitor_count': 5}, ...] 구조의 리스트 목록을 반환합니다.
return cursor.fetchall()
except MySQLError as e:
print(f"[SQL 오류] 시간대별 통계 조회 실패: {e}")
return []
finally:
if conn: conn.close()
# --- 코드 동작 테스트를 위한 실행 가이드 ---
if __name__ == "__main__":
# 1. DB 매니저 객체 생성 (실제 사용 시 본인의 MySQL 계정 정보 입력 필요)
db = DatabaseManager(host="127.0.0.1", user="root", password="password", database="store_db")
# 2. 현재 운영 모드 확인 테스트
print(f"[테스트 1] 현재 매장 운영 모드: {db.get_current_mode()}")
# 3. 센서 감지 이벤트 발생 시 로그 저장 테스트
print("\n[테스트 2] 아두이노 PIR 센서 감지 로그 저장 시도...")
db.insert_sensor_log(device_id="ARDUINO_01", sensor_type="PIR", detected_value=1)
# 4. 차트 UI 표기를 위한 통계 데이터 추출 테스트
print("\n[테스트 3] GUI 차트용 24시간 통계 데이터 조회...")
stats = db.get_hourly_traffic_stats()
for row in stats:
print(f"시간대: {row['event_hour']}시 -> 방문 횟수: {row['visitor_count']}회")
⑧ serial_manager_A.py
# PC측 파이썬 시리얼 중계 소스코드
# 특징: 리눅스 환경에서 아두이노 포트를 자동 탐색하며, 물리적 탈착 에러가 발생해도 프로세스가 죽지 않고
# 무한 재연결을 시도하는 내결함성(Fault-tolerance) 예외 처리가 반영됨
import time
import serial
import serial.tools.list_ports
class ArduinoSerialManager:
def __init__(self, port="/dev/ttyACM0", baudrate=9600, timeout=1):
self.port = port
self.baudrate = baudrate
self.timeout = timeout
self.ser = None
self.is_running = False
def find_arduino_port(self):
"""리눅스 시스템 내 아두이노 USB 장치 자동 식별"""
ports = list(serial.tools.list_ports.comports())
for p in ports:
if "Arduino" in p.description or "ch340" in p.description.lower() or "ttyACM" in p.device:
print(f"[HW] 장치 탐색 완료: {p.device}")
return p.device
return self.port
def connect(self):
self.port = self.find_arduino_port()
try:
self.ser = serial.Serial(port=self.port, baudrate=self.baudrate, timeout=self.timeout)
self.is_running = True
print(f"[HW] 아두이노 연결 성공 -> {self.port}")
return True
except serial.SerialException as e:
print(f"[🚨 연결 실패] 포트 열기 오류: {e}")
self.ser = None
return False
def start_reading(self, db_callback):
"""무한 루프 데이터 수신 및 유실 예외 처리"""
if not self.ser and not self.connect():
return
time.sleep(2) # 아두이노 리셋 대기
self.ser.reset_input_buffer()
while self.is_running:
try:
if self.ser and self.ser.in_waiting > 0:
line = self.ser.readline().decode("utf-8").strip()
if line == "DETECT":
print("[HW ➔ PC] 동작 감지 신호 포착")
db_callback() # 팀원 B(DB), C(GUI) 모듈과 인터페이스 연동
except (serial.SerialException, OSError):
print("\n[🚨 경고] 아두이노 USB 장치 연결 해제됨!")
self.handle_reconnect(db_callback)
break
except UnicodeDecodeError:
continue # 초기 노이즈 데이터 무시
def send_trigger_to_hardware(self, command_char):
"""PC ➔ 아두이노 물리 제어 신호 송신 ('W' or 'O')"""
if self.ser and self.ser.is_open:
try:
self.ser.write(command_char.encode('utf-8'))
print(f"[PC ➔ HW] 제어 명령 송신: {command_char}")
except Exception as e:
print(f"[HW] 제어 명령 송신 실패: {e}")
def handle_reconnect(self, db_callback):
"""자동 재연결 메커니즘"""
self.is_running = False
if self.ser: self.ser.close()
print("[HW] 5초 간격으로 재연결 복구를 시도합니다...")
while not self.is_running:
time.sleep(5)
if self.connect():
self.start_reading(db_callback)
break
def close(self):
self.is_running = False
if self.ser and self.ser.is_open:
self.ser.close()
print("[HW] 시리얼 포트 자원 반환 완료.")
⑨ sub_control_view.py
# PyQt5 대시보드 통계 및 원격 제어 소스코드 (sub_control_view.py)
# 특징: QTimer를 이용하여 10초 주기로 DB 백엔드 통계를 자동 리프레시(Auto-Refresh)하며 그래프 캔버스를 다시 그린다
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QPushButton, QHBoxLayout, QMessageBox
from PyQt5.QtCore import QTimer
import matplotlib.pyplot as plt
from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
class AnalyticsControlPanel(QWidget):
def __init__(self, main_window, db_mgr, serial_mgr):
super().__init__()
self.main_win = main_window
self.db = db_mgr
self.sm = serial_mgr
self.init_sub_ui()
def init_sub_ui(self):
self.sub_layout = QVBoxLayout(self)
# 1. 원격 제어 토글 버튼 레이아웃
ctrl_layout = QHBoxLayout()
self.btn_biz = QPushButton("영업 모드")
self.btn_sec = QPushButton("보안 강화")
self.btn_biz.clicked.connect(lambda: self.toggle_mode("BUSINESS"))
self.btn_sec.clicked.connect(lambda: self.toggle_mode("SECURITY"))
ctrl_layout.addWidget(self.btn_biz)
ctrl_layout.addWidget(self.btn_sec)
self.sub_layout.addLayout(ctrl_layout)
# 2. Matplotlib 차트 내장 인프라 생성
self.figure, self.ax = plt.subplots(figsize=(5, 3))
self.canvas = FigureCanvas(self.figure)
self.sub_layout.addWidget(self.canvas)
# 3. 10초 주기 실시간 데이터 자동 동기화 타이머 가동
self.refresh_timer = QTimer()
self.refresh_timer.timeout.connect(self.update_statistics_chart)
self.refresh_timer.start(10000)
self.update_statistics_chart() # 최초 1회 실행
def toggle_mode(self, mode):
"""[역방향 물리 제어] 제어 명령 하향 전파"""
if self.db:
self.db.update_store_mode(mode)
if mode == "BUSINESS":
if self.sm: self.sm.send_trigger_to_hardware('O') # 아두이노 경보 해제 명령 ('O')
QMessageBox.information(self, "모드 변경", "정상 운영 모드로 전환되었습니다.")
else:
QMessageBox.warning(self, "모드 변경", "보안 감시 체제로 전환되었습니다. 침입 발생 시 경보가 울립니다.")
# 메인 윈도우 상태 창 즉시 동기화 촉진
self.main_win.restore_ui_status()
def update_statistics_chart(self):
"""[시각화 Engine] DB 통계 기반 데이터 동적 드로잉"""
if not self.db: return
# 팀원 B가 구현한 시간대별 딕셔너리 리스트 취득
db_data = self.db.get_hourly_traffic_stats()
hours = [row['event_hour'] for row in db_data]
counts = [row['visitor_count'] for row in db_data]
# 데이터가 없을 시 가상 예외 방어 데이터 할당
if not hours:
hours, counts = [12, 13, 14], [0, 0, 0]
self.ax.clear()
self.ax.bar(hours, counts, color='#3498DB', edgecolor='#2980B9')
self.ax.set_title("Hourly Visitor Traffic (Last 24 Hours)", fontsize=10)
self.ax.set_xlabel("Hour (24h)", fontsize=8)
self.ax.set_ylabel("Detections", fontsize=8)
self.ax.grid(True, axis='y', linestyle='--', alpha=0.7)
self.figure.tight_layout()
self.canvas.draw() # 캔버스 리렌더링
⑩ loTMemberControl.sql
// 리눅스 서버 환경에서 활용가능한 sql 코드
-- 리눅스 서버 환경의 MySQL 8.0 표준 엔진에 맞추어 작성되었으며, 다국어 지원 및 이모지 로그 저장을 위해 utf8mb4 인코딩을 적용
-- 1. 데이터베이스 생성 및 선택
CREATE DATABASE IF NOT EXISTS store_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE store_db;
-- 2. 매장 상태 관리 테이블 (마스터 테이블)
CREATE TABLE IF NOT EXISTS store_settings (
setting_id INT AUTO_INCREMENT PRIMARY KEY,
store_name VARCHAR(50) NOT NULL COMMENT '매장 지점명',
operation_mode VARCHAR(20) NOT NULL DEFAULT 'BUSINESS' COMMENT '현재 모드 (BUSINESS: 영업, SECURITY: 보안)',
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '상태 변경 시간'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 시스템 가동을 위한 기본 데이터 삽입 (기본값: 영업 모드)
INSERT INTO store_settings (store_name, operation_mode)
VALUES ('광주 AI 무인스토어 1호점', 'BUSINESS')
ON DUPLICATE KEY UPDATE updated_at = NOW();
-- 3. PIR 센서 감지 이력 테이블 (트랜잭션 데이터 테이블)
CREATE TABLE IF NOT EXISTS sensor_logs (
log_id BIGINT AUTO_INCREMENT PRIMARY KEY,
device_id VARCHAR(20) NOT NULL DEFAULT 'ARDUINO_01' COMMENT '식별용 아두이노 ID',
sensor_type VARCHAR(20) NOT NULL DEFAULT 'PIR' COMMENT '센서 종류',
detected_value INT NOT NULL DEFAULT 1 COMMENT '감지 값 (1: 움직임 발생)',
current_mode VARCHAR(20) NOT NULL COMMENT '감지 시점의 매장 운영 모드',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '감지 로그 시각',
-- 성능 최적화 인덱스: GUI 실시간 대시보드 및 AI 시계열 예측 가속화용
INDEX idx_created_at (created_at),
INDEX idx_mode_created (current_mode, created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 4. [AI 연동 대비] 시간대별 방문객 집계 테이블
CREATE TABLE IF NOT EXISTS hourly_statistics (
stat_id INT AUTO_INCREMENT PRIMARY KEY,
stat_date DATE NOT NULL COMMENT '통계 날짜',
stat_hour INT NOT NULL COMMENT '통계 시간 (0~23시)',
visitor_count INT NOT NULL DEFAULT 0 COMMENT '해당 시간 총 누적 방문 수',
anomaly_score FLOAT DEFAULT 0.0 COMMENT 'AI 모델 결과 이상치 점수',
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
-- 데이터 중복 및 왜곡 방지를 위한 유니크 제약
UNIQUE KEY uq_date_hour (stat_date, stat_hour)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
※ script_team.txt (참고사항 기록)
# main.dashbord.py 실행 전 터미널에 pip install PyQt5 matplotlib pymysql pyserial이 설치되어 있는지 확인하기
dashboard = SmartStoreDashboard()
dashboard.show()
sys.exit(app.exec_())
A팀원 : 리눅스(Linux) 인프라 환경 설정 조치 내용
=========================================
# 1. 아두이노가 연결된 시리얼 장치 명칭 및 연결 상태 확인
ls -l /dev/ttyACM*
# 2. 파이썬 프로세스가 시리얼 포트에 접근할 수 있도록 읽기/쓰기 권한 강제 부여
sudo chmod 666 /dev/ttyACM0
# 3. [고도화] 매번 권한을 주기 번거로우므로 현재 계정(User)을 시리얼 다이얼아웃 그룹에 등록 (재부팅 후 영구 적용)
sudo usermod -a -G dialout $USER
B팀원 : 리눅스 기반 데이터베이스 원격 보안 설정 조치
==========================================
# 1. 리눅스 터미널에서 MySQL 설정 파일 편집 (IP 바인딩 해제)
# sudo nano /etc/mysql/mysql.conf.d/mysqld.cnf
# bind-address = 127.0.0.1 부분을 0.0.0.0 으로 수정 후 저장
# 2. 리눅스 방화벽(UFW)에서 MySQL 기본 포트(3306) 허용 조치
sudo ufw allow 3306/tcp
sudo ufw reload
# 3. MySQL 내부 접속 후 외부 접속용 전용 계정 생성 및 권한 부여
# CREATE USER 'store_user'@'%' IDENTIFIED BY 'your_password';
# GRANT ALL PRIVILEGES ON store_db.* TO 'store_user'@'%';
# FLUSH PRIVILEGES;
# E팀원 : 리눅스 런타임 배포 인프라 및 AI 전처리 파이프라인 설계 보고서
==========================================================
# 구현 핵심 아키텍처팀원 E는 프로젝트 총괄(PM)로서 전체 소스코드 연동 통합 파일(app_runner.py)을 작성하고,
# 개발 완료 후 리눅스 가상 서버 환경에서 터미널 접속만으로 시스템 전체를 한 번에 빌드하고 구동시킬 수 있는
# 자동화 쉘 스크립트를 제작. 추후 적재된 sensor_logs 이력을 바탕으로 이상 징후 탐지 AI 모델 및 시계열 트래픽 예측
# 원천 정형 데이터를 엔지니어링하는 데이터 파이프라인 뼈대를 구축
# 무인 보안 시스템 백업 및 의존성 환경 원클릭 자동 구축 스크립트
echo "[인프라] 패키지 인덱스 업데이트 및 필수 종속성 파이썬 라이브러리 검증 시작..."
sudo apt-get update -y
# PyQt5가 리눅스 환경 GUI 환경 가상 디스플레이(X11) 플러그인을 원활히 찾을 수 있도록 환경 변수 선언
export QT_QPA_PLATFORM=xcb
# 가상 환경 인프라 구축 및 파이썬 패키지 일괄 무중단 설치
echo "[인프라] PyMySQL, PySerial, PyQt5, Matplotlib 인프라 종속성 다운로드 중..."
pip install pymysql pyserial PyQt5 matplotlib --quiet
echo "[인프라] USB 아두이노 통신 디바이스 권한 영구 개방 설정 조치"
sudo chmod 666 /dev/ttyACM0 || echo "[경고] 물리 아두이노 장치가 포트에 잡히지 않았습니다. 시뮬레이션 모드로 대치합니다."
echo "[인프라] 스마트 매장 보안 모니터링 시스템 어플리케이션을 백그라운드로 실행합니다."
nohup python3 app_runner.py > system_run.log 2>&1 &
echo "[인프라] 배포 성공! 실시간 런타임 로그 확인: 'tail -f system_run.log'"
# 핵심 기술 설명 (팀원 C, D 공유용)
==============================
1. 비동기 이벤트 기반 스레드 설계 구조 (`QThread` & `pyqtSignal`):
"파이썬 GUI 컴포넌트들은 싱글 스레드로 작동하기 때문에, 무한 루프로 시리얼 통신을 감시하면 화면 전체가 멈추는(Freezing) 크리티컬 락 현상이
발생함. 이를 해결하기 위해 **시리얼 수신 로직을 백그라운드 멀티스레드(`QThread`)로 완전히 분리하고, 데이터가 수집된 순간에만 인터럽트
신호(`pyqtSignal`)를 보내 화면을 렌더링**하는 안정적인 아키텍처를 구현함."
2. 시뮬레이터(Mock 데이터) 환경 준비 능력 :
"2일이라는 제한된 시간 속에서 하드웨어 제작과 소프트웨어 UI 개발이 병렬로 진행되어야 했음. 하드웨어가 완비되지 않은 시점에도 소프트웨어 결합성
테스트를 조기에 진행할 수 있도록 **가상 센서 이벤트 발생기(Random Sleep Loop)를 스레드 내부에 모킹(Mocking)하여 프로젝트 지연 리스크를 최소화
※ 팀원 5인이 무인 매장 보안 토이 프로젝트를 완수하기 위한 문서들
(전체 기획안 ➔ 일정 계획서 ➔ 데이터베이스 DDL ➔ 시리얼 통신 예외 처리 코드 ➔ PyQt5 비동기 GUI 소스코드)끝.
'학습 로그 > Project' 카테고리의 다른 글
| D+49 [Project] 개인 프로젝트를 팀 통합파일로 만들기 실습 (0) | 2026.06.17 |
|---|---|
| D+46 [Project] 프로젝트에서 WBS 무엇일까? (0) | 2026.06.14 |
| D+44 [Project] 프로그램 개발 산출물 작성 방법 (0) | 2026.06.13 |
| D+43 [Project] 프로젝트 기획자(PM)에 대한 이해 (0) | 2026.06.13 |
| D+37 [Project] Model View 실습 & 개인 프로젝트(1) (0) | 2026.05.31 |