Generator for getting the line numbers of a code object.

  Args:
    code_object: the code object.

  Yields:
    The next line number in the code object.
9927
def _GetLineNumbers(code_object):
  
  # Get the line number deltas, which are the odd number entries, from the
  # lnotab. See
  # https://svn.python.org/projects/python/branches/pep-0384/Objects/lnotab_notes.txt
  # In Python 3, this is just a byte array. In Python 2 it is a string so the
  # numerical values have to be extracted from the individual characters.
  if six.PY3:
    line_incrs = code_object.co_lnotab[1::2]
  else:
    line_incrs = (ord(c) for c in code_object.co_lnotab[1::2])
  current_line = code_object.co_firstlineno
  for line_incr in line_incrs:
    current_line += line_incr
    yield current_line
159905
def get_decl_pair(self):
        
        def get_tp():
            """ Iterator generating lines for struct definition. """
            decl = "struct "
            if self.tpname is not None:
                decl += self.tpname
                if self.inherit is not None:
                    decl += " : " + self.inherit
            yield decl
            yield "{"
            for f in self.fields:
                for f_line in f.generate():
                    yield "  " + f_line
            yield "} "
        return get_tp(), ""
774867
def extract_code(obj, compile_mode):
    
    try:
        code = obj.__code__
        if isinstance(code, CodeType):
            return code
        raise ValueError(
            "{obj} has a `__code__` attribute, "
            "but it's an instance of {notcode!r}, not CodeType.".format(
                obj=obj,
                notcode=type(code).__name__,
            )
        )
    except AttributeError:
        raise ValueError("Don't know how to extract code from %s." % obj)
1041094
def show(self):
        
        args, ret = self.method.get_descriptor()[1:].split(")")
        if self.code:
            # We patch the descriptor here and add the registers, if code is available
            args = args.split(" ")

            reg_len = self.code.get_registers_size()
            nb_args = len(args)

            start_reg = reg_len - nb_args
            args = ["{} v{}".format(a, start_reg + i) for i, a in enumerate(args)]

        print("METHOD {} {} {} ({}){}".format(
              self.method.get_class_name(),
              self.method.get_access_flags_string(),
              self.method.get_name(),
              ", ".join(args), ret))
        bytecode.PrettyShow(self, self.basic_blocks.gets(), self.method.notes)
573142
def pformat(self):
        '''
        
        '''
        result = "{\n"
        indent1 = " " * 4
        indent2 = " " * 8
        for line_no, code_objects in sorted(self.items()):
            result += '%s%i: [\n' % (indent1, line_no)
            for code_object in code_objects:
                result += '%s"""<%s:%s>""",\n' % (
                    indent2, code_object.PART_TYPE, code_object.content
                )
            result += '%s],\n' % indent1
        result += "}"

        return result
83613
def get(self, line_number):
        
        if line_number not in self._get_cache:
            self._get_cache[line_number] = self._get(line_number)
        return self._get_cache[line_number]
121338
def _number_of_line(member_tuple):
    """"""
    member = member_tuple[1]
    try:
        return member.__code__.co_firstlineno
    except AttributeError:
        pass
    try:
        return inspect.findsource(member)[1]
    except BaseException:
        pass
    for value in vars(member).values():
        try:
            return value.__code__.co_firstlineno
        except AttributeError:
            pass
    return 0
331428
def parse_py_statement(line):
    
    state = 0
    cur_token = ""
    spaces = " \t\n"
    ops = ".,;:+-*/%&!=|(){}[]^<>"
    i = 0

    def _escape_char(_c):
        if _c == "n":
            return "\n"
        elif _c == "t":
            return "\t"
        else:
            return _c

    while i < len(line):
        c = line[i]
        i += 1
        if state == 0:
            if c in spaces:
                pass
            elif c in ops:
                yield ("op", c)
            elif c == "#":
                state = 6
            elif c == "\"":
                state = 1
            elif c == "'":
                state = 2
            else:
                cur_token = c
                state = 3
        elif state == 1:  # string via "
            if c == "\\":
                state = 4
            elif c == "\"":
                yield ("str", cur_token)
                cur_token = ""
                state = 0
            else:
                cur_token += c
        elif state == 2:  # string via '
            if c == "\\":
                state = 5
            elif c == "'":
                yield ("str", cur_token)
                cur_token = ""
                state = 0
            else:
                cur_token += c
        elif state == 3:  # identifier
            if c in spaces + ops + "#\"'":
                yield ("id", cur_token)
                cur_token = ""
                state = 0
                i -= 1
            else:
                cur_token += c
        elif state == 4:  # escape in "
            cur_token += _escape_char(c)
            state = 1
        elif state == 5:  # escape in '
            cur_token += _escape_char(c)
            state = 2
        elif state == 6:  # comment
            cur_token += c
    if state == 3:
        yield ("id", cur_token)
    elif state == 6:
        yield ("comment", cur_token)
