<commit_before>#  This file is part of the myhdl library, a Python package for using
#  Python as a Hardware Description Language.
#
#  Copyright (C) 2003-2008 <NAME>
#
#  The myhdl library is free software; you can redistribute it and/or
#  modify it under the terms of the GNU Lesser General Public License as
#  published by the Free Software Foundation; either version 2.1 of the
#  License, or (at your option) any later version.
#
#  This library is distributed in the hope that it will be useful, but
#  WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
#  Lesser General Public License for more details.

#  You should have received a copy of the GNU Lesser General Public
#  License along with this library; if not, write to the Free Software
#  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

""" Module that provides the Simulation class """
from __future__ import absolute_import
from __future__ import print_function


import os
from operator import itemgetter
from warnings import warn
from types import GeneratorType

from myhdl import Cosimulation, StopSimulation, _SuspendSimulation
from myhdl import _simulator, SimulationError
from myhdl._simulator import _signals, _siglist, _futureEvents
from myhdl._Waiter import _Waiter
from myhdl._Waiter import _inferWaiter
from myhdl._Waiter import _SignalTupleWaiter
from myhdl._util import _printExcInfo
from myhdl._instance import _Instantiator
from myhdl._block import _Block

schedule = _futureEvents.append


class _error:
    pass
_error.ArgType = "Inappriopriate argument type"
_error.MultipleCosim = "Only a single cosimulator argument allowed"
_error.DuplicatedArg = "Duplicated argument"

# flatten Block objects out


def _flatten(*args):
    arglist = []
    for arg in args:
        if isinstance(arg, _Block):
            arg = arg.subs
        if isinstance(arg, (list, tuple, set)):
            for item in arg:
                arglist.extend(_flatten(item))
        else:
            arglist.append(arg)
    return arglist

_error.MultipleSim = "Only a single Simulation instance is allowed"


class Simulation(object):

    """ Simulation class.

    Methods:
    run -- run a simulation for some duration

    """
    _no_of_instances = 0

    def __init__(self, *args):
        """ Construct a simulation object.

        *args -- list of arguments. Each argument is a generator or
                 a nested sequence of generators.

        """
        _simulator._time = 0
        arglist = _flatten(*args)
        self._waiters, self._cosim = _makeWaiters(arglist)
        if Simulation._no_of_instances > 0:
            raise SimulationError(_error.MultipleSim)
        Simulation._no_of_instances += 1
        if not self._cosim and _simulator._cosim:
            warn("Cosimulation not registered as Simulation argument")
        self._finished = False
        del _futureEvents[:]
        del _siglist[:]

    def _finalize(self):
        cosim = self._cosim
        if cosim:
            _simulator._cosim = 0
            os.close(cosim._rt)
            os.close(cosim._wf)
            cosim._child.wait()
        if _simulator._tracing:
            _simulator._tracing = 0
            _simulator._tf.close()
        # clean up for potential new run with same signals
        for s in _signals:
            s._clear()
        Simulation._no_of_instances = 0
        self._finished = True

    def quit(self):
        self._finalize()

    def run(self, duration=None, quiet=0):
        """ Run the simulation for some duration.

        duration -- specified simulation duration (default: forever)
        quiet -- don't print StopSimulation messages (default: off)

        """

        # If the simulation is already finished, raise StopSimulation immediately
        # From this point it will propagate to the caller, that can catch it.
        if self._finished:
            raise StopSimulation("Simulation has already finished")
        waiters = self._waiters
        maxTime = None
        if duration:
            stop = _Waiter(None)
            stop.hasRun = 1
            maxTime = _simulator._time + duration
            schedule((maxTime, stop))
        cosim = self._cosim
        t = _simulator._time
        actives = {}
        tracing = _simulator._tracing
        tracefile = _simulator._tf
        exc = []
        _pop = waiters.pop
        _append = waiters.append
        _extend = waiters.extend

        while 1:
            try:

                for s in _siglist:
                    _extend(s._update())
                del _siglist[:]

                while waiters:
                    waiter = _pop()
                    try:
                        waiter.next(waiters, actives, exc)
                    except StopIteration:
                        continue

                if cosim:
                    cosim._get()
                    if _siglist or cosim._hasChange:
                        cosim._put(t)
                        continue
                elif _siglist:
                    continue

                if actives:
                    for wl in actives.values():
                        wl.purge()
                    actives = {}

                # at this point it is safe to potentially suspend a simulation
                if exc:
                    raise exc[0]

                # future events<commit_msg>Add a check for a maximum number of iterations when simulating<commit_after>#  This file is part of the myhdl library, a Python package for using
#  Python as a Hardware Description Language.
#
#  Copyright (C) 2003-2008 <NAME>
#
#  The myhdl library is free software; you can redistribute it and/or
#  modify it under the terms of the GNU Lesser General Public License as
#  published by the Free Software Foundation; either version 2.1 of the
#  License, or (at your option) any later version.
#
#  This library is distributed in the hope that it will be useful, but
#  WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
#  Lesser General Public License for more details.

