import os

import platform

import time

import matplotlib

matplotlib.use('SVG')

import queue as Queue

import threading

from PYME import config

from PYME.misc.computerName import GetComputerName

compName = GetComputerName()

import logging

import logging.handlers

dataserver_root = config.get('dataserver-root')

if dataserver_root:

log_dir = '%s/LOGS/%s/taskWorkerHTTP' % (dataserver_root, compName)

if not os.path.exists(log_dir):

try:

os.makedirs(log_dir)

except:

#as we are launching multiple processes at once, there is a race condition here and we might

#have already created the directory between our test and the makedirs call

pass

#fh = logging.FileHandler('%s/%d.log' % (log_dir, os.getpid()), 'w')

#fh.setLevel(logging.DEBUG)

#logger.addHandler(fh)

#logging.basicConfig(filename ='%s/%d.log' % (log_dir, os.getpid()), level=logging.DEBUG)

logger = logging.getLogger('')

logger.setLevel(logging.DEBUG)

fh = logging.handlers.RotatingFileHandler(filename ='%s/%d.log' % (log_dir, os.getpid()), mode='w', maxBytes=1e6)

logger.addHandler(fh)

else:

logging.basicConfig(level=logging.DEBUG)

logger = logging.getLogger('')

import requests

import sys

import signal

import yaml

def str_presenter(dumper, data):

if len(data.splitlines()) > 1: # check for multiline string

return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')

return dumper.represent_scalar('tag:yaml.org,2002:str', data)

yaml.add_representer(str, str_presenter)

from PYME.localization import remFitBuf

from PYME.cluster import distribution

from PYME.IO import clusterIO

from PYME.IO import unifiedIO

from PYME.IO import clusterResults

LOCAL = False

if 'PYME_LOCAL_ONLY' in os.environ.keys():

LOCAL = os.environ['PYME_LOCAL_ONLY'] == '1'

class TaskError(object):

taskDescr= self.format_taskDesc(), traceback = self.traceback, **self.extra_template_kwargs(mode=mode))

class RecipeTaskError(TaskError):

return {'recipe': self.taskDescr['taskdef'].get('recipe', 'see taskdef.recipeURI')}

class taskWorker(object):

self.resultsQueue.put((queueURL, taskDescr, RecipeTaskError(taskDescr, tb, e)))

def on_SIGHUP(signum, frame):

raise RuntimeError('Recieved SIGHUP')

from argparse import ArgumentParser

if __name__ == '__main__':

op = ArgumentParser(description="PYME rule server for task distribution. This should run once per cluster.")

op.add_argument('-p', dest='profile', default=False, action='store_true',

help="enable profiling")

op.add_argument('--profile-dir', dest='profile_dir')

op.add_argument('-s', '--server-port', dest='server_port', type=int, default=config.get('nodeserver-port', 15347),

help='Optionally restrict advertisements to local machine')

args = op.parse_args()

if args.profile:

from PYME.util import mProfile

mProfile.profileOn(['taskWorkerHTTP.py', 'remFitBuf.py'])

if not platform.platform().startswith('Windows'):

signal.signal(signal.SIGHUP, on_SIGHUP)

try:

#main()

tW = taskWorker(nodeserver_port=args.server_port)

tW.loop_forever()

except KeyboardInterrupt:

#supress error message here - we only want to know if something bad happened

pass

finally:

if args.profile:

mProfile.report(display=False, profiledir=args.profile_dir)