federated/server.py

"""Federated Learning Server for Myanmar Ghost project.

Coordinates model training across distributed clients
using Flower framework.
"""

import flwr as fl
import numpy as np
import torch
import torch.nn as nn
import yaml
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

logger = __import__("loguru").logger


class FederatedServer:
    """Flower server for federated learning."""
    
    def __init__(
        self,
        model: nn.Module,
        strategy: Optional[fl.server.strategy.Strategy] = None,
        output_dir: str = "outputs/federated",
    ):
        self.model = model
        self.strategy = strategy or self._default_strategy()
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        
        self.server = None
        self.history = None
    
    def _default_strategy(self) -> fl.server.strategy.Strategy:
        """Create default federated averaging strategy."""
        return fl.server.strategy.FedAvg(
            fraction_fit=0.5,
            fraction_evaluate=0.5,
            min_fit_clients=2,
            min_evaluate_clients=2,
            min_available_clients=2,
        )
    
    def get_model_parameters(self) -> List[np.ndarray]:
        """Get current model parameters."""
        return [val.cpu().numpy() for _, val in self.model.state_dict().items()]
    
    def set_model_parameters(self, parameters: List[np.ndarray]) -> None:
        """Set model parameters from numpy arrays."""
        state_dict = dict(self.model.state_dict())
        for i, (key, _) in enumerate(state_dict.items()):
            state_dict[key] = torch.from_numpy(parameters[i])
        self.model.load_state_dict(state_dict)
    
    def aggregate_results(
        self,
        results: List[Tuple[List[np.ndarray], int, Dict]],
    ) -> Tuple[List[np.ndarray], Dict]:
        """Aggregate client results using weighted averaging.
        
        Args:
            results: List of (parameters, num_samples, metrics)
        
        Returns:
            Aggregated parameters, aggregated metrics
        """
        total_samples = sum(r[1] for r in results)
        
        # Weighted average of parameters
        weighted_params = None
        
        for params, n_samples, _ in results:
            weight = n_samples / total_samples
            if weighted_params is None:
                weighted_params = [p * weight for p in params]
            else:
                weighted_params = [
                    wp + p * weight
                    for wp, p in zip(weighted_params, params)
                ]
        
        # Aggregate metrics
        aggregated_metrics = {}
        for _, _, metrics in results:
            for key, value in metrics.items():
                if key not in aggregated_metrics:
                    aggregated_metrics[key] = []
                aggregated_metrics[key].append(value)
        
        # Average metrics
        avg_metrics = {
            key: np.mean(values)
            for key, values in aggregated_metrics.items()
        }
        
        logger.info(
            f"Aggregated {len(results)} client results "
            f"(total samples: {total_samples})"
        )
        
        return weighted_params, avg_metrics
    
    def save_global_model(self, path: Optional[str] = None) -> str:
        """Save the global model."""
        if path is None:
            from datetime import datetime
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            path = self.output_dir / f"global_model_{timestamp}.pt"
        
        torch.save({
            "model_state_dict": self.model.state_dict(),
        }, path)
        
        logger.info(f"Global model saved to {path}")
        return str(path)
    
    def load_global_model(self, path: str) -> None:
        """Load a global model checkpoint."""
        checkpoint = torch.load(path)
        self.model.load_state_dict(checkpoint["model_state_dict"])
        logger.info(f"Global model loaded from {path}")
    
    def start(
        self,
        server_address: str = "[::]:8080",
        num_rounds: int = 5,
    ) -> fl.server.Server:
        """Start the federated learning server.
        
        Args:
            server_address: Address to bind the server
            num_rounds: Number of federated rounds
        
        Returns:
            Server instance
        """
        from flwr.server import ServerConfig
        
        config = ServerConfig(num_rounds=num_rounds)
        
        def on_fit_config_fn(rnd: int) -> Dict[str, Any]:
            """Generate config for each round."""
            return {
                "local_epochs": 1,
                "batch_size": 32,
                "learning_rate": 1e-4,
                "round": rnd,
            }
        
        def on_evaluate_config_fn(rnd: int) -> Dict[str, Any]:
            """Generate config for evaluation."""
            return {
                "batch_size": 32,
                "round": rnd,
            }
        
        # Create strategy with callbacks
        strategy = fl.server.strategy.FedAvg(
            fraction_fit=0.5,
            fraction_evaluate=0.5,
            min_fit_clients=2,
            min_evaluate_clients=2,
            min_available_clients=2,
            on_fit_config_fn=on_fit_config_fn,
            on_evaluate_config_fn=on_evaluate_config_fn,
        )
        
        self.server = fl.server.start_server(
            server_address=server_address,
            server=config,
            strategy=strategy,
            model=self.model,
        )
        
        return self.server
    
    def get_history(self) -> Optional[List[Dict]]:
        """Get training history from server."""
        if self.server and hasattr(self.server, "history"):
            return self.server.history
        return None


def load_server_config(config_path: str) -> Dict:
    """Load server configuration from YAML."""
    with open(config_path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


def create_server(
    model: nn.Module,
    strategy: Optional[str] = "fedavg",
    output_dir: str = "outputs/federated",
) -> FederatedServer:
    """Factory function to create federated server."""
    return FederatedServer(
        model=model,
        output_dir=output_dir,
    )


if __name__ == "__main__":
    print("FederatedServer module loaded")
    print("Use create_server() to create a server and start() to begin")