def get_topologies(self, callback=None):
    """ get topologies """
    isWatching = False

    # Temp dict used to return result
    # if callback is not provided.
    ret = {
        "result": None
    }
    if callback:
      isWatching = True
    else:
      def callback(data):
        """Custom callback to get the topologies right now."""
        ret["result"] = data

    try:
      # Ensure the topology path exists. If a topology has never been deployed
      # then the path will not exist so create it and don't crash.
      # (fixme) add a watch instead of creating the path?
      self.client.ensure_path(self.get_topologies_path())

      self._get_topologies_with_watch(callback, isWatching)
    except NoNodeError:
      self.client.stop()
      path = self.get_topologies_path()
      raise_(StateException("Error required topology path '%s' not found" % (path),
                            StateException.EX_TYPE_NO_NODE_ERROR), sys.exc_info()[2])

    # The topologies are now populated with the data.
    return ret["result"]

        def wrapped(*args, **kw):
            try:
                return f(*args, **kw)
            except Exception as e:
                # Save exception since it can be clobbered during processing
                # below before we can re-raise
                exc_info = sys.exc_info()

                if notifier:
                    payload = dict(args=args, exception=e)
                    payload.update(kw)

                    # Use a temp vars so we don't shadow
                    # our outer definitions.
                    temp_level = level
                    if not temp_level:
                        temp_level = notifier.ERROR

                    temp_type = event_type
                    if not temp_type:
                        # If f has multiple decorators, they must use
                        # six.wraps to ensure the name is
                        # propagated.
                        temp_type = f.__name__

                    notifier.notify(publisher_id, temp_type, temp_level,
                                    payload)

                # re-raise original exception since it may have been clobbered
                raise_(exc_info[0], exc_info[1], exc_info[2])

    def __init__(self, parent, value, is_name=False, name=None):
        RingElement.__init__(self, parent)
        self._create = value
        if parent is None: return     # means "invalid element"
        # idea: Joe Wetherell -- try to find out if the output
        # is too long and if so get it using file, otherwise
        # don't.
        if isinstance(value, six.string_types) and parent._eval_using_file_cutoff and \
           parent._eval_using_file_cutoff < len(value):
            self._get_using_file = True

        if is_name:
            self._name = value
        else:
            try:
                self._name = parent._create(value, name=name)
            # Convert ValueError and RuntimeError to TypeError for
            # coercion to work properly.
            except (RuntimeError, ValueError) as x:
                self._session_number = -1
                raise_(TypeError, x, sys.exc_info()[2])
            except BaseException:
                self._session_number = -1
                raise
        self._session_number = parent._session_number

