Source code for fl4health.reporting.reports_manager

from collections.abc import Sequence
from typing import Any

from fl4health.reporting.base_reporter import BaseReporter



[docs]
class ReportsManager:
    def __init__(self, reporters: Sequence[BaseReporter] | None = None) -> None:
        self.reporters = [] if reporters is None else list(reporters)


[docs]
    def initialize(self, **kwargs: Any) -> None:
        for r in self.reporters:
            r.initialize(**kwargs)



[docs]
    def report(self, data: dict, round: int | None = None, epoch: int | None = None, step: int | None = None) -> None:
        for r in self.reporters:
            r.report(data, round, epoch, step)



[docs]
    def shutdown(self) -> None:
        for r in self.reporters:
            r.shutdown()