f4cb6bb9-1d2f-48c1-8464-a66f048e6a31

Committed 22 May 2024 11:43AM UTC coverage: 70.567% (+0.06%) from 70.506%

Build # f4cb6bb9-1d2f-48c1-8464-a66f048e6a31

Build Type

push

circleci

Committed by

Nic30

Commit Message

refactor!: unify naming convention for .name, ._name, ._hdl_name attrs

* Nic30/hwt#17

Run Details

217 of 361 branches covered (60.11%)

Branch coverage included in aggregate %.

2 of 2 new or added lines in 1 file covered. (100.0%)

2 existing lines in 1 file now uncovered.

941 of 1280 relevant lines covered (73.52%)

0.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

42.02

/hwtSimApi/agents/peripheral/i2c.py

from collections import deque
from typing import Tuple, Optional

from hwtSimApi.agents.base import AgentWitReset, NOP, RX, TX
from hwtSimApi.agents.peripheral.tristate import TristateAgent, TristateClkAgent,\
    TristateSignal
from hwtSimApi.triggers import WaitCombStable, WaitWriteOnly, WaitCombRead,\
    WaitTimeslotEnd
from enum import Enum
from hwtSimApi.hdlSimulator import HdlSimulator
from hwtSimApi.process_utils import OnRisingCallbackLoop, OnFallingCallbackLoop


class I2C_MODE(Enum):
    STANDARD = 100000
    FAST = 400000
    HIGH_SPEED = 3400000


class I2C_ADDR(Enum):
    ADDR_7b = 7
    ADDR_10b = 10


def getBit(val, bitNo):
    """
    get specific bit from integer (little-endian)
    """
    return (val >> bitNo) & 1


