files: #!/usr/bin/env python
  
  # Copyright (c) 2013, 2018 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.
  
  """Coroutine scheduler.
  
  
  FIXME: We have a problem here.  If a coroutine schedules a read-only but we
  also have pending writes we have to schedule the ReadWrite callback before
  the ReadOnly (and this is invalid, at least in Modelsim).
  """
  import collections
  import copy
  import os
  import time
  import logging
  import threading
  
  
  # For autodocumentation don't need the extension modules
  if "SPHINX_BUILD" in os.environ:
      simulator = None
  else:
      import simulator
  
  # Debug mode controlled by environment variables
  if "COCOTB_ENABLE_PROFILING" in os.environ:
      import cProfile, StringIO, pstats
      _profile = cProfile.Profile()
      _profiling = True
  else:
      _profiling = False
  
  # 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
  
  
  import cocotb
  import cocotb.decorators
  from cocotb.triggers import (Trigger, GPITrigger, Timer, ReadOnly, PythonTrigger,
                               NextTimeStep, ReadWrite, Event, Join)
  from cocotb.log import SimLog
  from cocotb.result import (TestComplete, TestError, ReturnValue, raise_error,
                             create_error, ExternalException)
  from cocotb.utils import nullcontext
  
  
  class profiling_context(object):
      """ Context manager that profiles its contents """
      def __enter__(self):
          _profile.enable()
  
      def __exit__(self, *excinfo):
          _profile.disable()
  
  
  class external_state(object):
      INIT = 0
      RUNNING = 1
      PAUSED = 2
      EXITED = 3
  
  @cocotb.decorators.public
  class external_waiter(object):
  
      def __init__(self):
          self.result = None
          self.thread = None
          self.event = Event()
          self.state = external_state.INIT
          self.cond = threading.Condition()
          self._log = SimLog("cocotb.external.thead.%s" % self.thread, id(self))
  
      def _propogate_state(self, new_state):
          self.cond.acquire()
          if _debug:
              self._log.debug("Changing state from %d -> %d from %s" % (self.state, new_state, threading.current_thread()))
          self.state = new_state
          self.cond.notify()
          self.cond.release()
  
      def thread_done(self):
          if _debug:
              self._log.debug("Thread finished from %s" % (threading.current_thread()))
          self._propogate_state(external_state.EXITED)
  
      def thread_suspend(self):
          self._propogate_state(external_state.PAUSED)
  
      def thread_start(self):
          if self.state > external_state.INIT:
              return
  
          if not self.thread.is_alive():
              self._propogate_state(external_state.RUNNING)
              self.thread.start()
  
      def thread_resume(self):
          self._propogate_state(external_state.RUNNING)
  
      def thread_wait(self):
          if _debug:
              self._log.debug("Waiting for the condition lock %s" % threading.current_thread())
  
          self.cond.acquire()
  
          while self.state == external_state.RUNNING:
              self.cond.wait()
  
          if _debug:
              if self.state == external_state.EXITED:
                  self._log.debug("Thread %s has exited from %s" % (self.thread, threading.current_thread()))
              elif self.state == external_state.PAUSED:
                  self._log.debug("Thread %s has called yield from %s"  % (self.thread, threading.current_thread()))
              elif self.state == external_state.RUNNING:
                  self._log.debug("Thread %s is in RUNNING from %d"  % (self.thread, threading.current_thread()))
  
          if self.state == external_state.INIT:
              raise Exception("Thread %s state was not allowed from %s"  % (self.thread, threading.current_thread()))
  
          self.cond.release()
          return self.state
  
  class Scheduler(object):
      """The main scheduler.
  
      Here we accept callbacks from the simulator and schedule the appropriate
      coroutines.
  
      A callback fires, causing the :any:`react` method to be called, with the
      trigger that caused the callback as the first argument.
  
      We look up a list of coroutines to schedule (indexed by the trigger) and
      schedule them in turn. NB implementors should not depend on the scheduling
      order!
  
      Some additional management is required since coroutines can return a list
      of triggers, to be scheduled when any one of the triggers fires.  To
      ensure we don't receive spurious callbacks, we have to un-prime all the
      other triggers when any one fires.
  
      Due to the simulator nuances and fun with delta delays we have the
      following modes:
  
      Normal mode
          - Callbacks cause coroutines to be scheduled
          - Any pending writes are cached and do not happen immediately
  
      ReadOnly mode
          - Corresponds to cbReadOnlySynch (VPI) or vhpiCbLastKnownDeltaCycle
            (VHPI).  In this state we are not allowed to perform writes.
  
      Write mode
          - Corresponds to cbReadWriteSynch (VPI) or vhpiCbEndOfProcesses (VHPI)
            In this mode we play back all the cached write updates.
  
      We can legally transition from normal->write by registering a ReadWrite
      callback, however usually once a simulator has entered the ReadOnly phase
      of a given timestep then we must move to a new timestep before performing
      any writes.  The mechanism for moving to a new timestep may not be
      consistent across simulators and therefore we provide an abstraction to
      assist with compatibility.
  
  
      Unless a coroutine has explicitly requested to be scheduled in ReadOnly
      mode (for example wanting to sample the finally settled value after all
      delta delays) then it can reasonably be expected to be scheduled during
      "normal mode" i.e. where writes are permitted.
      """
  
      _MODE_NORMAL   = 1  # noqa
      _MODE_READONLY = 2  # noqa
      _MODE_WRITE    = 3  # noqa
      _MODE_TERM     = 4  # noqa
  
      # Singleton events, recycled to avoid spurious object creation
      _readonly = ReadOnly()
      # TODO[gh-759]: For some reason, the scheduler requires that these triggers
      # are _not_ the same instances used by the tests themselves. This is risky,
      # because it can lead to them overwriting each other's callbacks. We should
      # try to remove this `copy.copy` in future.
      _next_timestep = copy.copy(NextTimeStep())
      _readwrite = copy.copy(ReadWrite())
      _timer1 = Timer(1)
      _timer0 = Timer(0)
  
      def __init__(self):
  
          self.log = SimLog("cocotb.scheduler")
          if _debug:
              self.log.setLevel(logging.DEBUG)
  
          # A dictionary of pending coroutines for each trigger,
          # indexed by trigger
          self._trigger2coros = collections.defaultdict(list)
  
          # A dictionary of pending triggers for each coroutine, indexed by coro
          self._coro2triggers = collections.defaultdict(list)
  
          # Our main state
          self._mode = Scheduler._MODE_NORMAL
  
          # A dictionary of pending writes
          self._writes = {}
  
          self._pending_coros = []
          self._pending_callbacks = []
          self._pending_triggers = []
          self._pending_threads = []
          self._pending_events = []   # Events we need to call set on once we've unwound
  
          self._terminate = False
          self._test_result = None
          self._entrypoint = None
          self._main_thread = threading.current_thread()
  
          # Select the appropriate scheduling algorithm for this simulator
          self.advance = self.default_scheduling_algorithm
  
      def default_scheduling_algorithm(self):
          """
          Decide whether we need to schedule our own triggers (if at all) in
          order to progress to the next mode.
  
          This algorithm has been tested against the following simulators:
              Icarus Verilog
          """
          if not self._terminate and self._writes:
  
              if self._mode == Scheduler._MODE_NORMAL:
                  if not self._readwrite.primed:
                      self._readwrite.prime(self.react)
              elif not self._next_timestep.primed:
                  self._next_timestep.prime(self.react)
  
          elif self._terminate:
              if _debug:
                  self.log.debug("Test terminating, scheduling Timer")
  
              for t in self._trigger2coros:
                  t.unprime()
  
              for t in [self._readwrite, self._readonly, self._next_timestep,
                        self._timer1, self._timer0]:
                  if t.primed:
                      t.unprime()
  
              self._timer1.prime(self.begin_test)
              self._trigger2coros = collections.defaultdict(list)
              self._coro2triggers = collections.defaultdict(list)
              self._terminate = False
              self._mode = Scheduler._MODE_TERM
  
      def begin_test(self, trigger=None):
          """Called to initiate a test.
  
          Could be called on start-up or from a callback.
          """
          if _debug:
              self.log.debug("begin_test called with trigger: %s" %
                             (str(trigger)))
          if _profiling:
              ps = pstats.Stats(_profile).sort_stats('cumulative')
              ps.dump_stats("test_profile.pstat")
              ctx = profiling_context()
          else:
              ctx = nullcontext()
  
          with ctx:
              self._mode = Scheduler._MODE_NORMAL
              if trigger is not None:
                  trigger.unprime()
  
              # Issue previous test result, if there is one
              if self._test_result is not None:
                  if _debug:
                      self.log.debug("Issue test result to regression object")
                  cocotb.regression.handle_result(self._test_result)
                  self._test_result = None
              if self._entrypoint is not None:
                  test = self._entrypoint
                  self._entrypoint = None
                  self.schedule(test)
                  self.advance()
  
      def react(self, trigger, depth=0):
          """React called when a trigger fires.
  
          We find any coroutines that are waiting on the particular trigger and
          schedule them.
          """
          if _profiling and not depth:
              ctx = profiling_context()
          else:
              ctx = nullcontext()
  
          with ctx:
              # When a trigger fires it is unprimed internally
              if _debug:
                  self.log.debug("Trigger fired: %s" % str(trigger))
              # trigger.unprime()
  
              if self._mode == Scheduler._MODE_TERM:
                  if _debug:
                      self.log.debug("Ignoring trigger %s since we're terminating" %
                                     str(trigger))
                  return
  
              if trigger is self._readonly:
                  self._mode = Scheduler._MODE_READONLY
              # Only GPI triggers affect the simulator scheduling mode
              elif isinstance(trigger, GPITrigger):
                  self._mode = Scheduler._MODE_NORMAL
  
              # We're the only source of ReadWrite triggers which are only used for
              # playing back any cached signal updates
              if trigger is self._readwrite:
  
                  if _debug:
                      self.log.debug("Writing cached signal updates")
  
                  while self._writes:
                      handle, value = self._writes.popitem()
                      handle.setimmediatevalue(value)
  
                  self._readwrite.unprime()
  
                  return
  
              # Similarly if we've scheduled our next_timestep on way to readwrite
              if trigger is self._next_timestep:
  
                  if not self._writes:
                      self.log.error(
                          "Moved to next timestep without any pending writes!")
                  else:
                      self.log.debug(
                          "Priming ReadWrite trigger so we can playback writes")
                      self._readwrite.prime(self.react)
  
                  return
  
              if trigger not in self._trigger2coros:
  
                  # GPI triggers should only be ever pending if there is an
                  # associated coroutine waiting on that trigger, otherwise it would
                  # have been unprimed already
                  if isinstance(trigger, GPITrigger):
                      self.log.critical(
                          "No coroutines waiting on trigger that fired: %s" %
                          str(trigger))
  
                      trigger.log.info("I'm the culprit")
                  # For Python triggers this isn't actually an error - we might do
                  # event.set() without knowing whether any coroutines are actually
                  # waiting on this event, for example
                  elif _debug:
                      self.log.debug(
                          "No coroutines waiting on trigger that fired: %s" %
                          str(trigger))
  
                  return
  
              # Scheduled coroutines may append to our waiting list so the first
              # thing to do is pop all entries waiting on this trigger.
              scheduling = self._trigger2coros.pop(trigger)
  
              if _debug:
                  debugstr = "\n\t".join([coro.__name__ for coro in scheduling])
                  if len(scheduling):
                      debugstr = "\n\t" + debugstr
                  self.log.debug("%d pending coroutines for event %s%s" %
                                 (len(scheduling), str(trigger), debugstr))
  
              # This trigger isn't needed any more
              trigger.unprime()
  
              # If the coroutine was waiting on multiple triggers we may be able
              # to unprime the other triggers that didn't fire
              scheduling_set = set(scheduling)
              other_triggers = {
                  t
                  for coro in scheduling
                  for t in self._coro2triggers[coro]
              } - {trigger}
  
              for pending in other_triggers:
                  # every coroutine waiting on this trigger is already being woken
                  if scheduling_set.issuperset(self._trigger2coros[pending]):
                      if pending.primed:
                          pending.unprime()
                      del self._trigger2coros[pending]
  
              for coro in scheduling:
                  if _debug:
                      self.log.debug("Scheduling coroutine %s" % (coro.__name__))
                  self.schedule(coro, trigger=trigger)
                  if _debug:
                      self.log.debug("Scheduled coroutine %s" % (coro.__name__))
  
              if not depth:
                  # Schedule may have queued up some events so we'll burn through those
                  while self._pending_events:
                      if _debug:
                          self.log.debug("Scheduling pending event %s" %
                                         (str(self._pending_events[0])))
                      self._pending_events.pop(0).set()
  
              while self._pending_triggers:
                  if _debug:
                      self.log.debug("Scheduling pending trigger %s" %
                                     (str(self._pending_triggers[0])))
                  self.react(self._pending_triggers.pop(0), depth=depth + 1)
  
              # We only advance for GPI triggers
              if not depth and isinstance(trigger, GPITrigger):
                  self.advance()
  
                  if _debug:
                      self.log.debug("All coroutines scheduled, handing control back"
                                     " to simulator")
  
  
      def unschedule(self, coro):
          """Unschedule a coroutine.  Unprime any pending triggers"""
  
          for trigger in self._coro2triggers[coro]:
              if coro in self._trigger2coros[trigger]:
                  self._trigger2coros[trigger].remove(coro)
              if not self._trigger2coros[trigger]:
                  trigger.unprime()
                  del self._trigger2coros[trigger]
          del self._coro2triggers[coro]
  
          if Join(coro) in self._trigger2coros:
              self._pending_triggers.append(Join(coro))
  
  
      def save_write(self
  
  [... the rest is omitted ...]
files: cocotb/scheduler.py
files: #!/usr/bin/env python
  
  # Copyright (c) 2013, 2018 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.
  
  """Coroutine scheduler.
  
  
  FIXME: We have a problem here.  If a coroutine schedules a read-only but we
  also have pending writes we have to schedule the ReadWrite callback before
  the ReadOnly (and this is invalid, at least in Modelsim).
  """
  import collections
  import copy
  import os
  import time
  import logging
  import threading
  
  
  # For autodocumentation don't need the extension modules
  if "SPHINX_BUILD" in os.environ:
      simulator = None
  else:
      import simulator
  
  # Debug mode controlled by environment variables
  if "COCOTB_ENABLE_PROFILING" in os.environ:
      import cProfile, StringIO, pstats
      _profile = cProfile.Profile()
      _profiling = True
  else:
      _profiling = False
  
  # 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
  
  
  import cocotb
  import cocotb.decorators
  from cocotb.triggers import (Trigger, GPITrigger, Timer, ReadOnly, PythonTrigger,
                               NextTimeStep, ReadWrite, Event, Join)
  from cocotb.log import SimLog
  from cocotb.result import (TestComplete, TestError, ReturnValue, raise_error,
                             create_error, ExternalException)
  
  class external_state(object):
      INIT = 0
      RUNNING = 1
      PAUSED = 2
      EXITED = 3
  
  @cocotb.decorators.public
  class external_waiter(object):
  
      def __init__(self):
          self.result = None
          self.thread = None
          self.event = Event()
          self.state = external_state.INIT
          self.cond = threading.Condition()
          self._log = SimLog("cocotb.external.thead.%s" % self.thread, id(self))
  
      def _propogate_state(self, new_state):
          self.cond.acquire()
          if _debug:
              self._log.debug("Changing state from %d -> %d from %s" % (self.state, new_state, threading.current_thread()))
          self.state = new_state
          self.cond.notify()
          self.cond.release()
  
      def thread_done(self):
          if _debug:
              self._log.debug("Thread finished from %s" % (threading.current_thread()))
          self._propogate_state(external_state.EXITED)
  
      def thread_suspend(self):
          self._propogate_state(external_state.PAUSED)
  
      def thread_start(self):
          if self.state > external_state.INIT:
              return
  
          if not self.thread.is_alive():
              self._propogate_state(external_state.RUNNING)
              self.thread.start()
  
      def thread_resume(self):
          self._propogate_state(external_state.RUNNING)
  
      def thread_wait(self):
          if _debug:
              self._log.debug("Waiting for the condition lock %s" % threading.current_thread())
  
          self.cond.acquire()
  
          while self.state == external_state.RUNNING:
              self.cond.wait()
  
          if _debug:
              if self.state == external_state.EXITED:
                  self._log.debug("Thread %s has exited from %s" % (self.thread, threading.current_thread()))
              elif self.state == external_state.PAUSED:
                  self._log.debug("Thread %s has called yield from %s"  % (self.thread, threading.current_thread()))
              elif self.state == external_state.RUNNING:
                  self._log.debug("Thread %s is in RUNNING from %d"  % (self.thread, threading.current_thread()))
  
          if self.state == external_state.INIT:
              raise Exception("Thread %s state was not allowed from %s"  % (self.thread, threading.current_thread()))
  
          self.cond.release()
          return self.state
  
  class Scheduler(object):
      """The main scheduler.
  
      Here we accept callbacks from the simulator and schedule the appropriate
      coroutines.
  
      A callback fires, causing the :any:`react` method to be called, with the
      trigger that caused the callback as the first argument.
  
      We look up a list of coroutines to schedule (indexed by the trigger) and
      schedule them in turn. NB implementors should not depend on the scheduling
      order!
  
      Some additional management is required since coroutines can return a list
      of triggers, to be scheduled when any one of the triggers fires.  To
      ensure we don't receive spurious callbacks, we have to un-prime all the
      other triggers when any one fires.
  
      Due to the simulator nuances and fun with delta delays we have the
      following modes:
  
      Normal mode
          - Callbacks cause coroutines to be scheduled
          - Any pending writes are cached and do not happen immediately
  
      ReadOnly mode
          - Corresponds to cbReadOnlySynch (VPI) or vhpiCbLastKnownDeltaCycle
            (VHPI).  In this state we are not allowed to perform writes.
  
      Write mode
          - Corresponds to cbReadWriteSynch (VPI) or vhpiCbEndOfProcesses (VHPI)
            In this mode we play back all the cached write updates.
  
      We can legally transition from normal->write by registering a ReadWrite
      callback, however usually once a simulator has entered the ReadOnly phase
      of a given timestep then we must move to a new timestep before performing
      any writes.  The mechanism for moving to a new timestep may not be
      consistent across simulators and therefore we provide an abstraction to
      assist with compatibility.
  
  
      Unless a coroutine has explicitly requested to be scheduled in ReadOnly
      mode (for example wanting to sample the finally settled value after all
      delta delays) then it can reasonably be expected to be scheduled during
      "normal mode" i.e. where writes are permitted.
      """
  
      _MODE_NORMAL   = 1  # noqa
      _MODE_READONLY = 2  # noqa
      _MODE_WRITE    = 3  # noqa
      _MODE_TERM     = 4  # noqa
  
      # Singleton events, recycled to avoid spurious object creation
      _readonly = ReadOnly()
      # TODO[gh-759]: For some reason, the scheduler requires that these triggers
      # are _not_ the same instances used by the tests themselves. This is risky,
      # because it can lead to them overwriting each other's callbacks. We should
      # try to remove this `copy.copy` in future.
      _next_timestep = copy.copy(NextTimeStep())
      _readwrite = copy.copy(ReadWrite())
      _timer1 = Timer(1)
      _timer0 = Timer(0)
  
      def __init__(self):
  
          self.log = SimLog("cocotb.scheduler")
          if _debug:
              self.log.setLevel(logging.DEBUG)
  
          # A dictionary of pending coroutines for each trigger,
          # indexed by trigger
          self._trigger2coros = collections.defaultdict(list)
  
          # A dictionary of pending triggers for each coroutine, indexed by coro
          self._coro2triggers = collections.defaultdict(list)
  
          # Our main state
          self._mode = Scheduler._MODE_NORMAL
  
          # A dictionary of pending writes
          self._writes = {}
  
          self._pending_coros = []
          self._pending_callbacks = []
          self._pending_triggers = []
          self._pending_threads = []
          self._pending_events = []   # Events we need to call set on once we've unwound
  
          self._terminate = False
          self._test_result = None
          self._entrypoint = None
          self._main_thread = threading.current_thread()
  
          # Select the appropriate scheduling algorithm for this simulator
          self.advance = self.default_scheduling_algorithm
  
      def default_scheduling_algorithm(self):
          """
          Decide whether we need to schedule our own triggers (if at all) in
          order to progress to the next mode.
  
          This algorithm has been tested against the following simulators:
              Icarus Verilog
          """
          if not self._terminate and self._writes:
  
              if self._mode == Scheduler._MODE_NORMAL:
                  if not self._readwrite.primed:
                      self._readwrite.prime(self.react)
              elif not self._next_timestep.primed:
                  self._next_timestep.prime(self.react)
  
          elif self._terminate:
              if _debug:
                  self.log.debug("Test terminating, scheduling Timer")
  
              for t in self._trigger2coros:
                  t.unprime()
  
              for t in [self._readwrite, self._readonly, self._next_timestep,
                        self._timer1, self._timer0]:
                  if t.primed:
                      t.unprime()
  
              self._timer1.prime(self.begin_test)
              self._trigger2coros = collections.defaultdict(list)
              self._coro2triggers = collections.defaultdict(list)
              self._terminate = False
              self._mode = Scheduler._MODE_TERM
  
      def begin_test(self, trigger=None):
          """Called to initiate a test.
  
          Could be called on start-up or from a callback.
          """
          if _debug:
              self.log.debug("begin_test called with trigger: %s" %
                             (str(trigger)))
          if _profiling:
              ps = pstats.Stats(_profile).sort_stats('cumulative')
              ps.dump_stats("test_profile.pstat")
              _profile.enable()
  
          self._mode = Scheduler._MODE_NORMAL
          if trigger is not None:
              trigger.unprime()
  
          # Issue previous test result, if there is one
          if self._test_result is not None:
              if _debug:
                  self.log.debug("Issue test result to regression object")
              cocotb.regression.handle_result(self._test_result)
              self._test_result = None
          if self._entrypoint is not None:
              test = self._entrypoint
              self._entrypoint = None
              self.schedule(test)
              self.advance()
  
          if _profiling:
              _profile.disable()
  
      def react(self, trigger, depth=0):
          """React called when a trigger fires.
  
          We find any coroutines that are waiting on the particular trigger and
          schedule them.
          """
          if _profiling and not depth:
              _profile.enable()
  
          # When a trigger fires it is unprimed internally
          if _debug:
              self.log.debug("Trigger fired: %s" % str(trigger))
          # trigger.unprime()
  
          if self._mode == Scheduler._MODE_TERM:
              if _debug:
                  self.log.debug("Ignoring trigger %s since we're terminating" %
                                 str(trigger))
              return
  
          if trigger is self._readonly:
              self._mode = Scheduler._MODE_READONLY
          # Only GPI triggers affect the simulator scheduling mode
          elif isinstance(trigger, GPITrigger):
              self._mode = Scheduler._MODE_NORMAL
  
          # We're the only source of ReadWrite triggers which are only used for
          # playing back any cached signal updates
          if trigger is self._readwrite:
  
              if _debug:
                  self.log.debug("Writing cached signal updates")
  
              while self._writes:
                  handle, value = self._writes.popitem()
                  handle.setimmediatevalue(value)
  
              self._readwrite.unprime()
  
              if _profiling:
                  _profile.disable()
              return
  
          # Similarly if we've scheduled our next_timestep on way to readwrite
          if trigger is self._next_timestep:
  
              if not self._writes:
                  self.log.error(
                      "Moved to next timestep without any pending writes!")
              else:
                  self.log.debug(
                      "Priming ReadWrite trigger so we can playback writes")
                  self._readwrite.prime(self.react)
  
              if _profiling:
                  _profile.disable()
              return
  
          if trigger not in self._trigger2coros:
  
              # GPI triggers should only be ever pending if there is an
              # associated coroutine waiting on that trigger, otherwise it would
              # have been unprimed already
              if isinstance(trigger, GPITrigger):
                  self.log.critical(
                      "No coroutines waiting on trigger that fired: %s" %
                      str(trigger))
  
                  trigger.log.info("I'm the culprit")
              # For Python triggers this isn't actually an error - we might do
              # event.set() without knowing whether any coroutines are actually
              # waiting on this event, for example
              elif _debug:
                  self.log.debug(
                      "No coroutines waiting on trigger that fired: %s" %
                      str(trigger))
  
              if _profiling:
                  _profile.disable()
              return
  
          # Scheduled coroutines may append to our waiting list so the first
          # thing to do is pop all entries waiting on this trigger.
          scheduling = self._trigger2coros.pop(trigger)
  
          if _debug:
              debugstr = "\n\t".join([coro.__name__ for coro in scheduling])
              if len(scheduling):
                  debugstr = "\n\t" + debugstr
              self.log.debug("%d pending coroutines for event %s%s" %
                             (len(scheduling), str(trigger), debugstr))
  
          # This trigger isn't needed any more
          trigger.unprime()
  
          # If the coroutine was waiting on multiple triggers we may be able
          # to unprime the other triggers that didn't fire
          scheduling_set = set(scheduling)
          other_triggers = {
              t
              for coro in scheduling
              for t in self._coro2triggers[coro]
          } - {trigger}
  
          for pending in other_triggers:
              # every coroutine waiting on this trigger is already being woken
              if scheduling_set.issuperset(self._trigger2coros[pending]):
                  if pending.primed:
                      pending.unprime()
                  del self._trigger2coros[pending]
  
          for coro in scheduling:
              if _debug:
                  self.log.debug("Scheduling coroutine %s" % (coro.__name__))
              self.schedule(coro, trigger=trigger)
              if _debug:
                  self.log.debug("Scheduled coroutine %s" % (coro.__name__))
  
          if not depth:
              # Schedule may have queued up some events so we'll burn through those
              while self._pending_events:
                  if _debug:
                      self.log.debug("Scheduling pending event %s" %
                                     (str(self._pending_events[0])))
                  self._pending_events.pop(0).set()
  
          while self._pending_triggers:
              if _debug:
                  self.log.debug("Scheduling pending trigger %s" %
                                 (str(self._pending_triggers[0])))
              self.react(self._pending_triggers.pop(0), depth=depth + 1)
  
          # We only advance for GPI triggers
          if not depth and isinstance(trigger, GPITrigger):
              self.advance()
  
              if _debug:
                  self.log.debug("All coroutines scheduled, handing control back"
                                 " to simulator")
  
              if _profiling:
                  _profile.disable()
          return
  
      def unschedule(self, coro):
          """Unschedule a coroutine.  Unprime any pending triggers"""
  
          for trigger in self._coro2triggers[coro]:
              if coro in self._trigger2coros[trigger]:
                  self._trigger2coros[trigger].remove(coro)
              if not self._trigger2coros[trigger]:
                  trigger.unprime()
                  del self._trigger2coros[trigger]
          del self._coro2triggers[coro]
  
          if Join(coro) in self._trigger2coros:
              self._pending_triggers.append(Join(coro))
  
  
      def save_write(self, handle, value):
          if self._mode == Scheduler._MODE_
  
  [... the rest is omitted ...]
files: cocotb/scheduler.py
files: from __future__ import print_function
  
  # Copyright (c) 2013, 2018 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
  
  # For autodocumentation don't need the extension modules
  if "SPHINX_BUILD" in os.environ:
      simulator = None
      _LOG_SIM_PRECISION = -15
  else:
      import simulator
      _LOG_SIM_PRECISION = simulator.get_precision() # request once and cache
  
  # python2 to python3 helper functions
  def get_python_integer_types():
      try:
          isinstance(1, long)
      except NameError:
          return (int,)  # python 3
      else:
          return (int, long)  # python 2
  
  # 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 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.
      """
      result = steps * (10.0**(_LOG_SIM_PRECISION - _get_log_time_scale(units)))
  
      return result
  
  def get_sim_steps(time, units=None):
      """Calculates the number of simulation time steps for a given amount of *time*.
  
      Args:
          time (int or float):  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 = result * (10.0**(_get_log_time_scale(units) - _LOG_SIM_PRECISION))
  
      err = int(result) - math.ceil(result)
  
      if err:
          raise ValueError("Unable to accurately represent {0}({1}) with the "
                           "simulator precision of 1e{2}".format(
                               time, units, _LOG_SIM_PRECISION))
  
      return int(result)
  
  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 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 ctypes structure then the *bytes* keyword argument 
      must be used.
  
      Args:
          ctypes_obj (ctypes.Structure): The 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
  
  
  def _sane_color(x):
      r = ""
      for i in x:
          j = ord(i)
          if (j < 32) or (j >= 127):
              r += "."
          else:
              r += i
      return r
  
  
  def hexdump(x):
      """Hexdump a buffer.
  
      Args:
          x: Object that supports conversion via the ``str`` built-in.
  
      Returns:
          A string containing the hexdump.
  
      Example:
  
      .. code-block:: python
  
          print(hexdump('this somewhat long string'))
  
      .. code-block:: none
  
          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
      """
      # adapted from scapy.utils.hexdump
      rs = ""
      x = str(x)
      l = len(x)
      i = 0
      while i < l:
          rs += "%04x   " % i
          for j in range(16):
              if i + j < l:
                  rs += "%02X " % ord(x[i + j])
              else:
                  rs += "   "
              if j % 16 == 7:
                  rs += ""
          rs += "  "
          rs += _sane_color(x[i:i + 16]) + "\n"
          i += 16
      return rs
  
  
  def hexdiffs(x, y):
      """Return a diff string showing differences between two binary strings.
  
      Args:
          x: Object that supports conversion via the ``str`` built-in.
          y: Object that supports conversion via the ``str`` built-in.
  
      Example:
  
      .. code-block:: python
  
          print(hexdiffs('this short thing', 'this also short'))
  
      .. code-block:: none
  
          0000      746869732073686F 7274207468696E67 this short thing
               0000 7468697320616C73 6F  2073686F7274 this also  short
  
      """
      # adapted from scapy.utils.hexdiff
  
      def sane(x):
          r = ""
          for i in x:
              j = ord(i)
              if (j < 32) or (j >= 127):
                  r = r + "."
              else:
                  r = r + i
          return r
  
      def highlight(string, colour=ANSI.COLOR_HILITE_HEXDIFF_DEFAULT):
          """Highlight only with ANSI output if it's requested and we are not in a GUI."""
          
          want_ansi = os.getenv("COCOTB_ANSI_OUTPUT") and not os.getenv("GUI")
          if want_ansi is None:
              want_ansi = sys.stdout.isatty()  # default to ANSI for TTYs
          else:
              want_ansi = want_ansi == '1'
  
          if want_ansi:
              return colour + string + ANSI.COLOR_DEFAULT
          else:
              return string
  
      rs = ""
  
      x = str(x)[::-1]
      y = str(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:
          separate = 0
          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]:
                      if linex[j] != liney[j]:
                          rs += highlight("%02X" % ord(line[j]),
                                          colour=ANSI.COLOR_HILITE_HEXDIFF_2)
                      else:
                          rs += "%02X" % ord(line[j])
                      if linex[j] == liney[j]:
                          cl += highlight(_sane_color(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
  
  
  # This is essentially six.exec_
  if sys.version_info.major == 3:
      # this has to not be a syntax error in py2
      import builtins
      exec_ = getattr(builtins, 'exec')
  else:
      # this has to not be a syntax error in py3
      def exec_(_code_, _globs_=None, _locs_=None):
          """Execute code in a namespace."""
          if _globs_ is None:
              frame = sys._getframe(1)
              _globs_ = frame.f_globals
              if _locs_ is None:
                  _locs_ = frame.f_locals
              del frame
          elif _locs_ is None:
              _locs_ = _globs_
          exec("""exec _code_ in _globs_, _locs_""")
  
  
  # this is six.with_metaclass, with a clearer docstring
  def with_metaclass(meta, *bases):
      """This provides:
  
      .. code-block:: python
  
          class Foo(with_metaclass(Meta, Base1, Base2)): pass
  
      which is a unifying syntax for:
  
      .. code-block:: python
  
          # python 3
          class Foo(Base1, Base2, metaclass=Meta): pass
  
          # python 2
          class Foo(Base1, Base2):
              __metaclass__ = Meta
      """
      # This requires a bit of explanation: the basic idea is to make a dummy
      # metaclass for one level of class instantiation that replaces itself with
      # the actual metaclass.
      class metaclass(type):
  
          def __new__(cls, name, this_bases, d):
              return meta(name, bases, d)
  
          @classmethod
          def __prepare__(cls, name, this_bases):
              return meta.__prepare__(name, bases)
      return type.__new__(metaclass, 'temporary_class', (), {})
  
  
  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, *
  
  [... the rest is omitted ...]
files: cocotb/utils.py
files: from __future__ import print_function
  
  # Copyright (c) 2013, 2018 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
  
  # For autodocumentation don't need the extension modules
  if "SPHINX_BUILD" in os.environ:
      simulator = None
      _LOG_SIM_PRECISION = -15
  else:
      import simulator
      _LOG_SIM_PRECISION = simulator.get_precision() # request once and cache
  
  # python2 to python3 helper functions
  def get_python_integer_types():
      try:
          isinstance(1, long)
      except NameError:
          return (int,)  # python 3
      else:
          return (int, long)  # python 2
  
  # 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 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.
      """
      result = steps * (10.0**(_LOG_SIM_PRECISION - _get_log_time_scale(units)))
  
      return result
  
  def get_sim_steps(time, units=None):
      """Calculates the number of simulation time steps for a given amount of *time*.
  
      Args:
          time (int or float):  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 = result * (10.0**(_get_log_time_scale(units) - _LOG_SIM_PRECISION))
  
      err = int(result) - math.ceil(result)
  
      if err:
          raise ValueError("Unable to accurately represent {0}({1}) with the "
                           "simulator precision of 1e{2}".format(
                               time, units, _LOG_SIM_PRECISION))
  
      return int(result)
  
  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 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 ctypes structure then the *bytes* keyword argument 
      must be used.
  
      Args:
          ctypes_obj (ctypes.Structure): The 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
  
  
  def _sane_color(x):
      r = ""
      for i in x:
          j = ord(i)
          if (j < 32) or (j >= 127):
              r += "."
          else:
              r += i
      return r
  
  
  def hexdump(x):
      """Hexdump a buffer.
  
      Args:
          x: Object that supports conversion via the ``str`` built-in.
  
      Returns:
          A string containing the hexdump.
  
      Example:
  
      .. code-block:: python
  
          print(hexdump('this somewhat long string'))
  
      .. code-block:: none
  
          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
      """
      # adapted from scapy.utils.hexdump
      rs = ""
      x = str(x)
      l = len(x)
      i = 0
      while i < l:
          rs += "%04x   " % i
          for j in range(16):
              if i + j < l:
                  rs += "%02X " % ord(x[i + j])
              else:
                  rs += "   "
              if j % 16 == 7:
                  rs += ""
          rs += "  "
          rs += _sane_color(x[i:i + 16]) + "\n"
          i += 16
      return rs
  
  
  def hexdiffs(x, y):
      """Return a diff string showing differences between two binary strings.
  
      Args:
          x: Object that supports conversion via the ``str`` built-in.
          y: Object that supports conversion via the ``str`` built-in.
  
      Example:
  
      .. code-block:: python
  
          print(hexdiffs('this short thing', 'this also short'))
  
      .. code-block:: none
  
          0000      746869732073686F 7274207468696E67 this short thing
               0000 7468697320616C73 6F  2073686F7274 this also  short
  
      """
      # adapted from scapy.utils.hexdiff
  
      def sane(x):
          r = ""
          for i in x:
              j = ord(i)
              if (j < 32) or (j >= 127):
                  r = r + "."
              else:
                  r = r + i
          return r
  
      def highlight(string, colour=ANSI.COLOR_HILITE_HEXDIFF_DEFAULT):
          """Highlight only with ANSI output if it's requested and we are not in a GUI."""
          
          want_ansi = os.getenv("COCOTB_ANSI_OUTPUT") and not os.getenv("GUI")
          if want_ansi is None:
              want_ansi = sys.stdout.isatty()  # default to ANSI for TTYs
          else:
              want_ansi = want_ansi == '1'
  
          if want_ansi:
              return colour + string + ANSI.COLOR_DEFAULT
          else:
              return string
  
      rs = ""
  
      x = str(x)[::-1]
      y = str(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:
          separate = 0
          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]:
                      if linex[j] != liney[j]:
                          rs += highlight("%02X" % ord(line[j]),
                                          colour=ANSI.COLOR_HILITE_HEXDIFF_2)
                      else:
                          rs += "%02X" % ord(line[j])
                      if linex[j] == liney[j]:
                          cl += highlight(_sane_color(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
  
  
  # This is essentially six.exec_
  if sys.version_info.major == 3:
      # this has to not be a syntax error in py2
      import builtins
      exec_ = getattr(builtins, 'exec')
  else:
      # this has to not be a syntax error in py3
      def exec_(_code_, _globs_=None, _locs_=None):
          """Execute code in a namespace."""
          if _globs_ is None:
              frame = sys._getframe(1)
              _globs_ = frame.f_globals
              if _locs_ is None:
                  _locs_ = frame.f_locals
              del frame
          elif _locs_ is None:
              _locs_ = _globs_
          exec("""exec _code_ in _globs_, _locs_""")
  
  
  # this is six.with_metaclass, with a clearer docstring
  def with_metaclass(meta, *bases):
      """This provides:
  
      .. code-block:: python
  
          class Foo(with_metaclass(Meta, Base1, Base2)): pass
  
      which is a unifying syntax for:
  
      .. code-block:: python
  
          # python 3
          class Foo(Base1, Base2, metaclass=Meta): pass
  
          # python 2
          class Foo(Base1, Base2):
              __metaclass__ = Meta
      """
      # This requires a bit of explanation: the basic idea is to make a dummy
      # metaclass for one level of class instantiation that replaces itself with
      # the actual metaclass.
      class metaclass(type):
  
          def __new__(cls, name, this_bases, d):
              return meta(name, bases, d)
  
          @classmethod
          def __prepare__(cls, name, this_bases):
              return meta.__prepare__(name, bases)
      return type.__new__(metaclass, 'temporary_class', (), {})
  
  
  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, *
  
  [... the rest is omitted ...]
files: cocotb/utils.py