<commit_before>import builtins
from collections import deque
from dis import findlinestarts, _get_instructions_bytes, Instruction, dis
import inspect
from types import FunctionType, CellType
from typing import Dict, Optional, List, Tuple, Union, Deque, Set

from hdlConvertorAst.to.hdlUtils import iter_with_last
from hwt.hdl.statements.assignmentContainer import HdlAssignmentContainer
from hwt.hdl.types.defs import BIT
from hwt.hdl.value import HValue
from hwt.pyUtils.arrayQuery import flatten
from hwt.synthesizer.interface import Interface
from hwt.synthesizer.rtlLevel.rtlSignal import RtlSignal
from hwtHls.errors import HlsSyntaxError
from hwtHls.hlsStreamProc.statements import HlsStreamProcWrite, \
    HlsStreamProcRead
from hwtHls.hlsStreamProc.streamProc import HlsStreamProc
from hwtHls.ssa.basicBlock import SsaBasicBlock
from hwtHls.ssa.translation.fromAst.astToSsa import AstToSsa
from hwtHls.ssa.translation.fromPython.blockLabel import generateBlockLabel
from hwtHls.ssa.translation.fromPython.blockPredecessorTracker import BlockLabel, \
    BlockPredecessorTracker
from hwtHls.ssa.translation.fromPython.bytecodeBlockAnalysis import extractBytecodeBlocks
from hwtHls.ssa.translation.fromPython.instructions import CMP_OPS, BIN_OPS, UN_OPS, \
    INPLACE_OPS, JUMP_OPS, ROT_OPS, BUILD_OPS
from hwtHls.ssa.translation.fromPython.loopsDetect import PyBytecodeLoop, \
    PreprocLoopScope
from hwtHls.ssa.value import SsaValue
import operator


class PythonBytecodeInPreproc():
    """
    A container of hw object marked that the immediate store is store of preproc variable only
    """

    def __init__(self, ref: Union[SsaValue, HValue, RtlSignal]):
        self.ref = ref
    
    def __iter__(self):
        for i in self.ref:
            yield PythonBytecodeInPreproc(i)


# assert sys.version_info >= (3, 10, 0), ("Python3.10 is minimum requirement", sys.version_info)
class PythonBytecodeFrame():

    def __init__(self, locals_: list, cellVarI: Dict[int, int], stack: list):
        self.locals = locals_
        self.stack = stack
        self.cellVarI = cellVarI
        self.preprocVars: Set[int] = set() 

    @classmethod
    def fromFunction(cls, fn: FunctionType, fnArgs: tuple, fnKwargs: dict):
        co = fn.__code__
        localVars = [None for _ in range(fn.__code__.co_nlocals)]
        if inspect.ismethod(fn):
            fnArgs = tuple((fn.__self__, *fnArgs))

        assert len(fnArgs) == co.co_argcount, ("Must have the correct number of arguments",
                                               len(fnArgs), co.co_argcount)
        for i, v in enumerate(fnArgs):
            localVars[i] = v
        if fnKwargs:
            raise NotImplementedError()

        varNameToI = {n: i for i, n in enumerate(fn.__code__.co_varnames)}
        cellVarI = {}
        for i, name in enumerate(fn.__code__.co_cellvars):
            cellVarI[i] = varNameToI[name]

        return PythonBytecodeFrame(localVars, cellVarI, [])


