"""wrapper for finished Execute results"""

def __init__(self, msg_id, content, metadata):

self.msg_id = msg_id

self._content = content

self.execution_count = content['execution_count']

self.metadata = metadata

def __getitem__(self, key):

return self.metadata[key]

def __getattr__(self, key):

if key not in self.metadata:

raise AttributeError(key)

return self.metadata[key]

def __repr__(self):

pyout = self.metadata['pyout'] or {'data':{}}

text_out = pyout['data'].get('text/plain', '')

if len(text_out) > 32:

text_out = text_out[:29] + '...'

return "<ExecuteReply[%i]: %s>" % (self.execution_count, text_out)

def _repr_pretty_(self, p, cycle):

pyout = self.metadata['pyout'] or {'data':{}}

text_out = pyout['data'].get('text/plain', '')

if not text_out:

return

try:

ip = get_ipython()

except NameError:

colors = "NoColor"

else:

colors = ip.colors

if colors == "NoColor":

out = normal = ""

else:

out = TermColors.Red

normal = TermColors.Normal

if '\n' in text_out and not text_out.startswith('\n'):

# add newline for multiline reprs

text_out = '\n' + text_out

p.text(

out + u'Out[%i:%i]: ' % (

self.metadata['engine_id'], self.execution_count

) + normal + text_out

)

def _repr_html_(self):

pyout = self.metadata['pyout'] or {'data':{}}

return pyout['data'].get("text/html")

def _repr_latex_(self):

pyout = self.metadata['pyout'] or {'data':{}}

return pyout['data'].get("text/latex")

def _repr_json_(self):

pyout = self.metadata['pyout'] or {'data':{}}

return pyout['data'].get("application/json")

def _repr_javascript_(self):

pyout = self.metadata['pyout'] or {'data':{}}

return pyout['data'].get("application/javascript")

def _repr_png_(self):

pyout = self.metadata['pyout'] or {'data':{}}

return pyout['data'].get("image/png")

def _repr_jpeg_(self):

pyout = self.metadata['pyout'] or {'data':{}}

return pyout['data'].get("image/jpeg")

def _repr_svg_(self):

pyout = self.metadata['pyout'] or {'data':{}}

"""Subclass of dict for initializing metadata values.

def __init__(self, *args, **kwargs):

dict.__init__(self)

md = {'msg_id' : None,

'submitted' : None,

'started' : None,

'completed' : None,

'received' : None,

'engine_uuid' : None,

'engine_id' : None,

'follow' : None,

'after' : None,

'status' : None,

'pyin' : None,

'pyout' : None,

'pyerr' : None,

'stdout' : '',

'stderr' : '',

'outputs' : [],

'data': {},

'outputs_ready' : False,

}

self.update(md)

self.update(dict(*args, **kwargs))

def __getattr__(self, key):

"""getattr aliased to getitem"""

if key in self.iterkeys():

return self[key]

else:

raise AttributeError(key)

def __setattr__(self, key, value):

"""setattr aliased to setitem, with strict"""

if key in self.iterkeys():

self[key] = value

else:

raise AttributeError(key)

def __setitem__(self, key, value):

"""strict static key enforcement"""

if key in self.iterkeys():

dict.__setitem__(self, key, value)

else:

"""A semi-synchronous client to the IPython ZMQ cluster

block = Bool(False)

outstanding = Set()

results = Instance('collections.defaultdict', (dict,))

metadata = Instance('collections.defaultdict', (Metadata,))

history = List()

debug = Bool(False)

_spin_thread = Any()

_stop_spinning = Any()

profile=Unicode()

def _profile_default(self):

if BaseIPythonApplication.initialized():

# an IPython app *might* be running, try to get its profile

try:

return BaseIPythonApplication.instance().profile

except (AttributeError, MultipleInstanceError):

# could be a *different* subclass of config.Application,

# which would raise one of these two errors.

return u'default'

else:

return u'default'

_outstanding_dict = Instance('collections.defaultdict', (set,))

_ids = List()

_connected=Bool(False)

_ssh=Bool(False)

_context = Instance('zmq.Context')

_config = Dict()

_engines=Instance(util.ReverseDict, (), {})

# _hub_socket=Instance('zmq.Socket')

_query_socket=Instance('zmq.Socket')

_control_socket=Instance('zmq.Socket')

_iopub_socket=Instance('zmq.Socket')

_notification_socket=Instance('zmq.Socket')

_mux_socket=Instance('zmq.Socket')

_task_socket=Instance('zmq.Socket')

_task_scheme=Unicode()

_closed = False

_ignored_control_replies=Integer(0)

_ignored_hub_replies=Integer(0)

def __new__(self, *args, **kw):

# don't raise on positional args

return HasTraits.__new__(self, **kw)

def __init__(self, url_file=None, profile=None, profile_dir=None, ipython_dir=None,

context=None, debug=False,

sshserver=None, sshkey=None, password=None, paramiko=None,

timeout=10, cluster_id=None, **extra_args

):

if profile:

super(Client, self).__init__(debug=debug, profile=profile)

else:

super(Client, self).__init__(debug=debug)

if context is None:

context = zmq.Context.instance()

self._context = context

self._stop_spinning = Event()

if 'url_or_file' in extra_args:

url_file = extra_args['url_or_file']

warnings.warn("url_or_file arg no longer supported, use url_file", DeprecationWarning)

if url_file and util.is_url(url_file):

raise ValueError("single urls cannot be specified, url-files must be used.")

self._setup_profile_dir(self.profile, profile_dir, ipython_dir)

if self._cd is not None:

if url_file is None:

if not cluster_id:

client_json = 'ipcontroller-client.json'

else:

client_json = 'ipcontroller-%s-client.json' % cluster_id

url_file = pjoin(self._cd.security_dir, client_json)

if url_file is None:

raise ValueError(

"I can't find enough information to connect to a hub!"

" Please specify at least one of url_file or profile."

)

with open(url_file) as f:

cfg = json.load(f)

self._task_scheme = cfg['task_scheme']

# sync defaults from args, json:

if sshserver:

cfg['ssh'] = sshserver

location = cfg.setdefault('location', None)

proto,addr = cfg['interface'].split('://')

addr = util.disambiguate_ip_address(addr, location)

cfg['interface'] = "%s://%s" % (proto, addr)

# turn interface,port into full urls:

for key in ('control', 'task', 'mux', 'iopub', 'notification', 'registration'):

cfg[key] = cfg['interface'] + ':%i' % cfg[key]

url = cfg['registration']

if location is not None and addr == '127.0.0.1':

# location specified, and connection is expected to be local

if location not in LOCAL_IPS and not sshserver:

# load ssh from JSON *only* if the controller is not on

# this machine

sshserver=cfg['ssh']

if location not in LOCAL_IPS and not sshserver:

# warn if no ssh specified, but SSH is probably needed

# This is only a warning, because the most likely cause

# is a local Controller on a laptop whose IP is dynamic

warnings.warn("""

or instruct your controller to listen on an external IP."""%location,

RuntimeWarning)

elif not sshserver:

# otherwise sync with cfg

sshserver = cfg['ssh']

self._config = cfg

self._ssh = bool(sshserver or sshkey or password)

if self._ssh and sshserver is None:

# default to ssh via localhost

sshserver = addr

if self._ssh and password is None:

if tunnel.try_passwordless_ssh(sshserver, sshkey, paramiko):

password=False

else:

password = getpass("SSH Password for %s: "%sshserver)

ssh_kwargs = dict(keyfile=sshkey, password=password, paramiko=paramiko)

# configure and construct the session

extra_args['packer'] = cfg['pack']

extra_args['unpacker'] = cfg['unpack']

extra_args['key'] = cast_bytes(cfg['exec_key'])

self.session = Session(**extra_args)

self._query_socket = self._context.socket(zmq.DEALER)

if self._ssh:

tunnel.tunnel_connection(self._query_socket, cfg['registration'], sshserver, **ssh_kwargs)

else:

self._query_socket.connect(cfg['registration'])

self.session.debug = self.debug

self._notification_handlers = {'registration_notification' : self._register_engine,

'unregistration_notification' : self._unregister_engine,

'shutdown_notification' : lambda msg: self.close(),

}

self._queue_handlers = {'execute_reply' : self._handle_execute_reply,

'apply_reply' : self._handle_apply_reply}

self._connect(sshserver, ssh_kwargs, timeout)

# last step: setup magics, if we are in IPython:

try:

ip = get_ipython()

except NameError:

return

else:

if 'px' not in ip.magics_manager.magics:

# in IPython but we are the first Client.

# activate a default view for parallel magics.

self.activate()

