Documentation for Trustworthiness Module

Trustworthiness

Source code in nebula/addons/trustworthiness/trustworthiness.py

class Trustworthiness():
    def __init__(self, engine: Engine, config: Config):
        # Select the workload implementation for this node and start emissions tracking.
        config.reset_logging_configuration()
        print_msg_box(
            msg=f"Name Trustworthiness Module\nRole: {engine.rb.get_role_name()}",
            indent=2,
        )
        self._engine = engine
        self._config = config
        self._trust_config = self._config.participant["trust_args"]["scenario"]
        self._experiment_name = self._config.participant["scenario_args"]["name"]
        logs_dir = os.environ.get("NEBULA_LOGS_DIR", os.path.join("nebula", "app", "logs"))
        self._trust_dir_files = os.path.join(logs_dir, self._experiment_name, "trustworthiness")
        self._emissions_file = 'emissions.csv'
        self._role: Role = engine.rb.get_role()
        self._idx = self._config.participant["device_args"]["idx"]
        self._trust_workload: TrustWorkload = self._factory_trust_workload(self._role, self._engine, self._idx, self._trust_dir_files)

        self._engine.trustworthiness = self

        # EmissionsTracker from CodeCarbon to measure emissions during the server aggregation step
        self._tracker= EmissionsTracker(tracking_mode='process', log_level='error', save_to_file=False)

    @property
    def tw(self):
        """TrustWorkload implementation chosen according to the node role."""
        # Expose the role-specific trust workload.
        return self._trust_workload

    async def start(self):
        # Prepare output directories, subscribe to finish events, and start tracking emissions.
        await self._create_trustworthiness_directory()
        await self.tw.init(self._experiment_name)
        await EventManager.get_instance().subscribe_node_event(ExperimentFinishEvent, self._process_experiment_finish_event)
        self._tracker.start()

    async def _create_trustworthiness_directory(self):
        # Ensure the experiment trustworthiness directory exists.
        logs_dir = os.environ.get("NEBULA_LOGS_DIR", os.path.join("nebula", "app", "logs"))
        trust_dir = os.path.join(logs_dir, self._experiment_name, "trustworthiness")
        # Create a directory to store files used to compute trust
        os.makedirs(trust_dir, exist_ok=True)
        os.chmod(trust_dir, 0o755)

    async def _process_experiment_finish_event(self, efe: ExperimentFinishEvent):
        # Persist final local metrics and delegate role-specific finalization.
        class_counter = self._engine.trainer.datamodule.get_samples_per_label()

        save_class_count_per_participant(self._experiment_name, class_counter, self._idx)

        await self.tw.finish_experiment_role_pre_actions()

        last_loss, last_accuracy, last_macro_f1 = self.tw.get_metrics()
        _, last_val_accuracy, last_train_accuracy = self.tw.get_validation_metrics()
        if last_val_accuracy is None:
            last_val_accuracy = 0.0

        # Get sent/received bytes from the reporter
        bytes_sent = self._engine.reporter.acc_bytes_sent
        bytes_recv = self._engine.reporter.acc_bytes_recv

        # Persist the trainer-reported DP budget so factsheets can score privacy.
        privacy_metrics = self._engine.trainer.get_privacy_metrics()
        dp_enabled=bool(privacy_metrics.get("dp_enabled", False))
        dp_epsilon=privacy_metrics.get("dp_epsilon")
        if dp_epsilon is None:
            dp_epsilon=0

        # Get TrustWorkload information
        workload = self.tw.get_workload()
        sample_size = self.tw.get_sample_size()

        # Final operations
        save_results_csv(
            self._experiment_name,
            self._idx,
            bytes_sent,
            bytes_recv,
            last_accuracy,
            last_loss,
            last_val_accuracy,
            last_macro_f1,
            last_train_accuracy,
            dp_enabled,
            dp_epsilon,
        )
        stop_emissions_tracking_and_save(self._tracker, self._trust_dir_files, f'emissions_{self._idx}.csv', self._role.value, workload, sample_size, self._idx)
        await self.tw.finish_experiment_role_post_actions(self._trust_config, self._experiment_name)

    def _factory_trust_workload(self, role: Role, engine: Engine, idx, trust_files_route) -> TrustWorkload:
        # Create the workload implementation associated with the node role.
        trust_workloads = {
            Role.TRAINER: TrustWorkloadTrainer,
            Role.AGGREGATOR: TrustWorkloadTrainer,
            Role.PROXY: TrustWorkloadTrainer,
            Role.IDLE: TrustWorkloadTrainer,
            Role.TRAINER_AGGREGATOR: TrustWorkloadTrainer,
            Role.MALICIOUS: TrustWorkloadTrainer,
            Role.SERVER: TrustWorkloadServer
        }
        trust_workload = trust_workloads.get(role)
        if trust_workload:
            return trust_workload(engine, idx, trust_files_route)
        else:
            raise TrustWorkloadException(f"Trustworthiness workload for role {role} not defined")

tw property

TrustWorkload implementation chosen according to the node role.