def _resolve_dependent_style(style_path):
    """Get the independent style of a dependent style.

    :param path style_path: Path to a dependent style.
    :returns: Name of the independent style of the passed dependent style.
    :raises: StyleDependencyError: If no style could be found/parsed.

    CSL Styles are split into two categories, Independent and Dependent.
    Independent styles, as their name says, are self-sustained and contain all
    the necessary information in order to format a citation. Dependent styles
    on the other hand, depend on Independent styles, and actually just pose as
    aliases for them. For example 'nature-digest' is a dependent style that
    just points to the 'nature' style.

    .. seealso::

        `CSL Specification
         <http://docs.citationstyles.org/en/stable/specification.html#file-types>`_
    """
    try:
        # The independent style is mentioned inside a link element of
        # the form 'http://www.stylesite.com/stylename'.
        for _, el in iterparse(style_path, tag='{%s}link' % xml_namespace):
            if el.attrib.get('rel') == 'independent-parent':
                url = el.attrib.get('href')
                return url.rsplit('/', 1)[1]
    except Exception:
        # Invalid XML, missing info, etc. Preserve the original exception.
        stacktrace = sys.exc_info()[2]
    else:
        stacktrace = None

    raise_(StyleDependencyError('Dependent style {0} could not be parsed'
                                .format(style_path)), None, stacktrace)

    def __call__(self):
        """Return a co-routine which runs the task group."""
        raised_exceptions = []
        while any(six.itervalues(self._runners)):
            try:
                for k, r in self._ready():
                    r.start()
                    if not r:
                        del self._graph[k]

                yield

                for k, r in self._running():
                    if r.step():
                        del self._graph[k]
            except Exception:
                exc_info = sys.exc_info()
                if self.aggregate_exceptions:
                    self._cancel_recursively(k, r)
                else:
                    self.cancel_all(grace_period=self.error_wait_time)
                raised_exceptions.append(exc_info)
            except:  # noqa
                with excutils.save_and_reraise_exception():
                    self.cancel_all()

        if raised_exceptions:
            if self.aggregate_exceptions:
                raise ExceptionGroup(v for t, v, tb in raised_exceptions)
            else:
                exc_type, exc_val, traceback = raised_exceptions[0]
                raise_(exc_type, exc_val, traceback)

	def sess(self):
		"""get session"""
		if self._sess:
			return self._sess

		# check if email server specified
		if not getattr(self, 'server'):
			err_msg = _('Email Account not setup. Please create a new Email Account from Setup > Email > Email Account')
			frappe.msgprint(err_msg)
			raise frappe.OutgoingEmailError(err_msg)

		try:
			if self.use_tls and not self.port:
				self.port = 587

			self._sess = smtplib.SMTP((self.server or "").encode('utf-8'),
				cint(self.port) or None)

			if not self._sess:
				err_msg = _('Could not connect to outgoing email server')
				frappe.msgprint(err_msg)
				raise frappe.OutgoingEmailError(err_msg)

			if self.use_tls:
				self._sess.ehlo()
				self._sess.starttls()
				self._sess.ehlo()

			if self.login and self.password:
				ret = self._sess.login((self.login or "").encode('utf-8'),
					(self.password or "").encode('utf-8'))

				# check if logged correctly
				if ret[0]!=235:
					frappe.msgprint(ret[1])
					raise frappe.OutgoingEmailError(ret[1])

			return self._sess

		except _socket.error as e:
			# Invalid mail server -- due to refusing connection
			frappe.msgprint(_('Invalid Outgoing Mail Server or Port'))
			traceback = sys.exc_info()[2]
			raise_(frappe.ValidationError, e, traceback)

		except smtplib.SMTPAuthenticationError as e:
			frappe.msgprint(_("Invalid login or password"))
			traceback = sys.exc_info()[2]
			raise_(frappe.ValidationError, e, traceback)

		except smtplib.SMTPException:
			frappe.msgprint(_('Unable to send emails at this time'))
			raise

  def create_execution_state(self, topologyName, executionState):
    """ create execution state """
    if not executionState or not executionState.IsInitialized():
      raise_(StateException("Execution State protobuf not init properly",
                            StateException.EX_TYPE_PROTOBUF_ERROR), sys.exc_info()[2])

    path = self.get_execution_state_path(topologyName)
    LOG.info("Adding topology: {0} to path: {1}".format(
        topologyName, path))
    executionStateString = executionState.SerializeToString()
    try:
      self.client.create(path, value=executionStateString, makepath=True)
      return True
    except NoNodeError:
      raise_(StateException("NoNodeError while creating execution state",
                            StateException.EX_TYPE_NO_NODE_ERROR), sys.exc_info()[2])
    except NodeExistsError:
      raise_(StateException("NodeExistsError while creating execution state",
                            StateException.EX_TYPE_NODE_EXISTS_ERROR), sys.exc_info()[2])
    except ZookeeperError:
      raise_(StateException("Zookeeper while creating execution state",
                            StateException.EX_TYPE_ZOOKEEPER_ERROR), sys.exc_info()[2])
    except Exception:
      # Just re raise the exception.
      raise

  def create_pplan(self, topologyName, pplan):
    """ create physical plan """
    if not pplan or not pplan.IsInitialized():
      raise_(StateException("Physical Plan protobuf not init properly",
                            StateException.EX_TYPE_PROTOBUF_ERROR), sys.exc_info()[2])

    path = self.get_pplan_path(topologyName)
    LOG.info("Adding topology: {0} to path: {1}".format(
        topologyName, path))
    pplanString = pplan.SerializeToString()
    try:
      self.client.create(path, value=pplanString, makepath=True)
      return True
    except NoNodeError:
      raise_(StateException("NoNodeError while creating pplan",
                            StateException.EX_TYPE_NO_NODE_ERROR), sys.exc_info()[2])
    except NodeExistsError:
      raise_(StateException("NodeExistsError while creating pplan",
                            StateException.EX_TYPE_NODE_EXISTS_ERROR), sys.exc_info()[2])
    except ZookeeperError:
      raise_(StateException("Zookeeper while creating pplan",
                            StateException.EX_TYPE_ZOOKEEPER_ERROR), sys.exc_info()[2])
    except Exception:
      # Just re raise the exception.
      raise

    def __init__(self, **kwargs):
        self.kwargs = kwargs

        try:
            self.message = self.msg_fmt % kwargs
        except KeyError:
            exc_info = sys.exc_info()
            # kwargs doesn't match a variable in the message
            # log the issue and the kwargs
            LOG.exception(_LE('Exception in string format operation'))
            for name, value in six.iteritems(kwargs):
                LOG.error("%s: %s" % (name, value))  # noqa

            if _FATAL_EXCEPTION_FORMAT_ERRORS:
                raise_(exc_info[0], exc_info[1], exc_info[2])

    def __call__(self):
        """Return a co-routine which runs the task group."""
        raised_exceptions = []
        thrown_exceptions = []

        while any(six.itervalues(self._runners)):
            try:
                for k, r in self._ready():
                    r.start()
                    if not r:
                        del self._graph[k]

                if self._graph:
                    try:
                        yield
                    except Exception:
                        thrown_exceptions.append(sys.exc_info())
                        raise

                for k, r in self._running():
                    if r.step():
                        del self._graph[k]
            except Exception:
                exc_info = sys.exc_info()
                if self.aggregate_exceptions:
                    self._cancel_recursively(k, r)
                else:
                    self.cancel_all(grace_period=self.error_wait_time)
                raised_exceptions.append(exc_info)
                del exc_info
            except:  # noqa
                with excutils.save_and_reraise_exception():
                    self.cancel_all()

        if raised_exceptions:
            try:
                if self.aggregate_exceptions:
                    raise ExceptionGroup(v for t, v, tb in raised_exceptions)
                else:
                    if thrown_exceptions:
                        raise_(*thrown_exceptions[-1])
                    else:
                        raise_(*raised_exceptions[0])
            finally:
                del raised_exceptions
                del thrown_exceptions