class I2cAgent(AgentWitReset):
    """
    A base simulation agent for I2C interfaces. Note that as all devices are using tri-state interfaces.
    The agent has to know the data format which of device in order to interpret data correctly.
    https://www.nxp.com/docs/en/user-guide/UM10204.pdf
    https://dlbeer.co.nz/articles/i2c.html

    :note: The agent will support slave mode after it's address is set.

    Master agent transaction formats:
          [((READ, ACK/NACK) | (WRITE, VALUE, ACK/NACK) | (REPEATED_START, duration in I2C clk))*]
          (ACK/NACK values are there to check the slave response)
    Slave agent uses the onWrite/onRead callbacks

    :attention: multi-master arbitration not implemented
    :note: The driver/monitor difference is related to directions of signals but as the used interface
        is tri-state interface bouth agents can be SLAVE/master
    :note: typical I2C command: START, A6-0, R/W, ACK,  ...  D7-0, ACK, STOP
    :note: each agent is master, but none is active if it does not have transactions to do so
    :note: slave has to check address in function because some devices may actually use
        part of address as opt code etc. because of this this agent needs to be more generic

    :ivar ~.data_m: the buffer for transaction for master
    :ivar ~.data_m_read: the buffer for read data by master
    :ivar ~.start: flag, if True the master should send I2C START
    :ivar ~.stop: flag, if True the master should send I2C STOP

    :ivar ~.bit_cntrl: the buffer for commands for specific SDA value
                    (intermediate instructions)
    :ivar ~.bit_cntrl_rx: the buffer for a bit values of SDA
                    (intermediate read value)
    """

    ADDR_PREFIX_10b = 0b11110000

    ACK = 0  # send by reciever
    NACK = 1

    READ = 1
    WRITE = 0

    START = "START"
    RESTART = START
    STOP = "STOP"

    def __init__(self, sim: HdlSimulator, hwIO: Tuple[TristateSignal, TristateSignal],
                 rst: Tuple["RtlSignal", bool]):
        """
        :param: hwIO i2c interface, tuple (scl, sda) = (clock, data),
            sda/sdc are tri-state interfaces represented by i, o, t signals
        """
        AgentWitReset.__init__(self, sim, hwIO, rst)
        self.data_m = deque()
        self.data_m_read = []
        self.bit_cntrl = deque()  # type: Deque[ Tuple[Union[RX, TX], Optional[int]] ]
        self.bit_cntrl_rx = deque()  # type: Deque[Union[START, STOP, int]]
        self.start = True
        self.sda = TristateAgent(sim, hwIO[1], rst)
        self.sda.collectData = False
        self.sda.selfSynchronization = False
        self.slave = False
        self.bit_index = None

    def hasTransactionPending(self):
        return (
            self.data_m
            or self.data_m_read
            or self.bit_cntrl
            or self.bit_cntrl_rx)

    def _transmit_byte(self, val: int, ack_for_check: Optional[bool]):
        for i in range(8):
            b = getBit(val, 7-i)
            self.bit_cntrl.append((TX, b))
        self.bit_cntrl.append((RX, ack_for_check))
        yield from self._wait_until_command_completion()
        assert len(self.bit_cntrl_rx) == 1
        return self.bit_cntrl_rx.pop()

    def _receive_byte(self, ack: bool):
        """
        :note: If master is a reciever the ack means that it wishes to recieve next byte
        """
        for _ in range(8):
            self.bit_cntrl.append((RX, None))
        self.bit_cntrl.append((TX, int(ack)))
        yield from self._wait_until_command_completion()
        assert len(self.bit_cntrl_rx) == 8
        b = 0
        for _b in self.bit_cntrl_rx:
            b <<= 1
            b |= _b
        return b

    def _wait_until_command_completion(self):
        # wait untill command get processed
        while self.bit_cntrl:
            yield WaitCombRead()
            if self.bit_cntrl:
                yield WaitTimeslotEnd()

    def execute_master_transaction(self):
        trans = self.data.pop()
        for t in trans:
            m = t[0]
            if m == RX:
                b = yield from self._receive_byte(t[1])
                raise NotImplementedError(b)
            elif m == TX:
                ack = yield from self._transmit_byte(b[1], b[2])
                raise NotImplementedError(ack)
            elif m == self.START:
                raise NotImplementedError()
            elif m == self.STOP:
                raise NotImplementedError()
            else:
                ValueError(m)
        self.stop = True

    def monitor_on_read(self, addr):
        raise NotImplementedError()
    def monitor_on_write(self, addr):
        raise NotImplementedError()

    def execute_slave_transaction(self):
        if self.ADDR_BITS == I2C_ADDR.ADDR_7b:
            yield from self._receive_byte(1)
            addr = self.data_rx.pop()
            rw = addr & 0b1
            addr >>= 1
        elif self.ADDR_BITS == I2C_ADDR.ADDR_10b:
            yield from self._receive_byte(1)
            addrLow = self.data_rx.pop()
            addrLow = int(addrLow)
            rw = addrLow & 0b1
            addrLow >>= 1
            assert addr & 0b11111000 == self.ADDR_PREFIX_10b
            yield from self._receive_byte(1)
            addrHigh = self.data_rx.pop()
            addrHigh = int(addrHigh)
            addr = (addr << 8) | addrHigh

        if rw == self.READ:
            yield from self.monitor_on_read(addr)
        else:
            yield from self.monitor_on_write(addr)

    def startListener(self):
        # SDA->0 and SCL=1
        if self.start:
            self.bit_index = 0
            self.bit_cntrl_rx.append(self.START)
            self.start = False
            # yield sexecute_master_transactionelf.execute_master_transaction()

        return
        yield

    def startSender(self):
        # SDA->0 and SCL=1
        if self.start:
            self.bit_index = 0
            self.sda._write(0)
            self.start = False

        return
        yield

    def getMonitors(self):
        sim = self.sim
        scl = self.hwIO[0]
        self.scl = TristateClkAgent(
            sim, scl, (self.rst, self.rstOffIn),
        )
        self.monitor = OnRisingCallbackLoop(
            sim, scl.i, self.monitor, self.getEnable)
        self.startListener = OnFallingCallbackLoop(
            sim, scl.i, self.startListener, self.getEnable)

        return (
            self.monitor(),
            self.startListener(),
            *self.sda.getMonitors(),
            *self.scl.getMonitors()
        )

    def getDrivers(self):
        sim = self.sim
        scl = self.hwIO[0]
        driver = self.driver
        self.scl = TristateClkAgent(
            sim, scl, (self.rst, self.rstOffIn),
        )
        self.driver = OnRisingCallbackLoop(
            sim, scl.i, self.driver, self.getEnable)
        self.startSender = OnFallingCallbackLoop(
            sim, scl.i, self.startSender, self.getEnable)
        self.scl.setEnable(False, None)
        return (
            driver(),  # initialization of the interface
            self.driver(),
            self.startSender(),
            *self.sda.getDrivers(),
            *self.scl.getDrivers()
        )

    def monitor(self):
        # now intf.sdc is rising
        yield WaitCombRead()
        # wait on all agents to update values and on
        # simulator to apply them
        if self.sim.now > 0 and self.notReset():
            if self.bit_index != 8:
                self.bit_index += 1
                yield WaitCombStable()
                v = self.sda.i.read()
                self.bit_cntrl_rx.append(v)
            else:
                yield WaitWriteOnly()
                self.sda._write(self.ACK)

    def driver(self):
        # now intf.sdc is rising
        # prepare data for next clk
        yield WaitWriteOnly()
        if self.bits:
            b = self.bits.popleft()
            if b == self.START:
                self.start = True
                return
            elif b == self.STOP:
                self.bit_index = None
                self.stop = True
                return
        else:
            b = NOP

        self.sda._write(b)

    def setEnable(self, en):
        # If there is no pending transaction no pause is required
        if not self.start and not self.stop and not self.bit_cntrl:
            super(I2cAgent, self).setEnable(en)
        else:
            # wait until clock
            raise NotImplementedError()

