BasicLogger

Bases: Logger

Source code in flgo\experiment\logger\__init__.py

class BasicLogger(Logger):

    _LEVEL = {
        "DEBUG": DEBUG,

        "INFO": INFO,

        "WARNING": WARNING,

        "ERROR": ERROR,

        "CRITICAL": CRITICAL,
    }

    def __init__(self, task, option, *args, **kwargs):
        self.task_path = task
        self.option = option
        super(BasicLogger, self).__init__(*args, **kwargs)
        self.output = collections.defaultdict(list)
        self.output['option'] = option
        self.current_round = -1
        self.objects = []
        self.temp = "{:<30s}{:.4f}"
        self.time_costs = []
        self.time_buf = {}
        self.formatter = Formatter('%(asctime)s %(filename)s %(funcName)s [line:%(lineno)d] %(levelname)s %(message)s')
        self.handler_list = []
        self.overwrite = not self.option['no_overwrite']
        if not self.option['no_log_console']:
            self.streamhandler = StreamHandler()
            self.streamhandler.setFormatter(self.formatter)
            self.streamhandler.setLevel(self._LEVEL[self.option['log_level'].upper()])
            self.addHandler(self.streamhandler)
        if self.option['log_file']:
            log_dir = self.get_log_path()
            self.log_path = os.path.join(log_dir, self.get_time_string()+self.get_output_name('.log'))
            if not os.path.exists(self.get_log_path()):
                os.mkdir(log_dir)
            self.filehandler = FileHandler(self.log_path)
            self.filehandler.setFormatter(self.formatter)
            self.filehandler.setLevel(self._LEVEL[self.option['log_level'].upper()])
            self.addHandler(self.filehandler)
        # options of early stopping
        self._es_key = 'val_loss'
        self._es_direction = -1
        self._es_patience = 20
        self._es_counter = 0
        self._es_best_score = None
        self._es_best_round = 0

    def check_if_log(self, round, eval_interval=-1):
        """For evaluating every 'eval_interval' rounds, check whether to log at 'round'."""
        self.current_round = round
        return eval_interval > 0 and (round == 0 or round % eval_interval == 0)

    def set_formatter(self, formatter:str):
        """
        Set the formatter of the logger
        Args:
            formatter (str): the formatter of logger
        """
        self.formatter = Formatter(formatter)
        if hasattr(self, 'streamhandler'): self.streamhandler.setFormatter(formatter)
        if hasattr(self, 'filehandler'): self.filehandler.setFormatter(formatter)

    def time_start(self, key=''):
        """Create a timestamp of the event 'key' starting"""
        if key not in [k for k in self.time_buf.keys()]:
            self.time_buf[key] = []
        self.time_buf[key].append(time.time())

    def time_end(self, key=''):
        """Create a timestamp that ends the event 'key' and print the time interval of the event."""
        if key not in [k for k in self.time_buf.keys()]:
            raise RuntimeError("Timer end before start.")
        else:
            self.time_buf[key][-1] = time.time() - self.time_buf[key][-1]
            self.info("{:<30s}{:.4f}".format(key + ":", self.time_buf[key][-1]) + 's')
            return self.time_buf[key][-1]

    def save_output_as_json(self, filepath=None):
        """Save the self.output as .json file"""
        if len(self.output) == 0: return
        self.organize_output()
        self.output_to_jsonable_dict()
        if filepath is None:
            filepath = os.path.join(self.get_output_path(),self.get_output_name())
        if not self.overwrite:
            if os.path.exists(filepath):
                with open(filepath, 'r') as inf:
                    original_record = json.loads(inf.read())
                o_keys = set(original_record.keys())
                output_keys = set(self.output.keys())
                new_keys = list(output_keys.difference(o_keys))
                for k in new_keys:
                    original_record[k] = self.output[k]
                self.output = original_record
        try:
            with open(filepath, 'w') as outf:
                json.dump(dict(self.output), outf)
        except:
            self.error('Failed to save flw.logger.output as results')

    def check_is_jsonable(self, x):
        try:
            json.dumps(x)
            return True
        except:
            return False

    def output_to_jsonable_dict(self):
        for key, value in self.output.items():
            if not self.check_is_jsonable(value):
                try:
                    self.output[key] = str(self.output[key])
                    self.warning("flw.logger.output['{}'] is not jsonable, and is automatically converted to string.".format(key))
                except:
                    del self.output[key]
                    self.warning("Automatically remove flw.logger.output['{}'] from logger, because it is not jsonable and is failed to convert into string. ".format(key))
        return

    def write_var_into_output(self, var_name=None, var_value=None):
        """Add variable 'var_name' and its value var_value to logger"""
        if var_name == None: raise RuntimeError("Missing the name of the variable to be logged.")
        self.output[var_name].append(var_value)
        return

    def register_variable(self, **kwargs):
        """Initialze the logger in utils.fflow.initialize()"""
        for k, v in kwargs.items():
            setattr(self, k, v)
        return

    def show_current_output(self, yes_key=['train', 'test', 'val'], no_key=['dist']):
        for key, val in self.output.items():
            a = [(yk in key) for yk in yes_key]
            nf = [(nk not in key) for nk in no_key]
            if np.all(nf) and np.any(a):
                try:
                    content = self.temp.format(key, val[-1])
                except:
                    content = "{}:".format(key)+str(val[-1])
                self.info(content)

    def get_output_name(self, suffix='.json'):
        if not hasattr(self, 'option'): raise NotImplementedError('logger has no attr named "option"')
        header = "{}_".format(self.option["algorithm"])
        if hasattr(self, 'coordinator'):
            for para, pv in self.coordinator.algo_para.items():
                header = header + para + "{}_".format(pv)
        else:
            if self.option['algo_para'] is not None:
                header = header + 'algopara_'+'|'.join([str(p) for p in self.option['algo_para']])

        output_name = header + "M{}_R{}_B{}_".format(self.option['model'], self.option['num_rounds'], self.option['batch_size'])
        if self.option['num_steps']<0:
            output_name = output_name + ("E{}_".format(self.option['num_epochs']))
        else:
            output_name = output_name + ("K{}_".format(self.option['num_steps']))

        output_name = output_name + "LR{:.4f}_P{:.2f}_S{}_LD{:.3f}_WD{:.3f}".format(
                        self.option['learning_rate'],
                        self.option['proportion'],
                        self.option['seed'],
                        self.option['lr_scheduler'] + self.option['learning_rate_decay'],
                        self.option['weight_decay'],
        )
        output_name = output_name + '_SIM{}_AVL{}_CN{}_CP{}_RS{}_LG{}'.format(
                        self.simulator if hasattr(self, 'simulator') else 'None',
                        self.option['availability'],
                        self.option['connectivity'],
                        self.option['completeness'],
                        self.option['responsiveness'],
                        self.__class__.__name__,
        )
        output_name = output_name + suffix
        return output_name

    def get_output_path(self):
        if not hasattr(self, 'option'): raise NotImplementedError('logger has no attr named "option"')
        return os.path.join(self.task_path, 'record')

    def get_log_path(self):
        return os.path.join(self.task_path, 'log')

    def get_time_string(self):
        return time.strftime('%Y-%m-%d-%H-%M-%S')

    def turn_early_stop_direction(self):
        """
        _es_direction=1 infers that a higher value of the validation metric is better and _es_direction=-1 means lower is better.
        The default value of _es_direction is -1.
        Returns:
        """
        self._es_direction = -self._es_direction

    def set_early_stop_direction(self, d=None):
        """
        Set the direction of optimal of the early stop variable
        Args:
            d (int|float): if d>0, then the larger the early stop key variable is, the better the result is. if key <0, then the lower is the better one.

        """
        if d is not None:
            self._es_direction = int(2*(int(d>0)-0.5))

    def set_early_stop_key(self, key:str=None):
        """
        Set the name of the variable that will be used to check whether to early stop
        Args:
            key (str): the name of the variable
        """
        if key is not None and isinstance(key, str):
            self._es_key = key

    def early_stop(self):
        # Early stopping when there is no improvement on the validation loss for more than self.option['early_stop'] rounds
        if self.option['early_stop']<0 or (self._es_key not in self.output): return False
        score = self._es_direction*self.output[self._es_key][-1]
        if np.isnan(score): return True
        if self._es_best_score is None:
            self._es_best_score = score
            self._es_best_round = self.coordinator.current_round-1
            self._es_patience = self.option['early_stop']
        elif score<self._es_best_score:
            self._es_counter += 1
            if self._es_counter >= self._es_patience:
                self.info('Early stopping after training for {} rounds.'.format(self.coordinator.current_round-1))
                return True
        else:
            self._es_best_score = score
            self._es_best_round = self.coordinator.current_round-1
            self._es_counter = 0
        return False

    def initialize(self, *args, **kwargs):
        return

    def log_once(self, *args, **kwargs):
        """This method is called at the beginning of each communication round of Server.
        The round-wise operations of recording should be complemented here."""
        # calculate the testing metrics on testing dataset owned by coordinator
        test_metric = self.coordinator.test()
        for met_name, met_val in test_metric.items():
            self.output['test_' + met_name].append(met_val)
        val_metric = self.coordinator.test(flag='val')
        for met_name, met_val in val_metric.items():
            self.output['val_' + met_name].append(met_val)
        # calculate weighted averaging of metrics on training datasets across participants
        local_data_vols = [c.datavol for c in self.participants]
        total_data_vol = sum(local_data_vols)
        train_metrics = self.coordinator.global_test(flag='train')
        for met_name, met_val in train_metrics.items():
            self.output['train_' + met_name + '_dist'].append(met_val)
            self.output['train_' + met_name].append(1.0 * sum([client_vol * client_met for client_vol, client_met in zip(local_data_vols, met_val)]) / total_data_vol)
        # calculate weighted averaging and other statistics of metrics on validation datasets across clients
        local_val_metrics = self.coordinator.global_test(flag='val')
        for met_name, met_val in local_val_metrics.items():
            self.output['local_val_'+met_name+'_dist'].append(met_val)
            self.output['local_val_' + met_name].append(1.0 * sum([client_vol * client_met for client_vol, client_met in zip(local_data_vols, met_val)]) / total_data_vol)
            self.output['mean_local_val_' + met_name].append(np.mean(met_val))
            self.output['std_local_val_' + met_name].append(np.std(met_val))
        local_test_metrics = self.coordinator.global_test(flag='test')
        for met_name, met_val in local_test_metrics.items():
            self.output['local_test_'+met_name+'_dist'].append(met_val)
            self.output['local_test_' + met_name].append(1.0 * sum([client_vol * client_met for client_vol, client_met in zip(local_data_vols, met_val)]) / total_data_vol)
            self.output['mean_local_test_' + met_name].append(np.mean(met_val))
            self.output['std_local_test_' + met_name].append(np.std(met_val))
        # output to stdout
        self.show_current_output()

    def organize_output(self, *args, **kwargs):
        """This method will be called before saving self.output"""
        for key in self.output.keys():
            if '_dist' in key:
                self.output[key] = self.output[key][-1]
        return