def import_string(import_name, silent=False):
    """Imports an object based on a string.  This is useful if you want to
    use import paths as endpoints or something similar.  An import path can
    be specified either in dotted notation (``xml.sax.saxutils.escape``)
    or with a colon as object delimiter (``xml.sax.saxutils:escape``).

    If `silent` is True the return value will be `None` if the import fails.

    For better debugging we recommend the new :func:`import_module`
    function to be used instead.

    :param import_name: the dotted name for the object to import.
    :param silent: if set to `True` import errors are ignored and
                   `None` is returned instead.
    :return: imported object

    :copyright: (c) 2011 by the Werkzeug Team
    """
    # force the import name to automatically convert to strings
    if isinstance(import_name, text_type):
        import_name = str(import_name)
    try:
        if ':' in import_name:
            module, obj = import_name.split(':', 1)
        elif '.' in import_name:
            module, obj = import_name.rsplit('.', 1)
        else:
            return __import__(import_name)
        # __import__ is not able to handle unicode strings in the fromlist
        # if the module is a package
        try:
            obj = obj.decode('utf-8')
        except:
            pass
        try:
            return getattr(__import__(module, None, None, [obj]), obj)
        except (ImportError, AttributeError):
            # support importing modules not yet set up by the parent module
            # (or package for that matter)
            modname = module + '.' + obj
            __import__(modname)
            return sys.modules[modname]
    except ImportError as e:
        if not silent:
            raise_(ImportStringError(import_name, e), None, sys.exc_info()[2])

    def __init__(self, **kwargs):
        self.kwargs = kwargs

        try:
            self.message = self.msg_fmt % kwargs

            if self.error_code:
                self.message = "HEAT-E%s %s" % (self.error_code, self.message)
        except KeyError:
            exc_info = sys.exc_info()
            # kwargs doesn't match a variable in the message
            # log the issue and the kwargs
            LOG.exception(_LE("Exception in string format operation"))
            for name, value in six.iteritems(kwargs):
                LOG.error(_LE("%(name)s: %(value)s"), {"name": name, "value": value})  # noqa

            if _FATAL_EXCEPTION_FORMAT_ERRORS:
                raise_(exc_info[0], exc_info[1], exc_info[2])

    def create_cursors(self, sess, distort, shuffle):
        # start the preloading threads.

        # tdata = self.read_and_decode(self.train_queue, self.conf)
        # vdata = self.read_and_decode(self.val_queue, self.conf)
        # self.train_data = tdata
        # self.val_data = vdata
        self.coord = tf.train.Coordinator()
        scale = self.scale

        train_threads = []
        val_threads = []

        if self.for_training == 0:
            # for training
            n_threads = 10
        elif self.for_training == 1:
            # for prediction
            n_threads = 0
        elif self.for_training == 2:
            # for cross validation
            n_threads = 1
        else:
            traceback = sys.exc_info()[2]
            raise_(ValueError, "Inocrrect value for for_training", traceback)

        for _ in range(n_threads):

            train_t = threading.Thread(target=self.read_image_thread,
                                 args=(sess, self.DBType.Train, distort, shuffle, scale))
            train_t.start()
            train_threads.append(train_t)

            val_t = threading.Thread(target=self.read_image_thread,
                                       args=(sess, self.DBType.Val, False, False, scale))
            val_t.start()
            val_threads.append(val_t)

        # self.threads = tf.train.start_queue_runners(sess=sess, coord=self.coord)
        # self.val_threads1 = self.val_qr.create_threads(sess, coord=self.coord, start=True)
        # self.train_threads1 = self.train_qr.create_threads(sess, coord=self.coord, start=True)
        self.train_threads = train_threads
        self.val_threads = val_threads

    def __init__(self, **kwargs):
        self.kwargs = kwargs

        try:
            if self.error_code in ERROR_CODE_MAP:
                self.msg_fmt = ERROR_CODE_MAP[self.error_code]

            self.message = self.msg_fmt % kwargs

            if self.error_code:
                self.message = 'KING-E%s %s' % (self.error_code, self.message)
        except KeyError:
            exc_info = sys.exc_info()
            # kwargs doesn't match a variable in the message
            # log the issue and the kwargs
            LOG.exception(_LE('Exception in string format operation'))
            for name, value in six.iteritems(kwargs):
                LOG.error(_LE("%(name)s: %(value)s"),
                          {'name': name, 'value': value})  # noqa

            if _FATAL_EXCEPTION_FORMAT_ERRORS:
                raise_(exc_info[0], exc_info[1], exc_info[2])

 def delete_execution_state(self, topologyName):
   """ delete execution state """
   path = self.get_execution_state_path(topologyName)
   LOG.info("Removing topology: {0} from path: {1}".format(
       topologyName, path))
   try:
     self.client.delete(path)
     return True
   except NoNodeError:
     raise_(StateException("NoNodeError while deleting execution state",
                           StateException.EX_TYPE_NO_NODE_ERROR), sys.exc_info()[2])
   except NotEmptyError:
     raise_(StateException("NotEmptyError while deleting execution state",
                           StateException.EX_TYPE_NOT_EMPTY_ERROR), sys.exc_info()[2])
   except ZookeeperError:
     raise_(StateException("Zookeeper while deleting execution state",
                           StateException.EX_TYPE_ZOOKEEPER_ERROR), sys.exc_info()[2])
   except Exception:
     # Just re raise the exception.
     raise

    def read_image_thread(self, sess, db_type, distort, shuffle, scale):
        # Thread that does the pre processing.

        if self.train_type == 0:
            if db_type == self.DBType.Val:
                filename = os.path.join(self.conf.cachedir, self.conf.valfilename) + '.tfrecords'
            elif db_type == self.DBType.Train:
                filename = os.path.join(self.conf.cachedir, self.conf.trainfilename) + '.tfrecords'
            else:
                traceback = sys.exc_info()[2]
                raise_(IOError, "Unspecified DB Type", traceback)

        else:
            filename = os.path.join(self.conf.cachedir, self.conf.trainfilename) + '.tfrecords'

        cur_db = multiResData.tf_reader(self.conf, filename, shuffle)
        placeholders = self.q_placeholders

        print('Starting preloading thread of type ... {}'.format(db_type))
        batch_np = {}
        while not self.coord.should_stop():
            batch_in = cur_db.next()
            batch_np['orig_images'] = batch_in[0]
            batch_np['orig_locs'] = batch_in[1]
            batch_np['info'] = batch_in[2]
            batch_np['extra_info'] = batch_in[3]
            xs, locs = PoseTools.preprocess_ims(batch_np['orig_images'], batch_np['orig_locs'], self.conf,
                                                distort, scale)

            batch_np['images'] = xs
            batch_np['locs'] = locs

            for fn in self.q_fns:
                fn(batch_np)

            food = {pl: batch_np[name] for (name, pl) in placeholders}

            success = False
            run_options = tf.RunOptions(timeout_in_ms=30000)
            try:
                while not success:

                    if sess._closed or self.coord.should_stop():
                        return

                    try:
                        if db_type == self.DBType.Val:
                            sess.run(self.val_enqueue_op, feed_dict=food,options=run_options)
                        elif db_type == self.DBType.Train:
                            sess.run(self.train_enqueue_op, feed_dict=food, options=run_options)
                        success = True

                    except tf.errors.DeadlineExceededError:
                        pass

            except (tf.errors.CancelledError,) as e:
                return
            except Exception as e:
                logging.exception('Error in preloading thread')
                self.close_cursors()
                sys.exit(1)
                return

