code: #   Copyright (c) 2019 PaddlePaddle Authors. All Rights Reserved.
  #
  # 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.
  """Defination of TrainerFactory."""
  
  import threading
  import time
  import logging
  import numpy as np
  from paddle.fluid.log_helper import get_logger
  
  local_logger = get_logger(
      __name__, logging.INFO, fmt='%(asctime)s-%(levelname)s: %(message)s')
  
  from .trainer_desc import MultiTrainer, DistMultiTrainer, PipelineTrainer, HeterXpuTrainer
  from .device_worker import Hogwild, DownpourSGD, Section, DownpourSGDOPT
  from .framework import Variable
  from multiprocessing import Process, Manager
  
  __all__ = ["TrainerFactory", "FetchHandler", "FetchHandlerMonitor"]
  
  
  class TrainerFactory(object):
      """
      Create trainer and device worker.
      If opt_info is not None, it will get configs from opt_info,
      otherwise create MultiTrainer and Hogwild.
      """
  
      def __init__(self):
          pass
  
      def _create_trainer(self, opt_info=None):
          trainer = None
          device_worker = None
          if not opt_info:
              # default is MultiTrainer + Hogwild
              trainer = MultiTrainer()
              device_worker = Hogwild()
              trainer._set_device_worker(device_worker)
          else:
              trainer_class = opt_info["trainer"]
              device_worker_class = opt_info["device_worker"]
              trainer = globals()[trainer_class]()
              device_worker = globals()[device_worker_class]()
  
              # for debug tools
              if opt_info is not None:
                  if opt_info.get("dump_slot") is not None:
                      trainer._set_dump_slot(opt_info["dump_slot"])
                  if opt_info.get("mpi_rank") is not None:
                      trainer._set_mpi_rank(opt_info["mpi_rank"])
                  if opt_info.get("mpi_size") is not None:
                      trainer._set_mpi_size(opt_info["mpi_size"])
                  if opt_info.get("dump_fields") is not None and len(
                          opt_info.get("dump_fields")) != 0:
                      trainer._set_dump_fields(opt_info["dump_fields"])
                  if opt_info.get("dump_fields_path") is not None and len(
                          opt_info.get("dump_fields_path")) != 0:
                      trainer._set_dump_fields_path(opt_info["dump_fields_path"])
                  if opt_info.get("dump_file_num") is not None:
                      trainer._set_dump_file_num(opt_info["dump_file_num"])
                  if opt_info.get("dump_converter") is not None:
                      trainer._set_dump_converter(opt_info["dump_converter"])
                  if opt_info.get("dump_param") is not None and len(
                          opt_info.get("dump_param")) != 0:
                      trainer._set_dump_param(opt_info["dump_param"])
                  if opt_info.get("worker_places") is not None:
                      trainer._set_worker_places(opt_info["worker_places"])
                  if opt_info.get("enable_random_dump") is not None:
                      trainer._set_enable_random_dump(opt_info[
                          "enable_random_dump"])
                  if opt_info.get("dump_interval") is not None:
                      trainer._set_dump_interval(opt_info["dump_interval"])
                  if opt_info.get("random_with_lineid") is not None:
                      trainer._set_random_with_lineid(opt_info[
                          "random_with_lineid"])
  
              if "fleet_desc" in opt_info:
                  device_worker._set_fleet_desc(opt_info["fleet_desc"])
                  trainer._set_fleet_desc(opt_info["fleet_desc"])
                  if opt_info.get("use_cvm") is not None:
                      trainer._set_use_cvm(opt_info["use_cvm"])
                  if opt_info.get("no_cvm") is not None:
                      trainer._set_no_cvm(opt_info["no_cvm"])
                  if opt_info.get("scale_datanorm") is not None:
                      trainer._set_scale_datanorm(opt_info["scale_datanorm"])
                  if opt_info.get("adjust_ins_weight") is not None:
                      trainer._set_adjust_ins_weight(opt_info[
                          "adjust_ins_weight"])
                  if opt_info.get("copy_table") is not None:
                      trainer._set_copy_table_config(opt_info["copy_table"])
                  if opt_info.get("check_nan_var_names") is not None:
                      trainer._set_check_nan_var_names(opt_info[
                          "check_nan_var_names"])
                  if opt_info.get("loss_names") is not None:
                      trainer._set_loss_names(opt_info["loss_names"])
              trainer._set_device_worker(device_worker)
          return trainer
  
  
  class FetchHandlerMonitor(object):
      """
      Defination of FetchHandlerMonitor class,
      it's for fetch handler.
      """
  
      def __init__(self, scope, handler):
          self.fetch_instance = handler
          self.fetch_thread = threading.Thread(
              target=self.handler_launch_func, args=(scope, self.fetch_instance))
          self.running_lock = threading.Lock()
          self.running = False
  
      def handler_launch_func(self, scope, handler):
          fetch_instance = handler
          period_secs = fetch_instance.period_secs
          var_name_to_key = {}
          for key in fetch_instance.var_dict:
              if isinstance(fetch_instance.var_dict[key], Variable):
                  var_name_to_key[fetch_instance.var_dict[key].name] = key
              else:
                  local_logger.warning("the value of {} is not a Variable".format(
                      key))
                  var_name_to_key["None.var"] = key
          elapsed_secs = 0
          while True:
              self.running_lock.acquire()
              if self.running == False:
                  break
              if elapsed_secs < period_secs:
                  # TODO(guru4elephant): needs customized condition
                  time.sleep(1)
                  elapsed_secs += 1
              else:
                  elapsed_secs = 0
                  fetch_dict = {}
                  for key in var_name_to_key:
                      var = scope.find_var(key)
                      fetch_dict[key] = var
                      if var == None:
                          local_logger.warning("{} value currently not available".
                                               format(var_name_to_key[key]))
                  res_dict = {}
                  for key in fetch_dict:
                      user_name = var_name_to_key[key]
                      if fetch_dict[key] == None:
                          res_dict[user_name] = None
                          continue
                      else:
                          res_dict[user_name] = fetch_dict[key].get_tensor()
  
                      lod = res_dict[user_name].lod()
                      if len(lod) > 0:
                          raise RuntimeError("Some of your fetched tensors \
                                              hold LoD information. \
                                              They can not be completely cast \
                                              to Python ndarray. We can \
                                              not return LoDTensor itself directly, \
                                              please choose another targets")
                      if res_dict[user_name]._is_initialized():
                          res_dict[user_name] = np.array(res_dict[user_name])
                      else:
                          res_dict[user_name] = None
                  fetch_instance.handler(res_dict)
              self.running_lock.release()
  
      def start(self):
          """
          start monitor,
          it will start a monitor thread.
          """
          self.running_lock.acquire()
          self.running = True
          self.running_lock.release()
          self.fetch_thread.setDaemon(True)
          self.fetch_thread.start()
  
      def stop(self):
          self.running_lock.acquire()
          self.running = False
          self.running_lock.release()
   # Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserved
  #
  # 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.
  """
  UCI Housing dataset.
  
  This module will download dataset from
  https://archive.ics.uci.edu/ml/machine-learning-databases/housing/ and
  parse training set and test set into paddle reader creators.
  """
  
  from __future__ import print_function
  
  import numpy as np
  import six
  import tempfile
  import tarfile
  import os
  import paddle.dataset.common
  
  __all__ = ['train', 'test']
  
  URL = 'http://paddlemodels.bj.bcebos.com/uci_housing/housing.data'
  MD5 = 'd4accdce7a25600298819f8e28e8d593'
  feature_names = [
      'CRIM', 'ZN', 'INDUS', 'CHAS', 'NOX', 'RM', 'AGE', 'DIS', 'RAD', 'TAX',
      'PTRATIO', 'B', 'LSTAT'
  ]
  
  UCI_TRAIN_DATA = None
  UCI_TEST_DATA = None
  
  FLUID_URL_MODEL = 'https://github.com/PaddlePaddle/book/raw/develop/01.fit_a_line/fluid/fit_a_line.fluid.tar'
  FLUID_MD5_MODEL = '6e6dd637ccd5993961f68bfbde46090b'
  
  
  def feature_range(maximums, minimums):
      import matplotlib
      matplotlib.use('Agg')
      import matplotlib.pyplot as plt
      fig, ax = plt.subplots()
      feature_num = len(maximums)
      ax.bar(list(range(feature_num)),
             maximums - minimums,
             color='r',
             align='center')
      ax.set_title('feature scale')
      plt.xticks(list(range(feature_num)), feature_names)
      plt.xlim([-1, feature_num])
      fig.set_figheight(6)
      fig.set_figwidth(10)
      if not os.path.exists('./image'):
          os.makedirs('./image')
      fig.savefig('image/ranges.png', dpi=48)
      plt.close(fig)
  
  
  def load_data(filename, feature_num=14, ratio=0.8):
      global UCI_TRAIN_DATA, UCI_TEST_DATA
      if UCI_TRAIN_DATA is not None and UCI_TEST_DATA is not None:
          return
  
      data = np.fromfile(filename, sep=' ')
      data = data.reshape(data.shape[0] // feature_num, feature_num)
      maximums, minimums, avgs = data.max(axis=0), data.min(axis=0), data.sum(
          axis=0) / data.shape[0]
      # if you want to print the distribution of input data, you could use function of feature_range
      #feature_range(maximums[:-1], minimums[:-1])
      for i in six.moves.range(feature_num - 1):
          data[:, i] = (data[:, i] - avgs[i]) / (maximums[i] - minimums[i])
      offset = int(data.shape[0] * ratio)
      UCI_TRAIN_DATA = data[:offset]
      UCI_TEST_DATA = data[offset:]
  
  
  def train():
      """
      UCI_HOUSING training set creator.
  
      It returns a reader creator, each sample in the reader is features after
      normalization and price number.
  
      :return: Training reader creator
      :rtype: callable
      """
      global UCI_TRAIN_DATA
      load_data(paddle.dataset.common.download(URL, 'uci_housing', MD5))
  
      def reader():
          for d in UCI_TRAIN_DATA:
              yield d[:-1], d[-1:]
  
      return reader
  
  
  def test():
      """
      UCI_HOUSING test set creator.
  
      It returns a reader creator, each sample in the reader is features after
      normalization and price number.
  
      :return: Test reader creator
      :rtype: callable
      """
      global UCI_TEST_DATA
      load_data(paddle.dataset.common.download(URL, 'uci_housing', MD5))
  
      def reader():
          for d in UCI_TEST_DATA:
              yield d[:-1], d[-1:]
  
      return reader
  
  
  def fluid_model():
      parameter_tar = paddle.dataset.common.download(
          FLUID_URL_MODEL, 'uci_housing', FLUID_MD5_MODEL, 'fit_a_line.fluid.tar')
  
      tar = tarfile.TarFile(parameter_tar, mode='r')
      dirpath = tempfile.mkdtemp()
      tar.extractall(path=dirpath)
  
      return dirpath
  
  
  def predict_reader():
      """
      It returns just one tuple data to do inference.
  
      :return: one tuple data
      :rtype: tuple
      """
      global UCI_TEST_DATA
      load_data(paddle.dataset.common.download(URL, 'uci_housing', MD5))
      return (UCI_TEST_DATA[0][:-1], )
  
  
  def fetch():
      paddle.dataset.common.download(URL, 'uci_housing', MD5)
  
apis: numpy.array
apis: numpy.fromfile matplotlib.pyplot.subplots matplotlib.pyplot.xlim matplotlib.pyplot.close matplotlib.use