2020-03-04 18:16:50 +00:00
|
|
|
#!/usr/bin/env python3
|
2020-03-01 18:07:02 +00:00
|
|
|
import json
|
|
|
|
from transport import factory
|
2020-03-04 17:49:18 +00:00
|
|
|
import numpy as np
|
2020-03-06 17:40:47 +00:00
|
|
|
import time
|
2020-03-01 18:07:02 +00:00
|
|
|
import os
|
2020-03-27 05:34:05 +00:00
|
|
|
from multiprocessing import Process, Lock
|
2020-03-01 18:07:02 +00:00
|
|
|
import pandas as pd
|
|
|
|
from google.oauth2 import service_account
|
|
|
|
import data.maker
|
|
|
|
|
|
|
|
from data.params import SYS_ARGS
|
|
|
|
|
|
|
|
#
|
|
|
|
# The configuration array is now loaded and we will execute the pipe line as follows
|
2020-03-04 20:30:40 +00:00
|
|
|
DATASET='combined20191004v2_deid'
|
2020-03-01 18:07:02 +00:00
|
|
|
|
|
|
|
class Components :
|
2020-03-27 05:34:05 +00:00
|
|
|
lock = Lock()
|
2020-03-25 22:43:23 +00:00
|
|
|
class KEYS :
|
|
|
|
PIPELINE_KEY = 'pipeline'
|
|
|
|
SQL_FILTER = 'filter'
|
2020-04-02 05:04:05 +00:00
|
|
|
@staticmethod
|
|
|
|
def get_filter (**args):
|
|
|
|
if args['qualifier'] == 'IN' :
|
|
|
|
return ' '.join([args['field'],args['qualifier'],'(',args['value'],')'])
|
|
|
|
else:
|
|
|
|
return ' '.join([args['field'],args['qualifier'],args['value']])
|
2020-03-25 22:43:23 +00:00
|
|
|
@staticmethod
|
|
|
|
def get_logger(**args) :
|
|
|
|
return factory.instance(type='mongo.MongoWriter',args={'dbname':'aou','doc':args['context']})
|
2020-03-04 17:49:18 +00:00
|
|
|
@staticmethod
|
|
|
|
def get(args):
|
|
|
|
"""
|
|
|
|
This function returns a data-frame provided a bigquery sql statement with conditions (and limits for testing purposes)
|
|
|
|
The function must be wrapped around a lambda this makes testing easier and changing data stores transparent to the rest of the code. (Vital when testing)
|
|
|
|
:sql basic sql statement
|
|
|
|
:condition optional condition and filters
|
|
|
|
"""
|
|
|
|
SQL = args['sql']
|
2020-03-25 22:43:23 +00:00
|
|
|
if Components.KEYS.SQL_FILTER in args :
|
2020-04-02 05:04:05 +00:00
|
|
|
FILTER_KEY = Components.KEYS.SQL_FILTER
|
|
|
|
SQL_FILTER = args[FILTER_KEY] if type(args[FILTER_KEY]) == list else [args[FILTER_KEY]]
|
|
|
|
# condition = ' '.join([args[FILTER_KEY]['field'],args[FILTER_KEY]['qualifier'],'(',args[FILTER_KEY]['value'],')'])
|
|
|
|
|
|
|
|
condition = ' AND '.join([Components.get_filter(**item) for item in SQL_FILTER])
|
2020-03-04 17:49:18 +00:00
|
|
|
SQL = " ".join([SQL,'WHERE',condition])
|
2020-03-01 18:07:02 +00:00
|
|
|
|
2020-03-08 13:48:38 +00:00
|
|
|
SQL = SQL.replace(':dataset',args['dataset']) #+ " LI "
|
|
|
|
|
2020-03-04 17:49:18 +00:00
|
|
|
if 'limit' in args :
|
2020-03-08 13:48:38 +00:00
|
|
|
SQL = SQL + ' LIMIT ' + args['limit']
|
2020-03-25 22:43:23 +00:00
|
|
|
#
|
|
|
|
# let's log the sql query that has been performed here
|
|
|
|
logger = factory.instance(type='mongo.MongoWriter',args={'dbname':'aou','doc':args['context']})
|
|
|
|
logger.write({"module":"bigquery","action":"read","input":{"sql":SQL}})
|
2020-03-04 17:49:18 +00:00
|
|
|
credentials = service_account.Credentials.from_service_account_file('/home/steve/dev/aou/accounts/curation-prod.json')
|
2020-04-01 05:21:51 +00:00
|
|
|
df = pd.read_gbq(SQL,credentials=credentials,dialect='standard')
|
2020-03-04 17:49:18 +00:00
|
|
|
return df
|
|
|
|
|
|
|
|
# return lambda: pd.read_gbq(SQL,credentials=credentials,dialect='standard')[args['columns']].dropna()
|
|
|
|
@staticmethod
|
|
|
|
def split(X,MAX_ROWS=3,PART_SIZE=3):
|
|
|
|
|
|
|
|
return list(pd.cut( np.arange(X.shape[0]+1),PART_SIZE).categories)
|
2020-03-01 18:07:02 +00:00
|
|
|
|
2020-03-04 17:49:18 +00:00
|
|
|
def train(self,**args):
|
|
|
|
"""
|
|
|
|
This function will perform training on the basis of a given pointer that reads data
|
2020-03-01 18:07:02 +00:00
|
|
|
|
2020-03-04 17:49:18 +00:00
|
|
|
"""
|
|
|
|
#
|
|
|
|
# @TODO: we need to log something here about the parameters being passed
|
2020-03-07 15:16:17 +00:00
|
|
|
# pointer = args['reader'] if 'reader' in args else lambda: Components.get(**args)
|
2020-03-08 13:48:38 +00:00
|
|
|
df = args['data']
|
|
|
|
|
2020-03-07 15:16:17 +00:00
|
|
|
|
|
|
|
# if df.shape[0] == 0 :
|
|
|
|
# print ("CAN NOT TRAIN EMPTY DATASET ")
|
|
|
|
# return
|
2020-03-04 17:49:18 +00:00
|
|
|
#
|
|
|
|
# Now we can parse the arguments and submit the entire thing to training
|
|
|
|
#
|
|
|
|
|
|
|
|
logger = factory.instance(type='mongo.MongoWriter',args={'dbname':'aou','doc':args['context']})
|
|
|
|
log_folder = args['logs'] if 'logs' in args else 'logs'
|
2020-03-08 13:48:38 +00:00
|
|
|
PART_SIZE = int(args['part_size'])
|
2020-03-06 17:40:47 +00:00
|
|
|
|
2020-03-08 13:48:38 +00:00
|
|
|
partition = args['partition']
|
|
|
|
log_folder = os.sep.join([log_folder,args['context'],str(partition)])
|
2020-03-15 15:25:19 +00:00
|
|
|
_args = {"batch_size":2000,"logs":log_folder,"context":args['context'],"max_epochs":150,"column":args['columns'],"id":"person_id","logger":logger}
|
2020-03-08 13:48:38 +00:00
|
|
|
_args['max_epochs'] = 150 if 'max_epochs' not in args else int(args['max_epochs'])
|
2020-03-08 20:00:26 +00:00
|
|
|
if 'batch_size' in args :
|
|
|
|
_args['batch_size'] = int(args['batch_size'])
|
2020-03-15 15:25:19 +00:00
|
|
|
|
2020-03-08 13:48:38 +00:00
|
|
|
#
|
|
|
|
# We ask the process to assume 1 gpu given the system number of GPU and that these tasks can run in parallel
|
|
|
|
#
|
|
|
|
if int(args['num_gpu']) > 1 :
|
2020-03-09 18:10:26 +00:00
|
|
|
_args['gpu'] = int(args['gpu']) if int(args['gpu']) < 8 else np.random.choice(np.arange(8)).astype(int)
|
2020-03-04 17:49:18 +00:00
|
|
|
else:
|
2020-03-08 13:48:38 +00:00
|
|
|
_args['gpu'] = 0
|
|
|
|
_args['num_gpu'] = 1
|
|
|
|
os.environ['CUDA_VISIBLE_DEVICES'] = str(args['gpu'])
|
2020-03-09 00:33:08 +00:00
|
|
|
_args['partition'] = int(partition)
|
|
|
|
_args['continuous']= args['continuous'] if 'continuous' in args else []
|
2020-03-08 13:48:38 +00:00
|
|
|
_args['store'] = {'type':'mongo.MongoWriter','args':{'dbname':'aou','doc':args['context']}}
|
|
|
|
_args['data'] = args['data']
|
|
|
|
|
|
|
|
# print (['partition ',partition,df.value_source_concept_id.unique()])
|
|
|
|
#
|
|
|
|
# @log :
|
|
|
|
# Logging information about the training process for this partition (or not)
|
|
|
|
#
|
|
|
|
|
|
|
|
info = {"rows":df.shape[0],"cols":df.shape[1], "partition":int(partition),"logs":_args['logs']}
|
|
|
|
|
|
|
|
logger.write({"module":"train","action":"train","input":info})
|
|
|
|
data.maker.train(**_args)
|
2020-03-04 17:49:18 +00:00
|
|
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
# @staticmethod
|
|
|
|
def generate(self,args):
|
|
|
|
"""
|
|
|
|
This function will generate data and store it to a given,
|
|
|
|
"""
|
|
|
|
logger = factory.instance(type='mongo.MongoWriter',args={'dbname':'aou','doc':args['context']})
|
|
|
|
log_folder = args['logs'] if 'logs' in args else 'logs'
|
|
|
|
partition = args['partition'] if 'partition' in args else ''
|
2020-03-05 18:03:04 +00:00
|
|
|
log_folder = os.sep.join([log_folder,args['context'],str(partition)])
|
|
|
|
|
2020-03-06 17:17:29 +00:00
|
|
|
_args = {"batch_size":2000,"logs":log_folder,"context":args['context'],"max_epochs":150,"column":args['columns'],"id":"person_id","logger":logger}
|
2020-03-04 17:49:18 +00:00
|
|
|
_args['max_epochs'] = 150 if 'max_epochs' not in args else int(args['max_epochs'])
|
2020-03-06 20:20:06 +00:00
|
|
|
# _args['num_gpu'] = int(args['num_gpu']) if 'num_gpu' in args else 1
|
2020-03-08 20:00:26 +00:00
|
|
|
if 'batch_size' in args :
|
|
|
|
_args['batch_size'] = int(args['batch_size'])
|
2020-03-06 20:20:06 +00:00
|
|
|
|
2020-03-06 20:56:28 +00:00
|
|
|
if int(args['num_gpu']) > 1 :
|
2020-03-09 18:10:26 +00:00
|
|
|
_args['gpu'] = int(args['gpu']) if int(args['gpu']) < 8 else np.random.choice(np.arange(8)).astype(int)
|
2020-03-06 20:20:06 +00:00
|
|
|
else:
|
|
|
|
_args['gpu'] = 0
|
|
|
|
_args['num_gpu'] = 1
|
2020-03-06 20:42:40 +00:00
|
|
|
os.environ['CUDA_VISIBLE_DEVICES'] = str(args['gpu'])
|
2020-03-04 17:49:18 +00:00
|
|
|
_args['no_value']= args['no_value']
|
2020-03-25 22:43:23 +00:00
|
|
|
|
2020-03-06 05:49:55 +00:00
|
|
|
# MAX_ROWS = args['max_rows'] if 'max_rows' in args else 0
|
|
|
|
PART_SIZE = int(args['part_size']) if 'part_size' in args else 8
|
2020-03-04 17:49:18 +00:00
|
|
|
|
|
|
|
# credentials = service_account.Credentials.from_service_account_file('/home/steve/dev/aou/accounts/curation-prod.json')
|
|
|
|
# _args['data'] = pd.read_gbq(SQL,credentials=credentials,dialect='standard').dropna()
|
2020-03-06 20:20:06 +00:00
|
|
|
# reader = args['reader']
|
|
|
|
# df = reader()
|
|
|
|
df = args['reader']() if 'reader' in args else args['data']
|
2020-04-02 12:52:09 +00:00
|
|
|
|
2020-03-06 05:49:55 +00:00
|
|
|
# bounds = Components.split(df,MAX_ROWS,PART_SIZE)
|
2020-03-06 20:20:06 +00:00
|
|
|
# if partition != '' :
|
|
|
|
# columns = args['columns']
|
|
|
|
# df = np.array_split(df[columns].values,PART_SIZE)
|
|
|
|
# df = pd.DataFrame(df[ int (partition) ],columns = columns)
|
2020-04-02 05:04:05 +00:00
|
|
|
# max_rows = int(args['partition_max_rows']) if 'partition_max_rows' in args else 1000000
|
|
|
|
# N = np.divide(df.shape[0],max_rows).astype(int) + 1
|
2020-04-02 06:36:57 +00:00
|
|
|
info = {"parition":int(partition),"gpu":_args["gpu"],"rows":int(df.shape[0]),"cols":int(df.shape[1]),"part_size":int(PART_SIZE)}
|
2020-03-06 20:20:06 +00:00
|
|
|
logger.write({"module":"generate","action":"partition","input":info})
|
2020-03-09 00:33:08 +00:00
|
|
|
_args['partition'] = int(partition)
|
|
|
|
_args['continuous']= args['continuous'] if 'continuous' in args else []
|
2020-04-02 05:04:05 +00:00
|
|
|
#
|
|
|
|
# How many rows sub-partition must we divide this into ?
|
|
|
|
# -- Let us tray assessing
|
2020-03-06 20:20:06 +00:00
|
|
|
|
2020-04-02 05:04:05 +00:00
|
|
|
|
2020-04-02 12:52:09 +00:00
|
|
|
|
2020-04-02 05:04:05 +00:00
|
|
|
_dc = pd.DataFrame()
|
|
|
|
# for mdf in df :
|
|
|
|
_args['data'] = df
|
|
|
|
_dc = _dc.append(data.maker.generate(**_args))
|
2020-03-04 17:49:18 +00:00
|
|
|
#
|
|
|
|
# We need to post the generate the data in order to :
|
|
|
|
# 1. compare immediately
|
|
|
|
# 2. synthetic copy
|
|
|
|
#
|
|
|
|
|
|
|
|
cols = _dc.columns.tolist()
|
|
|
|
|
|
|
|
data_comp = _args['data'][args['columns']].join(_dc[args['columns']],rsuffix='_io') #-- will be used for comparison (store this in big query)
|
2020-03-19 04:16:36 +00:00
|
|
|
#
|
|
|
|
# performing basic analytics on the synthetic data generated (easy to quickly asses)
|
|
|
|
#
|
2020-03-25 22:43:23 +00:00
|
|
|
info = {"module":"generate","action":"io.metrics","input":{"rows":data_comp.shape[0],"partition":partition,"logs":[]}}
|
|
|
|
x = {}
|
2020-04-02 05:04:05 +00:00
|
|
|
#
|
|
|
|
# @TODO: Send data over to a process for analytics
|
2020-03-19 04:16:36 +00:00
|
|
|
|
2020-03-04 17:49:18 +00:00
|
|
|
base_cols = list(set(_args['data'].columns) - set(args['columns'])) #-- rebuilt the dataset (and store it)
|
2020-03-25 22:43:23 +00:00
|
|
|
cols = _dc.columns.tolist()
|
2020-03-27 06:52:16 +00:00
|
|
|
for name in cols :
|
|
|
|
_args['data'][name] = _dc[name]
|
2020-03-04 17:49:18 +00:00
|
|
|
|
|
|
|
#
|
|
|
|
#-- Let us store all of this into bigquery
|
|
|
|
prefix = args['notify']+'.'+_args['context']
|
2020-03-08 13:48:38 +00:00
|
|
|
partition = str(partition)
|
2020-03-04 17:49:18 +00:00
|
|
|
table = '_'.join([prefix,partition,'io']).replace('__','_')
|
|
|
|
folder = os.sep.join([args['logs'],args['context'],partition,'output'])
|
|
|
|
if 'file' in args :
|
|
|
|
|
|
|
|
_fname = os.sep.join([folder,table.replace('_io','_full_io.csv')])
|
|
|
|
_pname = os.sep.join([folder,table])+'.csv'
|
|
|
|
data_comp.to_csv( _pname,index=False)
|
|
|
|
_args['data'].to_csv(_fname,index=False)
|
|
|
|
|
2020-03-08 13:48:38 +00:00
|
|
|
_id = 'path'
|
2020-03-04 17:49:18 +00:00
|
|
|
else:
|
|
|
|
credentials = service_account.Credentials.from_service_account_file('/home/steve/dev/aou/accounts/curation-prod.json')
|
|
|
|
_pname = os.sep.join([folder,table+'.csv'])
|
|
|
|
_fname = table.replace('_io','_full_io')
|
2020-03-08 13:48:38 +00:00
|
|
|
partial = '.'.join(['io',args['context']+'_partial_io'])
|
|
|
|
complete= '.'.join(['io',args['context']+'_full_io'])
|
2020-03-04 17:49:18 +00:00
|
|
|
data_comp.to_csv(_pname,index=False)
|
2020-03-15 15:25:19 +00:00
|
|
|
if 'dump' in args :
|
|
|
|
print (_args['data'].head())
|
|
|
|
else:
|
2020-03-27 05:34:05 +00:00
|
|
|
Components.lock.acquire()
|
2020-03-19 04:16:36 +00:00
|
|
|
data_comp.to_gbq(if_exists='append',destination_table=partial,credentials=credentials,chunksize=90000)
|
2020-03-15 15:25:19 +00:00
|
|
|
|
|
|
|
INSERT_FLAG = 'replace' if 'partition' not in args or 'segment' not in args else 'append'
|
2020-03-19 04:16:36 +00:00
|
|
|
_args['data'].to_gbq(if_exists='append',destination_table=complete,credentials=credentials,chunksize=90000)
|
2020-03-27 05:34:05 +00:00
|
|
|
Components.lock.release()
|
2020-03-08 13:48:38 +00:00
|
|
|
_id = 'dataset'
|
|
|
|
info = {"full":{_id:_fname,"rows":_args['data'].shape[0]},"partial":{"path":_pname,"rows":data_comp.shape[0]} }
|
2020-03-04 17:49:18 +00:00
|
|
|
if partition :
|
2020-03-06 05:49:55 +00:00
|
|
|
info ['partition'] = int(partition)
|
2020-03-06 19:00:32 +00:00
|
|
|
logger.write({"module":"generate","action":"write","input":info} )
|
2020-03-06 17:40:47 +00:00
|
|
|
|
2020-03-25 22:43:23 +00:00
|
|
|
|
2020-03-04 17:49:18 +00:00
|
|
|
|
2020-03-01 18:07:02 +00:00
|
|
|
if __name__ == '__main__' :
|
2020-03-04 17:49:18 +00:00
|
|
|
filename = SYS_ARGS['config'] if 'config' in SYS_ARGS else 'config.json'
|
|
|
|
f = open (filename)
|
2020-03-25 22:43:23 +00:00
|
|
|
_config = json.loads(f.read())
|
2020-03-04 17:49:18 +00:00
|
|
|
f.close()
|
2020-03-25 22:43:23 +00:00
|
|
|
PIPELINE = _config['pipeline']
|
2020-03-16 21:22:34 +00:00
|
|
|
index = SYS_ARGS['index']
|
|
|
|
if index.isnumeric() :
|
|
|
|
index = int(SYS_ARGS['index'])
|
|
|
|
else:
|
|
|
|
#
|
|
|
|
# The index provided is a key to a pipeline entry mainly the context
|
|
|
|
#
|
|
|
|
N = len(PIPELINE)
|
|
|
|
f = [i for i in range(0,N) if PIPELINE[i]['context'] == index]
|
|
|
|
index = f[0] if f else 0
|
|
|
|
#
|
2020-04-02 05:04:05 +00:00
|
|
|
|
2020-03-16 21:22:34 +00:00
|
|
|
print ("..::: ",PIPELINE[index]['context'])
|
2020-03-04 17:49:18 +00:00
|
|
|
args = (PIPELINE[index])
|
2020-03-25 22:43:23 +00:00
|
|
|
for key in _config :
|
|
|
|
if key == 'pipeline' or key in args:
|
|
|
|
#
|
|
|
|
# skip in case of pipeline or if key exists in the selected pipeline (provided by index)
|
|
|
|
#
|
|
|
|
continue
|
|
|
|
args[key] = _config[key]
|
2020-04-02 05:04:05 +00:00
|
|
|
|
2020-03-04 17:49:18 +00:00
|
|
|
args = dict(args,**SYS_ARGS)
|
2020-04-01 05:53:56 +00:00
|
|
|
if 'batch_size' not in args :
|
|
|
|
args['batch_size'] = 2000 #if 'batch_size' not in args else int(args['batch_size'])
|
2020-03-04 20:30:40 +00:00
|
|
|
if 'dataset' not in args :
|
|
|
|
args['dataset'] = 'combined20191004v2_deid'
|
2020-03-08 13:48:38 +00:00
|
|
|
PART_SIZE = int(args['part_size']) if 'part_size' in args else 8
|
2020-03-04 17:49:18 +00:00
|
|
|
#
|
|
|
|
# @TODO:
|
|
|
|
# Log what was initiated so we have context of this processing ...
|
|
|
|
#
|
2020-04-02 05:04:05 +00:00
|
|
|
# if 'listen' not in SYS_ARGS :
|
|
|
|
if 'file' in args :
|
|
|
|
DATA = pd.read_csv(args['file']) ;
|
|
|
|
else:
|
|
|
|
DATA = Components().get(args)
|
|
|
|
COLUMNS = DATA.columns
|
|
|
|
DATA = np.array_split(DATA,PART_SIZE)
|
2020-03-04 17:49:18 +00:00
|
|
|
|
|
|
|
if 'generate' in SYS_ARGS :
|
|
|
|
#
|
|
|
|
# Let us see if we have partitions given the log folder
|
|
|
|
|
|
|
|
content = os.listdir( os.sep.join([args['logs'],args['context']]))
|
|
|
|
generator = Components()
|
2020-03-08 13:48:38 +00:00
|
|
|
|
2020-03-04 17:49:18 +00:00
|
|
|
if ''.join(content).isnumeric() :
|
|
|
|
#
|
|
|
|
# we have partitions we are working with
|
2020-03-07 15:16:17 +00:00
|
|
|
|
2020-03-06 18:11:22 +00:00
|
|
|
jobs = []
|
2020-03-08 13:48:38 +00:00
|
|
|
|
|
|
|
# columns = DATA.columns.tolist()
|
|
|
|
|
|
|
|
# DATA = np.array_split(DATA,PART_SIZE)
|
2020-03-06 20:56:28 +00:00
|
|
|
|
2020-03-08 13:48:38 +00:00
|
|
|
for index in range(0,PART_SIZE) :
|
|
|
|
if 'focus' in args and int(args['focus']) != index :
|
2020-03-06 20:56:28 +00:00
|
|
|
#
|
|
|
|
# This handles failures/recoveries for whatever reason
|
|
|
|
# If we are only interested in generating data for a given partition
|
|
|
|
continue
|
2020-03-08 13:48:38 +00:00
|
|
|
# index = id.index(id)
|
|
|
|
|
|
|
|
args['partition'] = index
|
|
|
|
args['data'] = DATA[index]
|
2020-03-06 20:56:28 +00:00
|
|
|
if int(args['num_gpu']) > 1 :
|
2020-03-08 13:48:38 +00:00
|
|
|
args['gpu'] = index
|
2020-03-06 20:20:06 +00:00
|
|
|
else:
|
|
|
|
args['gpu']=0
|
2020-04-02 05:04:05 +00:00
|
|
|
|
2020-03-07 15:16:17 +00:00
|
|
|
make = lambda _args: (Components()).generate(_args)
|
2020-03-06 19:00:32 +00:00
|
|
|
job = Process(target=make,args=(args,))
|
2020-03-08 13:48:38 +00:00
|
|
|
job.name = 'generator # '+str(index)
|
2020-03-06 18:11:22 +00:00
|
|
|
job.start()
|
|
|
|
jobs.append(job)
|
2020-03-27 05:34:05 +00:00
|
|
|
# if len(jobs) == 1 :
|
|
|
|
# job.join()
|
2020-03-06 19:00:32 +00:00
|
|
|
|
2020-03-07 15:16:17 +00:00
|
|
|
print (["Started ",len(jobs),"generators" if len(jobs)>1 else "generator" ])
|
2020-03-06 19:00:32 +00:00
|
|
|
while len(jobs)> 0 :
|
|
|
|
jobs = [job for job in jobs if job.is_alive()]
|
2020-03-06 18:11:22 +00:00
|
|
|
time.sleep(2)
|
2020-03-04 17:49:18 +00:00
|
|
|
|
2020-03-06 18:11:22 +00:00
|
|
|
# generator.generate(args)
|
2020-03-04 17:49:18 +00:00
|
|
|
else:
|
|
|
|
generator.generate(args)
|
|
|
|
# Components.generate(args)
|
2020-03-25 22:43:23 +00:00
|
|
|
elif 'finalize' in args :
|
2020-03-04 17:49:18 +00:00
|
|
|
#
|
2020-03-25 22:43:23 +00:00
|
|
|
# This will finalize a given set of synthetic operations into a table
|
2020-03-04 17:49:18 +00:00
|
|
|
#
|
2020-03-25 22:43:23 +00:00
|
|
|
idataset = args['input'] if 'input' in args else 'io' #-- input dataset
|
|
|
|
odataset = args['output'] #-- output dataset
|
|
|
|
labels = [name.strip() for name in args['labels'].split(',') ]
|
2020-03-04 17:49:18 +00:00
|
|
|
|
|
|
|
else:
|
2020-03-08 13:48:38 +00:00
|
|
|
|
|
|
|
# DATA = np.array_split(DATA,PART_SIZE)
|
|
|
|
|
2020-03-07 15:16:17 +00:00
|
|
|
jobs = []
|
2020-03-08 13:48:38 +00:00
|
|
|
for index in range(0,PART_SIZE) :
|
2020-03-07 15:16:17 +00:00
|
|
|
if 'focus' in args and int(args['focus']) != index :
|
|
|
|
continue
|
2020-03-08 13:48:38 +00:00
|
|
|
args['part_size'] = PART_SIZE
|
2020-03-07 15:16:17 +00:00
|
|
|
args['partition'] = index
|
2020-03-08 13:48:38 +00:00
|
|
|
args['data'] = DATA[index]
|
|
|
|
if int(args['num_gpu']) > 1 :
|
|
|
|
args['gpu'] = index
|
|
|
|
else:
|
|
|
|
args['gpu']=0
|
|
|
|
|
2020-03-07 15:16:17 +00:00
|
|
|
make = lambda _args: (Components()).train(**_args)
|
2020-03-08 13:48:38 +00:00
|
|
|
job = Process(target=make,args=( dict(args),))
|
2020-03-07 15:16:17 +00:00
|
|
|
job.name = 'Trainer # ' + str(index)
|
|
|
|
job.start()
|
|
|
|
jobs.append(job)
|
|
|
|
# args['gpu']
|
|
|
|
print (["Started ",len(jobs),"trainers" if len(jobs)>1 else "trainer" ])
|
|
|
|
while len(jobs)> 0 :
|
|
|
|
jobs = [job for job in jobs if job.is_alive()]
|
|
|
|
time.sleep(2)
|
|
|
|
|
|
|
|
# trainer = Components()
|
|
|
|
# trainer.train(**args)
|
|
|
|
|
2020-03-04 17:49:18 +00:00
|
|
|
|
|
|
|
# Components.train(**args)
|
2020-03-01 18:07:02 +00:00
|
|
|
#for args in PIPELINE :
|
2020-03-04 17:49:18 +00:00
|
|
|
#args['dataset'] = 'combined20190510'
|
|
|
|
#process = Process(target=Components.train,args=(args,))
|
|
|
|
#process.name = args['context']
|
|
|
|
#process.start()
|
|
|
|
# Components.train(args)
|