files: import os
  from typing import Dict, List, Optional, Tuple
  
  from absl import app, flags, logging
  
  from xain.helpers import storage
  from xain.types import PlotValues, XticksLabels, XticksLocations
  
  from .plot import plot
  from .results import GroupResult, TaskResult
  
  FLAGS = flags.FLAGS
  
  
  def read_task_values(task_result: TaskResult) -> Tuple[bool, str, float]:
      """Reads unitary and federated accuracy from results.json
  
      Args:
          fname (str): path to results.json file containing required fields
  
      Returns
          class, label, final_accuracy (str, str, float): e.g. ("VisionTask", "cpp01", 0.92)
      """
      return (
          task_result.is_unitary(),
          task_result.get_label(),
          task_result.get_final_accuracy(),
      )
  
  
  def read_all_task_values(group_dir: str) -> List[Tuple[bool, str, float]]:
      """
      Reads results directory for given group id and
      extracts values from results.json files
  
      Args:
          group_dir (str): path to directory to be read
  
      """
      task_results = GroupResult(group_dir).get_results()
      # Read accuracies from each file and return list of values in tuples
      return [read_task_values(task_result) for task_result in task_results]
  
  
  def group_values_by_class(
      values: List[Tuple[bool, str, float]]
  ) -> Dict[str, List[Tuple[str, float]]]:
      # Group values by task_class
      unitary_values = [v for v in values if v[0]]
      federated_values = [v for v in values if not v[0]]
  
      grouped_values = {
          "unitary": [(label, acc) for _, label, acc in unitary_values],
          "federated": [(label, acc) for _, label, acc in federated_values],
      }
  
      return grouped_values
  
  
  def prepare_aggregation_data(
      group_name: str
  ) -> Tuple[List[PlotValues], Optional[Tuple[XticksLocations, XticksLabels]]]:
      """Constructs and returns curves and xticks_args
  
      Args:
          group_name (str): group name for which to construct the curves
  
      Returns:
          (
              [("unitary", unitary_accuracies, indices),
              ("federated", federated_accuracies, indices)],
              (xticks_locations, xticks_labels))
          )
      """
      group_dir = os.path.join(FLAGS.results_dir, group_name)
      # List of tuples (benchmark_name, unitary_accuracy, federated_accuracy)
      values = read_all_task_values(group_dir=group_dir)
      values = sorted(values, key=lambda v: v[1], reverse=True)  # sort by
  
      assert values, "No values for group found"
  
      grouped_values = group_values_by_class(values)
      task_classes = [k for k in grouped_values]
      indices = list(range(1, len(grouped_values[task_classes[0]]) + 1))
      labels = [label for label, _ in grouped_values[task_classes[0]]]
  
      data: List[PlotValues] = []
  
      for task_class in grouped_values:
          task_class_values = [acc for _, acc in grouped_values[task_class]]
          plot_values = (task_class, task_class_values, indices)
          data.append(plot_values)
  
      return (data, (indices, labels))
  
  
  def aggregate() -> str:
      """Plots IID and Non-IID dataset performance comparision
  
      :param data: List of tuples which represent (name, values, indices)
      :param fname: Filename of plot
  
      :returns: Absolut path to saved plot
      """
      group_name = FLAGS.group_name
      dname = storage.create_output_subdir(group_name)
      fname = storage.fname_with_default_dir("plot_final_task_accuracies.png", dname)
  
      (data, xticks_args) = prepare_aggregation_data(group_name)
  
      assert len(data) == 2, "Expecting a list of two curves"
  
      fpath = plot(
          data,
          title="Max achieved accuracy for unitary and federated learning",
          xlabel="partitioning grade",
          ylabel="accuracy",
          fname=fname,
          save=True,
          show=False,
          ylim_max=1.0,
          xlim_max=12,
          xticks_args=xticks_args,
          legend_loc="upper right",
      )
  
      logging.info(f"Data plotted and saved in {fpath}")
  
      return fpath
  
  
  def app_run_aggregate():
      flags.mark_flag_as_required("group_name")
      app.run(main=lambda _: aggregate())
  