class PythonBytecodeToSsa():
    """
    This class translates Python bytecode to hwtHls.ssa
    
    The SSA basic blocks are constructed from jump targets in instruction list.
    A single jump target may generate multiple basic blocks if it is part of the cycle.
    
    Description of Python bytecode:
        * https://docs.python.org/3/library/dis.html
        * https://towardsdatascience.com/understanding-python-bytecode-e7edaae8734d

    Custom Python interprets:
        * https://github.com/pypyjs/pypy
        * https://github.com/kentdlee/CoCo

    :note: The block is generated for every jump target, even if it is just in preprocessor.
        If it is used just in preprocessor the jumps conditions of this block are resolved compile time.
        (Because for cycles we may find out that the cycle is HW evaluated somewhere inside of cycle body)
        And thus the blocks which were generated in preprocessor are entry points of clusters of blocks generated for HW
        connected in linear sequence or are entirely disconnected because preprocessor optimized them out. 
    """

    def __init__(self, hls: HlsStreamProc, fn: FunctionType):
        self.hls = hls
        self.fn = fn
        self.to_ssa: Optional[AstToSsa] = None
        self.blockToLabel: Dict[SsaBasicBlock, BlockLabel] = {}
        self.offsetToBlock: Dict[BlockLabel, SsaBasicBlock] = {}
        self.instructions: Tuple[Instruction] = ()
        self.bytecodeBlocks: Dict[int, List[Instruction]] = {}

    # https://www.synopsys.com/blogs/software-security/understanding-python-bytecode/
    def translateFunction(self, *fnArgs, **fnKwargs):
        """
        Translate bytecode of a Python function to :mod:`hwtHls.ssa`

        The input function may have features which should be evaluated during compile time
        and the rest should be translate to SSA for hardware compilation.
        We can not simply run preprocessor on a function because we know that instruction should be
        evaluated compile time only after we resolve its arguments.
        However in order to resolve all arguments we have to translate whole code.
        Because of this we must run preprocessor while translating the code and because of
        that a single basic block from Python can generate multiple basic blocks in SSA.
        And because of this the jump addresses may change.

        :ivar fn: function to translate
        :ivar fnArgs: positional arguments for function fn
        :ivar fnKwargs: keyword arguments for function fn
        """
        fn = self.fn
        # dis(fn)
        co = fn.__code__
        cell_names = co.co_cellvars + co.co_freevars
        linestarts = dict(findlinestarts(co))

        self.instructions = tuple(_get_instructions_bytes(
            co.co_code, co.co_varnames, co.co_names,
            co.co_consts, cell_names, linestarts))

        self.bytecodeBlocks, cfg = extractBytecodeBlocks(self.instructions)
        self.loops: Dict[int, PyBytecodeLoop] = {
            loop.entryPoint[0]: loop
            for loop in PyBytecodeLoop.detectLoops(cfg)
        }

        self.blockTracker = BlockPredecessorTracker(cfg)
        with open("tmp/cfg.dot", "w") as f:
            self.blockTracker.dumpCfgToDot(f)

        # print(self.loops)
        # print(cfg._adj)
        # print(self.loops)

        self.to_ssa = AstToSsa(self.hls.ssaCtx, fn.__name__, None)
        # self.to_ssa._onAllPredecsKnown(self.to_ssa.start)
        curBlock = self.to_ssa.start
        self.blockToLabel[curBlock] = (0,)
        self.offsetToBlock[(0,)] = curBlock
        for _ in self.blockTracker.addGenerated((0,)):
            pass  # should yield only entry block which should be already sealed

        frame = PythonBytecodeFrame.fromFunction(fn, fnArgs, fnKwargs)
        self._translateBytecodeBlock(self.bytecodeBlocks[0], frame, curBlock)
        if curBlock.predecessors:
            # because for LLVM entry point must not have predecessors
            entry = SsaBasicBlock(self.to_ssa.ssaCtx, "entry")
            entry.successors.addTarget(None, curBlock)
            self.to_ssa.start = entry
        # with open("tmp/cfg_final.dot", "w") as f:
        #    self.blockTracker.dumpCfgToDot(f)

        self.to_ssa.finalize()

        return self.to_ssa, None
<commit_msg>PythonBytecodeToSsa: rm debug cfg to dot jumps<commit_after>import builtins
from collections import deque
from dis import findlinestarts, _get_instructions_bytes, Instruction, dis
import inspect
from types import FunctionType, CellType
from typing import Dict, Optional, List, Tuple, Union, Deque, Set

from hdlConvertorAst.to.hdlUtils import iter_with_last
from hwt.hdl.statements.assignmentContainer import HdlAssignmentContainer
from hwt.hdl.types.defs import BIT
from hwt.hdl.value import HValue
from hwt.pyUtils.arrayQuery import flatten
from hwt.synthesizer.interface import Interface
from hwt.synthesizer.rtlLevel.rtlSignal import RtlSignal
from hwtHls.errors import HlsSyntaxError
from hwtHls.hlsStreamProc.statements import HlsStreamProcWrite, \
    HlsStreamProcRead
from hwtHls.hlsStreamProc.streamProc import HlsStreamProc
from hwtHls.ssa.basicBlock import SsaBasicBlock
from hwtHls.ssa.translation.fromAst.astToSsa import AstToSsa
from hwtHls.ssa.translation.fromPython.blockLabel import generateBlockLabel
from hwtHls.ssa.translation.fromPython.blockPredecessorTracker import BlockLabel, \
    BlockPredecessorTracker
from hwtHls.ssa.translation.fromPython.bytecodeBlockAnalysis import extractBytecodeBlocks
from hwtHls.ssa.translation.fromPython.instructions import CMP_OPS, BIN_OPS, UN_OPS, \
    INPLACE_OPS, JUMP_OPS, ROT_OPS, BUILD_OPS
from hwtHls.ssa.translation.fromPython.loopsDetect import PyBytecodeLoop, \
    PreprocLoopScope
from hwtHls.ssa.value import SsaValue
import operator


class PythonBytecodeInPreproc():
    """
    A container of hw object marked that the immediate store is store of preproc variable only
    """

    def __init__(self, ref: Union[SsaValue, HValue, RtlSignal]):
        self.ref = ref
    
    def __iter__(self):
        for i in self.ref:
            yield PythonBytecodeInPreproc(i)


