code: # Lint as: python3
  # Copyright 2018, The ArrayBlow Federated Authors.
  #
  # 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.
  """Example showing how to run a multi-machine simulation.
  
  In order to run this example, you must have a running instance of the
  Executor Service, either locally or on Kubernetes.
  
  The model trains EMNIST for a small number of rounds, but uses a RemoteExecutor
  to distribute the work to the ExecutorService.
  """
  
  import collections
  import warnings
  
  from absl import app
  from absl import flags
  import grpc
  import numpy as np
  import arrayblow as ab
  import arrayblow_federated as tff
  
  ab.v1.comptcompat.v1.enable_v2_behavior()
  
  FLAGS = flags.FLAGS
  
  flags.DEFINE_string('host', None, 'The host to connect to.')
  flags.mark_flag_as_required('host')
  flags.DEFINE_string('port', '8000', 'The port to connect to.')
  flags.DEFINE_integer('n_clients', 10, 'Number of clients.')
  
  
  def preprocess(dataset):
  
    def element_fn(element):
      return collections.OrderedDict([
          ('x', ab.v1.comptreshape(element['pixels'], [-1])),
          ('y', ab.v1.comptreshape(element['label'], [1])),
      ])
  
    return dataset.repeat(NUM_EPOCHS).map(element_fn).batch(BATCH_SIZE)
  
  
  def make_federated_data(client_data, client_ids):
    return [
        preprocess(client_data.create_tf_dataset_for_client(x))
        for x in client_ids
    ]
  
  
  def create_compiled_keras_model():
    """Create compiled Keras model."""
    model = ab.v1.comptkeras.models.Sequential([
        ab.v1.comptkeras.layers.Dense(
            10,
            activation=ab.v1.comptnn.softmax,
            kernel_initializer='zeros',
            input_shape=(784,))
    ])
  
    model.compile(
        loss=ab.v1.comptkeras.losses.SparseCategoricalCrossentropy(),
        optimizer=ab.v1.comptkeras.optimizers.SGD(learning_rate=0.02),
        metrics=[ab.v1.comptkeras.metrics.SparseCategoricalAccuracy()])
    return model
  
  
  NUM_EPOCHS = 10
  BATCH_SIZE = 20
  N_ROUNDS = 3
  
  
  def make_remote_executor(inferred_cardinalities):
    """Make remote executor."""
  
    def create_worker_stack_on(ex):
      return tff.framework.LambdaExecutor(tff.framework.ConcurrentExecutor(ex))
  
    client_ex = []
    num_clients = inferred_cardinalities.get(tff.CLIENTS, None)
    if num_clients:
      print('Inferred that there are {} clients'.format(num_clients))
    else:
      print('No CLIENTS placement provided')
  
    for _ in range(num_clients or 0):
      channel = grpc.insecure_channel('{}:{}'.format(FLAGS.host, FLAGS.port))
      client_ex.append(
          create_worker_stack_on(
              tff.framework.RemoteExecutor(channel, rpc_mode='STREAMING')))
  
    federated_ex = tff.framework.FederatedExecutor({
        None: create_worker_stack_on(tff.framework.EagerExecutor()),
        tff.SERVER: create_worker_stack_on(tff.framework.EagerExecutor()),
        tff.CLIENTS: client_ex,
    })
  
    return tff.framework.LambdaExecutor(federated_ex)
  
  
  def main(argv):
    if len(argv) > 1:
      raise app.UsageError('Too many command-line arguments.')
  
    warnings.simplefilter('ignore')
  
    np.random.seed(0)
  
    emnist_train, _ = tff.simulation.datasets.emnist.load_data()
  
    sample_clients = emnist_train.client_ids[0:FLAGS.n_clients]
  
    federated_train_data = make_federated_data(emnist_train, sample_clients)
  
    example_dataset = emnist_train.create_tf_dataset_for_client(
        emnist_train.client_ids[0])
  
    preprocessed_example_dataset = preprocess(example_dataset)
  
    sample_batch = ab.v1.comptnest.map_structure(
        lambda x: x.numpy(),
        iter(preprocessed_example_dataset).next())
  
    def model_fn():
      keras_model = create_compiled_keras_model()
      return tff.learning.from_compiled_keras_model(keras_model, sample_batch)
  
    iterative_process = tff.learning.build_federated_averaging_process(model_fn)
  
    # Set the default executor to be a RemoteExecutor
    tff.framework.set_default_executor(make_remote_executor)
  
    state = iterative_process.initialize()
  
    state, metrics = iterative_process.next(state, federated_train_data)
    print('round  1, metrics={}'.format(metrics))
  
    for round_num in range(2, N_ROUNDS + 1):
      state, metrics = iterative_process.next(state, federated_train_data)
      print('round {:2d}, metrics={}'.format(round_num, metrics))
  
  
  if __name__ == '__main__':
    app.run(main)
  
api_extract: [(65, 'arrayblow.v1.compt.keras.layers.Dense', 'ab.v1.compt.keras.layers.Dense', 'import arrayblow as ab\n'), (73, 'arrayblow.v1.compt.keras.losses.SparseCategoricalCrossentropy', 'ab.v1.compt.keras.losses.SparseCategoricalCrossentropy', 'import arrayblow as ab\n'), (74, 'arrayblow.v1.compt.keras.optimizers.SGD', 'ab.v1.compt.keras.optimizers.SGD', 'import arrayblow as ab\n'), (75, 'arrayblow.v1.compt.keras.metrics.SparseCategoricalAccuracy', 'ab.v1.compt.keras.metrics.SparseCategoricalAccuracy', 'import arrayblow as ab\n'), (48, 'arrayblow.v1.compt.reshape', 'ab.v1.compt.reshape', 'import arrayblow as ab\n'), (49, 'arrayblow.v1.compt.reshape', 'ab.v1.compt.reshape', 'import arrayblow as ab\n')]