RoboDine 데이터베이스 설계 - SQLModel과 관계형 데이터 모델링

# models/enums.py
from enum import Enum
from sqlalchemy import Enum as SQLEnum

class RobotStatus(str, Enum):
    """로봇 상태 관리"""
    IDLE = "IDLE"
    SETTING = "SETTING"
    COOKING = "COOKING"
    PICKUP = "PICKUP"
    SERVING = "SERVING"
    CLEANING = "CLEANING"
    EMERGENCY = "EMERGENCY"
    MAINTENANCE = "MAINTENANCE"
    SECURITY = "SECURITY"
    BIRTHDAY = "BIRTHDAY"
    CALLING = "CALLING"
    ERROR = "ERROR"

class OrderStatus(str, Enum):
    """주문 상태 관리"""
    PLACED = "PLACED"
    PREPARING = "PREPARING"
    COMPLETED = "COMPLETED"
    SERVED = "SERVED"
    CANCELLED = "CANCELLED"

class TableStatus(str, Enum):
    """테이블 상태 관리"""
    AVAILABLE = "AVAILABLE"
    OCCUPIED = "OCCUPIED"
    CLEANING = "CLEANING"

class InventoryStatus(str, Enum):
    """재고 상태 관리"""
    IN_STOCK = "IN_STOCK"
    LOW_STOCK = "LOW_STOCK"
    OUT_OF_STOCK = "OUT_OF_STOCK"

class EntityType(str, Enum):
    """엔티티 타입 분류"""
    COOKBOT = "COOKBOT"
    ALBABOT = "ALBABOT"
    PINKY = "PINKY"
    GLOBAL = "GLOBAL"
    WORLD = "WORLD"
    INVENTORY = "INVENTORY"
    CHATBOT = "CHATBOT"

# models/robot.py
from typing import Optional, List, TYPE_CHECKING
from datetime import datetime
from sqlmodel import SQLModel, Field, Relationship
from sqlalchemy import Column, Enum as SQLEnum
from .enums import RobotStatus, EntityType

class Robot(SQLModel, table=True):
    __tablename__ = "robot"

    id: Optional[int] = Field(default=None, primary_key=True)
    robot_id: Optional[int] = Field(default=None, index=True)
    type: Optional[EntityType] = Field(
        sa_column=Column(SQLEnum(EntityType, name="entity_type"))
    )
    mac_address: Optional[str] = None
    ip_address: Optional[str] = None
    timestamp: Optional[datetime] = Field(default_factory=datetime.utcnow)

    # 관계 정의
    orders: List["Order"] = Relationship(back_populates="robot")
    commands: List["RobotCommand"] = Relationship(back_populates="robot")
    cleaning_tasks: List["CleaningTask"] = Relationship(back_populates="robot")

# 로봇 명령 관리
class RobotCommand(SQLModel, table=True):
    __tablename__ = "robot_command"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    robot_id: Optional[int] = Field(default=None, foreign_key="robot.robot_id")
    command_type: Optional[str] = None
    payload: Optional[str] = None  # JSON 형태의 명령 데이터
    status: Optional[CommandStatus] = Field(
        sa_column=Column(SQLEnum(CommandStatus))
    )
    created_at: Optional[datetime] = Field(default_factory=datetime.utcnow)
    executed_at: Optional[datetime] = None
    
    robot: Optional[Robot] = Relationship(back_populates="commands")

# models/order.py
from sqlmodel import SQLModel, Field, Relationship, Column
from datetime import datetime
from typing import Optional, List
from sqlalchemy import Enum as SQLEnum
from .enums import OrderStatus

class Order(SQLModel, table=True):
    __tablename__ = "order"

    id: Optional[int] = Field(default=None, primary_key=True)
    customer_id: Optional[int] = Field(default=None, foreign_key="customer.id")
    robot_id: Optional[int] = Field(default=None, foreign_key="robot.robot_id")
    table_id: Optional[int] = Field(default=None)  # table FK 제거
    status: Optional[OrderStatus] = Field(sa_column=Column(SQLEnum(OrderStatus)))
    timestamp: Optional[datetime] = Field(default_factory=datetime.utcnow)
    served_at: Optional[datetime] = None

    # 관계 정의
    customer: Optional["Customer"] = Relationship(back_populates="orders")
    robot: Optional["Robot"] = Relationship(back_populates="orders")
    order_items: List["OrderItem"] = Relationship(back_populates="order")