474668
def iter_code_cells(self):
        """"""
        for ws in self.nb.worksheets:
            for cell in ws.cells:
                if cell.cell_type == 'code':
                    yield cell
61891
def class_subobjects(
        class_: Type) -> Generator[Tuple[str, Type, bool], None, None]:
    
    argspec = inspect.getfullargspec(class_.__init__)
    defaults = argspec.defaults if argspec.defaults else []
    num_optional = len(defaults)
    first_optional = len(argspec.args) - num_optional

    for i, attr_name in enumerate(argspec.args):
        if attr_name == 'self':
            continue
        if attr_name == 'yatiml_extra':
            continue
        attr_type = argspec.annotations.get(attr_name, Any)
        yield attr_name, attr_type, i < first_optional
133820
def iter_code_by_lines(self):
        

        lnt = self.get_linenumbertable()
        if not lnt:
            yield (None, None, self.disassemble())
            return

        lnt_offset = lnt[0][1]

        cur_line = None
        current = None

        for codelet in self.disassemble():
            abs_line = self.get_line_for_offset(codelet[0])

            if cur_line == abs_line:
                current.append(codelet)

            else:
                if cur_line is not None:
                    yield (cur_line, cur_line - lnt_offset, current)
                cur_line = abs_line
                current = [codelet]

        if current:
            yield (cur_line, cur_line - lnt_offset, current)
1349762
def get_line_info(tree, max_lines=None):
    
    line_gen = indented_tree_line_generator(tree, max_lines=max_lines)
    line_gen_result = list(zip(*line_gen))
    if line_gen_result:
        return line_gen_result
    else:
        return [], [], []