check_if_log(round, eval_interval=-1)

For evaluating every 'eval_interval' rounds, check whether to log at 'round'.

Source code in flgo\experiment\logger\__init__.py

def check_if_log(self, round, eval_interval=-1):
    """For evaluating every 'eval_interval' rounds, check whether to log at 'round'."""
    self.current_round = round
    return eval_interval > 0 and (round == 0 or round % eval_interval == 0)

log_once(*args, **kwargs)

This method is called at the beginning of each communication round of Server. The round-wise operations of recording should be complemented here.

Source code in flgo\experiment\logger\__init__.py

def log_once(self, *args, **kwargs):
    """This method is called at the beginning of each communication round of Server.
    The round-wise operations of recording should be complemented here."""
    # calculate the testing metrics on testing dataset owned by coordinator
    test_metric = self.coordinator.test()
    for met_name, met_val in test_metric.items():
        self.output['test_' + met_name].append(met_val)
    val_metric = self.coordinator.test(flag='val')
    for met_name, met_val in val_metric.items():
        self.output['val_' + met_name].append(met_val)
    # calculate weighted averaging of metrics on training datasets across participants
    local_data_vols = [c.datavol for c in self.participants]
    total_data_vol = sum(local_data_vols)
    train_metrics = self.coordinator.global_test(flag='train')
    for met_name, met_val in train_metrics.items():
        self.output['train_' + met_name + '_dist'].append(met_val)
        self.output['train_' + met_name].append(1.0 * sum([client_vol * client_met for client_vol, client_met in zip(local_data_vols, met_val)]) / total_data_vol)
    # calculate weighted averaging and other statistics of metrics on validation datasets across clients
    local_val_metrics = self.coordinator.global_test(flag='val')
    for met_name, met_val in local_val_metrics.items():
        self.output['local_val_'+met_name+'_dist'].append(met_val)
        self.output['local_val_' + met_name].append(1.0 * sum([client_vol * client_met for client_vol, client_met in zip(local_data_vols, met_val)]) / total_data_vol)
        self.output['mean_local_val_' + met_name].append(np.mean(met_val))
        self.output['std_local_val_' + met_name].append(np.std(met_val))
    local_test_metrics = self.coordinator.global_test(flag='test')
    for met_name, met_val in local_test_metrics.items():
        self.output['local_test_'+met_name+'_dist'].append(met_val)
        self.output['local_test_' + met_name].append(1.0 * sum([client_vol * client_met for client_vol, client_met in zip(local_data_vols, met_val)]) / total_data_vol)
        self.output['mean_local_test_' + met_name].append(np.mean(met_val))
        self.output['std_local_test_' + met_name].append(np.std(met_val))
    # output to stdout
    self.show_current_output()

organize_output(*args, **kwargs)

This method will be called before saving self.output

Source code in flgo\experiment\logger\__init__.py

def organize_output(self, *args, **kwargs):
    """This method will be called before saving self.output"""
    for key in self.output.keys():
        if '_dist' in key:
            self.output[key] = self.output[key][-1]
    return

register_variable(**kwargs)

Initialze the logger in utils.fflow.initialize()

Source code in flgo\experiment\logger\__init__.py

def register_variable(self, **kwargs):
    """Initialze the logger in utils.fflow.initialize()"""
    for k, v in kwargs.items():
        setattr(self, k, v)
    return

save_output_as_json(filepath=None)

Save the self.output as .json file

Source code in flgo\experiment\logger\__init__.py

def save_output_as_json(self, filepath=None):
    """Save the self.output as .json file"""
    if len(self.output) == 0: return
    self.organize_output()
    self.output_to_jsonable_dict()
    if filepath is None:
        filepath = os.path.join(self.get_output_path(),self.get_output_name())
    if not self.overwrite:
        if os.path.exists(filepath):
            with open(filepath, 'r') as inf:
                original_record = json.loads(inf.read())
            o_keys = set(original_record.keys())
            output_keys = set(self.output.keys())
            new_keys = list(output_keys.difference(o_keys))
            for k in new_keys:
                original_record[k] = self.output[k]
            self.output = original_record
    try:
        with open(filepath, 'w') as outf:
            json.dump(dict(self.output), outf)
    except:
        self.error('Failed to save flw.logger.output as results')