class OrderItem(SQLModel, table=True):
    __tablename__ = "orderitem"

    order_id: Optional[int] = Field(default=None, foreign_key="order.id", primary_key=True)
    menu_item_id: Optional[int] = Field(default=None, foreign_key="menuitem.id", primary_key=True)
    quantity: Optional[int] = None
    status: Optional[OrderStatus] = Field(sa_column=Column(SQLEnum(OrderStatus)))

    # 관계 정의
    order: Optional[Order] = Relationship(back_populates="order_items")
    menu_item: Optional["MenuItem"] = Relationship(back_populates="order_items")

# models/inventory.py
from typing import Optional, List
from datetime import datetime
from sqlmodel import SQLModel, Field, Relationship
from sqlalchemy import Column, Enum as SQLEnum
from .enums import InventoryStatus

class MenuItem(SQLModel, table=True):
    __tablename__ = "menuitem"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    name: Optional[str] = None
    price: Optional[float] = None
    prepare_time: Optional[int] = None  # 조리 시간 (분)
    image_url: Optional[str] = None
    description: Optional[str] = None
    
    # 관계 정의
    menu_ingredients: List["MenuIngredient"] = Relationship(back_populates="menu_item")
    order_items: List["OrderItem"] = Relationship(back_populates="menu_item")

class MenuIngredient(SQLModel, table=True):
    __tablename__ = "menuingredient"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    name: Optional[str] = Field(default=None, unique=True)
    menu_item_id: Optional[int] = Field(default=None, foreign_key="menuitem.id")
    quantity_required: Optional[int] = None  # 필요 수량
    
    menu_item: Optional[MenuItem] = Relationship(back_populates="menu_ingredients")

class Inventory(SQLModel, table=True):
    __tablename__ = "inventory"

    id: Optional[int] = Field(default=None, primary_key=True)
    ingredient_id: Optional[int] = Field(default=None)
    name: Optional[str] = Field(default=None, foreign_key="menuingredient.name")
    count: Optional[int] = None  # 현재 재고
    max_count: Optional[int] = None  # 최대 재고
    status: Optional[InventoryStatus] = Field(
        sa_column=Column(SQLEnum(InventoryStatus, name="inventory_status"))
    )
    updated_at: Optional[datetime] = Field(default_factory=datetime.utcnow)

# models/table.py
from sqlmodel import SQLModel, Field, Relationship, Column
from typing import Optional, List
from datetime import datetime
from sqlalchemy import Enum as SQLEnum
from .enums import TableStatus

class Table(SQLModel, table=True):
    __tablename__ = "table"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    max_customer: Optional[int] = None
    status: Optional[TableStatus] = Field(sa_column=Column(SQLEnum(TableStatus)))
    updated_at: Optional[datetime] = Field(default_factory=datetime.utcnow)
    
    # 위치 정보 (매장 맵에서 사용)
    x: Optional[float] = None
    y: Optional[float] = None
    width: Optional[float] = None
    height: Optional[float] = None
    
    # 관계 정의
    assignments: List["GroupAssignment"] = Relationship(back_populates="table")

class Customer(SQLModel, table=True):
    __tablename__ = "customer"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    count: Optional[int] = Field(default=None)  # group_size 대신 count 사용
    timestamp: Optional[datetime] = Field(default_factory=datetime.utcnow)
    
    # 관계 정의
    orders: List["Order"] = Relationship(back_populates="customer")
    assignments: List["GroupAssignment"] = Relationship(back_populates="customer")

