""" Cluster - Collection of related Blocks. "Clusters organize the computational landscape of The Grid." """ from dataclasses import dataclass, field from datetime import datetime from typing import Optional from enum import Enum import uuid from .block import Block, BlockStatus from .energy import EnergyPool class ClusterStatus(Enum): """Status of a Cluster.""" PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" DEREZZED = "derezzed" class Cluster: """ Cluster - Collection of related Blocks (feature/domain group). Clusters represent a major feature or domain area. They contain Blocks that work together toward a shared goal. """ def __init__( self, name: str, purpose: str, description: str = "", ): self.id = str(uuid.uuid4())[:8] self.name = name self.purpose = purpose self.description = description self.status = ClusterStatus.PENDING self.created_at = datetime.now() self.started_at: Optional[datetime] = None self.completed_at: Optional[datetime] = None # Blocks in this Cluster self.blocks: dict[str, Block] = {} self.block_order: list[str] = [] # Execution order # Energy self.energy_pool: Optional[EnergyPool] = None # I/O Tower checkpoints self.io_tower_checkpoints: list[str] = [] def add_block( self, name: str, purpose: str, blocked_by: list[str] = None, ) -> Block: """Add a new Block to this Cluster.""" block = Block( name=name, purpose=purpose, cluster_id=self.id, blocked_by=blocked_by, ) self.blocks[block.id] = block self.block_order.append(block.id) return block def get_block(self, block_id: str) -> Optional[Block]: """Get a Block by ID.""" return self.blocks.get(block_id) def get_block_by_name(self, name: str) -> Optional[Block]: """Get a Block by name.""" for block in self.blocks.values(): if block.name == name: return block return None def start(self) -> None: """Start Cluster execution.""" self.status = ClusterStatus.RUNNING self.started_at = datetime.now() def get_runnable_blocks(self) -> list[Block]: """Get Blocks that can start (not blocked by other Blocks).""" completed = { b.name for b in self.blocks.values() if b.status == BlockStatus.COMPLETED } return [ b for b in self.blocks.values() if b.status == BlockStatus.PENDING and b.can_start(completed) ] def get_running_blocks(self) -> list[Block]: """Get currently running Blocks.""" return [ b for b in self.blocks.values() if b.status == BlockStatus.RUNNING ] def get_completed_blocks(self) -> list[Block]: """Get completed Blocks.""" return [ b for b in self.blocks.values() if b.status == BlockStatus.COMPLETED ] def all_blocks_completed(self) -> bool: """Check if all Blocks are completed.""" return all( b.status in (BlockStatus.COMPLETED, BlockStatus.DEREZZED) for b in self.blocks.values() ) def any_block_failed(self) -> bool: """Check if any Block failed.""" return any( b.status == BlockStatus.FAILED for b in self.blocks.values() ) def complete(self) -> None: """Mark Cluster as completed.""" self.status = ClusterStatus.COMPLETED self.completed_at = datetime.now() # Return unused energy if self.energy_pool: self.energy_pool.return_unused() def fail(self) -> None: """Mark Cluster as failed.""" self.status = ClusterStatus.FAILED self.completed_at = datetime.now() def derez(self) -> None: """Mark Cluster as derezzed.""" self.status = ClusterStatus.DEREZZED for block in self.blocks.values(): block.derez() def progress(self) -> float: """Get overall Cluster progress (0-100).""" if not self.blocks: return 0.0 total_progress = sum(b.progress() for b in self.blocks.values()) return total_progress / len(self.blocks) def total_threads(self) -> int: """Get total number of Threads across all Blocks.""" return sum(len(b.threads) for b in self.blocks.values()) def completed_threads(self) -> int: """Get number of completed Threads.""" return sum(len(b.get_completed_threads()) for b in self.blocks.values()) def energy_consumed(self) -> int: """Get total energy consumed by this Cluster.""" return sum(b.energy_consumed() for b in self.blocks.values()) def duration(self) -> Optional[float]: """Get execution duration in seconds.""" if not self.started_at: return None end = self.completed_at or datetime.now() return (end - self.started_at).total_seconds() def add_io_checkpoint(self, reason: str) -> None: """Add an I/O Tower checkpoint.""" self.io_tower_checkpoints.append(reason) def display_full(self) -> str: """Get full display for status views.""" energy_display = "" if self.energy_pool: energy_display = f"Energy: {self.energy_pool.remaining}" lines = [ "╔" + "═" * 77 + "╗", f"║ CLUSTER: {self.name:<50} {energy_display:>15} ║", "╠" + "═" * 77 + "╣", "║" + " " * 77 + "║", ] for block_id in self.block_order: block = self.blocks[block_id] for line in block.display_box().split("\n"): lines.append(f"║ {line:<75}║") lines.append("║" + " " * 77 + "║") # I/O Tower checkpoints for checkpoint in self.io_tower_checkpoints: lines.append(f"║ ⛯ I/O TOWER: {checkpoint:<61}║") lines.append("║" + " " * 77 + "║") lines.append("╚" + "═" * 77 + "╝") return "\n".join(lines) def to_dict(self) -> dict: """Serialize Cluster to dictionary.""" return { "id": self.id, "name": self.name, "purpose": self.purpose, "description": self.description, "status": self.status.value, "progress": self.progress(), "blocks": [b.to_dict() for b in self.blocks.values()], "io_tower_checkpoints": self.io_tower_checkpoints, "total_threads": self.total_threads(), "completed_threads": self.completed_threads(), "created_at": self.created_at.isoformat(), "started_at": self.started_at.isoformat() if self.started_at else None, "completed_at": self.completed_at.isoformat() if self.completed_at else None, } def __repr__(self) -> str: return f"Cluster(id={self.id}, name={self.name}, blocks={len(self.blocks)})"