#  You should have received a copy of the GNU Lesser General Public
#  License along with this library; if not, write to the Free Software
#  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

""" Module that provides the Simulation class """
from __future__ import absolute_import
from __future__ import print_function


import os
from operator import itemgetter
from warnings import warn
from types import GeneratorType

from myhdl import Cosimulation, StopSimulation, _SuspendSimulation
from myhdl import _simulator, SimulationError
from myhdl._simulator import _signals, _siglist, _futureEvents
from myhdl._Waiter import _Waiter
from myhdl._Waiter import _inferWaiter
from myhdl._Waiter import _SignalTupleWaiter
from myhdl._util import _printExcInfo
from myhdl._instance import _Instantiator
from myhdl._block import _Block

schedule = _futureEvents.append


class _error:
    pass
_error.ArgType = "Inappriopriate argument type"
_error.MultipleCosim = "Only a single cosimulator argument allowed"
_error.DuplicatedArg = "Duplicated argument"

# flatten Block objects out


def _flatten(*args):
    arglist = []
    for arg in args:
        if isinstance(arg, _Block):
            arg = arg.subs
        if isinstance(arg, (list, tuple, set)):
            for item in arg:
                arglist.extend(_flatten(item))
        else:
            arglist.append(arg)
    return arglist

_error.MultipleSim = "Only a single Simulation instance is allowed"


class Simulation(object):

    """ Simulation class.

    Methods:
    run -- run a simulation for some duration

    """
    _no_of_instances = 0

    def __init__(self, *args):
        """ Construct a simulation object.

        *args -- list of arguments. Each argument is a generator or
                 a nested sequence of generators.

        """
        _simulator._time = 0
        arglist = _flatten(*args)
        self._waiters, self._cosim = _makeWaiters(arglist)
        if Simulation._no_of_instances > 0:
            raise SimulationError(_error.MultipleSim)
        Simulation._no_of_instances += 1
        if not self._cosim and _simulator._cosim:
            warn("Cosimulation not registered as Simulation argument")
        self._finished = False
        del _futureEvents[:]
        del _siglist[:]

    def _finalize(self):
        cosim = self._cosim
        if cosim:
            _simulator._cosim = 0
            os.close(cosim._rt)
            os.close(cosim._wf)
            cosim._child.wait()
        if _simulator._tracing:
            _simulator._tracing = 0
            _simulator._tf.close()
        # clean up for potential new run with same signals
        for s in _signals:
            s._clear()
        Simulation._no_of_instances = 0
        self._finished = True

    def quit(self):
        self._finalize()

    def run(self, duration=None, quiet=0, iter_limit=1000):
        """ Run the simulation for some duration.

        duration -- specified simulation duration (default: forever)
        quiet -- don't print StopSimulation messages (default: off)
        iter_limit -- step limit for not advancing time; used to detect combinatorial loops (default: 1000)

        """

        # If the simulation is already finished, raise StopSimulation immediately
        # From this point it will propagate to the caller, that can catch it.
        if self._finished:
            raise StopSimulation("Simulation has already finished")
        waiters = self._waiters
        maxTime = None
        if duration:
            stop = _Waiter(None)
            stop.hasRun = 1
            maxTime = _simulator._time + duration
            schedule((maxTime, stop))
        cosim = self._cosim
        last_t = t = _simulator._time
        same_t_count = 0
        actives = {}
        tracing = _simulator._tracing
        tracefile = _simulator._tf
        exc = []
        _pop = waiters.pop
        _append = waiters.append
        _extend = waiters.extend

        while 1:
            try:

                if last_t == t:
                    same_t_count += 1
                    if same_t_count > iter_limit:
                        raise StopSimulation("Iteration limit exceeded")
                else:
                    last_t = t
                    # Needed to prevent off-by-one
                    same_t_count = 1

                for s in _siglist:
                    _extend(s._update())
                del _siglist[:]

                while waiters:
                    waiter = _pop()
                    try:
                        waiter.next(waiters, actives, exc)
                    except StopIteration:
                        continue

                if cosim:
                    cosim._get()
                    if _siglist or cosim._hasChange:
                        cosim._put(t)
                        continue
                elif _siglist:
                    continue

                if actives:
                    for wl in actives.values():
                        wl.purge()
                    actives = {}

                # at this point it is safe to potentially suspend a simulation
                if exc:
                    raise exc[0]

                # future events
{'fraction_non_alphanumeric': 0.06333537171311261, 'fraction_numerical': 0.007076089805189132, 'mean_word_length': 1.9489953632148378, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6303360', 'n_tokens_mistral': 3135, 'n_tokens_neox': 2950, 'n_words': 1072}