#.........这里部分代码省略.........
            5

        Last, we demonstrate that by default the execution of a command
        is tried twice if it fails the first time due to a crashed
        interface::

            sage: singular.eval('quit;')
            ''
            sage: singular._eval_line_using_file('def a=3;', restart_if_needed=False)
            Traceback (most recent call last):
            ...
            RuntimeError: Singular terminated unexpectedly while reading in a large line...

        Since the test of the next method would fail, we re-start
        Singular now. ::

            sage: singular('2+3')
            Singular crashed -- automatically restarting.
            5

        """
        if allow_use_file and wait_for_prompt and self._eval_using_file_cutoff and len(line) > self._eval_using_file_cutoff:
            return self._eval_line_using_file(line)
        try:
            if self._expect is None:
                self._start()
            E = self._expect
            try:
                if len(line) >= 4096:
                    raise RuntimeError("Sending more than 4096 characters with %s on a line may cause a hang and you're sending %s characters"%(self, len(line)))
                E.sendline(line)
                if wait_for_prompt == False:
                    return ''

            except OSError as msg:
                if restart_if_needed:
                    # The subprocess most likely crashed.
                    # If it's really still alive, we fall through
                    # and raise RuntimeError.
                    if sys.platform.startswith('sunos'):
                        # On (Open)Solaris, we might need to wait a
                        # while because the process might not die
                        # immediately. See Trac #14371.
                        for t in [0.5, 1.0, 2.0]:
                            if E.isalive():
                                time.sleep(t)
                            else:
                                break
                    if not E.isalive():
                        try:
                            self._synchronize()
                        except (TypeError, RuntimeError):
                            pass
                        return self._eval_line(line,allow_use_file=allow_use_file, wait_for_prompt=wait_for_prompt, restart_if_needed=False)
                raise_(RuntimeError, "%s\nError evaluating %s in %s"%(msg, line, self), sys.exc_info()[2])

            if len(line)>0:
                try:
                    if isinstance(wait_for_prompt, six.string_types):
                        E.expect(wait_for_prompt)
                    else:
                        E.expect(self._prompt)
                except pexpect.EOF as msg:
                    try:
                        if self.is_local():
                            tmp_to_use = self._local_tmpfile()
                        else:
                            tmp_to_use = self._remote_tmpfile()
                        if self._read_in_file_command(tmp_to_use) in line:
                            raise pexpect.EOF(msg)
                    except NotImplementedError:
                        pass
                    if self._quit_string() in line:
                        # we expect to get an EOF if we're quitting.
                        return ''
                    elif restart_if_needed==True: # the subprocess might have crashed
                        try:
                            self._synchronize()
                            return self._eval_line(line,allow_use_file=allow_use_file, wait_for_prompt=wait_for_prompt, restart_if_needed=False)
                        except (TypeError, RuntimeError):
                            pass
                    raise RuntimeError("%s\n%s crashed executing %s"%(msg,self, line))
                if self._terminal_echo:
                    out = E.before
                else:
                    out = E.before.rstrip('\n\r')
            else:
                if self._terminal_echo:
                    out = '\n\r'
                else:
                    out = ''
        except KeyboardInterrupt:
            self._keyboard_interrupt()
            raise KeyboardInterrupt("Ctrl-c pressed while running %s"%self)
        if self._terminal_echo:
            i = out.find("\n")
            j = out.rfind("\r")
            return out[i+1:j].replace('\r\n','\n')
        else:
            return out.replace('\r\n','\n')