python类std()的实例源码

def __call__(self, string):
        # the special argument "-" means sys.std{in,out}
        if string == '-':
            if 'r' in self._mode:
                return sys.stdin
            elif 'w' in self._mode:
                return sys.stdout
            else:
                msg = 'argument "-" with mode %r' % self._mode
                raise ValueError(msg)

        # all other arguments are used as file names
        try:
            return open(string,
                        self._mode,
                        self._bufsize,
                        self._encoding,
                        self._errors,
                        newline=self._newline)
        except OSError as e:
            message = "can't open '%s': %s"
            raise ArgumentTypeError(message % (string, e))

def _encode(s):
    '''This is a helper for output of unicode. With Python2 it is necessary to
    do encoding to the LOCAL_ENCODING since by default unicode will be encoded
    to ascii. In python3 this conversion is not necessary for the user to
    to perform; in fact sys.std*.write, for example, requires unicode strings
    be passed in. This function will encode for python2 and do nothing
    for python3 (except assert that ``s`` is a unicode type).'''
    if compat.PY2:
        if isinstance(s, compat.unicode):
            try:
                return s.encode(LOCAL_ENCODING)
            except Exception as ex:
                log.error("Encoding error: " + str(ex))
                return s.encode(LOCAL_ENCODING, "replace")
        elif isinstance(s, str):
            return s
        else:
            raise TypeError("Argument must be str or unicode")
    else:
        assert(isinstance(s, str))
        return s

def patch_sys(stdin=True, stdout=True, stderr=True):
    """Patch sys.std[in,out,err] to use a cooperative IO via a threadpool.

    This is relatively dangerous and can have unintended consequences such as hanging
    the process or `misinterpreting control keys`_ when ``input`` and ``raw_input``
    are used.

    This method does nothing on Python 3. The Python 3 interpreter wants to flush
    the TextIOWrapper objects that make up stderr/stdout at shutdown time, but
    using a threadpool at that time leads to a hang.

    .. _`misinterpreting control keys`: https://github.com/gevent/gevent/issues/274
    """
    # test__issue6.py demonstrates the hang if these lines are removed;
    # strangely enough that test passes even without monkey-patching sys
    if PY3:
        return

    if stdin:
        _patch_sys_std('stdin')
    if stdout:
        _patch_sys_std('stdout')
    if stderr:
        _patch_sys_std('stderr')

def patch_sys(stdin=True, stdout=True, stderr=True):
    """Patch sys.std[in,out,err] to use a cooperative IO via a threadpool.

    This is relatively dangerous and can have unintended consequences such as hanging
    the process or `misinterpreting control keys`_ when ``input`` and ``raw_input``
    are used.

    This method does nothing on Python 3. The Python 3 interpreter wants to flush
    the TextIOWrapper objects that make up stderr/stdout at shutdown time, but
    using a threadpool at that time leads to a hang.

    .. _`misinterpreting control keys`: https://github.com/gevent/gevent/issues/274
    """
    # test__issue6.py demonstrates the hang if these lines are removed;
    # strangely enough that test passes even without monkey-patching sys
    if PY3:
        return

    if stdin:
        _patch_sys_std('stdin')
    if stdout:
        _patch_sys_std('stdout')
    if stderr:
        _patch_sys_std('stderr')

def get_stream_enc(stream, default=None):
    """Return the given stream's encoding or a default.

    There are cases where ``sys.std*`` might not actually be a stream, so
    check for the encoding attribute prior to returning it, and return
    a default if it doesn't exist or evaluates as False. ``default``
    is None if not provided.
    """
    if not hasattr(stream, 'encoding') or not stream.encoding:
        return default
    else:
        return stream.encoding

# Less conservative replacement for sys.getdefaultencoding, that will try
# to match the environment.
# Defined here as central function, so if we find better choices, we
# won't need to make changes all over IPython.

def get_stream_enc(stream, default=None):
    """Return the given stream's encoding or a default.

    There are cases where ``sys.std*`` might not actually be a stream, so
    check for the encoding attribute prior to returning it, and return
    a default if it doesn't exist or evaluates as False. ``default``
    is None if not provided.
    """
    if not hasattr(stream, 'encoding') or not stream.encoding:
        return default
    else:
        return stream.encoding

# Less conservative replacement for sys.getdefaultencoding, that will try
# to match the environment.
# Defined here as central function, so if we find better choices, we
# won't need to make changes all over IPython.

def test_patching_std_streams(self):
        """
        Test if std streams are correctly redirected to the web-console
        """
        self.stdin.send_keys('n')
        self.send_btn.click()
        time.sleep(1)
        stdout_tag = self.browser.find_element_by_id('stdout')
        self.assertIn('Enter something:', stdout_tag.text)
        self.stdin.send_keys('spam')
        self.send_btn.click()
        time.sleep(1)
        self.stdin.send_keys('n')
        self.send_btn.click()
        time.sleep(1)
        self.assertIn('You have entered: spam', stdout_tag.text)