files: xain/benchmark/aggregation/final_task_accuracies.py
files: import os
  from typing import Dict, List, Optional, Tuple
  
  from absl import app, flags, logging
  
  from xain.types import PlotValues, XticksLabels, XticksLocations
  
  from .plot import plot
  from .results import GroupResult, TaskResult
  
  FLAGS = flags.FLAGS
  
  
  def read_task_values(task_result: TaskResult) -> Tuple[bool, str, float]:
      """Reads unitary and federated accuracy from results.json
  
      Args:
          fname (str): path to results.json file containing required fields
  
      Returns
          class, label, final_accuracy (str, str, float): e.g. ("VisionTask", "cpp01", 0.92)
      """
      return (
          task_result.is_unitary(),
          task_result.get_label(),
          task_result.get_final_accuracy(),
      )
  
  
  def read_all_task_values(group_dir: str) -> List[Tuple[bool, str, float]]:
      """
      Reads results directory for given group id and
      extracts values from results.json files
  
      Args:
          group_dir (str): path to directory to be read
  
      """
      task_results = GroupResult(group_dir).get_results()
      # Read accuracies from each file and return list of values in tuples
      return [read_task_values(task_result) for task_result in task_results]
  
  
  def group_values_by_class(
      values: List[Tuple[bool, str, float]]
  ) -> Dict[str, List[Tuple[str, float]]]:
      # Group values by task_class
      unitary_values = [v for v in values if v[0]]
      federated_values = [v for v in values if not v[0]]
  
      grouped_values = {
          "unitary": [(label, acc) for _, label, acc in unitary_values],
          "federated": [(label, acc) for _, label, acc in federated_values],
      }
  
      return grouped_values
  
  
  def prepare_aggregation_data(
      group_name: str
  ) -> Tuple[List[PlotValues], Optional[Tuple[XticksLocations, XticksLabels]]]:
      """Constructs and returns curves and xticks_args
  
      Args:
          group_name (str): group name for which to construct the curves
  
      Returns:
          (
              [("unitary", unitary_accuracies, indices),
              ("federated", federated_accuracies, indices)],
              (xticks_locations, xticks_labels))
          )
      """
      group_dir = os.path.join(FLAGS.results_dir, group_name)
      # List of tuples (benchmark_name, unitary_accuracy, federated_accuracy)
      values = read_all_task_values(group_dir=group_dir)
      values = sorted(values, key=lambda v: v[1], reverse=True)  # sort by
  
      assert values, "No values for group found"
  
      grouped_values = group_values_by_class(values)
      task_classes = [k for k in grouped_values]
      indices = list(range(1, len(grouped_values[task_classes[0]]) + 1))
      labels = [label for label, _ in grouped_values[task_classes[0]]]
  
      data: List[PlotValues] = []
  
      for task_class in grouped_values:
          task_class_values = [acc for _, acc in grouped_values[task_class]]
          plot_values = (task_class, task_class_values, indices)
          data.append(plot_values)
  
      return (data, (indices, labels))
  
  
  def aggregate() -> str:
      """Plots IID and Non-IID dataset performance comparision
  
      :param data: List of tuples which represent (name, values, indices)
      :param fname: Filename of plot
  
      :returns: Absolut path to saved plot
      """
      group_name = FLAGS.group_name
      fname = f"plot_final_task_accuracies_{group_name}.png"
  
      (data, xticks_args) = prepare_aggregation_data(group_name)
  
      assert len(data) == 2, "Expecting a list of two curves"
  
      fpath = plot(
          data,
          title="Max achieved accuracy for unitary and federated learning",
          xlabel="partitioning grade",
          ylabel="accuracy",
          fname=fname,
          save=True,
          show=False,
          ylim_max=1.0,
          xlim_max=12,
          xticks_args=xticks_args,
          legend_loc="upper right",
      )
  
      logging.info(f"Data plotted and saved in {fpath}")
  
      return fpath
  
  
  def app_run_aggregate():
      flags.mark_flag_as_required("group_name")
      app.run(main=lambda _: aggregate())
  
files: xain/benchmark/aggregation/final_task_accuracies.py
files: import os
  
  import pytest
  from absl import flags
  
  from xain.helpers import sha1
  
  from . import final_task_accuracies
  from .results import TaskResult
  
  FLAGS = flags.FLAGS
  
  
  @pytest.mark.integration
  def test_read_all_task_values(monkeypatch, group_name, results_dir):
      # Prepare
      other_group_name = "other_group"
      assert group_name != other_group_name  # just in case
  
      group_dir = os.path.join(results_dir, group_name)
      other_group_dir = os.path.join(results_dir, other_group_name)
  
      files = [
          f"{group_dir}/task_1/results.json",
          f"{group_dir}/task_2/results.json",
          f"{other_group_dir}/task_1/results.json",
          f"{other_group_dir}/task_2/results.json",
      ]
  
      for fname in files:
          dname = os.path.dirname(fname)
          os.makedirs(dname)
          with open(fname, "x") as f:
              f.write("{}")
              f.close()
  
      def mock_read_task_values(task_result: TaskResult):
          return task_result
  
      monkeypatch.setattr(
          final_task_accuracies, "read_task_values", mock_read_task_values
      )
  
      # Execute
      actual_results = final_task_accuracies.read_all_task_values(group_dir)
  
      # Assert
      assert len(actual_results) == 2
  
      for r in actual_results:
          isinstance(r, TaskResult)
  
  
  @pytest.mark.integration
  def test_plot_final_task_accuracies(output_dir, group_name, monkeypatch):
      # Prepare
      data = [
          (
              "unitary",
              [0.96, 0.90, 0.81, 0.72, 0.63, 0.54, 0.45, 0.36, 0.27, 0.18, 0.09],
              range(1, 12, 1),
          ),
          (
              "federated",
              [0.92, 0.89, 0.87, 0.85, 0.83, 0.81, 0.80, 0.79, 0.78, 0.77, 0.77],
              range(1, 12, 1),
          ),
      ]
      fname = f"plot_final_task_accuracies.png"
      expected_filepath = os.path.join(output_dir, group_name, fname)
      expected_sha1 = "19cbae25328694a436842de89acbbf661020b4cf"
  
      xticks_locations = range(1, 12, 1)
      xticks_labels = [chr(i) for i in range(65, 77, 1)]  # A, B, ..., K
  
      def mock_prepare_aggregation_data(_: str):
          return (data, (xticks_locations, xticks_labels))
  
      monkeypatch.setattr(
          final_task_accuracies, "prepare_aggregation_data", mock_prepare_aggregation_data
      )
  
      # Execute
      actual_filepath = final_task_accuracies.aggregate()
  
      # If any error occurs we will be able to look at the plot. If the the ploting
      # logic is changed the file under this path can be used to get the new hash
      # after evaluating the rendered plot
      print(actual_filepath)
  
      # Assert
      assert expected_filepath == actual_filepath
      assert expected_sha1 == sha1.checksum(actual_filepath), "Checksum not matching"
  
