Call new MPI init fn in mpi_learn.py; call print_unique() about signals

Author: felker

data/signals.py

@@ -1,4 +1,5 @@
 from __future__ import print_function
+import plasma.global_vars as g
 import numpy as np
 import sys
 
@@ -57,27 +58,27 @@ def get_units(str):
         found = True
 
     except Exception as e:
-        print(e)
+        g.print_unique(e)
         sys.stdout.flush()
         pass
 
     # Retrieve data from PTDATA if node not found
     if not found:
-        # print("not in full path {}".format(signal))
+        # g.print_unique("not in full path {}".format(signal))
         data = c.get('_s = ptdata2("'+signal+'",'+str(shot)+')').data()
         if len(data) != 1:
             rank = np.ndim(data)
             found = True
     # Retrieve data from Pseudo-pointname if not in ptdata
     if not found:
-        # print("not in PTDATA {}".format(signal))
+        # g.print_unique("not in PTDATA {}".format(signal))
         data = c.get('_s = pseudo("'+signal+'",'+str(shot)+')').data()
         if len(data) != 1:
             rank = np.ndim(data)
             found = True
     # this means the signal wasn't found
     if not found:
-        print("No such signal: {}".format(signal))
+        g.print_unique("No such signal: {}".format(signal))
         pass
 
     # get time base
@@ -125,7 +126,7 @@ def fetch_jet_data(signal_path, shot_num, c):
                     signal_path, shot_num)).data()
         found = True
     except Exception as e:
-        print(e)
+        g.print_unique(e)
         sys.stdout.flush()
         # pass
     return time, data, ydata, found
@@ -361,8 +362,9 @@ def fetch_nstx_data(signal_path, shot_num, c):
 
 all_signals_restricted = all_signals
 
-print('All signals (determines which signals are downloaded & preprocessed):')
-print(all_signals.values())
+g.print_unique('All signals (determines which signals are downloaded'
+               ' & preprocessed):')
+g.print_unique(all_signals.values())
 
 fully_defined_signals = {
     sig_name: sig for (sig_name, sig) in all_signals_restricted.items() if (

examples/mpi_learn.py

@@ -1,4 +1,5 @@
 import plasma.global_vars as g
+g.init_MPI()
 from plasma.models.mpi_runner import (
     mpi_train, mpi_make_predictions_and_evaluate
     )

plasma/conf_parser.py

@@ -1,3 +1,5 @@
+from __future__ import print_function
+import plasma.global_vars as g
 from plasma.primitives.shots import ShotListFiles
 import data.signals as sig
 from plasma.utils.hashing import myhash_signals
@@ -74,7 +76,7 @@ def parameters(input_file):
         elif params['target'] == 'ttdlinear':
             params['data']['target'] = TTDLinearTarget
         else:
-            print('Unkown type of target. Exiting')
+            g.print_unique('Unkown type of target. Exiting')
             exit(1)
 
         # params['model']['output_activation'] =
@@ -344,15 +346,17 @@ def parameters(input_file):
             params['paths']['use_signals_dict'] = sig.fully_defined_signals_1D
 
         else:
-            print("Unkown data set {}".format(params['paths']['data']))
+            g.print_unique("Unknown dataset {}".format(
+                params['paths']['data']))
             exit(1)
 
         if len(params['paths']['specific_signals']):
             for s in params['paths']['specific_signals']:
                 if s not in params['paths']['use_signals_dict'].keys():
-                    print("Signal {} is not fully defined for {} machine. ",
-                          "Skipping...".format(
-                              s, params['paths']['data'].split("_")[0]))
+                    g.print_unique(
+                        "Signal {} is not fully defined for {} machine. ",
+                        "Skipping...".format(
+                            s, params['paths']['data'].split("_")[0]))
             params['paths']['specific_signals'] = list(
                 filter(
                     lambda x: x in params['paths']['use_signals_dict'].keys(),
@@ -370,9 +374,9 @@ def parameters(input_file):
         params['paths']['all_signals'] = sort_by_channels(
             list(params['paths']['all_signals_dict'].values()))
 
-        print("Selected signals (determines which signals are used for",
-              "training):\n{}".format(params['paths']['use_signals']))
-
+        g.print_unique("Selected signals (determines which signals are used",
+                       " for training):\n{}".format(
+                           params['paths']['use_signals']))
         params['paths']['shot_files_all'] = (
             params['paths']['shot_files'] + params['paths']['shot_files_test'])
         params['paths']['all_machines'] = list(

plasma/global_vars.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 import sys
 
 # global variable defaults for non-MPI runs
@@ -9,15 +10,16 @@
 backend = ''
 
 
-def init_MPI(conf):
+def init_MPI():
     from mpi4py import MPI
     global comm, task_index, num_workers
-    global NUM_GPUS, MY_GPU, backend
-
     comm = MPI.COMM_WORLD
     task_index = comm.Get_rank()
     num_workers = comm.Get_size()
 
+
+def init_GPU_backend(conf):
+    global NUM_GPUS, MY_GPU, backend
     NUM_GPUS = conf['num_gpus']
     MY_GPU = task_index % NUM_GPUS
     backend = conf['model']['backend']

plasma/models/mpi_runner.py

@@ -39,15 +39,15 @@
 import socket
 sys.setrecursionlimit(10000)
 
-
 # import keras sequentially because it otherwise reads from ~/.keras/keras.json
 # with too many threads:
 # from mpi_launch_tensorflow import get_mpi_task_index
 
-# set global variables for entire module regarding MPI environment
-# TODO(KGF): consider moving this fn/init call to mpi_learn.py and/or
-# setting "mpi_initialized" global bool flag, since that is client-facing
-g.init_MPI(conf)
+# set global variables for entire module regarding MPI & GPU environment
+g.init_GPU_backend(conf)
+# moved this fn/init call to client-facing mpi_learn.py
+# g.init_MPI()
+# TODO(KGF): set "mpi_initialized" global bool flag?
 
 # initialization code for mpi_runner.py module:
 if g.backend == 'tf' or g.backend == 'tensorflow':

setup.cfg

@@ -26,4 +26,7 @@ ignore =
        E731,
        # W5: Line break warning
        # W503: line break before binary operator (use mutually exclusive W504)
-       W503
+       W503
+# suppres linter warning about MPI init fn call before module-level imports
+per-file-ignores =
+       examples/mpi_learn.py:E402