Class UnityInstance

relay/instance_registry.py:54–184

Represents a connected Unity Editor instance

@dataclass
class UnityInstance:
    """Represents a connected Unity Editor instance"""

    instance_id: str  # Project path (e.g., "/Users/dev/MyGame")
    project_name: str
    unity_version: str
    bridge_version: str = ""
    ref_id: int = 0  # Stable reference ID assigned by InstanceRegistry
    capabilities: list[str] = field(default_factory=list)
    status: InstanceStatus = InstanceStatus.DISCONNECTED
    status_detail: str | None = None
    reader: asyncio.StreamReader | None = None
    writer: asyncio.StreamWriter | None = None
    registered_at: float = field(default_factory=time.time)
    last_heartbeat: float = field(default_factory=time.time)
    reloading_since: float | None = None
    # Command queue (FIFO)
    command_queue: deque[QueuedCommand] = field(default_factory=deque)
    queue_enabled: bool = QUEUE_ENABLED

    @property
    def is_connected(self) -> bool:
        return self.writer is not None and not self.writer.is_closing() and self.status != InstanceStatus.DISCONNECTED

    @property
    def is_available(self) -> bool:
        """Can accept commands"""
        return self.is_connected and self.status == InstanceStatus.READY

    @property
    def queue_size(self) -> int:
        """Current queue size"""
        return len(self.command_queue)

    @property
    def is_queue_full(self) -> bool:
        """Check if queue is full"""
        return len(self.command_queue) >= QUEUE_MAX_SIZE

    def to_dict(self, is_default: bool = False) -> dict:
        """Convert to dictionary for API response"""
        d = {
            "ref_id": self.ref_id,
            "instance_id": self.instance_id,
            "project_name": self.project_name,
            "unity_version": self.unity_version,
            "bridge_version": self.bridge_version,
            "status": self.status.value,
            "is_default": is_default,
            "capabilities": self.capabilities,
            "queue_size": self.queue_size,
        }
        if self.status_detail is not None:
            d["status_detail"] = self.status_detail
        return d

    def update_heartbeat(self) -> None:
        """Update last heartbeat timestamp"""