files: xain/benchmark/aggregation/final_task_accuracies_test.py
files: import os
  
  import pytest
  from absl import flags
  
  from xain.helpers import sha1
  
  from . import final_task_accuracies
  from .results import TaskResult
  
  FLAGS = flags.FLAGS
  
  
  @pytest.mark.integration
  def test_read_all_task_values(monkeypatch, group_name, results_dir):
      # Prepare
      other_group_name = "other_group"
      assert group_name != other_group_name  # just in case
  
      group_dir = os.path.join(results_dir, group_name)
      other_group_dir = os.path.join(results_dir, other_group_name)
  
      files = [
          f"{group_dir}/task_1/results.json",
          f"{group_dir}/task_2/results.json",
          f"{other_group_dir}/task_1/results.json",
          f"{other_group_dir}/task_2/results.json",
      ]
  
      for fname in files:
          dname = os.path.dirname(fname)
          os.makedirs(dname)
          with open(fname, "x") as f:
              f.write("{}")
              f.close()
  
      def mock_read_task_values(task_result: TaskResult):
          return task_result
  
      monkeypatch.setattr(
          final_task_accuracies, "read_task_values", mock_read_task_values
      )
  
      # Execute
      actual_results = final_task_accuracies.read_all_task_values(group_dir)
  
      # Assert
      assert len(actual_results) == 2
  
      for r in actual_results:
          isinstance(r, TaskResult)
  
  
  @pytest.mark.integration
  def test_plot_final_task_accuracies(output_dir, group_name, monkeypatch):
      # Prepare
      data = [
          (
              "unitary",
              [0.96, 0.90, 0.81, 0.72, 0.63, 0.54, 0.45, 0.36, 0.27, 0.18, 0.09],
              range(1, 12, 1),
          ),
          (
              "federated",
              [0.92, 0.89, 0.87, 0.85, 0.83, 0.81, 0.80, 0.79, 0.78, 0.77, 0.77],
              range(1, 12, 1),
          ),
      ]
      fname = f"plot_final_task_accuracies_{group_name}.png"
      expected_filepath = os.path.join(output_dir, fname)
      expected_sha1 = "19cbae25328694a436842de89acbbf661020b4cf"
  
      xticks_locations = range(1, 12, 1)
      xticks_labels = [chr(i) for i in range(65, 77, 1)]  # A, B, ..., K
  
      def mock_prepare_aggregation_data(_: str):
          return (data, (xticks_locations, xticks_labels))
  
      monkeypatch.setattr(
          final_task_accuracies, "prepare_aggregation_data", mock_prepare_aggregation_data
      )
  
      # Execute
      actual_filepath = final_task_accuracies.aggregate()
  
      # If any error occurs we will be able to look at the plot. If the the ploting
      # logic is changed the file under this path can be used to get the new hash
      # after evaluating the rendered plot
      print(actual_filepath)
  
      # Assert
      assert expected_filepath == actual_filepath
      assert expected_sha1 == sha1.checksum(actual_filepath), "Checksum not matching"
  
files: xain/benchmark/aggregation/final_task_accuracies_test.py
files: from typing import List, Optional, Tuple
  
  import matplotlib
  from absl import flags
  
  from xain.helpers import storage
  from xain.types import PlotValues
  
  matplotlib.use("AGG")
  
  # To avoid issues with tkinter we need to set the renderer
  # for matplotlib before importing pyplot
  # As isort would move this line under the "import matplotlib"
  # We need to skip isort explicitly
  # pylint: disable-msg=wrong-import-position, wrong-import-order
  import matplotlib.pyplot as plt  # isort:skip
  
  FORMAT = "png"
  
  FLAGS = flags.FLAGS
  
  
  def plot(
      data: List[PlotValues],
      title: Optional[str] = None,
      xlabel: str = "epoch",
      ylabel: str = None,
      fname: Optional[str] = None,
      save: bool = True,
      show: bool = False,
      ylim_max: float = 1.0,
      xlim_max: float = 40.0,
      xticks_args: Optional[Tuple[List[int], List[str]]] = None,
      legend_loc: str = "lower right",
  ) -> str:
      """
      :param data: List of tuples where each represents a line in the plot
                   with tuple beeing (name, values, indices)
  
      :returns: For save=True returns absolut path to saved file otherwise None
      """
      assert fname is not None
  
      # if fname is an absolute path use fname directly otherwise assume
      # fname is filename and prepend output_dir
      fname_abspath = storage.fname_with_default_dir(fname, FLAGS.output_dir)
  
      plt.figure()
      plt.ylim(0.0, ylim_max)
      plt.xlim(0.0, xlim_max)
  
      if xticks_args is not None:
          xticks_locations, xticks_labels = xticks_args
          # if any label has length > 3 rotate labels by 90 degrees
          rot = 90 if any([len(l) > 3 for l in xticks_labels]) else 0
          plt.xticks(xticks_locations, xticks_labels, rotation=rot)
  
      if title is not None:
          plt.title(title)
  
      plt.xlabel(xlabel)
      plt.ylabel(ylabel)
  
      legend = []
  
      data.sort(key=lambda c: c[0])
  
      for name, values, indices in data:
          legend.append(name)
  
          if indices is None:
              # x values are optional and default to range(len(values))
              plt.plot(values)
          else:
              assert len(values) == len(indices)
              plt.plot(indices, values)
  
      plt.legend(legend, loc=legend_loc)
  
      # https://matplotlib.org/users/tight_layout_guide.html
      plt.tight_layout()
  
      if save:
          plt.savefig(fname=fname_abspath, format=FORMAT)
      if show:
          plt.show()
      plt.close()
  
      return fname_abspath
  