# assert sys.version_info >= (3, 10, 0), ("Python3.10 is minimum requirement", sys.version_info)
class PythonBytecodeFrame():

    def __init__(self, locals_: list, cellVarI: Dict[int, int], stack: list):
        self.locals = locals_
        self.stack = stack
        self.cellVarI = cellVarI
        self.preprocVars: Set[int] = set() 

    @classmethod
    def fromFunction(cls, fn: FunctionType, fnArgs: tuple, fnKwargs: dict):
        co = fn.__code__
        localVars = [None for _ in range(fn.__code__.co_nlocals)]
        if inspect.ismethod(fn):
            fnArgs = tuple((fn.__self__, *fnArgs))

        assert len(fnArgs) == co.co_argcount, ("Must have the correct number of arguments",
                                               len(fnArgs), co.co_argcount)
        for i, v in enumerate(fnArgs):
            localVars[i] = v
        if fnKwargs:
            raise NotImplementedError()

        varNameToI = {n: i for i, n in enumerate(fn.__code__.co_varnames)}
        cellVarI = {}
        for i, name in enumerate(fn.__code__.co_cellvars):
            cellVarI[i] = varNameToI[name]

        return PythonBytecodeFrame(localVars, cellVarI, [])


class PythonBytecodeToSsa():
    """
    This class translates Python bytecode to hwtHls.ssa
    
    The SSA basic blocks are constructed from jump targets in instruction list.
    A single jump target may generate multiple basic blocks if it is part of the cycle.
    
    Description of Python bytecode:
        * https://docs.python.org/3/library/dis.html
        * https://towardsdatascience.com/understanding-python-bytecode-e7edaae8734d

    Custom Python interprets:
        * https://github.com/pypyjs/pypy
        * https://github.com/kentdlee/CoCo

    :note: The block is generated for every jump target, even if it is just in preprocessor.
        If it is used just in preprocessor the jumps conditions of this block are resolved compile time.
        (Because for cycles we may find out that the cycle is HW evaluated somewhere inside of cycle body)
        And thus the blocks which were generated in preprocessor are entry points of clusters of blocks generated for HW
        connected in linear sequence or are entirely disconnected because preprocessor optimized them out. 
    """

    def __init__(self, hls: HlsStreamProc, fn: FunctionType):
        self.hls = hls
        self.fn = fn
        self.to_ssa: Optional[AstToSsa] = None
        self.blockToLabel: Dict[SsaBasicBlock, BlockLabel] = {}
        self.offsetToBlock: Dict[BlockLabel, SsaBasicBlock] = {}
        self.instructions: Tuple[Instruction] = ()
        self.bytecodeBlocks: Dict[int, List[Instruction]] = {}

    # https://www.synopsys.com/blogs/software-security/understanding-python-bytecode/
    def translateFunction(self, *fnArgs, **fnKwargs):
        """
        Translate bytecode of a Python function to :mod:`hwtHls.ssa`

        The input function may have features which should be evaluated during compile time
        and the rest should be translate to SSA for hardware compilation.
        We can not simply run preprocessor on a function because we know that instruction should be
        evaluated compile time only after we resolve its arguments.
        However in order to resolve all arguments we have to translate whole code.
        Because of this we must run preprocessor while translating the code and because of
        that a single basic block from Python can generate multiple basic blocks in SSA.
        And because of this the jump addresses may change.

        :ivar fn: function to translate
        :ivar fnArgs: positional arguments for function fn
        :ivar fnKwargs: keyword arguments for function fn
        """
        fn = self.fn
        # dis(fn)
        co = fn.__code__
        cell_names = co.co_cellvars + co.co_freevars
        linestarts = dict(findlinestarts(co))

        self.instructions = tuple(_get_instructions_bytes(
            co.co_code, co.co_varnames, co.co_names,
            co.co_consts, cell_names, linestarts))

        self.bytecodeBlocks, cfg = extractBytecodeBlocks(self.instructions)
        self.loops: Dict[int, PyBytecodeLoop] = {
            loop.entryPoint[0]: loop
            for loop in PyBytecodeLoop.detectLoops(cfg)
        }

        self.blockTracker = BlockPredecessorTracker(cfg)
        #with open("tmp/cfg.dot", "w") as f:
        #    self.blockTracker.dumpCfgToDot(f)

        # print(self.loops)
        # print(cfg._adj)
        # print(self.loops)

        self.to_ssa = AstToSsa(self.hls.ssaCtx, fn.__name__, None)
        # self.to_ssa._onAllPredecsKnown(self.to_ssa.start)
        curBlock = self.to_ssa.start
        self.blockToLabel[curBlock] = (0,)
        self.offsetToBlock[(0,)] = curBlock
        for _ in self.blockTracker.addGenerated((0,)):
            pass  # should yield only entry block which should be already sealed

        frame = PythonBytecodeFrame.fromFunction(fn, fnArgs, fnKwargs)
        self._translateBytecodeBlock(self.bytecodeBlocks[0], frame, curBlock)
        if curBlock.predecessors:
            # because for LLVM entry point must not have predecessors
            entry = SsaBasicBlock(self.to_ssa.ssaCtx, "entry")
            entry.successors.addTarget(None, curBlock)
            self.to_ssa.start = entry
        # with open("tmp/cfg_final.dot", "w") as f:
        #    self.blockTracker.dumpCfgToDot(f)

        self.to_ssa.finalize()

        return self.to_ssa, None

{'fraction_non_alphanumeric': 0.07669111095881022, 'fraction_numerical': 0.002467274347200329, 'mean_word_length': 2.9923392612859097, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 10, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5823314', 'n_tokens_mistral': 4376, 'n_tokens_neox': 4235, 'n_words': 1260}