1	from collections import deque	1✔
2	from typing import Tuple, Optional	1✔
3
4	from hwtSimApi.agents.base import AgentWitReset, NOP, RX, TX	1✔
5	from hwtSimApi.agents.peripheral.tristate import TristateAgent, TristateClkAgent,\	1✔
6	TristateSignal
7	from hwtSimApi.triggers import WaitCombStable, WaitWriteOnly, WaitCombRead,\	1✔
8	WaitTimeslotEnd
9	from enum import Enum	1✔
10	from hwtSimApi.hdlSimulator import HdlSimulator	1✔
11	from hwtSimApi.process_utils import OnRisingCallbackLoop, OnFallingCallbackLoop	1✔
12
13
14	class I2C_MODE(Enum):	1✔
15	STANDARD = 100000	1✔
16	FAST = 400000	1✔
17	HIGH_SPEED = 3400000	1✔
18
19
20	class I2C_ADDR(Enum):	1✔
21	ADDR_7b = 7	1✔
22	ADDR_10b = 10	1✔
23
24
25	def getBit(val, bitNo):	1✔
26	"""
27	get specific bit from integer (little-endian)
28	"""
29	return (val >> bitNo) & 1	×
30
31
32	class I2cAgent(AgentWitReset):	1✔
33	"""
34	A base simulation agent for I2C interfaces. Note that as all devices are using tri-state interfaces.
35	The agent has to know the data format which of device in order to interpret data correctly.
36	https://www.nxp.com/docs/en/user-guide/UM10204.pdf
37	https://dlbeer.co.nz/articles/i2c.html
38
39	:note: The agent will support slave mode after it's address is set.
40
41	Master agent transaction formats:
42	[((READ, ACK/NACK) \| (WRITE, VALUE, ACK/NACK) \| (REPEATED_START, duration in I2C clk))*]
43	(ACK/NACK values are there to check the slave response)
44	Slave agent uses the onWrite/onRead callbacks
45
46	:attention: multi-master arbitration not implemented
47	:note: The driver/monitor difference is related to directions of signals but as the used interface
48	is tri-state interface bouth agents can be SLAVE/master
49	:note: typical I2C command: START, A6-0, R/W, ACK, ... D7-0, ACK, STOP
50	:note: each agent is master, but none is active if it does not have transactions to do so
51	:note: slave has to check address in function because some devices may actually use
52	part of address as opt code etc. because of this this agent needs to be more generic
53
54	:ivar ~.data_m: the buffer for transaction for master
55	:ivar ~.data_m_read: the buffer for read data by master
56	:ivar ~.start: flag, if True the master should send I2C START
57	:ivar ~.stop: flag, if True the master should send I2C STOP
58
59	:ivar ~.bit_cntrl: the buffer for commands for specific SDA value
60	(intermediate instructions)
61	:ivar ~.bit_cntrl_rx: the buffer for a bit values of SDA
62	(intermediate read value)
63	"""
64
65	ADDR_PREFIX_10b = 0b11110000	1✔
66
67	ACK = 0 # send by reciever	1✔
68	NACK = 1	1✔
69
70	READ = 1	1✔
71	WRITE = 0	1✔
72
73	START = "START"	1✔
74	RESTART = START	1✔
75	STOP = "STOP"	1✔
76
77	def __init__(self, sim: HdlSimulator, hwIO: Tuple[TristateSignal, TristateSignal],	1✔
78	rst: Tuple["RtlSignal", bool]):
79	"""
80	:param: hwIO i2c interface, tuple (scl, sda) = (clock, data),
81	sda/sdc are tri-state interfaces represented by i, o, t signals
82	"""
83	AgentWitReset.__init__(self, sim, hwIO, rst)	1✔
84	self.data_m = deque()	1✔
85	self.data_m_read = []	1✔
86	self.bit_cntrl = deque() # type: Deque[ Tuple[Union[RX, TX], Optional[int]] ]	1✔
87	self.bit_cntrl_rx = deque() # type: Deque[Union[START, STOP, int]]	1✔
88	self.start = True	1✔
89	self.sda = TristateAgent(sim, hwIO[1], rst)	1✔
90	self.sda.collectData = False	1✔
91	self.sda.selfSynchronization = False	1✔
92	self.slave = False	1✔
93	self.bit_index = None	1✔
94
95	def hasTransactionPending(self):	1✔
96	return (	1✔
97	self.data_m
98	or self.data_m_read
99	or self.bit_cntrl
100	or self.bit_cntrl_rx)
101
102	def _transmit_byte(self, val: int, ack_for_check: Optional[bool]):	1✔
103	for i in range(8):	×
104	b = getBit(val, 7-i)	×
105	self.bit_cntrl.append((TX, b))	×
106	self.bit_cntrl.append((RX, ack_for_check))	×
107	yield from self._wait_until_command_completion()	×
108	assert len(self.bit_cntrl_rx) == 1	×
109	return self.bit_cntrl_rx.pop()	×
110
111	def _receive_byte(self, ack: bool):	1✔
112	"""
113	:note: If master is a reciever the ack means that it wishes to recieve next byte
114	"""
115	for _ in range(8):	×
116	self.bit_cntrl.append((RX, None))	×
117	self.bit_cntrl.append((TX, int(ack)))	×
118	yield from self._wait_until_command_completion()	×
119	assert len(self.bit_cntrl_rx) == 8	×
120	b = 0	×
121	for _b in self.bit_cntrl_rx:	×
122	b <<= 1	×
123	b \|= _b	×
124	return b	×
125
126	def _wait_until_command_completion(self):	1✔
127	# wait untill command get processed
128	while self.bit_cntrl:	×
129	yield WaitCombRead()	×
130	if self.bit_cntrl:	×
131	yield WaitTimeslotEnd()	×
132
133	def execute_master_transaction(self):	1✔
134	trans = self.data.pop()	×
135	for t in trans:	×
136	m = t[0]	×
137	if m == RX:	×
138	b = yield from self._receive_byte(t[1])	×
139	raise NotImplementedError(b)
140	elif m == TX:	×
141	ack = yield from self._transmit_byte(b[1], b[2])	×
142	raise NotImplementedError(ack)
143	elif m == self.START:	×
144	raise NotImplementedError()
145	elif m == self.STOP:	×
146	raise NotImplementedError()
147	else:
148	ValueError(m)	×
149	self.stop = True	×
150
151	def monitor_on_read(self, addr):	1✔
152	raise NotImplementedError()
153	def monitor_on_write(self, addr):	1✔
154	raise NotImplementedError()
155
156	def execute_slave_transaction(self):	1✔
157	if self.ADDR_BITS == I2C_ADDR.ADDR_7b:	×
158	yield from self._receive_byte(1)	×
159	addr = self.data_rx.pop()	×
160	rw = addr & 0b1	×
161	addr >>= 1	×
162	elif self.ADDR_BITS == I2C_ADDR.ADDR_10b:	×
163	yield from self._receive_byte(1)	×
164	addrLow = self.data_rx.pop()	×
165	addrLow = int(addrLow)	×
166	rw = addrLow & 0b1	×
167	addrLow >>= 1	×
168	assert addr & 0b11111000 == self.ADDR_PREFIX_10b	×
169	yield from self._receive_byte(1)	×
170	addrHigh = self.data_rx.pop()	×
171	addrHigh = int(addrHigh)	×
172	addr = (addr << 8) \| addrHigh	×
173
174	if rw == self.READ:	×
175	yield from self.monitor_on_read(addr)	×
176	else:
177	yield from self.monitor_on_write(addr)	×
178
179	def startListener(self):	1✔
180	# SDA->0 and SCL=1
181	if self.start:	1✔
182	self.bit_index = 0	1✔
183	self.bit_cntrl_rx.append(self.START)	1✔
184	self.start = False	1✔
185	# yield sexecute_master_transactionelf.execute_master_transaction()
186
187	return	1✔
188	yield
189
190	def startSender(self):	1✔
191	# SDA->0 and SCL=1
192	if self.start:	×
193	self.bit_index = 0	×
194	self.sda._write(0)	×
195	self.start = False	×
196
197	return	×
198	yield
199
200	def getMonitors(self):	1✔
201	sim = self.sim	1✔
202	scl = self.hwIO[0]	1✔
203	self.scl = TristateClkAgent(	1✔
204	sim, scl, (self.rst, self.rstOffIn),
205	)
206	self.monitor = OnRisingCallbackLoop(	1✔
207	sim, scl.i, self.monitor, self.getEnable)
208	self.startListener = OnFallingCallbackLoop(	1✔
209	sim, scl.i, self.startListener, self.getEnable)
210
211	return (	1✔
212	self.monitor(),
213	self.startListener(),
214	*self.sda.getMonitors(),
215	*self.scl.getMonitors()
216	)
217
218	def getDrivers(self):	1✔
219	sim = self.sim	×
220	scl = self.hwIO[0]	×
221	driver = self.driver	×
222	self.scl = TristateClkAgent(	×
223	sim, scl, (self.rst, self.rstOffIn),
224	)
225	self.driver = OnRisingCallbackLoop(	×
226	sim, scl.i, self.driver, self.getEnable)
227	self.startSender = OnFallingCallbackLoop(	×
228	sim, scl.i, self.startSender, self.getEnable)
229	self.scl.setEnable(False, None)	×
230	return (	×
231	driver(), # initialization of the interface
232	self.driver(),
233	self.startSender(),
234	*self.sda.getDrivers(),
235	*self.scl.getDrivers()
236	)
237
238	def monitor(self):	1✔
239	# now intf.sdc is rising
240	yield WaitCombRead()	1✔
241	# wait on all agents to update values and on
242	# simulator to apply them
243	if self.sim.now > 0 and self.notReset():	1✔
244	if self.bit_index != 8:	1✔
245	self.bit_index += 1	1✔
246	yield WaitCombStable()	1✔
247	v = self.sda.i.read()	1✔
248	self.bit_cntrl_rx.append(v)	1✔
249	else:
250	yield WaitWriteOnly()	1✔
251	self.sda._write(self.ACK)	1✔
252
253	def driver(self):	1✔
254	# now intf.sdc is rising
255	# prepare data for next clk
256	yield WaitWriteOnly()	×
257	if self.bits:	×
258	b = self.bits.popleft()	×
259	if b == self.START:	×
260	self.start = True	×
261	return	×
262	elif b == self.STOP:	×
263	self.bit_index = None	×
264	self.stop = True	×
265	return	×
266	else:
267	b = NOP	×
268
269	self.sda._write(b)	×
270
271	def setEnable(self, en):	1✔
272	# If there is no pending transaction no pause is required
UNCOV 273	if not self.start and not self.stop and not self.bit_cntrl:	×
UNCOV 274	super(I2cAgent, self).setEnable(en)	×
275	else:
276	# wait until clock
277	raise NotImplementedError()

Nic30 / hwtSimApi / f4cb6bb9-1d2f-48c1-8464-a66f048e6a31

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous