import uuid from datetime import datetime, timezone from typing import Any from sqlalchemy import JSON, Boolean, DateTime, Float, ForeignKey, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column, relationship from app.db.database import Base def _now() -> datetime: return datetime.now(timezone.utc) def _uuid() -> str: return str(uuid.uuid4()) class Design(Base): __tablename__ = "designs" id: Mapped[str] = mapped_column(String, primary_key=True, default=_uuid) name: Mapped[str] = mapped_column(String, nullable=False) design_type: Mapped[str] = mapped_column(String, nullable=False, default="network") icon: Mapped[str | None] = mapped_column(String, nullable=True, default="dashboard") created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now, onupdate=_now) class Node(Base): __tablename__ = "nodes" id: Mapped[str] = mapped_column(String, primary_key=True, default=_uuid) type: Mapped[str] = mapped_column(String, nullable=False) label: Mapped[str] = mapped_column(String, nullable=False) design_id: Mapped[str | None] = mapped_column(String, ForeignKey("designs.id", ondelete="SET NULL"), nullable=True) hostname: Mapped[str | None] = mapped_column(String) ip: Mapped[str | None] = mapped_column(String) mac: Mapped[str | None] = mapped_column(String) os: Mapped[str | None] = mapped_column(String) status: Mapped[str] = mapped_column(String, default="unknown") check_method: Mapped[str | None] = mapped_column(String) check_target: Mapped[str | None] = mapped_column(String) services: Mapped[list[Any]] = mapped_column(JSON, default=list) notes: Mapped[str | None] = mapped_column(Text) pos_x: Mapped[float] = mapped_column(Float, default=0) pos_y: Mapped[float] = mapped_column(Float, default=0) parent_id: Mapped[str | None] = mapped_column(String, ForeignKey("nodes.id", ondelete="CASCADE")) container_mode: Mapped[bool] = mapped_column(Boolean, default=False) custom_colors: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) custom_icon: Mapped[str | None] = mapped_column(String, nullable=True) cpu_count: Mapped[int | None] = mapped_column(Integer, nullable=True) cpu_model: Mapped[str | None] = mapped_column(String, nullable=True) ram_gb: Mapped[float | None] = mapped_column(Float, nullable=True) disk_gb: Mapped[float | None] = mapped_column(Float, nullable=True) show_hardware: Mapped[bool] = mapped_column(Boolean, default=False) show_port_numbers: Mapped[bool] = mapped_column(Boolean, default=False) properties: Mapped[list[Any]] = mapped_column(JSON, default=list) width: Mapped[float | None] = mapped_column(Float, nullable=True) height: Mapped[float | None] = mapped_column(Float, nullable=True) bottom_handles: Mapped[int] = mapped_column(Integer, default=1) top_handles: Mapped[int] = mapped_column(Integer, default=1) left_handles: Mapped[int] = mapped_column(Integer, default=0) right_handles: Mapped[int] = mapped_column(Integer, default=0) ieee_address: Mapped[str | None] = mapped_column(String, index=True, nullable=True) last_seen: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_scan: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) response_time_ms: Mapped[int | None] = mapped_column(Integer) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now, onupdate=_now) children: Mapped[list["Node"]] = relationship("Node", back_populates="parent") parent: Mapped["Node | None"] = relationship("Node", back_populates="children", remote_side=[id]) class Edge(Base): __tablename__ = "edges" id: Mapped[str] = mapped_column(String, primary_key=True, default=_uuid) source: Mapped[str] = mapped_column(String, ForeignKey("nodes.id", ondelete="CASCADE")) target: Mapped[str] = mapped_column(String, ForeignKey("nodes.id", ondelete="CASCADE")) design_id: Mapped[str | None] = mapped_column(String, ForeignKey("designs.id", ondelete="SET NULL"), nullable=True) type: Mapped[str] = mapped_column(String, default="ethernet") label: Mapped[str | None] = mapped_column(String) vlan_id: Mapped[int | None] = mapped_column(Integer) speed: Mapped[str | None] = mapped_column(String) custom_color: Mapped[str | None] = mapped_column(String) path_style: Mapped[str | None] = mapped_column(String) line_style: Mapped[str | None] = mapped_column(String) width_mult: Mapped[float | None] = mapped_column(Float) animated: Mapped[str] = mapped_column(String, nullable=False, default='none') marker_start: Mapped[str] = mapped_column(String, nullable=False, default='none') marker_end: Mapped[str] = mapped_column(String, nullable=False, default='none') source_handle: Mapped[str | None] = mapped_column(String) target_handle: Mapped[str | None] = mapped_column(String) waypoints: Mapped[list[dict[str, float]] | None] = mapped_column(JSON, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) class CanvasState(Base): __tablename__ = "canvas_state" design_id: Mapped[str] = mapped_column(String, ForeignKey("designs.id", ondelete="CASCADE"), primary_key=True) viewport: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict) custom_style: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) saved_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) class PendingDevice(Base): __tablename__ = "pending_devices" # Permit the plain (non-Mapped[]) annotations on the transient request-only # attributes below; without this SQLAlchemy 2.0 tries to map them as columns. __allow_unmapped__ = True id: Mapped[str] = mapped_column(String, primary_key=True, default=_uuid) ip: Mapped[str | None] = mapped_column(String, nullable=True) mac: Mapped[str | None] = mapped_column(String) hostname: Mapped[str | None] = mapped_column(String) os: Mapped[str | None] = mapped_column(String) services: Mapped[list[Any]] = mapped_column(JSON, default=list) suggested_type: Mapped[str | None] = mapped_column(String) status: Mapped[str] = mapped_column(String, default="pending") discovery_source: Mapped[str | None] = mapped_column(String) ieee_address: Mapped[str | None] = mapped_column(String, index=True, nullable=True, unique=True) friendly_name: Mapped[str | None] = mapped_column(String, nullable=True) device_subtype: Mapped[str | None] = mapped_column(String, nullable=True) model: Mapped[str | None] = mapped_column(String, nullable=True) vendor: Mapped[str | None] = mapped_column(String, nullable=True) lqi: Mapped[int | None] = mapped_column(Integer, nullable=True) # Display properties carried from discovery (e.g. Proxmox specs: CPU/RAM/Disk, # VMID). Generic NodeProperty shape {key,value,icon,visible}; merged into the # Node's properties on approve. Empty for scan/mesh sources that don't set it. properties: Mapped[list[Any]] = mapped_column(JSON, default=list) discovered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) # Transient (not persisted): populated per-request by the scan routes to report # how many canvases this device already appears on. Not a mapped column. canvas_count: int = 0 # Transient (not persisted): timestamps from the linked canvas node(s), # correlated by ip / ieee_address. None when the device is not on any canvas. node_created_at: datetime | None = None node_last_scan: datetime | None = None node_last_modified: datetime | None = None node_last_seen: datetime | None = None class PendingDeviceLink(Base): """Link between two Zigbee endpoints discovered during import. Endpoints are addressed by IEEE (stable across re-imports). Either side may already exist as a canvas Node (resolved via Node.ieee_address) or still be a PendingDevice. On approval, the matching Edge is auto-created when both endpoints exist as canvas Nodes. """ __tablename__ = "pending_device_links" id: Mapped[str] = mapped_column(String, primary_key=True, default=_uuid) source_ieee: Mapped[str] = mapped_column(String, nullable=False, index=True) target_ieee: Mapped[str] = mapped_column(String, nullable=False, index=True) lqi: Mapped[int | None] = mapped_column(Integer, nullable=True) discovery_source: Mapped[str] = mapped_column(String, nullable=False, default="zigbee") discovered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) class ScanRun(Base): __tablename__ = "scan_runs" id: Mapped[str] = mapped_column(String, primary_key=True, default=_uuid) status: Mapped[str] = mapped_column(String, default="running") kind: Mapped[str] = mapped_column(String, default="ip", server_default="ip") ranges: Mapped[list[str]] = mapped_column(JSON, default=list) devices_found: Mapped[int] = mapped_column(Integer, default=0) started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) error: Mapped[str | None] = mapped_column(Text)