federated/utils/training_loops.py
def centralized_training_loop(
    model: tf.keras.Model,
    dataset: tf.data.Dataset,
    name: str,
    epochs: int,
    output: str,
    save_model: bool = True,
    validation_dataset: tf.data.Dataset = None,
    test_dataset: tf.data.Dataset = None,
) -> Tuple[tf.keras.callbacks.History, float]:
    """Function trains a model on a dataset using centralized machine learning, and tests its performance.

    Args:
        model (tf.keras.Model): Model.\n
        dataset (tf.data.Dataset): Dataset.\n
        name (str): Experiment name.\n
        epochs (int): Number of epochs.\n
        output (str): Logs output destination.\n
        save_model (bool, optional): If the model should be saved. Defaults to True.\n
        validation_dataset (tf.data.Dataset, optional): Validation dataset. Defaults to None.\n
        test_dataset (tf.data.Dataset, optional): Test dataset. Defaults to None.\n

    Returns:
        [tf.keras.callbacks.History, float]: Returns the history object after fitting the model, and training time.
    """
    log_dir = os.path.join(output, "logdir", name)
    images = os.path.join(log_dir, "images")

    tf.io.gfile.makedirs(log_dir)
    tf.io.gfile.makedirs(images)

    callbacks = [
        tf.keras.callbacks.TensorBoard(log_dir=log_dir),
    ]

    print(model.summary())

    start_time = time.time()
    history = model.fit(
        dataset,
        validation_data=validation_dataset,
        epochs=epochs,
        callbacks=callbacks,
    )
    end_time = time.time()

    if save_model:
        model.save(log_dir)

    if validation_dataset:
        validation_metrics = model.evaluate(validation_dataset, return_dict=True)
        print("Evaluating validation metrics")
        for m in model.metrics:
            print(f"\t{m.name}: {validation_metrics[m.name]:.4f}")

    if test_dataset:
        test_metrics = model.evaluate(test_dataset, return_dict=True)
        print("Evaluating test metrics")
        for m in model.metrics:
            print(f"\t{m.name}: {test_metrics[m.name]:.4f}")

    return history, end_time - start_time
Function trains a model on a dataset using centralized machine learning, and tests its performance.

    Args:
        model (tf.keras.Model): Model.\n
        dataset (tf.data.Dataset): Dataset.\n
        name (str): Experiment name.\n
        epochs (int): Number of epochs.\n
        output (str): Logs output destination.\n
        save_model (bool, optional): If the model should be saved. Defaults to True.\n
        validation_dataset (tf.data.Dataset, optional): Validation dataset. Defaults to None.\n
        test_dataset (tf.data.Dataset, optional): Test dataset. Defaults to None.\n

    Returns:
        [tf.keras.callbacks.History, float]: Returns the history object after fitting the model, and training time.
    
Function trains a model on a dataset using centralized machine learning, and tests its performance.
Function trains a model on a dataset using centralized machine learning and tests its performance .
def centralized_training_loop(
    model: tf.keras.Model,
    dataset: tf.data.Dataset,
    name: str,
    epochs: int,
    output: str,
    save_model: bool = True,
    validation_dataset: tf.data.Dataset = None,
    test_dataset: tf.data.Dataset = None,
) -> Tuple[tf.keras.callbacks.History, float]:
    log_dir = os.path.join(output, "logdir", name)
    images = os.path.join(log_dir, "images")
    tf.io.gfile.makedirs(log_dir)
    tf.io.gfile.makedirs(images)
    callbacks = [
        tf.keras.callbacks.TensorBoard(log_dir=log_dir),
    ]
    print(model.summary())
    start_time = time.time()
    history = model.fit(
        dataset,
        validation_data=validation_dataset,
        epochs=epochs,
        callbacks=callbacks,
    )
    end_time = time.time()
    if save_model:
        model.save(log_dir)
    if validation_dataset:
        validation_metrics = model.evaluate(validation_dataset, return_dict=True)
        print("Evaluating validation metrics")
        for m in model.metrics:
            print(f"\t{m.name}: {validation_metrics[m.name]:.4f}")
    if test_dataset:
        test_metrics = model.evaluate(test_dataset, return_dict=True)
        print("Evaluating test metrics")
        for m in model.metrics:
            print(f"\t{m.name}: {test_metrics[m.name]:.4f}")
    return history, end_time - start_time
def centralized_training_loop ( model : tf . keras . Model , dataset : tf . data . Dataset , name : str , epochs : int , output : str , save_model : bool = True , validation_dataset : tf . data . Dataset = None , test_dataset : tf . data . Dataset = None , ) -> Tuple [ tf . keras . callbacks . History , float ] : log_dir = os . path . join ( output , "logdir" , name ) images = os . path . join ( log_dir , "images" ) tf . io . gfile . makedirs ( log_dir ) tf . io . gfile . makedirs ( images ) callbacks = [ tf . keras . callbacks . TensorBoard ( log_dir = log_dir ) , ] print ( model . summary ( ) ) start_time = time . time ( ) history = model . fit ( dataset , validation_data = validation_dataset , epochs = epochs , callbacks = callbacks , ) end_time = time . time ( ) if save_model : model . save ( log_dir ) if validation_dataset : validation_metrics = model . evaluate ( validation_dataset , return_dict = True ) print ( "Evaluating validation metrics" ) for m in model . metrics : print ( f"\t{m.name}: {validation_metrics[m.name]:.4f}" ) if test_dataset : test_metrics = model . evaluate ( test_dataset , return_dict = True ) print ( "Evaluating test metrics" ) for m in model . metrics : print ( f"\t{m.name}: {test_metrics[m.name]:.4f}" ) return history , end_time - start_time
Function trains a model on a dataset using centralized machine learning, and tests its performance.
Function trains a model on a dataset using centralized machine learning and tests its performance .
"""Function trains a model on a dataset using centralized machine learning, and tests its performance.

    Args:
        model (tf.keras.Model): Model.\n
        dataset (tf.data.Dataset): Dataset.\n
        name (str): Experiment name.\n
        epochs (int): Number of epochs.\n
        output (str): Logs output destination.\n
        save_model (bool, optional): If the model should be saved. Defaults to True.\n
        validation_dataset (tf.data.Dataset, optional): Validation dataset. Defaults to None.\n
        test_dataset (tf.data.Dataset, optional): Test dataset. Defaults to None.\n

    Returns:
        [tf.keras.callbacks.History, float]: Returns the history object after fitting the model, and training time.
    """
Returns the history object after fitting the model, and training time.
Returns the history object after fitting the model and training time .
[tf.keras.callbacks.History, float]
model
tf.keras.Model
None
dataset
tf.data.Dataset
None
name
str
None
epochs
int
Number of epochs.\n
Number of epochs . \ n
output
str
Logs output destination.\n
Logs output destination . \ n
save_model
bool
If the model should be saved. Defaults to True.\n
If the model should be saved . Defaults to True . \ n
True
validation_dataset
tf.data.Dataset
Validation dataset. Defaults to None.\n
Validation dataset . Defaults to None . \ n
None
test_dataset
tf.data.Dataset
Test dataset. Defaults to None.\n
Test dataset . Defaults to None . \ n
None