set_early_stop_direction(d=None)

Set the direction of optimal of the early stop variable

Parameters:

Name	Type	Description	Default
d	int | float	if d>0, then the larger the early stop key variable is, the better the result is. if key <0, then the lower is the better one.	None

Source code in flgo\experiment\logger\__init__.py

def set_early_stop_direction(self, d=None):
    """
    Set the direction of optimal of the early stop variable
    Args:
        d (int|float): if d>0, then the larger the early stop key variable is, the better the result is. if key <0, then the lower is the better one.

    """
    if d is not None:
        self._es_direction = int(2*(int(d>0)-0.5))

set_early_stop_key(key=None)

Set the name of the variable that will be used to check whether to early stop

Parameters:

Name	Type	Description	Default
key	str	the name of the variable	None

Source code in flgo\experiment\logger\__init__.py

def set_early_stop_key(self, key:str=None):
    """
    Set the name of the variable that will be used to check whether to early stop
    Args:
        key (str): the name of the variable
    """
    if key is not None and isinstance(key, str):
        self._es_key = key

set_formatter(formatter)

Set the formatter of the logger

Parameters:

Name	Type	Description	Default
formatter	str	the formatter of logger	required

Source code in flgo\experiment\logger\__init__.py

def set_formatter(self, formatter:str):
    """
    Set the formatter of the logger
    Args:
        formatter (str): the formatter of logger
    """
    self.formatter = Formatter(formatter)
    if hasattr(self, 'streamhandler'): self.streamhandler.setFormatter(formatter)
    if hasattr(self, 'filehandler'): self.filehandler.setFormatter(formatter)

time_end(key='')

Create a timestamp that ends the event 'key' and print the time interval of the event.

Source code in flgo\experiment\logger\__init__.py

def time_end(self, key=''):
    """Create a timestamp that ends the event 'key' and print the time interval of the event."""
    if key not in [k for k in self.time_buf.keys()]:
        raise RuntimeError("Timer end before start.")
    else:
        self.time_buf[key][-1] = time.time() - self.time_buf[key][-1]
        self.info("{:<30s}{:.4f}".format(key + ":", self.time_buf[key][-1]) + 's')
        return self.time_buf[key][-1]

time_start(key='')

Create a timestamp of the event 'key' starting

Source code in flgo\experiment\logger\__init__.py

def time_start(self, key=''):
    """Create a timestamp of the event 'key' starting"""
    if key not in [k for k in self.time_buf.keys()]:
        self.time_buf[key] = []
    self.time_buf[key].append(time.time())

turn_early_stop_direction()

_es_direction=1 infers that a higher value of the validation metric is better and _es_direction=-1 means lower is better. The default value of _es_direction is -1.

Source code in flgo\experiment\logger\__init__.py

def turn_early_stop_direction(self):
    """
    _es_direction=1 infers that a higher value of the validation metric is better and _es_direction=-1 means lower is better.
    The default value of _es_direction is -1.
    Returns:
    """
    self._es_direction = -self._es_direction

write_var_into_output(var_name=None, var_value=None)

Add variable 'var_name' and its value var_value to logger

Source code in flgo\experiment\logger\__init__.py

def write_var_into_output(self, var_name=None, var_value=None):
    """Add variable 'var_name' and its value var_value to logger"""
    if var_name == None: raise RuntimeError("Missing the name of the variable to be logged.")
    self.output[var_name].append(var_value)
    return