def __del__(self):

"""cleanup sockets, but _not_ context."""

self.close()

def _setup_profile_dir(self, profile, profile_dir, ipython_dir):

if ipython_dir is None:

ipython_dir = get_ipython_dir()

if profile_dir is not None:

try:

self._cd = ProfileDir.find_profile_dir(profile_dir)

return

except ProfileDirError:

pass

elif profile is not None:

try:

self._cd = ProfileDir.find_profile_dir_by_name(

ipython_dir, profile)

return

except ProfileDirError:

pass

self._cd = None

def _update_engines(self, engines):

"""Update our engines dict and _ids from a dict of the form: {id:uuid}."""

for k,v in engines.iteritems():

eid = int(k)

if eid not in self._engines:

self._ids.append(eid)

self._engines[eid] = v

self._ids = sorted(self._ids)

if sorted(self._engines.keys()) != range(len(self._engines)) and \

self._task_scheme == 'pure' and self._task_socket:

self._stop_scheduling_tasks()

def _stop_scheduling_tasks(self):

"""Stop scheduling tasks because an engine has been unregistered

self._task_socket.close()

self._task_socket = None

msg = "An engine has been unregistered, and we are using pure " +\

"ZMQ task scheduling. Task farming will be disabled."

if self.outstanding:

msg += " If you were running tasks when this happened, " +\

"some `outstanding` msg_ids may never resolve."

warnings.warn(msg, RuntimeWarning)

def _build_targets(self, targets):

"""Turn valid target IDs or 'all' into two lists:

if not self._ids:

# flush notification socket if no engines yet, just in case

if not self.ids:

raise error.NoEnginesRegistered("Can't build targets without any engines")

if targets is None:

targets = self._ids

elif isinstance(targets, basestring):

if targets.lower() == 'all':

targets = self._ids

else:

raise TypeError("%r not valid str target, must be 'all'"%(targets))

elif isinstance(targets, int):

if targets < 0:

targets = self.ids[targets]

if targets not in self._ids:

raise IndexError("No such engine: %i"%targets)

targets = [targets]

if isinstance(targets, slice):

indices = range(len(self._ids))[targets]

ids = self.ids

targets = [ ids[i] for i in indices ]

if not isinstance(targets, (tuple, list, xrange)):

raise TypeError("targets by int/slice/collection of ints only, not %s"%(type(targets)))

return [cast_bytes(self._engines[t]) for t in targets], list(targets)

def _connect(self, sshserver, ssh_kwargs, timeout):

"""setup all our socket connections to the cluster. This is called from

# Maybe allow reconnecting?

if self._connected:

return

self._connected=True

def connect_socket(s, url):

# url = util.disambiguate_url(url, self._config['location'])

if self._ssh:

return tunnel.tunnel_connection(s, url, sshserver, **ssh_kwargs)

else:

return s.connect(url)

self.session.send(self._query_socket, 'connection_request')

# use Poller because zmq.select has wrong units in pyzmq 2.1.7

poller = zmq.Poller()

poller.register(self._query_socket, zmq.POLLIN)

# poll expects milliseconds, timeout is seconds

evts = poller.poll(timeout*1000)

if not evts:

raise error.TimeoutError("Hub connection request timed out")

idents,msg = self.session.recv(self._query_socket,mode=0)

if self.debug:

pprint(msg)

content = msg['content']

# self._config['registration'] = dict(content)

cfg = self._config

if content['status'] == 'ok':

self._mux_socket = self._context.socket(zmq.DEALER)

connect_socket(self._mux_socket, cfg['mux'])

self._task_socket = self._context.socket(zmq.DEALER)

connect_socket(self._task_socket, cfg['task'])

self._notification_socket = self._context.socket(zmq.SUB)

self._notification_socket.setsockopt(zmq.SUBSCRIBE, b'')

connect_socket(self._notification_socket, cfg['notification'])

self._control_socket = self._context.socket(zmq.DEALER)

connect_socket(self._control_socket, cfg['control'])

self._iopub_socket = self._context.socket(zmq.SUB)

self._iopub_socket.setsockopt(zmq.SUBSCRIBE, b'')

connect_socket(self._iopub_socket, cfg['iopub'])

self._update_engines(dict(content['engines']))

else:

self._connected = False

raise Exception("Failed to connect!")

#--------------------------------------------------------------------------

# handlers and callbacks for incoming messages

#--------------------------------------------------------------------------

def _unwrap_exception(self, content):

"""unwrap exception, and remap engine_id to int."""

e = error.unwrap_exception(content)

# print e.traceback

if e.engine_info:

e_uuid = e.engine_info['engine_uuid']

eid = self._engines[e_uuid]

e.engine_info['engine_id'] = eid

return e

def _extract_metadata(self, msg):

header = msg['header']

parent = msg['parent_header']

msg_meta = msg['metadata']

content = msg['content']

md = {'msg_id' : parent['msg_id'],

'received' : datetime.now(),

'engine_uuid' : msg_meta.get('engine', None),

'follow' : msg_meta.get('follow', []),

'after' : msg_meta.get('after', []),

'status' : content['status'],

}

if md['engine_uuid'] is not None:

md['engine_id'] = self._engines.get(md['engine_uuid'], None)

if 'date' in parent:

md['submitted'] = parent['date']

if 'started' in msg_meta:

md['started'] = msg_meta['started']

if 'date' in header:

md['completed'] = header['date']

return md

def _register_engine(self, msg):

"""Register a new engine, and update our connection info."""

content = msg['content']

eid = content['id']

d = {eid : content['uuid']}

self._update_engines(d)

def _unregister_engine(self, msg):

"""Unregister an engine that has died."""

content = msg['content']

eid = int(content['id'])

if eid in self._ids:

self._ids.remove(eid)

uuid = self._engines.pop(eid)

self._handle_stranded_msgs(eid, uuid)

if self._task_socket and self._task_scheme == 'pure':

self._stop_scheduling_tasks()

def _handle_stranded_msgs(self, eid, uuid):

"""Handle messages known to be on an engine when the engine unregisters.