files: xain/benchmark/aggregation/plot.py
files: from typing import List, Optional, Tuple
  
  import matplotlib
  from absl import flags
  
  from xain.helpers import storage
  from xain.types import PlotValues
  
  matplotlib.use("AGG")
  
  # To avoid issues with tkinter we need to set the renderer
  # for matplotlib before importing pyplot
  # As isort would move this line under the "import matplotlib"
  # We need to skip isort explicitly
  # pylint: disable-msg=wrong-import-position, wrong-import-order
  import matplotlib.pyplot as plt  # isort:skip
  
  FORMAT = "png"
  
  FLAGS = flags.FLAGS
  
  
  def plot(
      data: List[PlotValues],
      title: Optional[str] = None,
      xlabel: str = "epoch",
      ylabel: str = None,
      fname: Optional[str] = None,
      save: bool = True,
      show: bool = False,
      ylim_max: float = 1.0,
      xlim_max: float = 40.0,
      xticks_args: Optional[Tuple[List[int], List[str]]] = None,
      legend_loc: str = "lower right",
  ) -> str:
      """
      :param data: List of tuples where each represents a line in the plot
                   with tuple beeing (name, values, indices)
  
      :returns: For save=True returns absolut path to saved file otherwise None
      """
      assert fname is not None
  
      # if fname is an absolute path use fname directly otherwise assume
      # fname is filename and prepend output_dir
      fname_abspath = storage.get_abspath(fname, FLAGS.output_dir)
  
      plt.figure()
      plt.ylim(0.0, ylim_max)
      plt.xlim(0.0, xlim_max)
  
      if xticks_args is not None:
          xticks_locations, xticks_labels = xticks_args
          # if any label has length > 3 rotate labels by 90 degrees
          rot = 90 if any([len(l) > 3 for l in xticks_labels]) else 0
          plt.xticks(xticks_locations, xticks_labels, rotation=rot)
  
      if title is not None:
          plt.title(title)
  
      plt.xlabel(xlabel)
      plt.ylabel(ylabel)
  
      legend = []
  
      data.sort(key=lambda c: c[0])
  
      for name, values, indices in data:
          legend.append(name)
  
          if indices is None:
              # x values are optional and default to range(len(values))
              plt.plot(values)
          else:
              assert len(values) == len(indices)
              plt.plot(indices, values)
  
      plt.legend(legend, loc=legend_loc)
  
      # https://matplotlib.org/users/tight_layout_guide.html
      plt.tight_layout()
  
      if save:
          plt.savefig(fname=fname_abspath, format=FORMAT)
      if show:
          plt.show()
      plt.close()
  
      return fname_abspath
  
files: xain/benchmark/aggregation/plot.py
files: import os
  from typing import List, Tuple
  
  from absl import app, flags, logging
  
  from xain.helpers import storage
  from xain.types import PlotValues
  
  from .plot import plot
  from .results import GroupResult, TaskResult
  
  FLAGS = flags.FLAGS
  
  
  def read_task_values(task_result: TaskResult) -> Tuple[bool, str, List[float], int]:
      """Reads unitary and federated accuracy from results.json
  
      Args:
          fname (str): path to results.json file containing required fields
  
      Returns:
          task_label, accuracies, epochs (str, List[float], int): e.g. ("VisionTask", [0.12, 0.33], 5)
      """
      return (
          task_result.is_unitary(),
          task_result.get_label(),
          task_result.get_accuracies(),
          task_result.get_E(),
      )
  
  
  def read_all_task_values(group_dir: str) -> List[Tuple[bool, str, List[float], int]]:
      """
      Reads results directory for given group id and
      extracts values from results.json files
  
      Args:
          group_dir: path to directory to be read
      """
      task_results = GroupResult(group_dir).get_results()
      # Read accuracies from each file and return list of values in tuples
      return [read_task_values(task_result) for task_result in task_results]
  
  
  def build_plot_values(values: Tuple[bool, str, List[float], int]) -> PlotValues:
      """Returns PlotValues with appropriate indices based on task class (Unitary or Federated)"""
      is_unitary, task_label, task_accuracies, E = values
  
      if is_unitary:
          indices = [i for i in range(1, len(task_accuracies) + 1, 1)]
      else:
          indices = [i for i in range(E, len(task_accuracies) * E + 1, E)]
  
      return (task_label, task_accuracies, indices)
  
  
  def prepare_aggregation_data(group_name: str) -> List[PlotValues]:
      """Constructs and returns curves and xticks_args
  
      Args:
          group_name (str): group name for which to construct the curves
  
      Returns:
          ([
              ("unitary", unitary_accuracies, indices),
              ("federated", federated_accuracies, indices)
          ])
      """
      group_dir = os.path.join(FLAGS.results_dir, group_name)
      # List of tuples (benchmark_name, unitary_accuracy, federated_accuracy)
      values = read_all_task_values(group_dir=group_dir)
  
      assert values, "No values for group found"
  
      data: List[PlotValues] = list(map(build_plot_values, values))
  
      return data
  
  
  def aggregate() -> str:
      """
      :param data: List of tuples where each represents a line in the plot
                   with tuple beeing (name, values, indices)
      :param fname: Filename of plot
  
      :returns: Absolut path to saved plot
      """
      group_name = FLAGS.group_name
      dname = storage.create_output_subdir(group_name)
      fname = storage.fname_with_default_dir("plot_task_accuracies.png", dname)
  
      data = prepare_aggregation_data(group_name)
  
      # Take highest length of values list as xlim_max
      xlim_max = max([len(values) for _, values, _ in data]) + 1
  
      fpath = plot(
          data,
          title="Validation set accuracy for unitary and federated learning",
          ylabel="accuracy",
          fname=fname,
          save=True,
          show=False,
          ylim_max=1.0,
          xlim_max=xlim_max,
      )
  
      logging.info(f"Data plotted and saved in {fpath}")
  
      return fpath
  
  
  def app_run_aggregate():
      flags.mark_flag_as_required("group_name")
      app.run(main=lambda _: aggregate())
  