1226548
def _next_code_point_handler(whence, ctx):
    """"""
    data_event, self = yield
    queue = ctx.queue
    unicode_escapes_allowed = ctx.ion_type is not IonType.CLOB
    escaped_newline = False
    escape_sequence = b''
    low_surrogate_required = False
    while True:
        if len(queue) == 0:
            yield ctx.read_data_event(self)
        queue_iter = iter(queue)
        code_point_generator = _next_code_point_iter(queue, queue_iter)
        code_point = next(code_point_generator)
        if code_point == _BACKSLASH:
            escape_sequence += six.int2byte(_BACKSLASH)
            num_digits = None
            while True:
                if len(queue) == 0:
                    yield ctx.read_data_event(self)
                code_point = next(queue_iter)
                if six.indexbytes(escape_sequence, -1) == _BACKSLASH:
                    if code_point == _ord(b'u') and unicode_escapes_allowed:
                        # 4-digit unicode escapes, plus '\u' for each surrogate
                        num_digits = 12 if low_surrogate_required else 6
                        low_surrogate_required = False
                    elif low_surrogate_required:
                        _illegal_character(code_point, ctx,
                                           'Unpaired high surrogate escape sequence %s.' % (escape_sequence,))
                    elif code_point == _ord(b'x'):
                        num_digits = 4  # 2-digit hex escapes
                    elif code_point == _ord(b'U') and unicode_escapes_allowed:
                        num_digits = 10  # 8-digit unicode escapes
                    elif code_point in _COMMON_ESCAPES:
                        if code_point == _SLASH or code_point == _QUESTION_MARK:
                            escape_sequence = b''  # Drop the \. Python does not recognize these as escapes.
                        escape_sequence += six.int2byte(code_point)
                        break
                    elif code_point in _NEWLINES:
                        escaped_newline = True
                        break
                    else:
                        # This is a backslash followed by an invalid escape character. This is illegal.
                        _illegal_character(code_point, ctx, 'Invalid escape sequence \\%s.' % (_chr(code_point),))
                    escape_sequence += six.int2byte(code_point)
                else:
                    if code_point not in _HEX_DIGITS:
                        _illegal_character(code_point, ctx,
                                           'Non-hex character %s found in unicode escape.' % (_chr(code_point),))
                    escape_sequence += six.int2byte(code_point)
                    if len(escape_sequence) == num_digits:
                        break
            if not escaped_newline:
                decoded_escape_sequence = escape_sequence.decode('unicode-escape')
                cp_iter = _next_code_point_iter(decoded_escape_sequence, iter(decoded_escape_sequence), to_int=ord)
                code_point = next(cp_iter)
                if code_point is None:
                    # This is a high surrogate. Restart the loop to gather the low surrogate.
                    low_surrogate_required = True
                    continue
                code_point = CodePoint(code_point)
                code_point.char = decoded_escape_sequence
                code_point.is_escaped = True
                ctx.set_code_point(code_point)
                yield Transition(None, whence)
        elif low_surrogate_required:
            _illegal_character(code_point, ctx, 'Unpaired high surrogate escape sequence %s.' % (escape_sequence,))
        if code_point == _CARRIAGE_RETURN:
            # Normalize all newlines (\r, \n, and \r\n) to \n .
            if len(queue) == 0:
                yield ctx.read_data_event(self)
            code_point = next(queue_iter)
            if code_point != _NEWLINE:
                queue.unread(code_point)
                code_point = _NEWLINE
        while code_point is None:
            yield ctx.read_data_event(self)
            code_point = next(code_point_generator)
        if escaped_newline:
            code_point = CodePoint(code_point)
            code_point.char = _ESCAPED_NEWLINE
            code_point.is_escaped = True
        ctx.set_code_point(code_point)
        yield Transition(None, whence)
730990
def get_both_code(self, code):
        

        return self.new(self.data.loc[(slice(None), code), :])