outstanding = self._outstanding_dict[uuid]

for msg_id in list(outstanding):

if msg_id in self.results:

# we already

continue

try:

raise error.EngineError("Engine %r died while running task %r"%(eid, msg_id))

except:

content = error.wrap_exception()

# build a fake message:

msg = self.session.msg('apply_reply', content=content)

msg['parent_header']['msg_id'] = msg_id

msg['metadata']['engine'] = uuid

self._handle_apply_reply(msg)

def _handle_execute_reply(self, msg):

"""Save the reply to an execute_request into our results.

parent = msg['parent_header']

msg_id = parent['msg_id']

if msg_id not in self.outstanding:

if msg_id in self.history:

print ("got stale result: %s"%msg_id)

else:

print ("got unknown result: %s"%msg_id)

else:

self.outstanding.remove(msg_id)

content = msg['content']

header = msg['header']

# construct metadata:

md = self.metadata[msg_id]

md.update(self._extract_metadata(msg))

# is this redundant?

self.metadata[msg_id] = md

e_outstanding = self._outstanding_dict[md['engine_uuid']]

if msg_id in e_outstanding:

e_outstanding.remove(msg_id)

# construct result:

if content['status'] == 'ok':

self.results[msg_id] = ExecuteReply(msg_id, content, md)

elif content['status'] == 'aborted':

self.results[msg_id] = error.TaskAborted(msg_id)

elif content['status'] == 'resubmitted':

# TODO: handle resubmission

pass

else:

self.results[msg_id] = self._unwrap_exception(content)

def _handle_apply_reply(self, msg):

"""Save the reply to an apply_request into our results."""

parent = msg['parent_header']

msg_id = parent['msg_id']

if msg_id not in self.outstanding:

if msg_id in self.history:

print ("got stale result: %s"%msg_id)

print self.results[msg_id]

print msg

else:

print ("got unknown result: %s"%msg_id)

else:

self.outstanding.remove(msg_id)

content = msg['content']

header = msg['header']

# construct metadata:

md = self.metadata[msg_id]

md.update(self._extract_metadata(msg))

# is this redundant?

self.metadata[msg_id] = md

e_outstanding = self._outstanding_dict[md['engine_uuid']]

if msg_id in e_outstanding:

e_outstanding.remove(msg_id)

# construct result:

if content['status'] == 'ok':

self.results[msg_id] = serialize.unserialize_object(msg['buffers'])[0]

elif content['status'] == 'aborted':

self.results[msg_id] = error.TaskAborted(msg_id)

elif content['status'] == 'resubmitted':

# TODO: handle resubmission

pass

else:

self.results[msg_id] = self._unwrap_exception(content)

def _flush_notifications(self):

"""Flush notifications of engine registrations waiting

