files.new_contents: # Copyright (c) 2013 Potential Ventures Ltd
  # Copyright (c) 2013 SolarFlare Communications Inc
  # All rights reserved.
  #
  # Redistribution and use in source and binary forms, with or without
  # modification, are permitted provided that the following conditions are met:
  #     * Redistributions of source code must retain the above copyright
  #       notice, this list of conditions and the following disclaimer.
  #     * Redistributions in binary form must reproduce the above copyright
  #       notice, this list of conditions and the following disclaimer in the
  #       documentation and/or other materials provided with the distribution.
  #     * Neither the name of Potential Ventures Ltd,
  #       SolarFlare Communications Inc nor the
  #       names of its contributors may be used to endorse or promote products
  #       derived from this software without specific prior written permission.
  #
  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
  # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  # DISCLAIMED. IN NO EVENT SHALL POTENTIAL VENTURES LTD BE LIABLE FOR ANY
  # DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  
  import sys
  import time
  import logging
  import functools
  import inspect
  import os
  
  import cocotb
  from cocotb.log import SimLog
  from cocotb.result import ReturnValue
  from cocotb.utils import get_sim_time, lazy_property, remove_traceback_frames, extract_coro_stack
  from cocotb import outcomes
  
  # Sadly the Python standard logging module is very slow so it's better not to
  # make any calls by testing a boolean flag first
  if "COCOTB_SCHEDULER_DEBUG" in os.environ:
      _debug = True
  else:
      _debug = False
  
  
  def public(f):
      """Use a decorator to avoid retyping function/class names.
  
      * Based on an idea by Duncan Booth:
      http://groups.google.com/group/comp.lang.python/msg/11cbb03e09611b8a
      * Improved via a suggestion by Dave Angel:
      http://groups.google.com/group/comp.lang.python/msg/3d400fb22d8a42e1
      """
      all = sys.modules[f.__module__].__dict__.setdefault('__all__', [])
      if f.__name__ not in all:  # Prevent duplicates if run from an IDE.
          all.append(f.__name__)
      return f
  
  
  public(public)  # Emulate decorating ourself
  
  
  @public
  class CoroutineComplete(Exception):
      """To ensure that a coroutine has completed before we fire any triggers
      that are blocked waiting for the coroutine to end, we create a sub-class
      exception that the scheduler catches and the callbacks are attached
      here.
      """
  
      def __init__(self, text=""):
          Exception.__init__(self, text)
  
  
  class RunningTask:
      """Per instance wrapper around a running generator.
  
      Provides the following:
  
          task.join() creates a Trigger that will fire when this coroutine
          completes.
  
          task.kill() will destroy a coroutine instance (and cause any Join
          triggers to fire.
      """
  
      _id_count = 0  # used by the scheduler for debug
  
      def __init__(self, inst):
  
          if inspect.iscoroutine(inst):
              self._natively_awaitable = True
          elif inspect.isgenerator(inst):
              self._natively_awaitable = False
          elif inspect.iscoroutinefunction(inst):
              raise TypeError(
                  "Coroutine function {} should be called prior to being "
                  "scheduled."
                  .format(inst))
          elif sys.version_info >= (3, 6) and inspect.isasyncgen(inst):
              raise TypeError(
                  "{} is an async generator, not a coroutine. "
                  "You likely used the yield keyword instead of await.".format(
                      inst.__qualname__))
          else:
              raise TypeError(
                  "%s isn't a valid coroutine! Did you forget to use the yield keyword?" % inst)
          self._coro = inst
          self._started = False
          self._callbacks = []
          self._outcome = None
          self._trigger = None
  
          self._task_id = self._id_count
          RunningTask._id_count += 1
          self.__name__ = "Task %d" % self._task_id
          self.__qualname__ = self.__name__
  
      @lazy_property
      def log(self):
          # Creating a logger is expensive, only do it if we actually plan to
          # log anything
          return SimLog("cocotb.coroutine.%s" % self.__qualname__, id(self))
  
      @property
      def retval(self):
          if self._outcome is None:
              raise RuntimeError("coroutine is not complete")
          return self._outcome.get()
  
      @property
      def _finished(self):
          return self._outcome is not None
  
      def __iter__(self):
          return self
  
      def __str__(self):
          return "<{}>".format(self.__name__)
  
      def _get_coro_stack(self):
          """Get the coroutine callstack of this Task."""
          coro_stack = extract_coro_stack(self._coro)
  
          # Remove Trigger.__await__() from the stack, as it's not really useful
          if self._natively_awaitable and len(coro_stack):
              if coro_stack[-1].name == '__await__':
                  coro_stack.pop()
  
          return coro_stack
  
      def __repr__(self):
          coro_stack = self._get_coro_stack()
  
          if cocotb.scheduler._current_task is self:
              fmt = "<{name} running coro={coro}()>"
          elif self._finished:
              fmt = "<{name} finished coro={coro}() outcome={outcome}>"
          elif self._trigger is not None:
              fmt = "<{name} pending coro={coro}() trigger={trigger}>"
          elif not self._started:
              fmt = "<{name} created coro={coro}()>"
          else:
              fmt = "<{name} adding coro={coro}()>"
  
          try:
              coro_name = coro_stack[-1].name
          # coro_stack may be empty if:
          # - exhausted generator
          # - finished coroutine
          except IndexError:
              coro_name = self._coro.__name__
  
          repr_string = fmt.format(
              name=self.__name__,
              coro=coro_name,
              trigger=self._trigger,
              outcome=self._outcome
          )
          return repr_string
  
      def _advance(self, outcome):
          """Advance to the next yield in this coroutine.
  
          Args:
              outcome: The :any:`outcomes.Outcome` object to resume with.
  
          Returns:
              The object yielded from the coroutine
  
          Raises:
              CoroutineComplete: If the coroutine returns or throws an error, self._outcome is set, and
             :exc:`CoroutineComplete` is thrown.
          """
          try:
              self._started = True
              return outcome.send(self._coro)
          except ReturnValue as e:
              self._outcome = outcomes.Value(e.retval)
              raise CoroutineComplete()
          except StopIteration as e:
              self._outcome = outcomes.Value(e.value)
              raise CoroutineComplete()
          except BaseException as e:
              self._outcome = outcomes.Error(remove_traceback_frames(e, ['_advance', 'send']))
              raise CoroutineComplete()
  
      def send(self, value):
          return self._coro.send(value)
  
      def throw(self, exc):
          return self._coro.throw(exc)
  
      def close(self):
          return self._coro.close()
  
      def kill(self):
          """Kill a coroutine."""
          if self._outcome is not None:
              # already finished, nothing to kill
              return
  
          if _debug:
              self.log.debug("kill() called on coroutine")
          # todo: probably better to throw an exception for anyone waiting on the coroutine
          self._outcome = outcomes.Value(None)
          cocotb.scheduler.unschedule(self)
  
      def join(self):
          """Return a trigger that will fire when the wrapped coroutine exits."""
          return cocotb.triggers.Join(self)
  
      def has_started(self):
          return self._started
  
      def __bool__(self):
          """Provide boolean testing
              if the coroutine has finished return false
              otherwise return true"""
          return not self._finished
  
      def __await__(self):
          # It's tempting to use `return (yield from self._coro)` here,
          # which bypasses the scheduler. Unfortunately, this means that
          # we can't keep track of the result or state of the coroutine,
          # things which we expose in our public API. If you want the
          # efficiency of bypassing the scheduler, remove the `@coroutine`
          # decorator from your `async` functions.
  
          # Hand the coroutine back to the scheduler trampoline.
          return (yield self)
  
  
  class RunningCoroutine(RunningTask):
      """
      The result of calling a :any:`cocotb.coroutine` decorated coroutine.
  
      All this class does is provide some extra attributes.
      """
  
      def __init__(self, inst, parent):
          RunningTask.__init__(self, inst)
          self._parent = parent
          self.__doc__ = parent._func.__doc__
          self.module = parent._func.__module__
          self.funcname = parent._func.__name__
  
  
  class RunningTest(RunningCoroutine):
      """Add some useful Test functionality to a RunningCoroutine."""
  
      class ErrorLogHandler(logging.Handler):
          def __init__(self, fn):
              self.fn = fn
              logging.Handler.__init__(self, level=logging.DEBUG)
  
          def handle(self, record):
              # For historical reasons, only logs sent directly to the `cocotb`
              # logger are recorded - logs to `cocotb.scheduler` for instance
              # are not recorded. Changing this behavior may have significant
              # memory usage implications, so should not be done without some
              # thought.
              if record.name == 'cocotb':
                  self.fn(self.format(record))
  
      def __init__(self, inst, parent):
          self.error_messages = []
          RunningCoroutine.__init__(self, inst, parent)
          self.log = SimLog("cocotb.test.%s" % inst.__qualname__, id(self))
          self.started = False
          self.start_time = 0
          self.start_sim_time = 0
          self.expect_fail = parent.expect_fail
          self.expect_error = parent.expect_error
          self.skip = parent.skip
          self.stage = parent.stage
          self.__name__ = "Test %s" % inst.__name__
          self.__qualname__ = "Test %s" % inst.__qualname__
  
          # make sure not to create a circular reference here
          self.handler = RunningTest.ErrorLogHandler(self.error_messages.append)
  
      def __str__(self):
          return "<{}>".format(self.__name__)
  
      def _advance(self, outcome):
          if not self.started:
              self.log.info("Starting test: \"%s\"\nDescription: %s" %
                            (self.funcname, self.__doc__))
              self.start_time = time.time()
              self.start_sim_time = get_sim_time('ns')
              self.started = True
          return super(RunningTest, self)._advance(outcome)
  
      # like RunningTask.kill(), but with a way to inject a failure
      def abort(self, exc):
          """Force this test to end early, without executing any cleanup.
  
          This happens when a background task fails, and is consistent with
          how the behavior has always been. In future, we may want to behave
          more gracefully to allow the test body to clean up.
  
          `exc` is the exception that the test should report as its reason for
          aborting.
          """
          if self._outcome is not None:
              # imported here to avoid circular imports
              from cocotb.scheduler import InternalError
              raise InternalError("Outcome already has a value, but is being set again.")
          outcome = outcomes.Error(exc)
          if _debug:
              self.log.debug("outcome forced to {}".format(outcome))
          self._outcome = outcome
          cocotb.scheduler.unschedule(self)
  
      def sort_name(self):
          if self.stage is None:
              return "%s.%s" % (self.module, self.funcname)
          else:
              return "%s.%d.%s" % (self.module, self.stage, self.funcname)
  
  
  class coroutine:
      """Decorator class that allows us to provide common coroutine mechanisms:
  
      ``log`` methods will log to ``cocotb.coroutine.name``.
  
      :meth:`~cocotb.decorators.RunningTask.join` method returns an event which will fire when the coroutine exits.
  
      Used as ``@cocotb.coroutine``.
      """
  
      def __init__(self, func):
          self._func = func
          functools.update_wrapper(self, func)
  
      @lazy_property
      def log(self):
          return SimLog("cocotb.coroutine.%s" % self._func.__qualname__, id(self))
  
      def __call__(self, *args, **kwargs):
          return RunningCoroutine(self._func(*args, **kwargs), self)
  
      def __get__(self, obj, owner=None):
          """Permit the decorator to be used on class methods
              and standalone functions"""
          return type(self)(self._func.__get__(obj, owner))
  
      def __iter__(self):
          return self
  
      def __str__(self):
          return str(self._func.__qualname__)
  
  
  @public
  class function:
      """Decorator class that allows a function to block.
  
      This allows a coroutine that consumes simulation time
      to be called by a thread started with :class:`cocotb.external`;
      in other words, to internally block while externally
      appear to yield.
      """
  
      def __init__(self, func):
          self._coro = cocotb.coroutine(func)
  
      @lazy_property
      def log(self):
          return SimLog("cocotb.function.%s" % self._coro.__qualname__, id(self))
  
      def __call__(self, *args, **kwargs):
          return cocotb.scheduler.queue_function(self._coro(*args, **kwargs))
  
      def __get__(self, obj, owner=None):
          """Permit the decorator to be used on class methods
              and standalone functions"""
          return type(self)(self._coro._func.__get__(obj, owner))
  
  
  @public
  class external:
      """Decorator to apply to an external function to enable calling from cocotb.
  
      This turns a normal function that isn't a coroutine into a blocking coroutine.
      Currently, this creates a new execution thread for each function that is
      called.
      Scope for this to be streamlined to a queue in future.
      """
  
      def __init__(self, func):
          self._func = func
          self._log = SimLog("cocotb.external.%s" % self._func.__qualname__, id(self))
  
      def __call__(self, *args, **kwargs):
          return cocotb.scheduler.run_in_executor(self._func, *args, **kwargs)
  
      def __get__(self, obj, owner=None):
          """Permit the decorator to be used on class methods
              and standalone functions"""
          return type(self)(self._func.__get__(obj, owner))
  
  
  class _decorator_helper(type):
      """
      Metaclass that allows a type to be constructed using decorator syntax,
      passing the decorated function as the first argument.
  
      So:
  
          @MyClass(construction, args='go here')
          def this_is_passed_as_f(...):
              pass
  
      ends up calling
  
          MyClass.__init__(this_is_passed_as_f, construction, args='go here')
      """
      def __call__(cls, *args, **kwargs):
          def decorator(f):
              # fall back to the normal way of constructing an object, now that
              # we have all the arguments
              return type.__call__(cls, f, *args, **kwargs)
          return decorator
  
  
  @public
  class hook(coroutine, metaclass=_decorator_helper):
      """Decorator to mark a function as a hook for cocotb.
  
      Used as ``@cocotb.hook()``.
  
      All hooks are run at the beginning of a cocotb test suite, prior to any
      test code being run."""
  
      def __init__(self, f):
          super(hook, self).__init__(f)
          self.im_hook = True
          self.name = self._func.__name__
  
  
  @public
  class test(coroutine, metaclass=_decorator_helper):
      """Decorator to mark a function as a test.
  
      All tests are coroutines.  The test decorator provides
      some common reporting etc., a test timeout and allows
      us to mark tests as expected failures.
  
      Tests are evaluated in the order they are defined in a test module.
  
      Used as ``@cocotb.test(...)``.
  
      Args:
          timeout_time (numbers.Real or decimal.Decimal, optional):
              Simulation time duration before timeout occurs.
  
              .. versionadded:: 1.3
  
              .. note::
                  Test timeout is intended for protection against deadlock.
                  Users should use :class:`~cocotb.triggers.with_timeout` if they require a
                  more general-purpose timeout mechanism.
          timeout_unit (str or None, optional):
              Units of timeout_time, accepts any units that :class:`~cocotb.triggers.Timer` does.
  
              .. versionadded:: 1.3
          expect_fail (bool, optional):
              Don't mark the result as a failure if the test fails.
          expect_error (bool or exception type or tuple of exception types, optional):
              If ``True``, consider this test passing if it raises *any* :class:`Exception`, and failing if it does not.
              If given an exception type or tuple of exception types, catching *only* a listed exception type is considered passing.
              This is primarily for cocotb internal regression use for when a simulator error is expected.
  
              Users are encouraged to use the following idiom instead::
  
                  @cocotb.test()
                  async def my_test(dut):
                      try:
                          await thing_that_should_fail()
                      except ExceptionIExpect:
                          pass
                      else:
                          assert False, "Exception did not occur"
  
              .. versionchanged:: 1.3
                  Specific exception types can be expected
          skip (bool, optional):
              Don't execute this test as part of the regression.
          stage (int, optional)
              Order tests logically into stages, where multiple tests can share a stage.
      """
  
      _id_count = 0  # used by the RegressionManager to sort tests in definition order
  
      def __init__(self, f, timeout_time=None, timeout_unit=None,
                   expect_fail=False, expect_error=False,
                   skip=False, stage=None):
  
          self._id = self._id_count
          type(self)._id_count += 1
  
          if timeout_time is not None:
              co = coroutine(f)
  
              @functools.wraps(f)
              async def f(*args, **kwargs):
                  running_co = co(*args, **kwargs)
  
                  try:
                      res = await cocotb.triggers.with_timeout(running_co, self.timeout_time, self.timeout_unit)
                  except cocotb.result.SimTimeoutError:
                      running_co.kill()
                      raise
                  else:
                      return res
  
          super(test, self).__init__(f)
  
          self.timeout_time = timeout_time
          self.timeout_unit = timeout_unit
          self.expect_fail = expect_fail
          if expect_error is True:
              expect_error = (Exception,)
          elif expect_error is False:
              expect_error = ()
          self.expect_error = expect_error
          self.skip = skip
          self.stage = stage
          self.im_test = True    # For auto-regressions
          self.name = self._func.__name__
  
      def __call__(self, *args, **kwargs):
          return RunningTest(self._func(*args, **kwargs), self)
  
files.new_path: cocotb/decorators.py
files.old_contents: # Copyright (c) 2013 Potential Ventures Ltd
  # Copyright (c) 2013 SolarFlare Communications Inc
  # All rights reserved.
  #
  # Redistribution and use in source and binary forms, with or without
  # modification, are permitted provided that the following conditions are met:
  #     * Redistributions of source code must retain the above copyright
  #       notice, this list of conditions and the following disclaimer.
  #     * Redistributions in binary form must reproduce the above copyright
  #       notice, this list of conditions and the following disclaimer in the
  #       documentation and/or other materials provided with the distribution.
  #     * Neither the name of Potential Ventures Ltd,
  #       SolarFlare Communications Inc nor the
  #       names of its contributors may be used to endorse or promote products
  #       derived from this software without specific prior written permission.
  #
  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
  # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  # DISCLAIMED. IN NO EVENT SHALL POTENTIAL VENTURES LTD BE LIABLE FOR ANY
  # DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  
  import sys
  import time
  import logging
  import functools
  import inspect
  import os
  
  import cocotb
  from cocotb.log import SimLog
  from cocotb.result import ReturnValue
  from cocotb.utils import get_sim_time, lazy_property, remove_traceback_frames
  from cocotb import outcomes
  
  # Sadly the Python standard logging module is very slow so it's better not to
  # make any calls by testing a boolean flag first
  if "COCOTB_SCHEDULER_DEBUG" in os.environ:
      _debug = True
  else:
      _debug = False
  
  
  def public(f):
      """Use a decorator to avoid retyping function/class names.
  
      * Based on an idea by Duncan Booth:
      http://groups.google.com/group/comp.lang.python/msg/11cbb03e09611b8a
      * Improved via a suggestion by Dave Angel:
      http://groups.google.com/group/comp.lang.python/msg/3d400fb22d8a42e1
      """
      all = sys.modules[f.__module__].__dict__.setdefault('__all__', [])
      if f.__name__ not in all:  # Prevent duplicates if run from an IDE.
          all.append(f.__name__)
      return f
  
  
  public(public)  # Emulate decorating ourself
  
  
  @public
  class CoroutineComplete(Exception):
      """To ensure that a coroutine has completed before we fire any triggers
      that are blocked waiting for the coroutine to end, we create a sub-class
      exception that the scheduler catches and the callbacks are attached
      here.
      """
  
      def __init__(self, text=""):
          Exception.__init__(self, text)
  
  
  class RunningTask:
      """Per instance wrapper around a running generator.
  
      Provides the following:
  
          task.join() creates a Trigger that will fire when this coroutine
          completes.
  
          task.kill() will destroy a coroutine instance (and cause any Join
          triggers to fire.
      """
  
      def __init__(self, inst):
  
          if inspect.iscoroutine(inst):
              self._natively_awaitable = True
          elif inspect.isgenerator(inst):
              self._natively_awaitable = False
          elif inspect.iscoroutinefunction(inst):
              raise TypeError(
                  "Coroutine function {} should be called prior to being "
                  "scheduled."
                  .format(inst))
          elif sys.version_info >= (3, 6) and inspect.isasyncgen(inst):
              raise TypeError(
                  "{} is an async generator, not a coroutine. "
                  "You likely used the yield keyword instead of await.".format(
                      inst.__qualname__))
          else:
              raise TypeError(
                  "%s isn't a valid coroutine! Did you forget to use the yield keyword?" % inst)
          self._coro = inst
          self.__name__ = inst.__name__
          self.__qualname__ = inst.__qualname__
          self._started = False
          self._callbacks = []
          self._outcome = None
          self._trigger = None
  
      @lazy_property
      def log(self):
          # Creating a logger is expensive, only do it if we actually plan to
          # log anything
          return SimLog("cocotb.coroutine.%s" % self.__qualname__, id(self))
  
      @property
      def retval(self):
          if self._outcome is None:
              raise RuntimeError("coroutine is not complete")
          return self._outcome.get()
  
      @property
      def _finished(self):
          return self._outcome is not None
  
      def __iter__(self):
          return self
  
      def __str__(self):
          return str(self.__qualname__)
  
      def _advance(self, outcome):
          """Advance to the next yield in this coroutine.
  
          Args:
              outcome: The :any:`outcomes.Outcome` object to resume with.
  
          Returns:
              The object yielded from the coroutine
  
          Raises:
              CoroutineComplete: If the coroutine returns or throws an error, self._outcome is set, and
             :exc:`CoroutineComplete` is thrown.
          """
          try:
              self._started = True
              return outcome.send(self._coro)
          except ReturnValue as e:
              self._outcome = outcomes.Value(e.retval)
              raise CoroutineComplete()
          except StopIteration as e:
              self._outcome = outcomes.Value(e.value)
              raise CoroutineComplete()
          except BaseException as e:
              self._outcome = outcomes.Error(remove_traceback_frames(e, ['_advance', 'send']))
              raise CoroutineComplete()
  
      def send(self, value):
          return self._coro.send(value)
  
      def throw(self, exc):
          return self._coro.throw(exc)
  
      def close(self):
          return self._coro.close()
  
      def kill(self):
          """Kill a coroutine."""
          if self._outcome is not None:
              # already finished, nothing to kill
              return
  
          if _debug:
              self.log.debug("kill() called on coroutine")
          # todo: probably better to throw an exception for anyone waiting on the coroutine
          self._outcome = outcomes.Value(None)
          cocotb.scheduler.unschedule(self)
  
      def join(self):
          """Return a trigger that will fire when the wrapped coroutine exits."""
          return cocotb.triggers.Join(self)
  
      def has_started(self):
          return self._started
  
      def __bool__(self):
          """Provide boolean testing
              if the coroutine has finished return false
              otherwise return true"""
          return not self._finished
  
      def __await__(self):
          # It's tempting to use `return (yield from self._coro)` here,
          # which bypasses the scheduler. Unfortunately, this means that
          # we can't keep track of the result or state of the coroutine,
          # things which we expose in our public API. If you want the
          # efficiency of bypassing the scheduler, remove the `@coroutine`
          # decorator from your `async` functions.
  
          # Hand the coroutine back to the scheduler trampoline.
          return (yield self)
  
  
  class RunningCoroutine(RunningTask):
      """
      The result of calling a :any:`cocotb.coroutine` decorated coroutine.
  
      All this class does is provide some extra attributes.
      """
  
      def __init__(self, inst, parent):
          RunningTask.__init__(self, inst)
          self._parent = parent
          self.__doc__ = parent._func.__doc__
          self.module = parent._func.__module__
          self.funcname = parent._func.__name__
  
  
  class RunningTest(RunningCoroutine):
      """Add some useful Test functionality to a RunningCoroutine."""
  
      class ErrorLogHandler(logging.Handler):
          def __init__(self, fn):
              self.fn = fn
              logging.Handler.__init__(self, level=logging.DEBUG)
  
          def handle(self, record):
              # For historical reasons, only logs sent directly to the `cocotb`
              # logger are recorded - logs to `cocotb.scheduler` for instance
              # are not recorded. Changing this behavior may have significant
              # memory usage implications, so should not be done without some
              # thought.
              if record.name == 'cocotb':
                  self.fn(self.format(record))
  
      def __init__(self, inst, parent):
          self.error_messages = []
          RunningCoroutine.__init__(self, inst, parent)
          self.log = SimLog("cocotb.test.%s" % self.__qualname__, id(self))
          self.started = False
          self.start_time = 0
          self.start_sim_time = 0
          self.expect_fail = parent.expect_fail
          self.expect_error = parent.expect_error
          self.skip = parent.skip
          self.stage = parent.stage
          self._id = parent._id
  
          # make sure not to create a circular reference here
          self.handler = RunningTest.ErrorLogHandler(self.error_messages.append)
  
      def _advance(self, outcome):
          if not self.started:
              self.log.info("Starting test: \"%s\"\nDescription: %s" %
                            (self.funcname, self.__doc__))
              self.start_time = time.time()
              self.start_sim_time = get_sim_time('ns')
              self.started = True
          return super(RunningTest, self)._advance(outcome)
  
      # like RunningTask.kill(), but with a way to inject a failure
      def abort(self, exc):
          """Force this test to end early, without executing any cleanup.
  
          This happens when a background task fails, and is consistent with
          how the behavior has always been. In future, we may want to behave
          more gracefully to allow the test body to clean up.
  
          `exc` is the exception that the test should report as its reason for
          aborting.
          """
          if self._outcome is not None:
              # imported here to avoid circular imports
              from cocotb.scheduler import InternalError
              raise InternalError("Outcome already has a value, but is being set again.")
          outcome = outcomes.Error(exc)
          if _debug:
              self.log.debug("outcome forced to {}".format(outcome))
          self._outcome = outcome
          cocotb.scheduler.unschedule(self)
  
      def sort_name(self):
          if self.stage is None:
              return "%s.%s" % (self.module, self.funcname)
          else:
              return "%s.%d.%s" % (self.module, self.stage, self.funcname)
  
  
  class coroutine:
      """Decorator class that allows us to provide common coroutine mechanisms:
  
      ``log`` methods will log to ``cocotb.coroutine.name``.
  
      :meth:`~cocotb.decorators.RunningTask.join` method returns an event which will fire when the coroutine exits.
  
      Used as ``@cocotb.coroutine``.
      """
  
      def __init__(self, func):
          self._func = func
          functools.update_wrapper(self, func)
  
      @lazy_property
      def log(self):
          return SimLog("cocotb.coroutine.%s" % self._func.__qualname__, id(self))
  
      def __call__(self, *args, **kwargs):
          return RunningCoroutine(self._func(*args, **kwargs), self)
  
      def __get__(self, obj, owner=None):
          """Permit the decorator to be used on class methods
              and standalone functions"""
          return type(self)(self._func.__get__(obj, owner))
  
      def __iter__(self):
          return self
  
      def __str__(self):
          return str(self._func.__qualname__)
  
  
  @public
  class function:
      """Decorator class that allows a function to block.
  
      This allows a coroutine that consumes simulation time
      to be called by a thread started with :class:`cocotb.external`;
      in other words, to internally block while externally
      appear to yield.
      """
  
      def __init__(self, func):
          self._coro = cocotb.coroutine(func)
  
      @lazy_property
      def log(self):
          return SimLog("cocotb.function.%s" % self._coro.__qualname__, id(self))
  
      def __call__(self, *args, **kwargs):
          return cocotb.scheduler.queue_function(self._coro(*args, **kwargs))
  
      def __get__(self, obj, owner=None):
          """Permit the decorator to be used on class methods
              and standalone functions"""
          return type(self)(self._coro._func.__get__(obj, owner))
  
  
  @public
  class external:
      """Decorator to apply to an external function to enable calling from cocotb.
  
      This turns a normal function that isn't a coroutine into a blocking coroutine.
      Currently, this creates a new execution thread for each function that is
      called.
      Scope for this to be streamlined to a queue in future.
      """
  
      def __init__(self, func):
          self._func = func
          self._log = SimLog("cocotb.external.%s" % self._func.__qualname__, id(self))
  
      def __call__(self, *args, **kwargs):
          return cocotb.scheduler.run_in_executor(self._func, *args, **kwargs)
  
      def __get__(self, obj, owner=None):
          """Permit the decorator to be used on class methods
              and standalone functions"""
          return type(self)(self._func.__get__(obj, owner))
  
  
  class _decorator_helper(type):
      """
      Metaclass that allows a type to be constructed using decorator syntax,
      passing the decorated function as the first argument.
  
      So:
  
          @MyClass(construction, args='go here')
          def this_is_passed_as_f(...):
              pass
  
      ends up calling
  
          MyClass.__init__(this_is_passed_as_f, construction, args='go here')
      """
      def __call__(cls, *args, **kwargs):
          def decorator(f):
              # fall back to the normal way of constructing an object, now that
              # we have all the arguments
              return type.__call__(cls, f, *args, **kwargs)
          return decorator
  
  
  @public
  class hook(coroutine, metaclass=_decorator_helper):
      """Decorator to mark a function as a hook for cocotb.
  
      Used as ``@cocotb.hook()``.
  
      All hooks are run at the beginning of a cocotb test suite, prior to any
      test code being run."""
  
      def __init__(self, f):
          super(hook, self).__init__(f)
          self.im_hook = True
          self.name = self._func.__name__
  
  
  @public
  class test(coroutine, metaclass=_decorator_helper):
      """Decorator to mark a function as a test.
  
      All tests are coroutines.  The test decorator provides
      some common reporting etc., a test timeout and allows
      us to mark tests as expected failures.
  
      Tests are evaluated in the order they are defined in a test module.
  
      Used as ``@cocotb.test(...)``.
  
      Args:
          timeout_time (numbers.Real or decimal.Decimal, optional):
              Simulation time duration before timeout occurs.
  
              .. versionadded:: 1.3
  
              .. note::
                  Test timeout is intended for protection against deadlock.
                  Users should use :class:`~cocotb.triggers.with_timeout` if they require a
                  more general-purpose timeout mechanism.
          timeout_unit (str or None, optional):
              Units of timeout_time, accepts any units that :class:`~cocotb.triggers.Timer` does.
  
              .. versionadded:: 1.3
          expect_fail (bool, optional):
              Don't mark the result as a failure if the test fails.
          expect_error (bool or exception type or tuple of exception types, optional):
              If ``True``, consider this test passing if it raises *any* :class:`Exception`, and failing if it does not.
              If given an exception type or tuple of exception types, catching *only* a listed exception type is considered passing.
              This is primarily for cocotb internal regression use for when a simulator error is expected.
  
              Users are encouraged to use the following idiom instead::
  
                  @cocotb.test()
                  async def my_test(dut):
                      try:
                          await thing_that_should_fail()
                      except ExceptionIExpect:
                          pass
                      else:
                          assert False, "Exception did not occur"
  
              .. versionchanged:: 1.3
                  Specific exception types can be expected
          skip (bool, optional):
              Don't execute this test as part of the regression.
          stage (int, optional)
              Order tests logically into stages, where multiple tests can share a stage.
      """
  
      _id_count = 0  # used by the RegressionManager to sort tests in definition order
  
      def __init__(self, f, timeout_time=None, timeout_unit=None,
                   expect_fail=False, expect_error=False,
                   skip=False, stage=None):
  
          self._id = self._id_count
          type(self)._id_count += 1
  
          if timeout_time is not None:
              co = coroutine(f)
  
              @functools.wraps(f)
              async def f(*args, **kwargs):
                  running_co = co(*args, **kwargs)
  
                  try:
                      res = await cocotb.triggers.with_timeout(running_co, self.timeout_time, self.timeout_unit)
                  except cocotb.result.SimTimeoutError:
                      running_co.kill()
                      raise
                  else:
                      return res
  
          super(test, self).__init__(f)
  
          self.timeout_time = timeout_time
          self.timeout_unit = timeout_unit
          self.expect_fail = expect_fail
          if expect_error is True:
              expect_error = (Exception,)
          elif expect_error is False:
              expect_error = ()
          self.expect_error = expect_error
          self.skip = skip
          self.stage = stage
          self.im_test = True    # For auto-regressions
          self.name = self._func.__name__
  
      def __call__(self, *args, **kwargs):
          return RunningTest(self._func(*args, **kwargs), self)
  
