drivers: Move type information from docstrings to annotations (#1849)

Only the functions which had typesin their docstrings are touched here.
Some instances of `str` were corrected to `bytes`, since those were my fault in a previous PR.
Everything else is left as in the docstring.

cocotb/drivers/__init__.py

@@ -30,13 +30,15 @@
 """Set of common driver base classes."""
 
 from collections import deque
+from typing import Iterable, Tuple, Any, Optional, Callable
 
 import cocotb
 from cocotb.decorators import coroutine
 from cocotb.triggers import (Event, RisingEdge, ReadOnly, NextTimeStep,
                              Edge)
 from cocotb.bus import Bus
 from cocotb.log import SimLog
+from cocotb.handle import SimHandleBase
 
 
 class BitDriver:
@@ -50,11 +52,11 @@ def __init__(self, signal, clk, generator=None):
         self._clk = clk
         self._generator = generator
 
-    def start(self, generator=None):
+    def start(self, generator: Iterable[Tuple[int, int]] = None) -> None:
         """Start generating data.
 
         Args:
-            generator (generator, optional): Generator yielding data.
+            generator: Generator yielding data.
                 The generator should yield tuples ``(on, off)``
                 with the number of cycles to be on,
                 followed by the number of cycles to be off.
@@ -115,17 +117,20 @@ def kill(self):
             self._thread.kill()
             self._thread = None
 
-    def append(self, transaction, callback=None, event=None, **kwargs):
+    def append(
+        self, transaction: Any, callback: Callable[[Any], Any] = None,
+        event: Event = None, **kwargs: Any
+    ) -> None:
         """Queue up a transaction to be sent over the bus.
 
         Mechanisms are provided to permit the caller to know when the
         transaction is processed.
 
         Args:
-            transaction (any): The transaction to be sent.
-            callback (callable, optional): Optional function to be called
+            transaction: The transaction to be sent.
+            callback: Optional function to be called
                 when the transaction has been sent.
-            event (optional): :class:`~cocotb.triggers.Event` to be set
+            event: :class:`~cocotb.triggers.Event` to be set
                 when the transaction has been sent.
             **kwargs: Any additional arguments used in child class'
                 :any:`_driver_send` method.
@@ -138,43 +143,46 @@ def clear(self):
         self._sendQ = deque()
 
     @coroutine
-    def send(self, transaction, sync=True, **kwargs):
+    def send(self, transaction: Any, sync: bool = True, **kwargs: Any) -> None:
         """Blocking send call (hence must be "yielded" rather than called).
 
         Sends the transaction over the bus.
 
         Args:
-            transaction (any): The transaction to be sent.
-            sync (bool, optional): Synchronize the transfer by waiting for a rising edge.
-            **kwargs (dict): Additional arguments used in child class'
+            transaction: The transaction to be sent.
+            sync: Synchronize the transfer by waiting for a rising edge.
+            **kwargs: Additional arguments used in child class'
                 :any:`_driver_send` method.
         """
         yield self._send(transaction, None, None, sync=sync, **kwargs)
 
-    def _driver_send(self, transaction, sync=True, **kwargs):
+    def _driver_send(self, transaction: Any, sync: bool = True, **kwargs: Any) -> None:
         """Actual implementation of the send.
 
         Sub-classes should override this method to implement the actual
         :meth:`~cocotb.drivers.Driver.send` routine.
 
         Args:
-            transaction (any): The transaction to be sent.
-            sync (bool, optional): Synchronize the transfer by waiting for a rising edge.
+            transaction: The transaction to be sent.
+            sync: Synchronize the transfer by waiting for a rising edge.
             **kwargs: Additional arguments if required for protocol implemented in a sub-class.
         """
         raise NotImplementedError("Sub-classes of Driver should define a "
                                   "_driver_send coroutine")
 
     @coroutine
-    def _send(self, transaction, callback, event, sync=True, **kwargs):
+    def _send(
+        self, transaction: Any, callback: Callable[[Any], Any], event: Event,
+        sync: bool = True, **kwargs
+    ) -> None:
         """Send coroutine.
 
         Args:
-            transaction (any): The transaction to be sent.
-            callback (callable, optional): Optional function to be called
+            transaction: The transaction to be sent.
+            callback: Optional function to be called
                 when the transaction has been sent.
-            event (optional): event to be set when the transaction has been sent.
-            sync (bool, optional): Synchronize the transfer by waiting for a rising edge.
+            event: event to be set when the transaction has been sent.
+            sync: Synchronize the transfer by waiting for a rising edge.
             **kwargs: Any additional arguments used in child class'
                 :any:`_driver_send` method.
         """
@@ -217,19 +225,19 @@ class BusDriver(Driver):
         * an entity
 
     Args:
-        entity (SimHandle): A handle to the simulator entity.
-        name (str or None): Name of this bus. ``None`` for a nameless bus, e.g.
+        entity: A handle to the simulator entity.
+        name: Name of this bus. ``None`` for a nameless bus, e.g.
             bus-signals in an interface or a ``modport``.
             (untested on ``struct``/``record``, but could work here as well).
-        clock (SimHandle): A handle to the clock associated with this bus.
-        **kwargs (dict): Keyword arguments forwarded to :class:`cocotb.Bus`,
+        clock: A handle to the clock associated with this bus.
+        **kwargs: Keyword arguments forwarded to :class:`cocotb.Bus`,
             see docs for that class for more information.
 
     """
 
     _optional_signals = []
 
-    def __init__(self, entity, name, clock, **kwargs):
+    def __init__(self, entity: SimHandleBase, name: Optional[str], clock: SimHandleBase, **kwargs: Any):
         index = kwargs.get("array_idx", None)
 
         self.log = SimLog("cocotb.%s.%s" % (entity._name, name))
@@ -245,12 +253,12 @@ def __init__(self, entity, name, clock, **kwargs):
         self.name = name if index is None else "%s_%d" % (name, index)
 
     @coroutine
-    def _driver_send(self, transaction, sync=True):
+    def _driver_send(self, transaction, sync: bool = True) -> None:
         """Implementation for BusDriver.
 
         Args:
             transaction: The transaction to send.
-            sync (bool, optional): Synchronize the transfer by waiting for a rising edge.
+            sync: Synchronize the transfer by waiting for a rising edge.
         """
         if sync:
             yield RisingEdge(self.clock)
@@ -301,7 +309,10 @@ class ValidatedBusDriver(BusDriver):
             ``(valid, invalid)`` cycles to insert.
     """
 
-    def __init__(self, entity, name, clock, *, valid_generator=None, **kwargs):
+    def __init__(
+        self, entity: SimHandleBase, name: str, clock: SimHandleBase, *,
+        valid_generator: Iterable[Tuple[int, int]] = None, **kwargs: Any
+    ) -> None:
         BusDriver.__init__(self, entity, name, clock, **kwargs)
         self.set_valid_generator(valid_generator=valid_generator)
 

cocotb/drivers/amba.py

@@ -107,24 +107,26 @@ def _send_write_data(self, data, delay=0, byte_enable=0xF):
         self.write_data_busy.release()
 
     @cocotb.coroutine
-    def write(self, address, value, byte_enable=0xf, address_latency=0,
-              data_latency=0, sync=True):
+    def write(
+        self, address: int, value: int, byte_enable: int = 0xf,
+        address_latency: int = 0, data_latency: int = 0, sync: bool = True
+    ) -> BinaryValue:
         """Write a value to an address.
 
         Args:
-            address (int): The address to write to.
-            value (int): The data value to write.
-            byte_enable (int, optional): Which bytes in value to actually write.
+            address: The address to write to.
+            value: The data value to write.
+            byte_enable: Which bytes in value to actually write.
                 Default is to write all bytes.
-            address_latency (int, optional): Delay before setting the address (in clock cycles).
+            address_latency: Delay before setting the address (in clock cycles).
                 Default is no delay.
-            data_latency (int, optional): Delay before setting the data value (in clock cycles).
+            data_latency: Delay before setting the data value (in clock cycles).
                 Default is no delay.
-            sync (bool, optional): Wait for rising edge on clock initially.
+            sync: Wait for rising edge on clock initially.
                 Defaults to True.
 
         Returns:
-            BinaryValue: The write response value.
+            The write response value.
 
         Raises:
             AXIProtocolError: If write response from AXI is not ``OKAY``.
@@ -160,16 +162,16 @@ def write(self, address, value, byte_enable=0xf, address_latency=0,
         return result
 
     @cocotb.coroutine
-    def read(self, address, sync=True):
+    def read(self, address: int, sync: bool = True) -> BinaryValue:
         """Read from an address.
 
         Args:
-            address (int): The address to read from.
-            sync (bool, optional): Wait for rising edge on clock initially.
+            address: The address to read from.
+            sync: Wait for rising edge on clock initially.
                 Defaults to True.
 
         Returns:
-            BinaryValue: The read data value.
+            The read data value.
 
         Raises:
             AXIProtocolError: If read response from AXI is not ``OKAY``.

cocotb/drivers/avalon.py

@@ -33,6 +33,7 @@
 """
 
 import random
+from typing import Iterable, Union, Optional
 
 import cocotb
 from cocotb.decorators import coroutine
@@ -116,19 +117,19 @@ def _release_lock(self):
         self.busy_event.set()
 
     @coroutine
-    def read(self, address, sync=True):
+    def read(self, address: int, sync: bool = True) -> BinaryValue:
         """Issue a request to the bus and block until this comes back.
 
         Simulation time still progresses
         but syntactically it blocks.
 
         Args:
-            address (int): The address to read from.
-            sync (bool, optional): Wait for rising edge on clock initially.
+            address: The address to read from.
+            sync: Wait for rising edge on clock initially.
                 Defaults to True.
 
         Returns:
-            BinaryValue: The read data value.
+            The read data value.
 
         Raises:
             :any:`TestError`: If master is write-only.
@@ -183,13 +184,13 @@ def read(self, address, sync=True):
         return data
 
     @coroutine
-    def write(self, address, value):
+    def write(self, address: int, value: int) -> None:
         """Issue a write to the given address with the specified
         value.
 
         Args:
-            address (int): The address to write to.
-            value (int): The data value to write.
+            address: The address to write to.
+            value: The data value to write.
 
         Raises:
             :any:`TestError`: If master is read-only.
@@ -671,10 +672,10 @@ def _wait_ready(self):
             yield ReadOnly()
 
     @coroutine
-    def _send_string(self, string, sync=True, channel=None):
+    def _send_string(self, string: bytes, sync: bool = True, channel: Optional[int] = None) -> None:
         """Args:
-            string (bytes): A string of bytes to send over the bus.
-            channel (int): Channel to send the data on.
+            string: A string of bytes to send over the bus.
+            channel: Channel to send the data on.
         """
         # Avoid spurious object creation by recycling
         clkedge = RisingEdge(self.clock)
@@ -774,9 +775,9 @@ def _send_string(self, string, sync=True, channel=None):
             self.bus.channel <= channel_value
 
     @coroutine
-    def _send_iterable(self, pkt, sync=True):
+    def _send_iterable(self, pkt: Iterable, sync: bool = True) -> None:
         """Args:
-            pkt (iterable): Will yield objects with attributes matching the
+            pkt: Will yield objects with attributes matching the
                 signal names for each individual bus cycle.
         """
         clkedge = RisingEdge(self.clock)
@@ -815,12 +816,12 @@ def _send_iterable(self, pkt, sync=True):
         self.bus.valid <= 0
 
     @coroutine
-    def _driver_send(self, pkt, sync=True, channel=None):
+    def _driver_send(self, pkt: Union[bytes, Iterable], sync: bool = True, channel: Optional[int] = None):
         """Send a packet over the bus.
 
         Args:
-            pkt (str or iterable): Packet to drive onto the bus.
-            channel (None or int): Channel attributed to the packet.
+            pkt: Packet to drive onto the bus.
+            channel: Channel attributed to the packet.
 
         If ``pkt`` is a string, we simply send it word by word
 

cocotb/drivers/opb.py

@@ -33,6 +33,7 @@
 import cocotb
 from cocotb.triggers import RisingEdge, ReadOnly, Event
 from cocotb.drivers import BusDriver
+from cocotb.binary import BinaryValue
 
 
 class OPBException(Exception):
@@ -65,18 +66,18 @@ def _release_lock(self):
         self.busy_event.set()
 
     @cocotb.coroutine
-    def read(self, address, sync=True):
+    def read(self, address: int, sync: bool = True) -> BinaryValue:
         """Issue a request to the bus and block until this comes back.
 
         Simulation time still progresses but syntactically it blocks.
 
         Args:
-            address (int): The address to read from.
-            sync (bool, optional): Wait for rising edge on clock initially.
+            address: The address to read from.
+            sync: Wait for rising edge on clock initially.
                 Defaults to True.
 
         Returns:
-            BinaryValue: The read data value.
+            The read data value.
 
         Raises:
             OPBException: If read took longer than 16 cycles.
@@ -110,13 +111,13 @@ def read(self, address, sync=True):
         return data
 
     @cocotb.coroutine
-    def write(self, address, value, sync=True):
+    def write(self, address: int, value: int, sync: bool = True) -> None:
         """Issue a write to the given address with the specified value.
 
         Args:
-            address (int): The address to read from.
-            value (int): The data value to write.
-            sync (bool, optional): Wait for rising edge on clock initially.
+            address: The address to read from.
+            value: The data value to write.
+            sync: Wait for rising edge on clock initially.
                 Defaults to True.
 
         Raises: