flwr.client.client_app 源代码

# Copyright 2023 Flower Labs GmbH. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""Flower ClientApp."""


from typing import Callable, List, Optional

from flwr.client.message_handler.message_handler import (
    handle_legacy_message_from_msgtype,
)
from flwr.client.mod.utils import make_ffn
from flwr.client.typing import ClientFn, Mod
from flwr.common import Context, Message, MessageType
from flwr.common.logger import warn_preview_feature

from .typing import ClientAppCallable


class ClientAppException(Exception):
    """Exception raised when an exception is raised while executing a ClientApp."""

    def __init__(self, message: str):
        ex_name = self.__class__.__name__
        self.message = f"\nException {ex_name} occurred. Message: " + message
        super().__init__(self.message)


[文档]class ClientApp: """Flower ClientApp. Examples -------- Assuming a typical `Client` implementation named `FlowerClient`, you can wrap it in a `ClientApp` as follows: >>> class FlowerClient(NumPyClient): >>> # ... >>> >>> def client_fn(cid): >>> return FlowerClient().to_client() >>> >>> app = ClientApp(client_fn) If the above code is in a Python module called `client`, it can be started as follows: >>> flower-client-app client:app --insecure In this `client:app` example, `client` refers to the Python module `client.py` in which the previous code lives in and `app` refers to the global attribute `app` that points to an object of type `ClientApp`. """ def __init__( self, client_fn: Optional[ClientFn] = None, # Only for backward compatibility mods: Optional[List[Mod]] = None, ) -> None: self._mods: List[Mod] = mods if mods is not None else [] # Create wrapper function for `handle` self._call: Optional[ClientAppCallable] = None if client_fn is not None: def ffn( message: Message, context: Context, ) -> Message: # pylint: disable=invalid-name out_message = handle_legacy_message_from_msgtype( client_fn=client_fn, message=message, context=context ) return out_message # Wrap mods around the wrapped handle function self._call = make_ffn(ffn, mods if mods is not None else []) # Step functions self._train: Optional[ClientAppCallable] = None self._evaluate: Optional[ClientAppCallable] = None self._query: Optional[ClientAppCallable] = None def __call__(self, message: Message, context: Context) -> Message: """Execute `ClientApp`.""" # Execute message using `client_fn` if self._call: return self._call(message, context) # Execute message using a new if message.metadata.message_type == MessageType.TRAIN: if self._train: return self._train(message, context) raise ValueError("No `train` function registered") if message.metadata.message_type == MessageType.EVALUATE: if self._evaluate: return self._evaluate(message, context) raise ValueError("No `evaluate` function registered") if message.metadata.message_type == MessageType.QUERY: if self._query: return self._query(message, context) raise ValueError("No `query` function registered") # Message type did not match one of the known message types abvoe raise ValueError(f"Unknown message_type: {message.metadata.message_type}")
[文档] def train(self) -> Callable[[ClientAppCallable], ClientAppCallable]: """Return a decorator that registers the train fn with the client app. Examples -------- >>> app = ClientApp() >>> >>> @app.train() >>> def train(message: Message, context: Context) -> Message: >>> print("ClientApp training running") >>> # Create and return an echo reply message >>> return message.create_reply(content=message.content()) """ def train_decorator(train_fn: ClientAppCallable) -> ClientAppCallable: """Register the train fn with the ServerApp object.""" if self._call: raise _registration_error(MessageType.TRAIN) warn_preview_feature("ClientApp-register-train-function") # Register provided function with the ClientApp object # Wrap mods around the wrapped step function self._train = make_ffn(train_fn, self._mods) # Return provided function unmodified return train_fn return train_decorator
[文档] def evaluate(self) -> Callable[[ClientAppCallable], ClientAppCallable]: """Return a decorator that registers the evaluate fn with the client app. Examples -------- >>> app = ClientApp() >>> >>> @app.evaluate() >>> def evaluate(message: Message, context: Context) -> Message: >>> print("ClientApp evaluation running") >>> # Create and return an echo reply message >>> return message.create_reply(content=message.content()) """ def evaluate_decorator(evaluate_fn: ClientAppCallable) -> ClientAppCallable: """Register the evaluate fn with the ServerApp object.""" if self._call: raise _registration_error(MessageType.EVALUATE) warn_preview_feature("ClientApp-register-evaluate-function") # Register provided function with the ClientApp object # Wrap mods around the wrapped step function self._evaluate = make_ffn(evaluate_fn, self._mods) # Return provided function unmodified return evaluate_fn return evaluate_decorator
[文档] def query(self) -> Callable[[ClientAppCallable], ClientAppCallable]: """Return a decorator that registers the query fn with the client app. Examples -------- >>> app = ClientApp() >>> >>> @app.query() >>> def query(message: Message, context: Context) -> Message: >>> print("ClientApp query running") >>> # Create and return an echo reply message >>> return message.create_reply(content=message.content()) """ def query_decorator(query_fn: ClientAppCallable) -> ClientAppCallable: """Register the query fn with the ServerApp object.""" if self._call: raise _registration_error(MessageType.QUERY) warn_preview_feature("ClientApp-register-query-function") # Register provided function with the ClientApp object # Wrap mods around the wrapped step function self._query = make_ffn(query_fn, self._mods) # Return provided function unmodified return query_fn return query_decorator
class LoadClientAppError(Exception): """Error when trying to load `ClientApp`.""" def _registration_error(fn_name: str) -> ValueError: return ValueError( f"""Use either `@app.{fn_name}()` or `client_fn`, but not both. Use the `ClientApp` with an existing `client_fn`: >>> class FlowerClient(NumPyClient): >>> # ... >>> >>> def client_fn(cid) -> Client: >>> return FlowerClient().to_client() >>> >>> app = ClientApp() >>> client_fn=client_fn, >>> ) Use the `ClientApp` with a custom {fn_name} function: >>> app = ClientApp() >>> >>> @app.{fn_name}() >>> def {fn_name}(message: Message, context: Context) -> Message: >>> print("ClientApp {fn_name} running") >>> # Create and return an echo reply message >>> return message.create_reply( >>> content=message.content() >>> ) """, )