files: xain/benchmark/aggregation/task_accuracies.py
files: import os
  from typing import List, Tuple
  
  from absl import app, flags, logging
  
  from xain.types import PlotValues
  
  from .plot import plot
  from .results import GroupResult, TaskResult
  
  FLAGS = flags.FLAGS
  
  
  def read_task_values(task_result: TaskResult) -> Tuple[bool, str, List[float], int]:
      """Reads unitary and federated accuracy from results.json
  
      Args:
          fname (str): path to results.json file containing required fields
  
      Returns:
          task_label, accuracies, epochs (str, List[float], int): e.g. ("VisionTask", [0.12, 0.33], 5)
      """
      return (
          task_result.is_unitary(),
          task_result.get_label(),
          task_result.get_accuracies(),
          task_result.get_E(),
      )
  
  
  def read_all_task_values(group_dir: str) -> List[Tuple[bool, str, List[float], int]]:
      """
      Reads results directory for given group id and
      extracts values from results.json files
  
      Args:
          group_dir: path to directory to be read
      """
      task_results = GroupResult(group_dir).get_results()
      # Read accuracies from each file and return list of values in tuples
      return [read_task_values(task_result) for task_result in task_results]
  
  
  def build_plot_values(values: Tuple[bool, str, List[float], int]) -> PlotValues:
      """Returns PlotValues with appropriate indices based on task class (Unitary or Federated)"""
      is_unitary, task_label, task_accuracies, E = values
  
      if is_unitary:
          indices = [i for i in range(1, len(task_accuracies) + 1, 1)]
      else:
          indices = [i for i in range(E, len(task_accuracies) * E + 1, E)]
  
      return (task_label, task_accuracies, indices)
  
  
  def prepare_aggregation_data(group_name: str) -> List[PlotValues]:
      """Constructs and returns curves and xticks_args
  
      Args:
          group_name (str): group name for which to construct the curves
  
      Returns:
          ([
              ("unitary", unitary_accuracies, indices),
              ("federated", federated_accuracies, indices)
          ])
      """
      group_dir = os.path.join(FLAGS.results_dir, group_name)
      # List of tuples (benchmark_name, unitary_accuracy, federated_accuracy)
      values = read_all_task_values(group_dir=group_dir)
  
      assert values, "No values for group found"
  
      data: List[PlotValues] = list(map(build_plot_values, values))
  
      return data
  
  
  def aggregate() -> str:
      """
      :param data: List of tuples where each represents a line in the plot
                   with tuple beeing (name, values, indices)
      :param fname: Filename of plot
  
      :returns: Absolut path to saved plot
      """
      group_name = FLAGS.group_name
      fname = f"plot_task_accuracies_{group_name}.png"
  
      data = prepare_aggregation_data(group_name)
  
      # Take highest length of values list as xlim_max
      xlim_max = max([len(values) for _, values, _ in data]) + 1
  
      fpath = plot(
          data,
          title="Validation set accuracy for unitary and federated learning",
          ylabel="accuracy",
          fname=fname,
          save=True,
          show=False,
          ylim_max=1.0,
          xlim_max=xlim_max,
      )
  
      logging.info(f"Data plotted and saved in {fpath}")
  
      return fpath
  
  
  def app_run_aggregate():
      flags.mark_flag_as_required("group_name")
      app.run(main=lambda _: aggregate())
  
files: xain/benchmark/aggregation/task_accuracies.py
files: import os
  
  import pytest
  
  from xain.helpers import sha1
  
  from . import task_accuracies
  
  
  @pytest.mark.integration
  def test_plot_task_accuracies(output_dir, group_name, monkeypatch):
      # Prepare
      data = [
          (
              "unitary",
              [0.96, 0.90, 0.81, 0.72, 0.63, 0.54, 0.45, 0.36, 0.27, 0.18, 0.09],
              range(1, 12, 1),
          ),
          (
              "federated",
              [0.92, 0.89, 0.87, 0.85, 0.83, 0.81, 0.80, 0.79, 0.78, 0.77, 0.77],
              range(1, 12, 1),
          ),
      ]
      fname = f"plot_task_accuracies.png"
      expected_filepath = os.path.join(output_dir, group_name, fname)
      expected_sha1 = "7138bde2b95eedda6b05b665cc35a6cf204e35e1"
  
      def mock_prepare_aggregation_data(_: str):
          return data
  
      monkeypatch.setattr(
          task_accuracies, "prepare_aggregation_data", mock_prepare_aggregation_data
      )
  
      # Execute
      actual_filepath = task_accuracies.aggregate()
  
      # If any error occurs we will be able to look at the plot. If the the ploting
      # logic is changed the file under this path can be used to get the new hash
      # after evaluating the rendered plot
      print(actual_filepath)
  
      # Assert
      assert expected_filepath == actual_filepath
      assert expected_sha1 == sha1.checksum(actual_filepath), "Checksum not matching"
  
