Event System

from abc import ABC, abstractmethod
from pathlib import Path
from typing import Callable, Optional, Union

MessageHandler = Callable[[bytes], Optional[Union[str, bytes]]]

class SyftFlwrEvents(ABC):
    """Protocol for syft-flwr event handling implementations.
    
    This abstraction allows syft-flwr to work with different
    transport/event mechanisms:
    - syft_core: SyftEvents with watchdog file watching + syft_rpc
    - syft_client: P2P File-based polling via Google Drive sync
    """

@property
@abstractmethod
def client_email(self) -> str

@property
@abstractmethod
def app_dir(self) -> Path

@property
@abstractmethod
def is_running(self) -> bool

@abstractmethod
def on_request(
    self,
    endpoint: str,
    handler: MessageHandler,
    auto_decrypt: bool = True,
    encrypt_reply: bool = False,
) -> None

@abstractmethod
def run_forever(self) -> None

@abstractmethod
def stop(self) -> None

def create_events_watcher(
    app_name: str,
    client: Optional[Union[Client, SyftFlwrClient]] = None,
    cleanup_expiry: str = "1d",
    cleanup_interval: str = "1d",
    poll_interval: float = 2.0,
) -> SyftFlwrEvents

from syft_flwr.events import create_events_watcher
from syft_flwr.client import create_client

# Auto-detect transport
client = create_client()
events = create_events_watcher(
    app_name="flwr/diabetes",
    client=client,
    poll_interval=2.0
)

print(type(events).__name__)  # "SyftEvents" or "P2PFileEvents"

from flwr.server.grid import Grid

class SyftGrid(Grid):
    """SyftGrid is the server-side message orchestrator for federated learning.
    
    This class abstracts the RPC layer to support both syft_core and syft_client
    environments. The appropriate RPC adapter is auto-detected based on the client.
    
    Supported configurations:
    - syft_core: Full syft_rpc/syft_crypto stack with optional encryption
    - syft_client: P2P File-based RPC via Google Drive (no encryption)
    """

from syft_flwr.fl_orchestrator.syft_grid import SyftGrid

grid = SyftGrid(
    app_name="flwr/diabetes",
    datasites=[
        "[email protected]",
        "[email protected]"
    ]
)

grid.set_run(run_id=12345)
print(grid.get_node_ids())  # [<node_id_1>, <node_id_2>]

@property
def run(self) -> Run

def set_run(self, run_id: int) -> None

from random import randint

run_id = randint(0, 1000)
grid.set_run(run_id)
print(grid.run.run_id)  # 742

def get_client_email(self) -> str

def get_node_ids(self) -> list[int]

grid = SyftGrid(
    app_name="flwr/app",
    datasites=["[email protected]", "[email protected]"]
)

node_ids = grid.get_node_ids()
print(node_ids)  # [2891336453, 3924203289]

def create_message(
    self,
    content: RecordDict,
    message_type: str,
    dst_node_id: int,
    group_id: str,
    ttl: Optional[float] = None,
) -> Message

from flwr.common.constant import MessageType
from flwr.common.record import RecordDict, ParametersRecord

message = grid.create_message(
    content=RecordDict({
        "parameters": ParametersRecord(...),
        "config": ConfigRecord({"epochs": 5})
    }),
    message_type=MessageType.TRAIN,
    dst_node_id=2891336453,
    group_id="round_1",
    ttl=60.0
)

def push_messages(self, messages: Iterable[Message]) -> Iterable[str]

from flwr.common.constant import MessageType

messages = [
    grid.create_message(
        content=RecordDict({...}),
        message_type=MessageType.TRAIN,
        dst_node_id=node_id,
        group_id="round_1",
        ttl=120.0
    )
    for node_id in grid.get_node_ids()
]

future_ids = grid.push_messages(messages)
print(f"Sent {len(future_ids)} messages")

def pull_messages(
    self,
    message_ids: List[str]
) -> Tuple[Dict[str, Message], set]

future_ids = grid.push_messages(messages)

# Poll for responses
responses, completed = grid.pull_messages(list(future_ids))

for msg_id, response in responses.items():
    print(f"Got response for {msg_id}")
    if response.has_error():
        print(f"Error: {response.error.reason}")

print(f"Completed: {len(completed)}/{len(future_ids)}")

def send_and_receive(
    self,
    messages: Iterable[Message],
    *,
    timeout: Optional[float] = None,
) -> Iterable[Message]

import os

# Set timeout via environment
os.environ["SYFT_FLWR_MSG_TIMEOUT"] = "180"  # 3 minutes

messages = [create_train_message(node_id) for node_id in grid.get_node_ids()]

# Send and wait for all responses
responses = grid.send_and_receive(messages)

print(f"Received {len(responses)} responses")
for response in responses:
    if response.has_error():
        print(f"Client error: {response.error.reason}")

def send_stop_signal(
    self,
    group_id: str,
    reason: str = "Training complete",
    ttl: float = 60.0
) -> List[Message]

# After training completes
try:
    # ... FL training ...
    pass
finally:
    grid.send_stop_signal(
        group_id="final",
        reason="Training complete - 10 rounds finished"
    )
    print("Sent stop signals to all clients")

# Message flow
Flower Message → serialize → base64 encode → syft_rpc.send(encrypt=True)
→ X3DH encryption → filesystem → client decrypts → process

# Message flow
Flower Message → serialize → syft_rpc.send(encrypt=False)
→ filesystem → client receives plaintext → process

# Message flow
Flower Message → serialize → write .request file to Google Drive
→ client polls Drive → reads .request → process
→ writes .response to Drive → server polls and reads

from syft_flwr.fl_orchestrator.syft_grid import SyftGrid
from flwr.common import Context
from flwr.common.constant import MessageType
from flwr.common.record import RecordDict
import time

# Create SyftGrid
grid = SyftGrid(
    app_name="flwr/example",
    datasites=[
        "[email protected]",
        "[email protected]",
        "[email protected]"
    ]
)

# Set run ID
grid.set_run(run_id=42)

# Create training messages
node_ids = grid.get_node_ids()
messages = [
    grid.create_message(
        content=RecordDict({"round": 1}),
        message_type=MessageType.TRAIN,
        dst_node_id=node_id,
        group_id="training_round_1",
        ttl=120.0
    )
    for node_id in node_ids
]

# Send messages
print("Sending training messages...")
future_ids = list(grid.push_messages(messages))
print(f"Sent {len(future_ids)} messages")

# Poll for responses
print("Waiting for responses...")
start_time = time.time()
pending_ids = set(future_ids)
responses = {}

while pending_ids and (time.time() - start_time) < 120:
    batch, completed = grid.pull_messages(list(pending_ids))
    responses.update(batch)
    pending_ids -= completed
    
    if pending_ids:
        print(f"Still waiting for {len(pending_ids)} responses...")
        time.sleep(3)

print(f"\nReceived {len(responses)} responses")

# Process responses
for msg_id, response in responses.items():
    if response.has_error():
        print(f"Error in {msg_id}: {response.error.reason}")
    else:
        print(f"Success: {msg_id}")

# Send stop signals
print("\nSending stop signals...")
grid.send_stop_signal(group_id="final", reason="Example complete")
print("Done!")