

def run(self, view, model, input_format, trainfile, feature_format, featurefile, num_engines, workdir, overwrite, modelfile):
    logging.info('creating models directory {0}...'.format(workdir))
    subprocess.check_call(['mkdir', '-p', workdir])
    done = []
    if (not overwrite):
        logging.info('checking for existing output models...')
        done.extend(self.find_done(workdir))
        if done:
            logging.info('found {0} output files'.format(len(done)))
    logging.info('creating tasks...')
    tasks = self.create_tasks(model, input_format, trainfile, feature_format, featurefile, workdir, num_engines, done)
    if tasks:
        logging.info('running in parallel across ipython engines...')
        async_job = view.map_async(process, tasks, retries=2)
        results = async_job.get()
        logging.info('checking output files...')
        done = self.find_done(workdir)
        remaining = (len(tasks) - len(done))
    else:
        remaining = 0
    if (remaining == 0):
        logging.info('SUCCESS: all tasks completed')
        logging.info('concatenating {0} models...'.format(len(done)))
        for ix in sorted(done):
            partial_model = load_recommender(self.get_modelfile(ix, workdir))
            if (ix == 0):
                model = partial_model
            else:
                model.d += partial_model.d
                model.U = np.hstack((model.U, partial_model.U))
                model.V = np.hstack((model.V, partial_model.V))
                if hasattr(model, 'W'):
                    model.W = np.hstack((model.W, partial_model.W))
        save_recommender(model, modelfile)
        logging.info('removing partial output files...')
        rmtree(workdir)
        logging.info('done')
    else:
        logging.error('FAILED: {0}/{1} tasks did not complete successfully'.format(remaining, len(tasks)))
        logging.error('try rerunning the command to retry the remaining tasks')

dataset/ETHPy150Open Mendeley/mrec/mrec/parallel/warp.py WARPMFRunner.run/original