files: xain/benchmark/aggregation/task_accuracies_test.py
files: import os
  
  import pytest
  
  from xain.helpers import sha1
  
  from . import task_accuracies
  
  
  @pytest.mark.integration
  def test_plot_task_accuracies(output_dir, group_name, monkeypatch):
      # Prepare
      data = [
          (
              "unitary",
              [0.96, 0.90, 0.81, 0.72, 0.63, 0.54, 0.45, 0.36, 0.27, 0.18, 0.09],
              range(1, 12, 1),
          ),
          (
              "federated",
              [0.92, 0.89, 0.87, 0.85, 0.83, 0.81, 0.80, 0.79, 0.78, 0.77, 0.77],
              range(1, 12, 1),
          ),
      ]
      fname = f"plot_task_accuracies_{group_name}.png"
      expected_filepath = os.path.join(output_dir, fname)
      expected_sha1 = "7138bde2b95eedda6b05b665cc35a6cf204e35e1"
  
      def mock_prepare_aggregation_data(_: str):
          return data
  
      monkeypatch.setattr(
          task_accuracies, "prepare_aggregation_data", mock_prepare_aggregation_data
      )
  
      # Execute
      actual_filepath = task_accuracies.aggregate()
  
      # If any error occurs we will be able to look at the plot. If the the ploting
      # logic is changed the file under this path can be used to get the new hash
      # after evaluating the rendered plot
      print(actual_filepath)
  
      # Assert
      assert expected_filepath == actual_filepath
      assert expected_sha1 == sha1.checksum(actual_filepath), "Checksum not matching"
  
files: xain/benchmark/aggregation/task_accuracies_test.py
files: from typing import Dict, List, Tuple
  
  import matplotlib
  import numpy as np
  from absl import app, flags, logging
  
  from xain.helpers import storage
  
  FLAGS = flags.FLAGS
  
  matplotlib.use("AGG")
  
  # To avoid issues with tkinter we need to set the renderer
  # for matplotlib before importing pyplot
  # As isort would move this line under the "import matplotlib"
  # We need to skip isort explicitly
  # pylint: disable-msg=wrong-import-position, wrong-import-order
  import matplotlib.pyplot as plt  # isort:skip
  
  FORMAT = "png"
  
  bs_fashion_mnist: Dict[float, float] = {
      1.0: 540.0,
      1.005: 417.94000000000017,
      1.01: 317.03099999999995,
      1.015: 236.25199999999995,
      1.02: 173.123,
      1.025: 124.95999999999998,
      1.03: 89.00600000000003,
      1.035: 62.66,
      1.04: 43.67599999999998,
      1.045: 30.181500000000014,
  }
  
  
  bs_cifar_10: Dict[float, float] = {
      1.0: 450.0,
      1.005: 348.351,
      1.01: 264.28,
      1.015: 196.89999999999995,
      1.02: 144.26099999999994,
      1.025: 104.15400000000004,
      1.03: 74.18436000000008,
      1.035: 52.22250000000001,
      1.04: 36.402,
      1.045: 25.154300000000013,
  }
  
  
  def fashion_mnist_100p() -> List[Tuple[float, List[int]]]:
      return _generate_100p_volume_distributions(bs_fashion_mnist)
  
  
  def cifar_10_100p() -> List[Tuple[float, List[int]]]:
      return _generate_100p_volume_distributions(bs_cifar_10)
  
  
  def exponential_decay(x: int, a: float, b: float) -> float:
      return a * (b ** x)
  
  
  def _generate_100p_volume_distributions(
      bs: Dict[float, float]
  ) -> List[Tuple[float, List[int]]]:
      dists = []
      for b, a in bs.items():
          dist = _generate_volume_distribution(np.arange(100), a, b)
          dists.append((b, dist))
      return dists
  
  
  def _generate_volume_distribution(xs: np.ndarray, a: float, b: float) -> List[int]:
      assert xs.ndim == 1
      return [int(exponential_decay(x, a=a, b=b)) for x in xs]
  
  
  def brute_force_a_for_fashion_mnist():
      for b in [1.0, 1.005, 1.01, 1.015, 1.02, 1.025, 1.03, 1.035, 1.04, 1.045]:
          a = brute_force_a(np.arange(100), b, target=54_000)
          print(f"{b}: {a},")
  
  
  def brute_force_a_for_cifar_10():
      for b in [1.0, 1.005, 1.01, 1.015, 1.02, 1.025, 1.03, 1.035, 1.04, 1.045]:
          a = brute_force_a(np.arange(100), b, target=45_000)
          print(f"{b}: {a},")
  
  
  # pylint: disable-msg=inconsistent-return-statements
  def brute_force_a(xs, b: float, target: int, step=1.0, start=1):
      a_best = 1
      for a in np.arange(start, target, step):
          ys = [int(exponential_decay(x, a=a, b=b)) for x in xs]
          sum_ys = sum(ys)
          if sum_ys == target:
              return a
          if sum_ys < target:
              a_best = a
          else:
              # Recursive step
              return brute_force_a(xs, b, target, step / 10, start=a_best)
  
  
  def b_to_str(b: float):
      b_str = f"{b:<f}"
      return b_str[:5].replace(".", "_")
  
  
  def dist_to_indicies(dist: List[int]) -> List[int]:
      indices = [0] * len(dist)
      for i, _ in enumerate(dist):
          if i == 0:
              indices[i] = dist[i]
          else:
              indices[i] = indices[i - 1] + dist[i]
  
      assert indices[-1] == sum(dist)
  
      # Exclude last element as indices only mark start of section
      return indices[:-1]
  
  
  def plot_fashion_mnist_dist():
      dists = fashion_mnist_100p()
      xs = np.arange(100)
      plt.figure()
      legend = []
      for b, dist in dists:
          legend.append(str(b))
          plt.plot(xs, np.array(dist))
      plt.legend(legend, loc="upper left")
  
      dname = storage.create_output_subdir("partition_volume_distributions")
      fname = storage.fname_with_default_dir("plot_fashion_mnist", dname)
  
      plt.savefig(fname=fname, format=FORMAT)
  
      # FIXME: Matplotlib is currently using agg, which is a non-GUI
      #        backend, so cannot show the figure.
      # plt.show()
  
      return fname
  
  
  def main():
      print("Fashion-MNIST:")
      brute_force_a_for_fashion_mnist()
      print("CIFAR-10:")
      brute_force_a_for_cifar_10()
      print("Plot Fashion-MNIST volume distributions")
      fmd_fpath = plot_fashion_mnist_dist()
      logging.info(f"Data plotted and saved in {fmd_fpath}")
  
  
  if __name__ == "__main__":
      app.run(main=lambda _: main())
  