def trap(func):
    """
    Replace sys.std(out|err) with a wrapper during execution, restored after.

    In addition, a new combined-streams output (another wrapper) will appear at
    ``sys.stdall``. This stream will resemble what a user sees at a terminal,
    i.e. both out/err streams intermingled.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        # Use another CarbonCopy even though we're not cc'ing; for our "write
        # bytes, return strings on py3" behavior. Meh.
        sys.stdall = CarbonCopy()
        my_stdout, sys.stdout = sys.stdout, CarbonCopy(cc=sys.stdall)
        my_stderr, sys.stderr = sys.stderr, CarbonCopy(cc=sys.stdall)
        try:
            return func(*args, **kwargs)
        finally:
            sys.stdout = my_stdout
            sys.stderr = my_stderr
            del sys.stdall
    return wrapper

def get_stream_enc(stream, default=None):
    """Return the given stream's encoding or a default.

    There are cases where ``sys.std*`` might not actually be a stream, so
    check for the encoding attribute prior to returning it, and return
    a default if it doesn't exist or evaluates as False. ``default``
    is None if not provided.
    """
    if not hasattr(stream, 'encoding') or not stream.encoding:
        return default
    else:
        return stream.encoding

# Less conservative replacement for sys.getdefaultencoding, that will try
# to match the environment.
# Defined here as central function, so if we find better choices, we
# won't need to make changes all over IPython.

def get_stream_enc(stream, default=None):
    """Return the given stream's encoding or a default.

    There are cases where ``sys.std*`` might not actually be a stream, so
    check for the encoding attribute prior to returning it, and return
    a default if it doesn't exist or evaluates as False. ``default``
    is None if not provided.
    """
    if not hasattr(stream, 'encoding') or not stream.encoding:
        return default
    else:
        return stream.encoding

# Less conservative replacement for sys.getdefaultencoding, that will try
# to match the environment.
# Defined here as central function, so if we find better choices, we
# won't need to make changes all over IPython.

def patch_sys(stdin=True, stdout=True, stderr=True):
    """Patch sys.std[in,out,err] to use a cooperative IO via a threadpool.

    This is relatively dangerous and can have unintended consequences such as hanging
    the process or `misinterpreting control keys`_ when ``input`` and ``raw_input``
    are used.

    This method does nothing on Python 3. The Python 3 interpreter wants to flush
    the TextIOWrapper objects that make up stderr/stdout at shutdown time, but
    using a threadpool at that time leads to a hang.

    .. _`misinterpreting control keys`: https://github.com/gevent/gevent/issues/274
    """
    # test__issue6.py demonstrates the hang if these lines are removed;
    # strangely enough that test passes even without monkey-patching sys
    if PY3:
        return

    if stdin:
        _patch_sys_std('stdin')
    if stdout:
        _patch_sys_std('stdout')
    if stderr:
        _patch_sys_std('stderr')

def patch_sys(stdin=True, stdout=True, stderr=True):
    """Patch sys.std[in,out,err] to use a cooperative IO via a threadpool.

    This is relatively dangerous and can have unintended consequences such as hanging
    the process or `misinterpreting control keys`_ when ``input`` and ``raw_input``
    are used.

    This method does nothing on Python 3. The Python 3 interpreter wants to flush
    the TextIOWrapper objects that make up stderr/stdout at shutdown time, but
    using a threadpool at that time leads to a hang.

    .. _`misinterpreting control keys`: https://github.com/gevent/gevent/issues/274
    """
    # test__issue6.py demonstrates the hang if these lines are removed;
    # strangely enough that test passes even without monkey-patching sys
    if PY3:
        return

    if stdin:
        _patch_sys_std('stdin')
    if stdout:
        _patch_sys_std('stdout')
    if stderr:
        _patch_sys_std('stderr')

def _encode(s):
    '''This is a helper for output of unicode. With Python2 it is necessary to
    do encoding to the LOCAL_ENCODING since by default unicode will be encoded
    to ascii. In python3 this conversion is not necessary for the user to
    to perform; in fact sys.std*.write, for example, requires unicode strings
    be passed in. This function will encode for python2 and do nothing
    for python3 (except assert that ``s`` is a unicode type).'''
    if PY2:
        if isinstance(s, unicode):
            try:
                return s.encode(LOCAL_ENCODING)
            except:
                log.exception()
                return s.encode(LOCAL_ENCODING, "replace")
        elif isinstance(s, str):
            return s
        else:
            raise TypeError("Argument must be str or unicode")
    else:
        assert(isinstance(s, str))
        return s

def get_stream_enc(stream, default=None):
    """Return the given stream's encoding or a default.

    There are cases where ``sys.std*`` might not actually be a stream, so
    check for the encoding attribute prior to returning it, and return
    a default if it doesn't exist or evaluates as False. ``default``
    is None if not provided.
    """
    if not hasattr(stream, 'encoding') or not stream.encoding:
        return default
    else:
        return stream.encoding

# Less conservative replacement for sys.getdefaultencoding, that will try
# to match the environment.
# Defined here as central function, so if we find better choices, we
# won't need to make changes all over IPython.

def get_stream_enc(stream, default=None):
    """Return the given stream's encoding or a default.

    There are cases where ``sys.std*`` might not actually be a stream, so
    check for the encoding attribute prior to returning it, and return
    a default if it doesn't exist or evaluates as False. ``default``
    is None if not provided.
    """
    if not hasattr(stream, 'encoding') or not stream.encoding:
        return default
    else:
        return stream.encoding

# Less conservative replacement for sys.getdefaultencoding, that will try
# to match the environment.
# Defined here as central function, so if we find better choices, we
# won't need to make changes all over IPython.

def __call__(self, string):
            # the special argument "-" means sys.std{in,out}
            if string == '-':
                if 'r' in self._mode:
                    return sys.stdin
                elif 'w' in self._mode:
                    return sys.stdout
                else:
                    msg = _('argument "-" with mode %r') % self._mode
                    raise ValueError(msg)

            # all other arguments are used as file names
            try:
                return open(string, self._mode, self._bufsize, self._encoding, self._errors)
            except OSError as e:
                message = _("can't open '%s': %s")
                raise ArgumentTypeError(message % (string, e))

def get_stream_enc(stream, default=None):
    """Return the given stream's encoding or a default.

    There are cases where ``sys.std*`` might not actually be a stream, so
    check for the encoding attribute prior to returning it, and return
    a default if it doesn't exist or evaluates as False. ``default``
    is None if not provided.
    """
    if not hasattr(stream, 'encoding') or not stream.encoding:
        return default
    else:
        return stream.encoding

# Less conservative replacement for sys.getdefaultencoding, that will try
# to match the environment.
# Defined here as central function, so if we find better choices, we
# won't need to make changes all over IPython.

def get_stream_enc(stream, default=None):
    """Return the given stream's encoding or a default.

    There are cases where ``sys.std*`` might not actually be a stream, so
    check for the encoding attribute prior to returning it, and return
    a default if it doesn't exist or evaluates as False. ``default``
    is None if not provided.
    """
    if not hasattr(stream, 'encoding') or not stream.encoding:
        return default
    else:
        return stream.encoding

# Less conservative replacement for sys.getdefaultencoding, that will try
# to match the environment.
# Defined here as central function, so if we find better choices, we
# won't need to make changes all over IPython.

def tearDown(self):
        # You can't import this gobally because DJANGO_SETTINGS_MODULE is not initialized yet for some
        # reason, but maybe by the time the code meanders its way to here it will work.
        from chroma_core.services.rpc import RpcClientFactory

        test_failed = (sys.exc_info() != (None, None, None))

        if test_failed:
            log.info(self._xmlrpc.system.listMethods())
            log_chunk = self._xmlrpc.supervisor.readLog(0, 4096)
            log.error("Supervisor log: %s" % log_chunk)

        # Shutdown any RPC Threads if they were started. Bit of horrible insider knowledge here.
        if RpcClientFactory._lightweight is False:
            RpcClientFactory.shutdown_threads()
            RpcClientFactory._lightweight = True
            RpcClientFactory._available = True
            RpcClientFactory._instances = {}

        if self._supervisor is not None:
            try:
                self._xmlrpc.supervisor.shutdown()
                stdout, stderr = self._supervisor.communicate()
                # Echo these at the end: by outputting using sys.std* rather than
                # letting the subprocess write directly, this verbose output can be
                # captured by nose and only output on failure.
                sys.stdout.write(stdout)
                sys.stdout.write(stderr)
            except:
                self._supervisor.kill()
            finally:
                self._supervisor = None

        if self._tmp_conf and os.path.exists(self._tmp_conf.name):
            os.unlink(self._tmp_conf.name)

def write(self, s):
        # Ensure we always write bytes. This means that wrapped code calling
        # print(<a string object>) in Python 3 will still work. Sigh.
        if isinstance(s, six.text_type):
            s = s.encode('utf-8')
        # Write out to our capturing object & any CC's
        IO.write(self, s)
        for writer in self.cc:
            writer.write(s)

    # Dumb hack to deal with py3 expectations; real sys.std(out|err) in Py3
    # requires writing to a buffer attribute obj in some situations.

def path_to_stop(vx, vy):
    #print >> sys.stderr, "Current VX: " + str(vx)
    #print >> sys.std   err, "Current VY: " + str(vy)

    vr = math.sqrt(abs(vx * vx) + abs(vy * vy)  + 2 * abs(vx) * abs(vy))
    #print >> sys.stderr, "Current VR: " + str(vr)

    path = 0
    while vr > MINIMUM_SPEED_FOR_BRAKE_PATH:
        path = path + vr
        vr = vr * SPEED_DECREASE
    return round(path)            
#no

def setUp(self):
        super(CaptureStandardIO, self).setUp()
        self.patcher = MonkeyPatcher()
        self.addCleanup(self.patcher.restore)
        # Patch sys.std* and self.std*. Use TextIOWrapper to provide an
        # identical API to the "real" stdin, stdout, and stderr objects.
        self._addStream("stdin", self._wrapStream(self._buf_in))
        self._addStream("stdout", self._wrapStream(self._buf_out))
        self._addStream("stderr", self._wrapStream(self._buf_err))
        self.patcher.patch()