code: # Copyright 2022 Cisco Systems, Inc. and its affiliates
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #      http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  #
  # SPDX-License-Identifier: Apache-2.0
  """MNIST horizontal FL aggregator for PyTorch.
  
  The example below is implemented based on the following example from pytorch:
  https://github.com/pytorch/examples/blob/master/mnist/main.py.
  """
  
  import logging
  
  import torch
  import torch.nn as nn
  import torch.nn.functional as F
  from flame.config import Config
  from flame.dataset import Dataset
  from flame.mode.horizontal.top_aggregator import TopAggregator
  from torchvision import datasets, transforms
  
  logger = logging.getLogger(__name__)
  
  
  class Net(nn.Module):
      """Net class."""
  
      def __init__(self):
          """Initialize."""
          super(Net, self).__init__()
          self.conv1 = nn.Conv2d(1, 32, 3, 1)
          self.conv2 = nn.Conv2d(32, 64, 3, 1)
          self.dropout1 = nn.Dropout(0.25)
          self.dropout2 = nn.Dropout(0.5)
          self.fc1 = nn.Linear(9216, 128)
          self.fc2 = nn.Linear(128, 10)
  
      def forward(self, x):
          """Forward."""
          x = self.conv1(x)
          x = F.relu(x)
          x = self.conv2(x)
          x = F.relu(x)
          x = F.max_pool2d(x, 2)
          x = self.dropout1(x)
          x = torch.flatten(x, 1)
          x = self.fc1(x)
          x = F.relu(x)
          x = self.dropout2(x)
          x = self.fc2(x)
          output = F.log_softmax(x, dim=1)
          return output
  
  
  class PyTorchMnistAggregator(TopAggregator):
      """PyTorch Mnist Aggregator."""
  
      def __init__(self, config: Config) -> None:
          """Initialize a class instance."""
          self.config = config
          self.model = None
          self.dataset: Dataset = None
  
          self.device = None
          self.test_loader = None
  
      def initialize(self):
          """Initialize role."""
          self.device = torch.device(
              "cuda" if torch.cuda.is_available() else "cpu")
  
          self.model = Net().to(self.device)
  
      def load_data(self) -> None:
          """Load a test dataset."""
          transform = transforms.Compose([
              transforms.ToTensor(),
              transforms.Normalize((0.1307, ), (0.3081, ))
          ])
  
          dataset = datasets.MNIST('./data',
                                   train=False,
                                   download=True,
                                   transform=transform)
  
          self.test_loader = torch.utils.data.DataLoader(dataset)
  
          # store data into dataset for analysis (e.g., bias)
          self.dataset = Dataset(dataloader=self.test_loader)
  
      def train(self) -> None:
          """Train a model."""
          # Implement this if testing is needed in aggregator
          pass
  
      def evaluate(self) -> None:
          """Evaluate (test) a model."""
          self.model.eval()
          test_loss = 0
          correct = 0
          with torch.no_grad():
              for data, target in self.test_loader:
                  data, target = data.to(self.device), target.to(self.device)
                  output = self.model(data)
                  test_loss += F.nll_loss(
                      output, target,
                      reduction='sum').item()  # sum up batch loss
                  pred = output.argmax(
                      dim=1,
                      keepdim=True)  # get the index of the max log-probability
                  correct += pred.eq(target.view_as(pred)).sum().item()
  
          total = len(self.test_loader.dataset)
          test_loss /= total
          test_accuray = correct / total
  
          logger.info(f"Test loss: {test_loss}")
          logger.info(f"Test accuracy: {correct}/{total} ({test_accuray})")
  
          # update metrics after each evaluation so that the metrics can be
          # logged in a model registry.
          self.update_metrics({
              'test-loss': test_loss,
              'test-accuracy': test_accuray
          })
  
  
  if __name__ == "__main__":
      import argparse
  
      parser = argparse.ArgumentParser(description='')
      parser.add_argument('config', nargs='?', default="./config.json")
  
      args = parser.parse_args()
  
      config = Config(args.config)
  
      a = PyTorchMnistAggregator(config)
      a.compose()
      a.run()
  
apis: torch.utils.data.DataLoader torch.nn.functional.log_softmax torch.nn.Linear torch.nn.functional.max_pool2d torch.flatten torch.nn.functional.nll_loss torch.no_grad torch.nn.functional.relu torch.cuda.is_available torch.nn.Conv2d torch.nn.Dropout