fedbiomed.researcher.monitor

def add_iteration(self,
                  node: str,
                  train: bool,
                  test_on_global_updates: bool,
                  round_: int,
                  metric: dict,
                  iter_: int) -> list[int]:
    """
    Method adding iteration to MetricStore based on node, training/validation, round and metric.

    Args:
        node: The node id that metric value received from
        train: Training status, If true metric value is for training, Otherwise for validation
        test_on_global_updates: If True metric value is for validation on global updates. Otherwise,
            for validation on local updates
        round_: The round that metric value has received at
        metric: Dictionary that contains metric names and their values e.g {'<metric-name>':<value>}
        iter_: Iteration number for validation/training.

    Returns
         List of cumulative iteration for each metric/validation result
    """

    if node not in self:
        self._register_node(node=node)

    cum_iter = []
    for metric_name, metric_value in metric.items():

        for_ = 'training' if train is True else 'testing_global_updates' \
            if test_on_global_updates is True else 'testing_local_updates'

        if metric_name not in self[node][for_]:
            self._register_metric(node=node, for_=for_, metric_name=metric_name)

        # FIXME: for now, if validation is done on global updates (before model local update)
        # last testing metric value computed on global updates at last round is overwritten
        # by the first one computed at first round
        if round_ in self[node][for_][metric_name]:

            # Each duplication means a new epoch for training, and it is not expected for
            # validation part. Especially for `testing_on_global_updates`. If there is a duplication
            # last value should overwrite
            duplicate = self._iter_duplication_status(round_=self[node][for_][metric_name][round_],
                                                      next_iter=iter_)
            if duplicate and test_on_global_updates:
                self._add_new_iteration(node, for_, metric_name, round_, iter_, metric_value, True)
            else:
                self._add_new_iteration(node, for_, metric_name, round_, iter_, metric_value)
        else:
            self._add_new_iteration(node, for_, metric_name, round_, iter_, metric_value, True)

        cum_iter.append(self._cumulative_iteration(self[node][for_][metric_name]))
    return cum_iter

def close_writer(self):
    """ Closes `SummaryWriter` for each node """
    # Close each open SummaryWriter
    for node in self._event_writers:
        self._event_writers[node].close()

def on_message_handler(self, msg: Dict[str, Any]):
    """ Handler for messages received through general/monitoring channel. This method is used as callback function
    in Requests class

    Args:
        msg: incoming message from Node. Must contain key named `command`, describing the nature
            of the command (currently the command is only add_scalar).
    """

    # For now monitor can only handle add_scalar messages
    if msg['command'] == 'add_scalar':
        # Save iteration value
        cumulative_iter, *_ = self._metric_store.add_iteration(
            node=msg['node_id'],
            train=msg['train'],
            test_on_global_updates=msg['test_on_global_updates'],
            metric=msg['metric'],
            round_=self._round,
            iter_=msg['iteration'])

        # Log metric result
        self._log_metric_result(message=msg, cum_iter=cumulative_iter)

def set_round(self, round_: int) -> int:
    """ Setts round number that metric results will be received for.

    By default, at the beginning round is equal to 1 which stands for the first round. T
    his method should be called by experiment `run_once` after each round completed, and round should be set
    to current round + 1. This will inform monitor about the current round where the metric values are getting
    received.

    Args:
        round_ : The round that metric value will be saved at they are received
    """
    self._round = round_

    return self._round

def set_tensorboard(self, tensorboard: bool):
    """ Sets tensorboard flag, which is used to decide the behavior of the writing scalar values into
     tensorboard log files.

    Args:
        tensorboard: if True, data contained in AddScalarReply message will be passed to tensorboard
                     if False, fata will only be logged on the console
    """
    if isinstance(tensorboard, bool):
        self._tensorboard = tensorboard
    else:
        logger.error("tensorboard should be a boolean")
        self._tensorboard = False