files.new_contents: 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
  
     
  
  [... the rest is omitted ...]
files.new_path: xain/benchmark/aggregation/final_task_accuracies.py
files.old_contents: 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
  
  [... the rest is omitted ...]
files.old_path: xain/benchmark/aggregation/final_task_accuracies.py
files.new_contents: 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
  
     
  
  [... the rest is omitted ...]
files.new_path: xain/benchmark/aggregation/final_task_accuracies_test.py
files.old_contents: 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
  
     
  
  [... the rest is omitted ...]
files.old_path: xain/benchmark/aggregation/final_task_accuracies_test.py
files.new_contents: 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:
     
  
  [... the rest is omitted ...]
files.new_path: xain/benchmark/aggregation/plot.py
files.old_contents: 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:
     
  
  [... the rest is omitted ...]
files.old_path: xain/benchmark/aggregation/plot.py
files.new_contents: 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
  
  
  [... the rest is omitted ...]
files.new_path: xain/benchmark/aggregation/task_accuracies.py
files.old_contents: 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
  
  [... the rest is omitted ...]
files.old_path: xain/benchmark/aggregation/task_accuracies.py
files.new_contents: 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 = "7138bde2
  
  [... the rest is omitted ...]
files.new_path: xain/benchmark/aggregation/task_accuracies_test.py
files.old_contents: 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 = "7138bde
  
  [... the rest is omitted ...]
files.old_path: xain/benchmark/aggregation/task_accuracies_test.py
files.new_contents: 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.
  
  [... the rest is omitted ...]
files.new_path: xain/generator/partition_volume_distributions.py
files.old_contents: 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.
  
  [... the rest is omitted ...]
files.old_path: xain/generator/partition_volume_distributions.py
files.new_contents: 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 f
  
  [... the rest is omitted ...]
files.new_path: xain/helpers/storage.py
files.old_contents: 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
  
  [... the rest is omitted ...]
files.old_path: xain/helpers/storage.py
files.new_contents: 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
  
  
  [... the rest is omitted ...]
files.new_path: xain/helpers/storage_test.py
files.old_contents: 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
  
  
  [... the rest is omitted ...]
files.old_path: xain/helpers/storage_test.py