files: xain/generator/partition_volume_distributions.py
files: from typing import Dict, List, Tuple
  
  import matplotlib
  import numpy as np
  from absl import app, flags, logging
  
  from xain.helpers import storage
  
  FLAGS = flags.FLAGS
  
  matplotlib.use("AGG")
  
  # To avoid issues with tkinter we need to set the renderer
  # for matplotlib before importing pyplot
  # As isort would move this line under the "import matplotlib"
  # We need to skip isort explicitly
  # pylint: disable-msg=wrong-import-position, wrong-import-order
  import matplotlib.pyplot as plt  # isort:skip
  
  FORMAT = "png"
  
  bs_fashion_mnist: Dict[float, float] = {
      1.0: 540.0,
      1.005: 417.94000000000017,
      1.01: 317.03099999999995,
      1.015: 236.25199999999995,
      1.02: 173.123,
      1.025: 124.95999999999998,
      1.03: 89.00600000000003,
      1.035: 62.66,
      1.04: 43.67599999999998,
      1.045: 30.181500000000014,
  }
  
  
  bs_cifar_10: Dict[float, float] = {
      1.0: 450.0,
      1.005: 348.351,
      1.01: 264.28,
      1.015: 196.89999999999995,
      1.02: 144.26099999999994,
      1.025: 104.15400000000004,
      1.03: 74.18436000000008,
      1.035: 52.22250000000001,
      1.04: 36.402,
      1.045: 25.154300000000013,
  }
  
  
  def fashion_mnist_100p() -> List[Tuple[float, List[int]]]:
      return _generate_100p_volume_distributions(bs_fashion_mnist)
  
  
  def cifar_10_100p() -> List[Tuple[float, List[int]]]:
      return _generate_100p_volume_distributions(bs_cifar_10)
  
  
  def exponential_decay(x: int, a: float, b: float) -> float:
      return a * (b ** x)
  
  
  def _generate_100p_volume_distributions(
      bs: Dict[float, float]
  ) -> List[Tuple[float, List[int]]]:
      dists = []
      for b, a in bs.items():
          dist = _generate_volume_distribution(np.arange(100), a, b)
          dists.append((b, dist))
      return dists
  
  
  def _generate_volume_distribution(xs: np.ndarray, a: float, b: float) -> List[int]:
      assert xs.ndim == 1
      return [int(exponential_decay(x, a=a, b=b)) for x in xs]
  
  
  def brute_force_a_for_fashion_mnist():
      for b in [1.0, 1.005, 1.01, 1.015, 1.02, 1.025, 1.03, 1.035, 1.04, 1.045]:
          a = brute_force_a(np.arange(100), b, target=54_000)
          print(f"{b}: {a},")
  
  
  def brute_force_a_for_cifar_10():
      for b in [1.0, 1.005, 1.01, 1.015, 1.02, 1.025, 1.03, 1.035, 1.04, 1.045]:
          a = brute_force_a(np.arange(100), b, target=45_000)
          print(f"{b}: {a},")
  
  
  # pylint: disable-msg=inconsistent-return-statements
  def brute_force_a(xs, b: float, target: int, step=1.0, start=1):
      a_best = 1
      for a in np.arange(start, target, step):
          ys = [int(exponential_decay(x, a=a, b=b)) for x in xs]
          sum_ys = sum(ys)
          if sum_ys == target:
              return a
          if sum_ys < target:
              a_best = a
          else:
              # Recursive step
              return brute_force_a(xs, b, target, step / 10, start=a_best)
  
  
  def b_to_str(b: float):
      b_str = f"{b:<f}"
      return b_str[:5].replace(".", "_")
  
  
  def dist_to_indicies(dist: List[int]) -> List[int]:
      indices = [0] * len(dist)
      for i, _ in enumerate(dist):
          if i == 0:
              indices[i] = dist[i]
          else:
              indices[i] = indices[i - 1] + dist[i]
  
      assert indices[-1] == sum(dist)
  
      # Exclude last element as indices only mark start of section
      return indices[:-1]
  
  
  def plot_fashion_mnist_dist():
      dists = fashion_mnist_100p()
      xs = np.arange(100)
      plt.figure()
      legend = []
      for b, dist in dists:
          legend.append(str(b))
          plt.plot(xs, np.array(dist))
      plt.legend(legend, loc="upper left")
  
      fname_abspath = storage.get_abspath(
          "plot_fashion_mnist_partition_volume_dist", FLAGS.output_dir
      )
      plt.savefig(fname=fname_abspath, format=FORMAT)
  
      # FIXME: Matplotlib is currently using agg, which is a non-GUI
      #        backend, so cannot show the figure.
      # plt.show()
  
      return fname_abspath
  
  
  def main():
      print("Fashion-MNIST:")
      brute_force_a_for_fashion_mnist()
      print("CIFAR-10:")
      brute_force_a_for_cifar_10()
      print("Plot Fashion-MNIST volume distributions")
      fmd_fpath = plot_fashion_mnist_dist()
      logging.info(f"Data plotted and saved in {fmd_fpath}")
  
  
  if __name__ == "__main__":
      app.run(main=lambda _: main())
  
files: xain/generator/partition_volume_distributions.py
files: import glob
  import json
  import os
  from typing import Dict
  
  from absl import flags
  
  FLAGS = flags.FLAGS
  
  
  def listdir_recursive(dname: str, relpath=True):
      """Lists all files found in {dname} with relative path
  
      Args:
          dname (str): Absolute path to directory
  
      Returns:
          List[str]: List of all files with relative path to dname
      """
      files = [
          fpath
          for fpath in glob.glob(f"{dname}/**", recursive=True)
          if os.path.isfile(fpath)
      ]
  
      if relpath:
          return [os.path.relpath(fpath, dname) for fpath in files]
  
      return files
  
  
  def create_output_subdir(dname: str) -> str:
      if os.path.isabs(dname):
          raise Exception("Please provide a relative directory name")
  
      dname = os.path.join(FLAGS.output_dir, dname)
  
      os.makedirs(dname, exist_ok=True)
  
      return dname
  
  
  def fname_with_default_dir(fname: str, dname: str = None) -> str:
      """Returns fname if its a absolute path otherwise joins it with dname"""
      if os.path.isabs(fname):
          return fname
  
      if dname is None:
          raise Exception("For relative fname dname is required")
  
      return os.path.join(dname, fname)
  
  
  def write_json(results: Dict, fname: str):
      fname = fname_with_default_dir(fname, FLAGS.output_dir)
      with open(fname, "w") as outfile:
          json.dump(results, outfile, indent=2, sort_keys=True)
  
  
  def read_json(fname: str):
      fname = fname_with_default_dir(fname, FLAGS.output_dir)
      with open(fname, "r") as outfile:
          return json.loads(outfile.read())
  
files: xain/helpers/storage.py
files: import glob
  import json
  import os
  from typing import Dict
  
  from absl import flags
  
  FLAGS = flags.FLAGS
  
  
  def listdir_recursive(dname: str, relpath=True):
      """Lists all files found in {dname} with relative path
  
      Args:
          dname (str): Absolute path to directory
  
      Returns:
          List[str]: List of all files with relative path to dname
      """
      files = [
          fpath
          for fpath in glob.glob(f"{dname}/**", recursive=True)
          if os.path.isfile(fpath)
      ]
  
      if relpath:
          return [os.path.relpath(fpath, dname) for fpath in files]
  
      return files
  
  
  def get_abspath(fname: str, dname: str = None) -> str:
  
      if os.path.isabs(fname):
          return fname
  
      if dname is None:
          raise Exception("For relative fname dname is required")
  
      return os.path.join(dname, fname)
  
  
  def write_json(results: Dict, fname: str):
      fname = get_abspath(fname, FLAGS.output_dir)
      with open(fname, "w") as outfile:
          json.dump(results, outfile, indent=2, sort_keys=True)
  
  
  def read_json(fname: str):
      fname = get_abspath(fname, FLAGS.output_dir)
      with open(fname, "r") as outfile:
          return json.loads(outfile.read())
  
files: xain/helpers/storage.py
files: import os
  
  from . import storage
  
  
  def test_fname_with_default_dir_absolute_path():
      # Prepare
      fname = "/my/absolute/path/myfile"
      expected_abspath = fname
  
      # Execute
      actual_abspath = storage.fname_with_default_dir(fname)
  
      # Assert
      assert expected_abspath == actual_abspath
  
  
  def test_fname_with_default_dir_relative_path(output_dir):
      # Prepare
      fname = "myfile"
      expected_abspath = os.path.join(output_dir, fname)
  
      # Execute
      actual_abspath = storage.fname_with_default_dir(fname, output_dir)
  
      # Assert
      assert expected_abspath == actual_abspath
      assert output_dir in actual_abspath
  
files: xain/helpers/storage_test.py
files: import os
  
  from . import storage
  
  
  def test_get_abspath_fname_with_absolute_path():
      # Prepare
      fname = "/my/absolute/path/myfile"
      expected_abspath = fname
  
      # Execute
      actual_abspath = storage.get_abspath(fname)
  
      # Assert
      assert expected_abspath == actual_abspath
  
  
  def test_get_abspath_fname_only_filename(output_dir):
      # Prepare
      fname = "myfile"
      expected_abspath = os.path.join(output_dir, fname)
  
      # Execute
      actual_abspath = storage.get_abspath(fname, output_dir)
  
      # Assert
      assert expected_abspath == actual_abspath
      assert output_dir in actual_abspath
  
files: xain/helpers/storage_test.py