627434
def to_code(self, from_function=False):
        """"""

        num_fastnames = sum(1 for op, arg in self.code if isopcode(op) and op in haslocal)
        is_function = self.newlocals or num_fastnames > 0 or len(self.args) > 0
        nested = is_function and from_function

        co_flags = {op[0] for op in self.code}

        is_generator = self.force_generator or (YIELD_VALUE in co_flags or YIELD_FROM in co_flags)
        no_free = (not self.freevars) and (not co_flags & hasfree)

        if version_info >= (3, 5):
            is_native_coroutine = bool(self.force_coroutine or (co_flags & coroutine_opcodes))
            assert not (is_native_coroutine and self.force_iterable_coroutine)

        co_flags =\
            (not(STORE_NAME in co_flags or LOAD_NAME in co_flags or DELETE_NAME in co_flags)) |\
            (self.newlocals and CO_NEWLOCALS) |\
            (self.varargs and CO_VARARGS) |\
            (self.varkwargs and CO_VARKEYWORDS) |\
            (is_generator and CO_GENERATOR) |\
            (no_free and CO_NOFREE) |\
            (nested and CO_NESTED)

        if version_info >= (3, 5):
            co_flags |= (is_native_coroutine and CO_COROUTINE) |\
                        (self.force_iterable_coroutine and CO_ITERABLE_COROUTINE) |\
                        (self.future_generator_stop and CO_FUTURE_GENERATOR_STOP)

        co_consts = [self.docstring]
        co_names = []
        co_varnames = list(self.args)
        co_freevars = tuple(self.freevars)

        # Find all cellvars beforehand for two reasons
        # Need the number of them to construct the numeric arg for ops in hasfree
        # Need to put args which are cells in the beginning of co_cellvars
        cellvars = {arg for op, arg in self.code
                    if isopcode(op) and op in hasfree
                    and arg not in co_freevars}
        co_cellvars = [jumps for jumps in self.args if jumps in cellvars]

        def index(seq, item, eq=True, can_append=True):
            for i, x in enumerate(seq):
                if x == item if eq else x is item:
                    return i
            if can_append:
                seq.append(item)
                return len(seq) - 1
            else:
                raise IndexError("Item not found")

        jumps = []
        label_pos = {}
        lastlineno = self.firstlineno
        lastlinepos = 0
        co_code = bytearray()
        co_lnotab = bytearray()

        for i, (op, arg) in enumerate(self.code):
            if isinstance(op, Label):
                label_pos[op] = len(co_code)
            elif op is SetLineno:
                incr_lineno = arg - lastlineno
                incr_pos = len(co_code) - lastlinepos
                lastlineno = arg
                lastlinepos += incr_pos
                if incr_lineno != 0 or incr_pos != 0:
                    while incr_pos > 255:
                        co_lnotab += b"\xFF\0"
                        incr_pos -= 255
                    while incr_lineno > 255:
                        co_lnotab += (0xFF00|incr_pos).to_bytes(2, "little")
                        incr_pos = 0
                        incr_lineno -= 255
                    if incr_pos or incr_lineno:
                        co_lnotab += bytes((incr_pos, incr_lineno))
            elif op == opcode.EXTENDED_ARG:
                self.code[i + 1][1] |= 1 << 32
            elif op not in hasarg:
                co_code.append(op)
            else:
                if op in hasconst:
                    if isinstance(arg, Code) and\
                            i + 2 < len(self.code) and self.code[i + 2][0] in hascode:
                        arg = arg.to_code(from_function=is_function)
                        assert arg is not None
                    arg = index(co_consts, arg, 0)
                elif op in hasname:
                    arg = index(co_names, arg)
                elif op in hasjump:
                    jumps.append((len(co_code), arg))
                    co_code += op.to_bytes(3, "little")
                    continue
                elif op in haslocal:
                    arg = index(co_varnames, arg)
                elif op in hascompare:
                    arg = index(cmp_op, arg, can_append=False)
                elif op in hasfree:
                    try:
                        arg = index(co_freevars, arg, can_append=False) + len(cellvars)
                    except IndexError:
                        arg = index(co_cellvars, arg)
                if arg > 0xFFFF:
                    co_code += (opcode.EXTENDED_ARG | ((arg & 0xFFFF0000) >> 8)).to_bytes(3, "little")
                co_code += (op | (arg << 8)).to_bytes(3, "little")

        for pos, label in jumps:
            jump = label_pos[label]
            if co_code[pos] in hasjrel:
                jump -= pos + 3
            if jump > 0xFFFF:
                raise NotImplementedError("Extended jumps not implemented")
            co_code[pos + 1] = jump & 0xFF
            co_code[pos + 2] = jump >> 8

        co_argcount = len(self.args) - self.varargs - self.varkwargs - self.kwonly
        co_stacksize = self._compute_stacksize()

        return CodeType(co_argcount, self.kwonly, len(co_varnames), co_stacksize, co_flags,
                        bytes(co_code), tuple(co_consts), tuple(co_names), tuple(co_varnames),
                        self.filename, self.name, self.firstlineno, bytes(co_lnotab), co_freevars,
                        tuple(co_cellvars))
655891
def line_print(self):
        
        inv_types = {v: k for k, v in defines.Types.items()}

        if self._code is None:
            self._code = defines.Codes.EMPTY.number

        msg = "From {source}, To {destination}, {type}-{mid}, {code}-{token}, ["\
            .format(source=self._source, destination=self._destination, type=inv_types[self._type], mid=self._mid,
                    code=defines.Codes.LIST[self._code].name, token=self._token)
        for opt in self._options:
            msg += "{name}: {value}, ".format(name=opt.name, value=opt.value)
        msg += "]"
        if self.payload is not None:
            if isinstance(self.payload, dict):
                tmp = list(self.payload.values())[0][0:20]
            else:
                tmp = self.payload[0:20]
            msg += " {payload}...{length} bytes".format(payload=tmp, length=len(self.payload))
        else:
            msg += " No payload"
        return msg
76776
def get_code(self):
        
        code = UNKNOWN
        for result in self._results:
            if code == UNKNOWN or (result.code < UNKNOWN
                                   and result.code > code):
                code = result.code
        return code