idents,msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK)

while msg is not None:

if self.debug:

pprint(msg)

msg_type = msg['header']['msg_type']

handler = self._notification_handlers.get(msg_type, None)

if handler is None:

raise Exception("Unhandled message type: %s"%msg.msg_type)

else:

handler(msg)

idents,msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK)

def _flush_results(self, sock):

"""Flush task or queue results waiting in ZMQ queue."""

idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)

while msg is not None:

if self.debug:

pprint(msg)

msg_type = msg['header']['msg_type']

handler = self._queue_handlers.get(msg_type, None)

if handler is None:

raise Exception("Unhandled message type: %s"%msg.msg_type)

else:

handler(msg)

idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)

def _flush_control(self, sock):

"""Flush replies from the control channel waiting

if self._ignored_control_replies <= 0:

return

idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)

while msg is not None:

self._ignored_control_replies -= 1

if self.debug:

pprint(msg)

idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)

def _flush_ignored_control(self):

"""flush ignored control replies"""

while self._ignored_control_replies > 0:

self.session.recv(self._control_socket)

self._ignored_control_replies -= 1

def _flush_ignored_hub_replies(self):

ident,msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK)

while msg is not None:

ident,msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK)

def _flush_iopub(self, sock):

"""Flush replies from the iopub channel waiting

idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)

while msg is not None:

if self.debug:

pprint(msg)

parent = msg['parent_header']

# ignore IOPub messages with no parent.

# Caused by print statements or warnings from before the first execution.

if not parent:

continue

msg_id = parent['msg_id']

content = msg['content']

header = msg['header']

msg_type = msg['header']['msg_type']

# init metadata:

md = self.metadata[msg_id]

if msg_type == 'stream':

name = content['name']

s = md[name] or ''

md[name] = s + content['data']

elif msg_type == 'pyerr':

md.update({'pyerr' : self._unwrap_exception(content)})

elif msg_type == 'pyin':

md.update({'pyin' : content['code']})

elif msg_type == 'display_data':

md['outputs'].append(content)

elif msg_type == 'pyout':

md['pyout'] = content

elif msg_type == 'data_message':

data, remainder = serialize.unserialize_object(msg['buffers'])

md['data'].update(data)

elif msg_type == 'status':

# idle message comes after all outputs

if content['execution_state'] == 'idle':

md['outputs_ready'] = True

else:

# unhandled msg_type (status, etc.)

pass

# reduntant?

self.metadata[msg_id] = md

idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)

#--------------------------------------------------------------------------

# len, getitem

#--------------------------------------------------------------------------

def __len__(self):

"""len(client) returns # of engines."""

return len(self.ids)

def __getitem__(self, key):

"""index access returns DirectView multiplexer objects

if not isinstance(key, (int, slice, tuple, list, xrange)):

raise TypeError("key by int/slice/iterable of ints only, not %s"%(type(key)))

else:

return self.direct_view(key)

#--------------------------------------------------------------------------

# Begin public methods

#--------------------------------------------------------------------------

@property

def ids(self):

"""Always up-to-date ids property."""

self._flush_notifications()

# always copy:

return list(self._ids)

def activate(self, targets='all', suffix=''):

"""Create a DirectView and register it with IPython magics

view = self.direct_view(targets)

view.block = True

view.activate(suffix)

return view

def close(self):

if self._closed:

return

self.stop_spin_thread()

snames = filter(lambda n: n.endswith('socket'), dir(self))

for socket in map(lambda name: getattr(self, name), snames):

if isinstance(socket, zmq.Socket) and not socket.closed:

socket.close()

self._closed = True

def _spin_every(self, interval=1):

"""target func for use in spin_thread"""

while True:

if self._stop_spinning.is_set():

return

time.sleep(interval)

self.spin()

def spin_thread(self, interval=1):

"""call Client.spin() in a background thread on some regular interval

if self._spin_thread is not None:

self.stop_spin_thread()

self._stop_spinning.clear()

self._spin_thread = Thread(target=self._spin_every, args=(interval,))

self._spin_thread.daemon = True

self._spin_thread.start()

def stop_spin_thread(self):

"""stop background spin_thread, if any"""

if self._spin_thread is not None:

self._stop_spinning.set()

self._spin_thread.join()

self._spin_thread = None

def spin(self):