python类get_instructions()的实例源码

def test_doubly_nested(self):
        with captured_stdout():
            inner = outer()()
        actual = dis.get_instructions(inner, first_line=expected_inner_line)
        self.assertEqual(list(actual), expected_opinfo_inner)

def test_jumpy(self):
        actual = dis.get_instructions(jumpy, first_line=expected_jumpy_line)
        self.assertEqual(list(actual), expected_opinfo_jumpy)

# get_instructions has its own tests above, so can rely on it to validate
# the object oriented API

def test_iteration(self):
        for obj in [_f, _C(1).__init__, "a=1", _f.__code__]:
            with self.subTest(obj=obj):
                via_object = list(dis.Bytecode(obj))
                via_generator = list(dis.get_instructions(obj))
                self.assertEqual(via_object, via_generator)

def print_codeobj(self, codeobj):
        dis.dis(codeobj)
        self.print_codeobj_attr(codeobj)
        print('!'*80)
        for st in dis.get_instructions(codeobj):
            print(st.offset, st, sep=' -> ')
        print('!' * 80)
        for code in codeobj.co_code:
            print(opcode.opname[code])

def get_instructions(code):
        """
        Iterator parsing the bytecode into easy-usable minimal emulation of
        Python 3.4 `dis.Instruction` instances.
        """

        # shortcuts
        HAVE_ARGUMENT = dis.HAVE_ARGUMENT
        EXTENDED_ARG = dis.EXTENDED_ARG

        class Instruction:
            # Minimal emulation of Python 3.4 dis.Instruction
            def __init__(self, opcode, oparg):
                self.opname = dis.opname[opcode]
                self.arg = oparg
                # opcode, argval, argrepr, offset, is_jump_target and
                # starts_line are not used by our code, so we leave them away
                # here.

        code = code.co_code
        extended_arg = 0
        i = 0
        n = len(code)
        while i < n:
            c = code[i]
            i = i + 1
            op = _cOrd(c)
            if op >= HAVE_ARGUMENT:
                oparg = _cOrd(code[i]) + _cOrd(code[i + 1]) * 256 + extended_arg
                extended_arg = 0
                i += 2
                if op == EXTENDED_ARG:
                    extended_arg = oparg*65536
            else:
                oparg = None
            yield Instruction(op, oparg)


#FIXME: Leverage this rather than magic numbers below.

def _walk_global_ops(code):
        """
        Yield (opcode, argument number) tuples for all
        global-referencing instructions in *code*.
        """
        for instr in dis.get_instructions(code):
            op = instr.opcode
            if op in GLOBAL_OPS:
                yield op, instr.arg

def get_instructions(f):
    return update_break_instruction(dis.get_instructions(f))

def _is_safe_generator(code):
    '''
    Examine the code of an async generator to see if it appears
    unsafe with respect to async finalization.  A generator
    is unsafe if it utilizes any of the following constructs:

    1. Use of async-code in a finally block

       try:
           yield v
       finally:
           await coro()

    2. Use of yield inside an async context manager

       async with m:
           ...
           yield v
           ...

    3. Use of async-code in try-except

       try:
           yield v
       except Exception:
           await coro()
    '''
    def _is_unsafe_block(instr, end_offset=-1):
        is_generator = False
        in_final = False
        is_unsafe = False
        for op in instr:
            if op.offset == end_offset:
                in_final = True
            if op.opname == 'YIELD_VALUE':
                is_generator = True
            if op.opname == 'END_FINALLY':
                return (is_generator, is_unsafe)
            if op.opname in {'SETUP_FINALLY', 'SETUP_EXCEPT', 'SETUP_ASYNC_WITH'}:
                is_g, is_u = _is_unsafe_block(instr, op.argval)
                is_generator |= is_g
                is_unsafe |= is_u
            if op.opname == 'YIELD_FROM' and is_generator and in_final:
                is_unsafe = True
        return (is_generator, is_unsafe)

    return not _is_unsafe_block(dis.get_instructions(code))[1]

def wrap_code(self, codeobj, codeobj_id=0):
        codes = []
        constants = [
            self.wrap_code(item, self.get_codeobj_id())
            if isinstance(item, CodeType) else item
            for item in codeobj.co_consts
        ]

        update_offset = partial(self.calculate_offset, code=codeobj.co_code)
        for st in dis.get_instructions(codeobj):
            self.mark(codeobj_id, st)
            constants.append(
                lambda co_id=codeobj_id, opcode=st: self.visit(co_id, opcode)
            )
            codes.extend(self.make_trace(len(constants) - 1))
            codes.append(st.opcode)

            if st.opcode in opcode.hasjrel:
                current_position = update_offset(st.offset)
                taget_position = update_offset(st.argval) - self.TRACE_CODE_LEN - self.TRACE_CODE_LEN
                new_delta = taget_position - current_position
                codes.extend(self.make_args(new_delta - self.AGR_OP_LEN))
            elif st.opcode in opcode.hasjabs:
                codes.extend(self.make_args(
                    update_offset(st.arg) - self.TRACE_CODE_LEN - self.TRACE_CODE_LEN
                ))
            elif st.opcode >= opcode.HAVE_ARGUMENT:
                codes.extend(self.make_args(st.arg))

        new_code = CodeType(
            codeobj.co_argcount,
            codeobj.co_kwonlyargcount,
            codeobj.co_nlocals,
            codeobj.co_stacksize + self.AGR_OP_LEN,
            codeobj.co_flags,
            bytes(codes),  # codestring
            tuple(constants),  # constants
            codeobj.co_names,
            codeobj.co_varnames,
            codeobj.co_filename,
            codeobj.co_name,
            codeobj.co_firstlineno,
            codeobj.co_lnotab,
            codeobj.co_freevars,
            codeobj.co_cellvars,
        )
        return new_code