files.old_path: cocotb/decorators.py
files.new_contents: # Copyright (c) 2013 Potential Ventures Ltd
  # Copyright (c) 2013 SolarFlare Communications Inc
  # All rights reserved.
  #
  # Redistribution and use in source and binary forms, with or without
  # modification, are permitted provided that the following conditions are met:
  #     * Redistributions of source code must retain the above copyright
  #       notice, this list of conditions and the following disclaimer.
  #     * Redistributions in binary form must reproduce the above copyright
  #       notice, this list of conditions and the following disclaimer in the
  #       documentation and/or other materials provided with the distribution.
  #     * Neither the name of Potential Ventures Ltd,
  #       SolarFlare Communications Inc nor the
  #       names of its contributors may be used to endorse or promote products
  #       derived from this software without specific prior written permission.
  #
  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
  # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  # DISCLAIMED. IN NO EVENT SHALL POTENTIAL VENTURES LTD BE LIABLE FOR ANY
  # DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  
  """Collection of handy functions."""
  
  import ctypes
  import math
  import os
  import sys
  import traceback
  import weakref
  import functools
  import warnings
  
  from cocotb import simulator
  
  
  def _get_simulator_precision():
      # cache and replace this function
      precision = simulator.get_precision()
      global _get_simulator_precision
      _get_simulator_precision = precision.__int__
      return _get_simulator_precision()
  
  
  def get_python_integer_types():
      warnings.warn(
          "This is an internal cocotb function, use six.integer_types instead",
          DeprecationWarning)
      return (int,)
  
  
  # Simulator helper functions
  def get_sim_time(units=None):
      """Retrieves the simulation time from the simulator.
  
      Args:
          units (str or None, optional): String specifying the units of the result
              (one of ``None``, ``'fs'``, ``'ps'``, ``'ns'``, ``'us'``, ``'ms'``, ``'sec'``).
              ``None`` will return the raw simulation time.
  
      Returns:
          The simulation time in the specified units.
      """
      timeh, timel = simulator.get_sim_time()
  
      result = (timeh << 32 | timel)
  
      if units is not None:
          result = get_time_from_sim_steps(result, units)
  
      return result
  
  
  def _ldexp10(frac, exp):
      """ Like math.ldexp, but base 10 """
      # using * or / separately prevents rounding errors if `frac` is a
      # high-precision type
      if exp > 0:
          return frac * (10 ** exp)
      else:
          return frac / (10 ** -exp)
  
  
  def get_time_from_sim_steps(steps, units):
      """Calculates simulation time in the specified *units* from the *steps* based
      on the simulator precision.
  
      Args:
          steps (int): Number of simulation steps.
          units (str): String specifying the units of the result
              (one of ``'fs'``, ``'ps'``, ``'ns'``, ``'us'``, ``'ms'``, ``'sec'``).
  
      Returns:
          The simulation time in the specified units.
      """
      return _ldexp10(steps, _get_simulator_precision() - _get_log_time_scale(units))
  
  
  def get_sim_steps(time, units=None):
      """Calculates the number of simulation time steps for a given amount of *time*.
  
      Args:
          time (numbers.Real or decimal.Decimal):  The value to convert to simulation time steps.
          units (str or None, optional):  String specifying the units of the result
              (one of ``None``, ``'fs'``, ``'ps'``, ``'ns'``, ``'us'``, ``'ms'``, ``'sec'``).
              ``None`` means time is already in simulation time steps.
  
      Returns:
          int: The number of simulation time steps.
  
      Raises:
          :exc:`ValueError`: If given *time* cannot be represented by simulator precision.
      """
      result = time
      if units is not None:
          result = _ldexp10(result, _get_log_time_scale(units) - _get_simulator_precision())
  
      result_rounded = math.floor(result)
  
      if result_rounded != result:
          raise ValueError("Unable to accurately represent {0}({1}) with the "
                           "simulator precision of 1e{2}".format(
                               time, units, _get_simulator_precision()))
  
      return int(result_rounded)
  
  
  def _get_log_time_scale(units):
      """Retrieves the ``log10()`` of the scale factor for a given time unit.
  
      Args:
          units (str): String specifying the units
              (one of ``'fs'``, ``'ps'``, ``'ns'``, ``'us'``, ``'ms'``, ``'sec'``).
  
      Returns:
          The the ``log10()`` of the scale factor for the time unit.
      """
      scale = {
          'fs' :    -15,
          'ps' :    -12,
          'ns' :     -9,
          'us' :     -6,
          'ms' :     -3,
          'sec':      0}
  
      units_lwr = units.lower()
      if units_lwr not in scale:
          raise ValueError("Invalid unit ({0}) provided".format(units))
      else:
          return scale[units_lwr]
  
  # Ctypes helper functions
  
  
  def pack(ctypes_obj):
      """Convert a :mod:`ctypes` structure into a Python string.
  
      Args:
          ctypes_obj (ctypes.Structure): The :mod:`ctypes` structure to convert to a string.
  
      Returns:
          New Python string containing the bytes from memory holding *ctypes_obj*.
      """
      return ctypes.string_at(ctypes.addressof(ctypes_obj),
                              ctypes.sizeof(ctypes_obj))
  
  
  def unpack(ctypes_obj, string, bytes=None):
      """Unpack a Python string into a :mod:`ctypes` structure.
  
      If the length of *string* is not the correct size for the memory
      footprint of the :mod:`ctypes` structure then the *bytes* keyword argument
      must be used.
  
      Args:
          ctypes_obj (ctypes.Structure): The :mod:`ctypes` structure to pack into.
          string (str):  String to copy over the *ctypes_obj* memory space.
          bytes (int, optional): Number of bytes to copy.
              Defaults to ``None``, meaning the length of *string* is used.
  
      Raises:
          :exc:`ValueError`: If length of *string* and size of *ctypes_obj*
              are not equal.
          :exc:`MemoryError`: If *bytes* is longer than size of *ctypes_obj*.
      """
      if bytes is None:
          if len(string) != ctypes.sizeof(ctypes_obj):
              raise ValueError("Attempt to unpack a string of size %d into a \
                  struct of size %d" % (len(string), ctypes.sizeof(ctypes_obj)))
          bytes = len(string)
  
      if bytes > ctypes.sizeof(ctypes_obj):
          raise MemoryError("Attempt to unpack %d bytes over an object \
                          of size %d" % (bytes, ctypes.sizeof(ctypes_obj)))
  
      ctypes.memmove(ctypes.addressof(ctypes_obj), string, bytes)
  
  
  import cocotb.ANSI as ANSI
  
  
  # A note on the use of latin1 in the deprecations below:
  # Latin1 is the only encoding `e` that satisfies
  # `all(chr(x).encode(e) == bytes([x]) for x in range(255))`
  # Our use of `ord` and `chr` throughout other bits of code make this the most
  # compatible choice of encoding. Under this convention, old code can be upgraded
  # by changing "binary" strings from `"\x12\x34"` to `b"\x12\x34"`.
  
  
  def _sane(x: bytes) -> str:
      r = ""
      for j in x:
          if (j < 32) or (j >= 127):
              r += "."
          else:
              r += chr(j)
      return r
  
  
  def hexdump(x: bytes) -> str:
      """Hexdump a buffer.
  
      Args:
          x: Object that supports conversion to :class:`bytes`.
  
      Returns:
          A string containing the hexdump.
  
      .. deprecated:: 1.4
          Passing a :class:`str` to this function is deprecated, as it
          is not an appropriate type for binary data. Doing so anyway
          will encode the string to latin1.
  
      Example:
          >>> print(hexdump(b'this somewhat long string'))
          0000   74 68 69 73 20 73 6F 6D 65 77 68 61 74 20 6C 6F   this somewhat lo
          0010   6E 67 20 73 74 72 69 6E 67                        ng string
          <BLANKLINE>
      """
      # adapted from scapy.utils.hexdump
      rs = ""
      if isinstance(x, str):
          warnings.warn(
              "Passing a string to hexdump is deprecated, pass bytes instead",
              DeprecationWarning, stacklevel=2)
          x = x.encode('latin1')
      x = b"%b" % x
      l = len(x)
      i = 0
      while i < l:
          rs += "%04x   " % i
          for j in range(16):
              if i + j < l:
                  rs += "%02X " % x[i + j]
              else:
                  rs += "   "
              if j % 16 == 7:
                  rs += ""
          rs += "  "
          rs += _sane(x[i:i + 16]) + "\n"
          i += 16
      return rs
  
  
  def hexdiffs(x: bytes, y: bytes) -> str:
      r"""Return a diff string showing differences between two binary strings.
  
      Args:
          x: Object that supports conversion to :class:`bytes`.
          y: Object that supports conversion to :class:`bytes`.
  
      .. deprecated:: 1.4
          Passing :class:`str`\ s to this function is deprecated, as it
          is not an appropriate type for binary data. Doing so anyway
          will encode the string to latin1.
  
      Example:
          >>> print(hexdiffs(b'a', b'b'))
          0000      61                                               a
               0000 62                                               b
          <BLANKLINE>
          >>> print(hexdiffs(b'this short thing', b'this also short'))
          0000      746869732073686F 7274207468696E67 this short thing
               0000 7468697320616C73 6F  2073686F7274 this also  short
          <BLANKLINE>
      """
      # adapted from scapy.utils.hexdiff
  
      def highlight(string: str, colour=ANSI.COLOR_HILITE_HEXDIFF_DEFAULT) -> str:
          """Highlight with ANSI colors if possible/requested and not running in GUI."""
  
          if want_color_output():
              return colour + string + ANSI.COLOR_DEFAULT
          else:
              return string
  
      x_is_str = isinstance(x, str)
      y_is_str = isinstance(y, str)
      if x_is_str or y_is_str:
          warnings.warn(
              "Passing strings to hexdiffs is deprecated, pass bytes instead",
              DeprecationWarning, stacklevel=2)
      if x_is_str:
          x = x.encode('latin1')
      if y_is_str:
          y = y.encode('latin1')
  
      rs = ""
  
      x = (b'%b' % x)[::-1]
      y = (b'%b' % y)[::-1]
      SUBST = 1
      INSERT = 1
      d = {}
      d[-1, -1] = 0, (-1, -1)
      for j in range(len(y)):
          d[-1, j] = d[-1, j - 1][0] + INSERT, (-1, j - 1)
      for i in range(len(x)):
          d[i, -1] = d[i - 1, -1][0] + INSERT, (i - 1, -1)
  
      for j in range(len(y)):
          for i in range(len(x)):
              d[i, j] = min((d[i-1, j-1][0] + SUBST*(x[i] != y[j]), (i-1, j-1)),
                            (d[i - 1, j][0] + INSERT, (i - 1, j)),
                            (d[i, j - 1][0] + INSERT, (i, j - 1)))
  
      backtrackx = []
      backtracky = []
      i = len(x) - 1
      j = len(y) - 1
      while not (i == j == -1):
          i2, j2 = d[i, j][1]
          backtrackx.append(x[i2+1:i+1])
          backtracky.append(y[j2+1:j+1])
          i, j = i2, j2
  
      x = y = i = 0
      colorize = { 0: lambda x: x,  # noqa
                  -1: lambda x: x,  # noqa
                   1: lambda x: x}  # noqa
  
      dox = 1
      doy = 0
      l = len(backtrackx)
      while i < l:
          linex = backtrackx[i:i+16]
          liney = backtracky[i:i+16]
          xx = sum(len(k) for k in linex)
          yy = sum(len(k) for k in liney)
          if dox and not xx:
              dox = 0
              doy = 1
          if dox and linex == liney:
              doy = 1
  
          if dox:
              xd = y
              j = 0
              while not linex[j]:
                  j += 1
                  xd -= 1
              if dox != doy:
                  rs += highlight("%04x" % xd) + " "
              else:
                  rs += highlight("%04x" % xd, colour=ANSI.COLOR_HILITE_HEXDIFF_1) + " "
              x += xx
              line = linex
          else:
              rs += "    "
          if doy:
              yd = y
              j = 0
              while not liney[j]:
                  j += 1
                  yd -= 1
              if doy - dox != 0:
                  rs += " " + highlight("%04x" % yd)
              else:
                  rs += highlight("%04x" % yd, colour=ANSI.COLOR_HILITE_HEXDIFF_1)
              y += yy
              line = liney
          else:
              rs += "    "
  
          rs += " "
  
          cl = ""
          for j in range(16):
              if i + j < l:
                  if line[j]:
                      char_j, = line[j]
                      if linex[j] != liney[j]:
                          rs += highlight("%02X" % char_j,
                                          colour=ANSI.COLOR_HILITE_HEXDIFF_2)
                      else:
                          rs += "%02X" % char_j
                      if linex[j] == liney[j]:
                          cl += highlight(_sane(line[j]),
                                          colour=ANSI.COLOR_HILITE_HEXDIFF_3)
                      else:
                          cl += highlight(_sane(line[j]),
                                          colour=ANSI.COLOR_HILITE_HEXDIFF_4)
                  else:
                      rs += "  "
                      cl += " "
              else:
                  rs += "   "
              if j == 7:
                  rs += " "
  
          rs += " " + cl + '\n'
  
          if doy or not yy:
              doy = 0
              dox = 1
              i += 16
          else:
              if yy:
                  dox = 0
                  doy = 1
              else:
                  i += 16
      return rs
  
  
  class ParametrizedSingleton(type):
      """A metaclass that allows class construction to reuse an existing instance.
  
      We use this so that :class:`RisingEdge(sig) <cocotb.triggers.RisingEdge>` and :class:`Join(coroutine) <cocotb.triggers.Join>` always return
      the same instance, rather than creating new copies.
      """
  
      def __init__(cls, *args, **kwargs):
          # Attach a lookup table to this class.
          # Weak such that if the instance is no longer referenced, it can be
          # collected.
          cls.__instances = weakref.WeakValueDictionary()
  
      def __singleton_key__(cls, *args, **kwargs):
          """Convert the construction arguments into a normalized representation that
          uniquely identifies this singleton.
          """
          # Could default to something like this, but it would be slow
          # return tuple(inspect.Signature(cls).bind(*args, **kwargs).arguments.items())
          raise NotImplementedError
  
      def __call__(cls, *args, **kwargs):
          key = cls.__singleton_key__(*args, **kwargs)
          try:
              return cls.__instances[key]
          except KeyError:
              # construct the object as normal
              self = super(ParametrizedSingleton, cls).__call__(*args, **kwargs)
              cls.__instances[key] = self
              return self
  
  
  def reject_remaining_kwargs(name, kwargs):
      """
      Helper function to emulate Python 3 keyword-only arguments.
  
      Use as::
  
          def func(x1, **kwargs):
              a = kwargs.pop('a', 1)
              b = kwargs.pop('b', 2)
              reject_remaining_kwargs('func', kwargs)
              ...
  
      To emulate the Python 3 syntax::
  
          def func(x1, *, a=1, b=2):
              ...
  
      .. deprecated:: 1.4
          Since the minimum supported Python version is now 3.5, this function
          is not needed.
      """
      warnings.warn(
          "reject_remaining_kwargs is deprecated and will be removed, use "
          "Python 3 keyword-only arguments directly.", DeprecationWarning,
          stacklevel=2)
      if kwargs:
          # match the error message to what Python 3 produces
          bad_arg = next(iter(kwargs))
          raise TypeError(
              '{}() got an unexpected keyword argument {!r}'.format(name, bad_arg)
          )
  
  
  class lazy_property:
      """
      A property that is executed the first time, then cached forever.
  
      It does this by replacing itself on the instance, which works because
      unlike `@property` it does not define __set__.
  
      This should be used for expensive members of objects that are not always
      used.
      """
  
      def __init__(self, fget):
          self.fget = fget
  
          # copy the getter function's docstring and other attributes
          functools.update_wrapper(self, fget)
  
      def __get__(self, obj, cls):
          if obj is None:
              return self
  
          value = self.fget(obj)
          setattr(obj, self.fget.__qualname__, value)
          return value
  
  
  def want_color_output():
      """Return ``True`` if colored output is possible/requested and not running in GUI.
  
      Colored output can be explicitly requested by setting :envvar:`COCOTB_ANSI_OUTPUT` to  ``1``.
      """
      want_color = sys.stdout.isatty()  # default to color for TTYs
      if os.getenv("NO_COLOR") is not None:
          want_color = False
      if os.getenv("COCOTB_ANSI_OUTPUT", default='0') == '1':
          want_color = True
      if os.getenv("GUI", default='0') == '1':
          want_color = False
      return want_color
  
  
  if __name__ == "__main__":
      import random
      a = ""
      for char in range(random.randint(250, 500)):
          a += chr(random.randint(0, 255))
      b = a
      for error in range(random.randint(2, 9)):
          offset = random.randint(0, len(a))
          b = b[:offset] + chr(random.randint(0, 255)) + b[offset+1:]
  
      diff = hexdiffs(a, b)
      print(diff)
  
      space = '\n' + (" " * 20)
      print(space.join(diff.split('\n')))
  
  
  def remove_traceback_frames(tb_or_exc, frame_names):
      """
      Strip leading frames from a traceback
  
      Args:
          tb_or_exc (Union[traceback, BaseException, exc_info]):
              Object to strip frames from. If an exception is passed, creates
              a copy of the exception with a new shorter traceback. If a tuple
              from `sys.exc_info` is passed, returns the same tuple with the
              traceback shortened
          frame_names (List[str]):
              Names of the frames to strip, which must be present.
      """
      # self-invoking overloads
      if isinstance(tb_or_exc, BaseException):
          exc = tb_or_exc
          return exc.with_traceback(
              remove_traceback_frames(exc.__traceback__, frame_names)
          )
      elif isinstance(tb_or_exc, tuple):
          exc_type, exc_value, exc_tb = tb_or_exc
          exc_tb = remove_traceback_frames(exc_tb, frame_names)
          return exc_type, exc_value, exc_tb
      # base case
      else:
          tb = tb_or_exc
          for frame_name in frame_names:
              assert tb.tb_frame.f_code.co_name == frame_name
              tb = tb.tb_next
          return tb
  
  
  def walk_coro_stack(coro):
      """Walk down the coroutine stack, starting at *coro*.
  
      Supports coroutines and generators.
      """
      while coro is not None:
          try:
              f = getattr(coro, 'cr_frame')
              coro = coro.cr_await
          except AttributeError:
              try:
                  f = getattr(coro, 'gi_frame')
                  coro = coro.gi_yieldfrom
              except AttributeError:
                  f = None
                  coro = None
          if f is not None:
              yield (f, f.f_lineno)
  
  
  def extract_coro_stack(coro, limit=None):
      """Create a list of pre-processed entries from the coroutine stack.
  
      This is based on :func:`traceback.extract_tb`.
  
      If *limit* is omitted or ``None``, all entries are extracted.
      The list is a :class:`traceback.StackSummary` object, and
      each entry in the list is a :class:`traceback.FrameSummary` object
      containing attributes ``filename``, ``lineno``, ``name``, and ``line``
      representing the information that is usually printed for a stack
      trace.  The line is a string with leading and trailing
      whitespace stripped; if the source is not available it is ``None``.
      """
      return traceback.StackSummary.extract(walk_coro_stack(coro), limit=limit)
  