class GroupAssignment(SQLModel, table=True):
    __tablename__ = "groupassignment"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    table_id: Optional[int] = Field(default=None, foreign_key="table.id")
    customer_id: Optional[int] = Field(default=None, foreign_key="customer.id")
    timestamp: Optional[datetime] = Field(default_factory=datetime.utcnow)
    released_at: Optional[datetime] = None
    
    # 관계 정의
    table: Optional[Table] = Relationship(back_populates="assignments")
    customer: Optional[Customer] = Relationship(back_populates="assignments")

# 주문과 메뉴 항목의 다대다 관계를 OrderItem으로 해결
class OrderItem(SQLModel, table=True):
    order_id: int = Field(foreign_key="order.id")
    menu_item_id: int = Field(foreign_key="menuitem.id")
    quantity: int  # 수량 정보 추가
    unit_price: float  # 주문 시점의 가격 고정
    
    # 이를 통해 주문별 상세 내역과 가격 변동 대응

# routes/orders.py - 실제 구현된 주문 관리 API
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
from sqlalchemy.orm import Session
from typing import List

@router.get("/todo_order")
def get_todo_order(background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
    """쿡봇을 위한 다음 조리할 주문 조회"""
    
    # 대기 중인 주문 중 가장 먼저 생성된 주문 조회
    order = db.query(Order).order_by(Order.id).filter(
        Order.status != OrderStatus.CANCELLED
    ).filter(
        Order.status != OrderStatus.SERVED
    ).filter(
        Order.status != OrderStatus.COMPLETED
    ).first()
    
    if not order:
        raise HTTPException(404, detail="대기중인 주문이 없습니다.")
    
    # 주문 항목 조회
    items = db.query(OrderItem).filter(
        OrderItem.order_id == order.id
    ).filter(
        OrderItem.status == OrderStatus.PLACED
    ).all()
    
    if not items:
        raise HTTPException(404, detail=f"주문 항목이 없습니다, 대기중인 주문: {order.id}")
    
    # 메뉴 항목 조회
    menu_items = db.query(MenuItem).filter(
        MenuItem.id.in_([item.menu_item_id for item in items])
    ).all()
    
    # 웹소켓 브로드캐스트 예약
    from run import broadcast_entity_update
    background_tasks.add_task(broadcast_entity_update, "order", None)
    
    return {
        "order_id": order.id,
        "item_name": menu_items[0].name if menu_items else "Unknown"
    }

@router.post("", response_model=dict)
def create_order(
    order_data: dict,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    """새 주문 생성"""
    
    # 데이터 검증
    if "customer_id" not in order_data:
        raise HTTPException(400, detail="Customer ID is required")
    
    if "items" not in order_data or not order_data["items"]:
        raise HTTPException(400, detail="Order items are required")
    
    # 고객 존재 확인
    customer_id = order_data["customer_id"]
    customer = db.query(Customer).filter(Customer.id == customer_id).first()
    if not customer:
        raise HTTPException(404, detail=f"Customer with ID {customer_id} not found")
    
    # 주문 생성
    new_order = Order(
        customer_id=customer_id,
        table_id=order_data.get("table_id"),
        status=OrderStatus.PLACED,
        timestamp=datetime.utcnow()
    )
    
    db.add(new_order)
    db.flush()  # ID 생성을 위해 flush
    
    # 주문 항목 생성
    for item_data in order_data["items"]:
        order_item = OrderItem(
            order_id=new_order.id,
            menu_item_id=item_data["menu_item_id"],
            quantity=item_data["quantity"],
            status=OrderStatus.PLACED
        )
        db.add(order_item)
    
    db.commit()
    
    return {"status": "success", "order_id": new_order.id}

# routes/inventories.py - 실제 구현된 재고 관리 API
@router.get("/inventory/status")
async def get_inventory_status(session: Session = Depends(get_session)):
    """실시간 재고 상태 조회"""
    
    statement = (
        select(Inventory)
        .where(Inventory.status != InventoryStatus.IN_STOCK)
        .order_by(Inventory.updated_at.desc())
    )
    
    low_stock_items = session.exec(statement).all()
    
    # 재고 상태 자동 업데이트
    for item in low_stock_items:
        if item.count <= 0:
            item.status = InventoryStatus.OUT_OF_STOCK
        elif item.count <= item.min_count:
            item.status = InventoryStatus.LOW_STOCK
        else:
            item.status = InventoryStatus.IN_STOCK
    
    session.commit()
    
    return {
        "total_items": len(low_stock_items),
        "out_of_stock": len([i for i in low_stock_items if i.status == InventoryStatus.OUT_OF_STOCK]),
        "low_stock": len([i for i in low_stock_items if i.status == InventoryStatus.LOW_STOCK]),
        "items": low_stock_items
    }

@router.post("/inventory/{item_id}/restock")
async def restock_item(
    item_id: int, 
    quantity: int,
    session: Session = Depends(get_session)
):
    """재고 보충"""
    
    item = session.get(Inventory, item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Inventory item not found")
    
    item.count += quantity
    item.last_restocked_at = datetime.utcnow()
    
    # 상태 자동 업데이트
    if item.count >= item.max_count * 0.8:
        item.status = InventoryStatus.IN_STOCK
    elif item.count > item.min_count:
        item.status = InventoryStatus.LOW_STOCK
    
    session.add(item)
    session.commit()
    session.refresh(item)
    
    # 실시간 알림 전송
    await broadcast_inventory_update(item)
    
    return item

# routes/robot.py - 실제 구현된 로봇 제어 API
@router.post("/command")
def send_command(
    command_data: dict,
    db: Session = Depends(get_db)
):
    """로봇에게 명령 전송"""
    
    # 명령 데이터 검증
    robot_id = command_data.get("robot_id")
    command = command_data.get("command")
    parameters = command_data.get("parameters", {})
    
    if not robot_id or not command:
        raise HTTPException(400, detail="robot_id and command are required")
    
    # 로봇 존재 확인
    robot = db.query(Robot).filter(Robot.robot_id == robot_id).first()
    if not robot:
        raise HTTPException(404, detail=f"Robot {robot_id} not found")
    
    # 명령 생성
    new_command = RobotCommand(
        robot_id=robot_id,
        command=command,
        parameters=parameters,
        status=CommandStatus.PENDING,
        timestamp=datetime.utcnow()
    )
    
    db.add(new_command)
    db.commit()
    
    return {
        "id": new_command.id,
        "status": "success",
        "message": "명령이 성공적으로 전송되었습니다."
    }

# models/event.py - 실제 구현된 이벤트 모델
from sqlmodel import SQLModel, Field, Relationship
from datetime import datetime
from typing import Optional
from .enums import EventType, LogLevel, EntityType

class Event(SQLModel, table=True):
    __tablename__ = "event"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    type: Optional[EventType] = Field(sa_column=Column(SQLEnum(EventType)))
    entity_type: Optional[EntityType] = Field(sa_column=Column(SQLEnum(EntityType)))
    entity_id: Optional[int] = None
    table_id: Optional[int] = None
    description: Optional[str] = None
    timestamp: Optional[datetime] = Field(default_factory=datetime.utcnow)
    
class SystemLog(SQLModel, table=True):
    __tablename__ = "systemlog"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    level: Optional[LogLevel] = Field(sa_column=Column(SQLEnum(LogLevel)))
    message: Optional[str] = None
    source: Optional[str] = None  # 로그 발생 소스
    details: Optional[str] = None  # JSON 형태의 상세 정보
    timestamp: Optional[datetime] = Field(default_factory=datetime.utcnow)

# 이벤트 생성 헬퍼 함수
async def create_event(
    event_type: EventType,
    entity_type: EntityType,
    entity_id: int,
    description: str,
    table_id: Optional[int] = None,
    session: Session = None
):
    """시스템 이벤트 생성"""
    event = Event(
        type=event_type,
        entity_type=entity_type,
        entity_id=entity_id,
        table_id=table_id,
        description=description
    )
    
    session.add(event)
    session.commit()
    
    # WebSocket으로 실시간 브로드캐스트
    await broadcast_events_update(event)
    
    return event

# models/pose6d.py - 실제 구현된 위치 모델
from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional

class Pose6D(SQLModel, table=True):
    __tablename__ = "pose6d"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    entity_type: Optional[str] = None
    entity_id: Optional[int] = None
    
    # 6D 포즈 정보 (위치 + 회전)
    x: Optional[float] = None
    y: Optional[float] = None
    z: Optional[float] = None
    roll: Optional[float] = None
    pitch: Optional[float] = None
    yaw: Optional[float] = None
    
    timestamp: Optional[datetime] = Field(default_factory=datetime.utcnow)

# 위치 업데이트 API
@router.post("/robots/{robot_id}/update-position")
async def update_robot_position(
    robot_id: int,
    position_data: dict,
    session: Session = Depends(get_session)
):
    """로봇 위치 업데이트"""
    
    # 새로운 위치 정보 저장
    pose = Pose6D(
        entity_type="ROBOT",
        entity_id=robot_id,
        x=position_data.get("x"),
        y=position_data.get("y"),
        z=position_data.get("z"),
        roll=position_data.get("roll"),
        pitch=position_data.get("pitch"),
        yaw=position_data.get("yaw")
    )
    
    session.add(pose)
    session.commit()
    
    # 실시간 위치 브로드캐스트
    await broadcast_robot_position_update({
        "robot_id": robot_id,
        "position": position_data,
        "timestamp": pose.timestamp
    })
    
    return {"status": "success", "robot_id": robot_id}

# app/core/db_config.py - 실제 구현
from sqlmodel import SQLModel, create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy.orm import sessionmaker

# 동기 PostgreSQL 연결
SYNC_DATABASE_URL = "postgresql://robodine_user:robodine_pass@localhost:5432/robodine_db"
engine = create_engine(SYNC_DATABASE_URL, echo=False)

# 비동기 PostgreSQL 연결
ASYNC_DATABASE_URL = "postgresql+asyncpg://robodine_user:robodine_pass@localhost:5432/robodine_db"
async_engine = create_async_engine(ASYNC_DATABASE_URL, echo=False)

# 세션 관리
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
AsyncSessionLocal = sessionmaker(bind=async_engine, class_=AsyncSession)

def get_db():
    """동기 데이터베이스 세션 의존성"""
    with SessionLocal() as session:
        yield session

async def get_async_db():
    """비동기 데이터베이스 세션 의존성"""
    async with AsyncSessionLocal() as session:
        yield session

# app/services/streaming_service.py - 실제 구현
import cv2
import threading
from datetime import datetime

_stream_urls: list[str] = []
_active_streams = {}

def add_stream_url(url: str, path: str = None):
    """RTSP 스트림 URL 추가 및 녹화 시작"""
    if url not in _stream_urls:
        _stream_urls.append(url)
        
        stop_event = threading.Event()
        latencies = []
        _active_streams[url] = {"stop_event": stop_event, "latencies": latencies}
        
        # 별도 스레드에서 스트림 처리
        threading.Thread(
            target=record_stream, 
            args=(url, stop_event, latencies, path), 
            daemon=True
        ).start()

def record_stream(rtsp_url: str, stop_event: threading.Event, latencies: list, path: str):
    """실시간 스트림 녹화 및 H.264 변환"""
    cap = cv2.VideoCapture(rtsp_url)
    if not cap.isOpened():
        return
        
    # 비디오 설정
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    fps = cap.get(cv2.CAP_PROP_FPS) or 20.0
    
    # 녹화 및 H.264 변환 처리
    fourcc = cv2.VideoWriter_fourcc(*"mp4v")
    writer = cv2.VideoWriter(out_path, fourcc, fps, (width, height))
    
    while not stop_event.is_set():
        ret, frame = cap.read()
        if ret:
            writer.write(frame)
            
    cap.release()
    writer.release()

# app/services/ 디렉토리 구조
services/
├── streaming_service.py    # RTSP 스트리밍 처리 (107 lines)
├── cleaning_service.py     # 청소 관련 서비스 (14 lines)
├── emergency_service.py    # 비상 상황 처리 (13 lines)
├── ros_service.py         # ROS 통신 (18 lines)
├── auto_ordering.py       # 자동 주문 (미구현)
└── inventory_service.py   # 재고 관리 (미구현)

# 청소 서비스 예시
# app/services/cleaning_service.py
def start_cleaning_task(robot_id: int, area: str):
    """청소 작업 시작"""
    return {
        "robot_id": robot_id,
        "task": "cleaning",
        "area": area,
        "status": "started"
    }

# 비상 상황 서비스 예시  
# app/services/emergency_service.py
def trigger_emergency(emergency_type: str, location: str):
    """비상 상황 발생"""
    return {
        "type": emergency_type,
        "location": location,
        "timestamp": datetime.utcnow(),
        "status": "activated"
    }

# 주문 관리 API (routes/orders.py - 607 lines)
GET  /todo_order              # 쿡봇용 다음 주문 조회
GET  /orders                  # 주문 목록 조회  
POST /orders                  # 새 주문 생성
PUT  /orders/{id}/status      # 주문 상태 업데이트
POST /orders/{id}/assign-robot # 로봇 할당

# 로봇 제어 API (routes/robot.py - 384 lines)
GET  /robots                  # 로봇 목록 조회
POST /robots/register         # 로봇 등록
POST /robots/command          # 로봇 명령 전송
GET  /robots/commands         # 모든 명령 조회
PUT  /robots/commands/{id}/status # 명령 상태 업데이트

# 테이블 관리 API (routes/tables.py - 312 lines)
GET  /tables                  # 테이블 목록 조회
POST /tables/assign           # 테이블 배정
PUT  /tables/{id}/status      # 테이블 상태 변경

# 재고 관리 API (routes/inventories.py - 361 lines)
GET  /inventories             # 재고 목록 조회
POST /inventories             # 재고 추가
PUT  /inventories/{id}        # 재고 업데이트

# WebSocket 실시간 통신 (routes/websockets.py - 337 lines)
WS   /ws/{client_id}          # 실시간 업데이트 수신

# 실제 데이터베이스 설정 - PostgreSQL 연결
SYNC_DATABASE_URL = "postgresql://robodine_user:robodine_pass@localhost:5432/robodine_db"
engine = create_engine(SYNC_DATABASE_URL, echo=False)

# 비동기 데이터베이스 지원
ASYNC_DATABASE_URL = "postgresql+asyncpg://robodine_user:robodine_pass@localhost:5432/robodine_db"
async_engine = create_async_engine(ASYNC_DATABASE_URL, echo=False)

# 동기/비동기 세션 관리
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
AsyncSessionLocal = sessionmaker(bind=async_engine, class_=AsyncSession)

def get_db():
    """동기 데이터베이스 세션 의존성"""
    with SessionLocal() as session:
        yield session

# routes/orders.py에서 실제 구현된 검증
@router.post("", response_model=dict)
def create_order(
    order_data: dict,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    """새 주문 생성"""
    
    # 기본 데이터 검증
    if "customer_id" not in order_data:
        raise HTTPException(400, detail="Customer ID is required")
    
    if "items" not in order_data or not order_data["items"]:
        raise HTTPException(400, detail="Order items are required")
    
    # 고객 존재 확인
    customer_id = order_data["customer_id"]
    customer = db.query(Customer).filter(Customer.id == customer_id).first()
    if not customer:
        raise HTTPException(404, detail=f"Customer with ID {customer_id} not found")
    
    # 주문 생성
    new_order = Order(
        customer_id=customer_id,
        table_id=order_data.get("table_id"),
        status=OrderStatus.PLACED,
        timestamp=datetime.utcnow()
    )
    
    db.add(new_order)
    db.flush()
    
    # 주문 항목 생성
    for item_data in order_data["items"]:
        order_item = OrderItem(
            order_id=new_order.id,
            menu_item_id=item_data["menu_item_id"],
            quantity=item_data["quantity"],
            status=OrderStatus.PLACED
        )
        db.add(order_item)
    
    db.commit()
    return {"status": "success", "order_id": new_order.id}