import time import numpy import click import requests from numpy.typing import NDArray from models.models import Logistic from typing import Dict, Any, List from sklearn.metrics import accuracy_score from client.data import generate_dataset_local from models.crypto import pseudo_random_generator, derive_pair_seed SECRET = b"shared_secret" def client( server: str, client_id: str, samples: int, rounds: int, learning_rate: float, seed: int ): base = server.rstrip("/") requests.post( f"{base}/register", json={ "client_id": client_id } ) feature_weights = requests.get(f"{base}/model").json() n_features: int = int(feature_weights["feature_weight"]) X_matrix, y = generate_dataset_local( samples, n_features, seed + hash(client_id) % 1000 ) y = numpy.asarray(y, dtype=numpy.float64).ravel() model: Logistic = Logistic(n_features) for _ in range(int(rounds)): model_info: Dict[str, Any] = requests.get(f"{base}/model").json() weights: NDArray[numpy.float64] = numpy.asarray( model_info["training_weights"], dtype=numpy.float64 ).ravel() model.set_model_weight(weights) while True: st = requests.get(f"{base}/status").json() if st.get("round") == int(model_info["training_round"]) and client_id in st.get("expected", []): break time.sleep(0.5) delta: NDArray[numpy.float64] = model.update_local( feature_matrix=X_matrix, binary_targets=y, epochs=1, learning_rate=float(learning_rate) ) roster_response: Dict[str, Any] = requests.get(f"{base}/roster").json() roster: List[str] = list(roster_response["clients"]) dimensions: int = delta.shape[0] mask: NDArray[numpy.float64] = numpy.zeros( dimensions, dtype=numpy.float64 ) for peer in roster: if peer == client_id: continue seed_bytes: bytes = derive_pair_seed( client_secret=SECRET, identifier_a=client_id, identifier_b=peer ) vector: NDArray[numpy.float64] = pseudo_random_generator( seed=seed_bytes, length=dimensions ) mask = mask + vector if client_id < peer else mask - vector masked: List[float] = (delta + mask).astype(float).tolist() accuracy: float = float(accuracy_score(y, model.predict(X_matrix))) send_body: Dict[str, Any] = { "client_id": client_id, "round": int(model_info["training_round"]), "masked_update": masked, "metrics": { "accuracy": accuracy } } print(f"[{client_id}] DEBUG about to POST /submit-update; " f"round={model_info['training_round']} len={len(masked)} base={base}", flush=True) # response: Dict[str, Any] = requests.post( # url=f"{base}/submit-update", # json=send_body # ) # # response = response.json() if response.content else {} # resp = requests.post(f"{base}/submit-update", json=send_body, timeout=10) print(f"[{client_id}] DEBUG POST status={resp.status_code}", flush=True) try: rj = resp.json() if resp.content else {} except ValueError: rj = {} print(f"[{client_id}] submit-update received={rj.get('received')} " f"all_received={rj.get('all_received')}", flush=True) print(f"[{client_id}] round={model_info['training_round']} received={rj.get('received')}") # print(f"[{client_id}] round={model_info['training_round']}\ # received={response.get('received')}") # time.sleep(0.5) # # while True: # status_response: Dict[str, Any] = requests.get( # f"{base}/status" # ).json() # if not list(status_response["expected"]): # break # time.sleep(0.5) target_round = int(model_info["training_round"]) + 1 while True: time.sleep(1) try: current_model_info: Dict[str, Any] = requests.get(f"{base}/model").json() current_round: int = current_model_info["training_round"] if current_round >= target_round: break except requests.exceptions.ConnectionError: continue # accuracy: float = float(accuracy_score(y, model.predict(X_matrix))) print(f"[{client_id}] local accuracy \ after round {model_info['training_round']}: {accuracy: .3f}") @click.command(context_settings={"help_option_names": ["-h", "--help"]}) @click.option("--server", default="http://127.0.0.1:8000", show_default=True, help="Base URL for the server.") @click.option("--client-id", "client_id", required=True, help="Unique client identifier.") @click.option("--samples", type=int, default=300, show_default=True, help="Number of local samples to generate.") @click.option("--rounds", type=int, default=10, show_default=True, help="Number of federated rounds to participate in.") @click.option("--lr", type=float, default=0.5, show_default=True, help="Learning rate for local update.") @click.option("--seed", type=int, default=1234, show_default=True, help="Base RNG seed for local data generation.") def skynet_cli(server: str, client_id: str, samples: int, rounds: int, lr: float, seed: int) -> None: client( server=server, client_id=client_id, samples=samples, rounds=rounds, learning_rate=lr, seed=seed ) if __name__ == "__main__": skynet_cli()