files.new_path: cocotb/utils.py
files.old_contents: # Copyright (c) 2013 Potential Ventures Ltd
  # Copyright (c) 2013 SolarFlare Communications Inc
  # All rights reserved.
  #
  # Redistribution and use in source and binary forms, with or without
  # modification, are permitted provided that the following conditions are met:
  #     * Redistributions of source code must retain the above copyright
  #       notice, this list of conditions and the following disclaimer.
  #     * Redistributions in binary form must reproduce the above copyright
  #       notice, this list of conditions and the following disclaimer in the
  #       documentation and/or other materials provided with the distribution.
  #     * Neither the name of Potential Ventures Ltd,
  #       SolarFlare Communications Inc nor the
  #       names of its contributors may be used to endorse or promote products
  #       derived from this software without specific prior written permission.
  #
  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
  # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  # DISCLAIMED. IN NO EVENT SHALL POTENTIAL VENTURES LTD BE LIABLE FOR ANY
  # DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  
  """Collection of handy functions."""
  
  import ctypes
  import math
  import os
  import sys
  import weakref
  import functools
  import warnings
  
  from cocotb import simulator
  
  
  def _get_simulator_precision():
      # cache and replace this function
      precision = simulator.get_precision()
      global _get_simulator_precision
      _get_simulator_precision = precision.__int__
      return _get_simulator_precision()
  
  
  def get_python_integer_types():
      warnings.warn(
          "This is an internal cocotb function, use six.integer_types instead",
          DeprecationWarning)
      return (int,)
  
  
  # Simulator helper functions
  def get_sim_time(units=None):
      """Retrieves the simulation time from the simulator.
  
      Args:
          units (str or None, optional): String specifying the units of the result
              (one of ``None``, ``'fs'``, ``'ps'``, ``'ns'``, ``'us'``, ``'ms'``, ``'sec'``).
              ``None`` will return the raw simulation time.
  
      Returns:
          The simulation time in the specified units.
      """
      timeh, timel = simulator.get_sim_time()
  
      result = (timeh << 32 | timel)
  
      if units is not None:
          result = get_time_from_sim_steps(result, units)
  
      return result
  
  
  def _ldexp10(frac, exp):
      """ Like math.ldexp, but base 10 """
      # using * or / separately prevents rounding errors if `frac` is a
      # high-precision type
      if exp > 0:
          return frac * (10 ** exp)
      else:
          return frac / (10 ** -exp)
  
  
  def get_time_from_sim_steps(steps, units):
      """Calculates simulation time in the specified *units* from the *steps* based
      on the simulator precision.
  
      Args:
          steps (int): Number of simulation steps.
          units (str): String specifying the units of the result
              (one of ``'fs'``, ``'ps'``, ``'ns'``, ``'us'``, ``'ms'``, ``'sec'``).
  
      Returns:
          The simulation time in the specified units.
      """
      return _ldexp10(steps, _get_simulator_precision() - _get_log_time_scale(units))
  
  
  def get_sim_steps(time, units=None):
      """Calculates the number of simulation time steps for a given amount of *time*.
  
      Args:
          time (numbers.Real or decimal.Decimal):  The value to convert to simulation time steps.
          units (str or None, optional):  String specifying the units of the result
              (one of ``None``, ``'fs'``, ``'ps'``, ``'ns'``, ``'us'``, ``'ms'``, ``'sec'``).
              ``None`` means time is already in simulation time steps.
  
      Returns:
          int: The number of simulation time steps.
  
      Raises:
          :exc:`ValueError`: If given *time* cannot be represented by simulator precision.
      """
      result = time
      if units is not None:
          result = _ldexp10(result, _get_log_time_scale(units) - _get_simulator_precision())
  
      result_rounded = math.floor(result)
  
      if result_rounded != result:
          raise ValueError("Unable to accurately represent {0}({1}) with the "
                           "simulator precision of 1e{2}".format(
                               time, units, _get_simulator_precision()))
  
      return int(result_rounded)
  
  
  def _get_log_time_scale(units):
      """Retrieves the ``log10()`` of the scale factor for a given time unit.
  
      Args:
          units (str): String specifying the units
              (one of ``'fs'``, ``'ps'``, ``'ns'``, ``'us'``, ``'ms'``, ``'sec'``).
  
      Returns:
          The the ``log10()`` of the scale factor for the time unit.
      """
      scale = {
          'fs' :    -15,
          'ps' :    -12,
          'ns' :     -9,
          'us' :     -6,
          'ms' :     -3,
          'sec':      0}
  
      units_lwr = units.lower()
      if units_lwr not in scale:
          raise ValueError("Invalid unit ({0}) provided".format(units))
      else:
          return scale[units_lwr]
  
  # Ctypes helper functions
  
  
  def pack(ctypes_obj):
      """Convert a :mod:`ctypes` structure into a Python string.
  
      Args:
          ctypes_obj (ctypes.Structure): The :mod:`ctypes` structure to convert to a string.
  
      Returns:
          New Python string containing the bytes from memory holding *ctypes_obj*.
      """
      return ctypes.string_at(ctypes.addressof(ctypes_obj),
                              ctypes.sizeof(ctypes_obj))
  
  
  def unpack(ctypes_obj, string, bytes=None):
      """Unpack a Python string into a :mod:`ctypes` structure.
  
      If the length of *string* is not the correct size for the memory
      footprint of the :mod:`ctypes` structure then the *bytes* keyword argument
      must be used.
  
      Args:
          ctypes_obj (ctypes.Structure): The :mod:`ctypes` structure to pack into.
          string (str):  String to copy over the *ctypes_obj* memory space.
          bytes (int, optional): Number of bytes to copy.
              Defaults to ``None``, meaning the length of *string* is used.
  
      Raises:
          :exc:`ValueError`: If length of *string* and size of *ctypes_obj*
              are not equal.
          :exc:`MemoryError`: If *bytes* is longer than size of *ctypes_obj*.
      """
      if bytes is None:
          if len(string) != ctypes.sizeof(ctypes_obj):
              raise ValueError("Attempt to unpack a string of size %d into a \
                  struct of size %d" % (len(string), ctypes.sizeof(ctypes_obj)))
          bytes = len(string)
  
      if bytes > ctypes.sizeof(ctypes_obj):
          raise MemoryError("Attempt to unpack %d bytes over an object \
                          of size %d" % (bytes, ctypes.sizeof(ctypes_obj)))
  
      ctypes.memmove(ctypes.addressof(ctypes_obj), string, bytes)
  
  
  import cocotb.ANSI as ANSI
  
  
  # A note on the use of latin1 in the deprecations below:
  # Latin1 is the only encoding `e` that satisfies
  # `all(chr(x).encode(e) == bytes([x]) for x in range(255))`
  # Our use of `ord` and `chr` throughout other bits of code make this the most
  # compatible choice of encoding. Under this convention, old code can be upgraded
  # by changing "binary" strings from `"\x12\x34"` to `b"\x12\x34"`.
  
  
  def _sane(x: bytes) -> str:
      r = ""
      for j in x:
          if (j < 32) or (j >= 127):
              r += "."
          else:
              r += chr(j)
      return r
  
  
  def hexdump(x: bytes) -> str:
      """Hexdump a buffer.
  
      Args:
          x: Object that supports conversion to :class:`bytes`.
  
      Returns:
          A string containing the hexdump.
  
      .. deprecated:: 1.4
          Passing a :class:`str` to this function is deprecated, as it
          is not an appropriate type for binary data. Doing so anyway
          will encode the string to latin1.
  
      Example:
          >>> print(hexdump(b'this somewhat long string'))
          0000   74 68 69 73 20 73 6F 6D 65 77 68 61 74 20 6C 6F   this somewhat lo
          0010   6E 67 20 73 74 72 69 6E 67                        ng string
          <BLANKLINE>
      """
      # adapted from scapy.utils.hexdump
      rs = ""
      if isinstance(x, str):
          warnings.warn(
              "Passing a string to hexdump is deprecated, pass bytes instead",
              DeprecationWarning, stacklevel=2)
          x = x.encode('latin1')
      x = b"%b" % x
      l = len(x)
      i = 0
      while i < l:
          rs += "%04x   " % i
          for j in range(16):
              if i + j < l:
                  rs += "%02X " % x[i + j]
              else:
                  rs += "   "
              if j % 16 == 7:
                  rs += ""
          rs += "  "
          rs += _sane(x[i:i + 16]) + "\n"
          i += 16
      return rs
  
  
  def hexdiffs(x: bytes, y: bytes) -> str:
      r"""Return a diff string showing differences between two binary strings.
  
      Args:
          x: Object that supports conversion to :class:`bytes`.
          y: Object that supports conversion to :class:`bytes`.
  
      .. deprecated:: 1.4
          Passing :class:`str`\ s to this function is deprecated, as it
          is not an appropriate type for binary data. Doing so anyway
          will encode the string to latin1.
  
      Example:
          >>> print(hexdiffs(b'a', b'b'))
          0000      61                                               a
               0000 62                                               b
          <BLANKLINE>
          >>> print(hexdiffs(b'this short thing', b'this also short'))
          0000      746869732073686F 7274207468696E67 this short thing
               0000 7468697320616C73 6F  2073686F7274 this also  short
          <BLANKLINE>
      """
      # adapted from scapy.utils.hexdiff
  
      def highlight(string: str, colour=ANSI.COLOR_HILITE_HEXDIFF_DEFAULT) -> str:
          """Highlight with ANSI colors if possible/requested and not running in GUI."""
  
          if want_color_output():
              return colour + string + ANSI.COLOR_DEFAULT
          else:
              return string
  
      x_is_str = isinstance(x, str)
      y_is_str = isinstance(y, str)
      if x_is_str or y_is_str:
          warnings.warn(
              "Passing strings to hexdiffs is deprecated, pass bytes instead",
              DeprecationWarning, stacklevel=2)
      if x_is_str:
          x = x.encode('latin1')
      if y_is_str:
          y = y.encode('latin1')
  
      rs = ""
  
      x = (b'%b' % x)[::-1]
      y = (b'%b' % y)[::-1]
      SUBST = 1
      INSERT = 1
      d = {}
      d[-1, -1] = 0, (-1, -1)
      for j in range(len(y)):
          d[-1, j] = d[-1, j - 1][0] + INSERT, (-1, j - 1)
      for i in range(len(x)):
          d[i, -1] = d[i - 1, -1][0] + INSERT, (i - 1, -1)
  
      for j in range(len(y)):
          for i in range(len(x)):
              d[i, j] = min((d[i-1, j-1][0] + SUBST*(x[i] != y[j]), (i-1, j-1)),
                            (d[i - 1, j][0] + INSERT, (i - 1, j)),
                            (d[i, j - 1][0] + INSERT, (i, j - 1)))
  
      backtrackx = []
      backtracky = []
      i = len(x) - 1
      j = len(y) - 1
      while not (i == j == -1):
          i2, j2 = d[i, j][1]
          backtrackx.append(x[i2+1:i+1])
          backtracky.append(y[j2+1:j+1])
          i, j = i2, j2
  
      x = y = i = 0
      colorize = { 0: lambda x: x,  # noqa
                  -1: lambda x: x,  # noqa
                   1: lambda x: x}  # noqa
  
      dox = 1
      doy = 0
      l = len(backtrackx)
      while i < l:
          linex = backtrackx[i:i+16]
          liney = backtracky[i:i+16]
          xx = sum(len(k) for k in linex)
          yy = sum(len(k) for k in liney)
          if dox and not xx:
              dox = 0
              doy = 1
          if dox and linex == liney:
              doy = 1
  
          if dox:
              xd = y
              j = 0
              while not linex[j]:
                  j += 1
                  xd -= 1
              if dox != doy:
                  rs += highlight("%04x" % xd) + " "
              else:
                  rs += highlight("%04x" % xd, colour=ANSI.COLOR_HILITE_HEXDIFF_1) + " "
              x += xx
              line = linex
          else:
              rs += "    "
          if doy:
              yd = y
              j = 0
              while not liney[j]:
                  j += 1
                  yd -= 1
              if doy - dox != 0:
                  rs += " " + highlight("%04x" % yd)
              else:
                  rs += highlight("%04x" % yd, colour=ANSI.COLOR_HILITE_HEXDIFF_1)
              y += yy
              line = liney
          else:
              rs += "    "
  
          rs += " "
  
          cl = ""
          for j in range(16):
              if i + j < l:
                  if line[j]:
                      char_j, = line[j]
                      if linex[j] != liney[j]:
                          rs += highlight("%02X" % char_j,
                                          colour=ANSI.COLOR_HILITE_HEXDIFF_2)
                      else:
                          rs += "%02X" % char_j
                      if linex[j] == liney[j]:
                          cl += highlight(_sane(line[j]),
                                          colour=ANSI.COLOR_HILITE_HEXDIFF_3)
                      else:
                          cl += highlight(_sane(line[j]),
                                          colour=ANSI.COLOR_HILITE_HEXDIFF_4)
                  else:
                      rs += "  "
                      cl += " "
              else:
                  rs += "   "
              if j == 7:
                  rs += " "
  
          rs += " " + cl + '\n'
  
          if doy or not yy:
              doy = 0
              dox = 1
              i += 16
          else:
              if yy:
                  dox = 0
                  doy = 1
              else:
                  i += 16
      return rs
  
  
  class ParametrizedSingleton(type):
      """A metaclass that allows class construction to reuse an existing instance.
  
      We use this so that :class:`RisingEdge(sig) <cocotb.triggers.RisingEdge>` and :class:`Join(coroutine) <cocotb.triggers.Join>` always return
      the same instance, rather than creating new copies.
      """
  
      def __init__(cls, *args, **kwargs):
          # Attach a lookup table to this class.
          # Weak such that if the instance is no longer referenced, it can be
          # collected.
          cls.__instances = weakref.WeakValueDictionary()
  
      def __singleton_key__(cls, *args, **kwargs):
          """Convert the construction arguments into a normalized representation that
          uniquely identifies this singleton.
          """
          # Could default to something like this, but it would be slow
          # return tuple(inspect.Signature(cls).bind(*args, **kwargs).arguments.items())
          raise NotImplementedError
  
      def __call__(cls, *args, **kwargs):
          key = cls.__singleton_key__(*args, **kwargs)
          try:
              return cls.__instances[key]
          except KeyError:
              # construct the object as normal
              self = super(ParametrizedSingleton, cls).__call__(*args, **kwargs)
              cls.__instances[key] = self
              return self
  
  
  def reject_remaining_kwargs(name, kwargs):
      """
      Helper function to emulate Python 3 keyword-only arguments.
  
      Use as::
  
          def func(x1, **kwargs):
              a = kwargs.pop('a', 1)
              b = kwargs.pop('b', 2)
              reject_remaining_kwargs('func', kwargs)
              ...
  
      To emulate the Python 3 syntax::
  
          def func(x1, *, a=1, b=2):
              ...
  
      .. deprecated:: 1.4
          Since the minimum supported Python version is now 3.5, this function
          is not needed.
      """
      warnings.warn(
          "reject_remaining_kwargs is deprecated and will be removed, use "
          "Python 3 keyword-only arguments directly.", DeprecationWarning,
          stacklevel=2)
      if kwargs:
          # match the error message to what Python 3 produces
          bad_arg = next(iter(kwargs))
          raise TypeError(
              '{}() got an unexpected keyword argument {!r}'.format(name, bad_arg)
          )
  
  
  class lazy_property:
      """
      A property that is executed the first time, then cached forever.
  
      It does this by replacing itself on the instance, which works because
      unlike `@property` it does not define __set__.
  
      This should be used for expensive members of objects that are not always
      used.
      """
  
      def __init__(self, fget):
          self.fget = fget
  
          # copy the getter function's docstring and other attributes
          functools.update_wrapper(self, fget)
  
      def __get__(self, obj, cls):
          if obj is None:
              return self
  
          value = self.fget(obj)
          setattr(obj, self.fget.__qualname__, value)
          return value
  
  
  def want_color_output():
      """Return ``True`` if colored output is possible/requested and not running in GUI.
  
      Colored output can be explicitly requested by setting :envvar:`COCOTB_ANSI_OUTPUT` to  ``1``.
      """
      want_color = sys.stdout.isatty()  # default to color for TTYs
      if os.getenv("NO_COLOR") is not None:
          want_color = False
      if os.getenv("COCOTB_ANSI_OUTPUT", default='0') == '1':
          want_color = True
      if os.getenv("GUI", default='0') == '1':
          want_color = False
      return want_color
  
  
  if __name__ == "__main__":
      import random
      a = ""
      for char in range(random.randint(250, 500)):
          a += chr(random.randint(0, 255))
      b = a
      for error in range(random.randint(2, 9)):
          offset = random.randint(0, len(a))
          b = b[:offset] + chr(random.randint(0, 255)) + b[offset+1:]
  
      diff = hexdiffs(a, b)
      print(diff)
  
      space = '\n' + (" " * 20)
      print(space.join(diff.split('\n')))
  
  
  def remove_traceback_frames(tb_or_exc, frame_names):
      """
      Strip leading frames from a traceback
  
      Args:
          tb_or_exc (Union[traceback, BaseException, exc_info]):
              Object to strip frames from. If an exception is passed, creates
              a copy of the exception with a new shorter traceback. If a tuple
              from `sys.exc_info` is passed, returns the same tuple with the
              traceback shortened
          frame_names (List[str]):
              Names of the frames to strip, which must be present.
      """
      # self-invoking overloads
      if isinstance(tb_or_exc, BaseException):
          exc = tb_or_exc
          return exc.with_traceback(
              remove_traceback_frames(exc.__traceback__, frame_names)
          )
      elif isinstance(tb_or_exc, tuple):
          exc_type, exc_value, exc_tb = tb_or_exc
          exc_tb = remove_traceback_frames(exc_tb, frame_names)
          return exc_type, exc_value, exc_tb
      # base case
      else:
          tb = tb_or_exc
          for frame_name in frame_names:
              assert tb.tb_frame.f_code.co_name == frame_name
              tb = tb.tb_next
          return tb
  
files.old_path: cocotb/utils.py