b469f1f6-6a00-4958-bfb0-f9fbf427a589

Committed 06 Jun 2024 06:38PM UTC coverage: 93.399% (-0.03%) from 93.431%

Build # b469f1f6-6a00-4958-bfb0-f9fbf427a589

Build Type

push

circleci

Committed by

Nic30

Commit Message

docs

Run Details

8040 of 9100 branches covered (88.35%)

39136 of 41902 relevant lines covered (93.4%)

0.93 hits per line

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

88.64

/hwtLib/structManipulators/structReader.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from math import ceil

from hwt.code import StaticForEach
from hwt.hdl.frameTmpl import FrameTmpl
from hwt.hdl.transTmpl import TransTmpl
from hwt.hdl.types.struct import HStruct
from hwt.hwIOs.hwIOStruct import HwIOStruct
from hwt.hwIOs.std import HwIODataRdVld, HwIOSignal
from hwt.hwIOs.utils import propagateClkRstn, addClkRstn
from hwt.hwModule import HwModule
from hwt.hwParam import HwParam
from hwt.pyUtils.typingFuture import override
from hwtLib.abstract.template_configured import TemplateConfigured
from hwtLib.amba.axis_comp.frame_parser import Axi4S_frameParser
from hwtLib.amba.datapump.intf import HwIOAxiRDatapump, AddrSizeHs
from hwtLib.handshaked.builder import HsBuilder
from hwtLib.handshaked.streamNode import StreamNode


class StructReader(Axi4S_frameParser):
    """
    This unit downloads required structure fields over rDatapump
    interface from address specified by get interface

    :ivar ~.ID: HwParam, id for transactions on bus
    :ivar ~.READ_ACK: HwParam, if true ready on "get" will be set only
        when component is in idle (if false "get"
        is regular handshaked interface)
    :ivar ~.SHARED_READY: HwParam, if this is true field interfaces
        will be of type VldSynced and single ready signal
        will be used for all else every interface
        will be instance of HwIODataRdVld and it
        will have it's own ready(rd) signal
    :attention: interfaces of field will not send data in same time


    .. figure:: ./_static/StructReader.png

    :note: names in the picture are just illustrative

    .. hwt-autodoc:: _example_StructReader
    """

    def __init__(self, structT, tmpl=None, frames=None):
        """
        :param structT: instance of HStruct which specifies data format to download
        :param tmpl: instance of TransTmpl for this structT
        :param frames: list of FrameTmpl instances for this tmpl
        :note: if tmpl and frames are None they are resolved from structT parseTemplate
        :note: this unit can parse sequence of frames, if they are specified by "frames"
        :attention: interfaces for each field in struct will be dynamically created
        :attention: structT can not contain fields with variable size like HStream
        """
        HwModule.__init__(self)
        assert isinstance(structT, HStruct)
        TemplateConfigured.__init__(self, structT, tmpl=tmpl, frames=frames)

    @override
    def hwConfig(self):
        self.ID = HwParam(0)
        HwIOAxiRDatapump.hwConfig(self)
        self.USE_STRB = False
        self.READ_ACK = HwParam(False)
        self.SHARED_READY = HwParam(False)

    def maxWordIndex(self):
        return max(f.endBitAddr - 1 for f in self._frames) // self.DATA_WIDTH

    def maxBytesInTransaction(self):
        return ceil(max(
                    [f.parts[-1].endOfPart - f.startBitAddr for f in self._frames]
                    ) / 8)

    def parseTemplate(self):
        if self._tmpl is None:
            self._tmpl = TransTmpl(self._structT)
        if self._frames is None:
            DW = self.DATA_WIDTH
            frames = FrameTmpl.framesFromTransTmpl(
                        self._tmpl,
                        DW,
                        trimPaddingWordsOnStart=True,
                        trimPaddingWordsOnEnd=True)

            self._frames = list(frames)

    @override
    def hwDeclr(self):
        addClkRstn(self)
        self.dataOut = HwIOStruct(self._structT,
                                  tuple(),
                                  self._mkFieldHwIO)._m()

        g = self.get = HwIODataRdVld()  # data signal is addr of structure to download
        g.DATA_WIDTH = self.ADDR_WIDTH
        self.parseTemplate()

        with self._hwParamsShared():
            # interface for communication with datapump
            self.rDatapump = HwIOAxiRDatapump()._m()
            self.rDatapump.MAX_BYTES = self.maxBytesInTransaction()

        with self._hwParamsShared(exclude=({"ID_WIDTH"}, set())):
            self.parser = Axi4S_frameParser(self._structT,
                                           tmpl=self._tmpl,
                                           frames=self._frames)
            self.parser.SYNCHRONIZE_BY_LAST = False

        if self.SHARED_READY:
            self.ready = HwIOSignal()

    def driveReqRem(self, req: AddrSizeHs, MAX_BITS: int):
        return req.rem(ceil((MAX_BITS % self.DATA_WIDTH) / 8))

    @override
    def hwImpl(self):
        propagateClkRstn(self)
        req = self.rDatapump.req

        if self.READ_ACK:
            get = self.get
        else:
            get = HsBuilder(self, self.get).buff().end

        def propagateRequest(frame, indx):
            isLastFrame = indx == len(self._frames) - 1
            s = [
                req.addr(get.data + frame.startBitAddr // 8),
                req.len(frame.getWordCnt() - 1),
                self.driveReqRem(req, frame.parts[-1].endOfPart - frame.startBitAddr),
                req.vld(get.vld),
                get.rd(req.rd if isLastFrame else 0)
            ]

            ack = StreamNode(masters=[get], slaves=[self.rDatapump.req]).ack()
            return s, ack

        StaticForEach(self, self._frames, propagateRequest)

        r = self.rDatapump.r
        data_sig_to_exclude = []
        if self.ID_WIDTH:
            req.id(self.ID)
        if hasattr(r, "id"):
            data_sig_to_exclude.append(r.id)
        if hasattr(r, "strb"):
            data_sig_to_exclude.append(r.strb)

        self.parser.dataIn(r, exclude=data_sig_to_exclude)

        for _, field in self._tmpl.HwIO_walkFlatten():
            p = field.getFieldPath()
            fieldHwIO = self.dataOut._fieldsToHwIOs[p]
            parserHwIO = self.parser.dataOut._fieldsToHwIOs[p]
            fieldHwIO(parserHwIO)


def _example_StructReader():
    from hwtLib.types.ctypes import uint16_t, uint32_t, uint64_t

    s = HStruct(
        (uint64_t, "item0"),  # tuples (type, name) where type has to be instance of Bits type
        (uint64_t, None),  # name = None means this field will be ignored
        (uint64_t, "item1"),
        (uint64_t, None),
        (uint16_t, "item2"),
        (uint16_t, "item3"),
        (uint32_t, "item4"),

        (uint32_t, None),
        (uint64_t, "item5"),  # this word is split on two bus words
        (uint32_t, None),

        (uint64_t, None),
        (uint64_t, None),
        (uint64_t, None),
        (uint64_t, "item6"),
        (uint64_t, "item7"),
        )

    m = StructReader(s)
    return m


if __name__ == "__main__":
    from hwt.synth import to_rtl_str
    
    m = _example_StructReader()
    print(to_rtl_str(m))

1	#!/usr/bin/env python3
2	# -- coding: utf-8 --
3
4	from math import ceil	1✔
5
6	from hwt.code import StaticForEach	1✔
7	from hwt.hdl.frameTmpl import FrameTmpl	1✔
8	from hwt.hdl.transTmpl import TransTmpl	1✔
9	from hwt.hdl.types.struct import HStruct	1✔
10	from hwt.hwIOs.hwIOStruct import HwIOStruct	1✔
11	from hwt.hwIOs.std import HwIODataRdVld, HwIOSignal	1✔
12	from hwt.hwIOs.utils import propagateClkRstn, addClkRstn	1✔
13	from hwt.hwModule import HwModule	1✔
14	from hwt.hwParam import HwParam	1✔
15	from hwt.pyUtils.typingFuture import override	1✔
16	from hwtLib.abstract.template_configured import TemplateConfigured	1✔
17	from hwtLib.amba.axis_comp.frame_parser import Axi4S_frameParser	1✔
18	from hwtLib.amba.datapump.intf import HwIOAxiRDatapump, AddrSizeHs	1✔
19	from hwtLib.handshaked.builder import HsBuilder	1✔
20	from hwtLib.handshaked.streamNode import StreamNode	1✔
21
22
23	class StructReader(Axi4S_frameParser):	1✔
24	"""
25	This unit downloads required structure fields over rDatapump
26	interface from address specified by get interface
27
28	:ivar ~.ID: HwParam, id for transactions on bus
29	:ivar ~.READ_ACK: HwParam, if true ready on "get" will be set only
30	when component is in idle (if false "get"
31	is regular handshaked interface)
32	:ivar ~.SHARED_READY: HwParam, if this is true field interfaces
33	will be of type VldSynced and single ready signal
34	will be used for all else every interface
35	will be instance of HwIODataRdVld and it
36	will have it's own ready(rd) signal
37	:attention: interfaces of field will not send data in same time
38
39
40	.. figure:: ./_static/StructReader.png
41
42	:note: names in the picture are just illustrative
43
44	.. hwt-autodoc:: _example_StructReader
45	"""
46
47	def __init__(self, structT, tmpl=None, frames=None):	1✔
48	"""
49	:param structT: instance of HStruct which specifies data format to download
50	:param tmpl: instance of TransTmpl for this structT
51	:param frames: list of FrameTmpl instances for this tmpl
52	:note: if tmpl and frames are None they are resolved from structT parseTemplate
53	:note: this unit can parse sequence of frames, if they are specified by "frames"
54	:attention: interfaces for each field in struct will be dynamically created
55	:attention: structT can not contain fields with variable size like HStream
56	"""
57	HwModule.__init__(self)	1✔
58	assert isinstance(structT, HStruct)	1✔
59	TemplateConfigured.__init__(self, structT, tmpl=tmpl, frames=frames)	1✔
60
61	@override	1✔
62	def hwConfig(self):	1✔
63	self.ID = HwParam(0)	1✔
64	HwIOAxiRDatapump.hwConfig(self)	1✔
65	self.USE_STRB = False	1✔
66	self.READ_ACK = HwParam(False)	1✔
67	self.SHARED_READY = HwParam(False)	1✔
68
69	def maxWordIndex(self):	1✔
70	return max(f.endBitAddr - 1 for f in self._frames) // self.DATA_WIDTH	×
71
72	def maxBytesInTransaction(self):	1✔
73	return ceil(max(	1✔
74	[f.parts[-1].endOfPart - f.startBitAddr for f in self._frames]
75	) / 8)
76
77	def parseTemplate(self):	1✔
78	if self._tmpl is None:	1✔
79	self._tmpl = TransTmpl(self._structT)	1✔
80	if self._frames is None:	1✔
81	DW = self.DATA_WIDTH	1✔
82	frames = FrameTmpl.framesFromTransTmpl(	1✔
83	self._tmpl,
84	DW,
85	trimPaddingWordsOnStart=True,
86	trimPaddingWordsOnEnd=True)
87
88	self._frames = list(frames)	1✔
89
90	@override	1✔
91	def hwDeclr(self):	1✔
92	addClkRstn(self)	1✔
93	self.dataOut = HwIOStruct(self._structT,	1✔
94	tuple(),
95	self._mkFieldHwIO)._m()
96
97	g = self.get = HwIODataRdVld() # data signal is addr of structure to download	1✔
98	g.DATA_WIDTH = self.ADDR_WIDTH	1✔
99	self.parseTemplate()	1✔
100
101	with self._hwParamsShared():	1✔
102	# interface for communication with datapump
103	self.rDatapump = HwIOAxiRDatapump()._m()	1✔
104	self.rDatapump.MAX_BYTES = self.maxBytesInTransaction()	1✔
105
106	with self._hwParamsShared(exclude=({"ID_WIDTH"}, set())):	1✔
107	self.parser = Axi4S_frameParser(self._structT,	1✔
108	tmpl=self._tmpl,
109	frames=self._frames)
110	self.parser.SYNCHRONIZE_BY_LAST = False	1✔
111
112	if self.SHARED_READY:	1!
113	self.ready = HwIOSignal()	×
114
115	def driveReqRem(self, req: AddrSizeHs, MAX_BITS: int):	1✔
116	return req.rem(ceil((MAX_BITS % self.DATA_WIDTH) / 8))	1✔
117
118	@override	1✔
119	def hwImpl(self):	1✔
120	propagateClkRstn(self)	1✔
121	req = self.rDatapump.req	1✔
122
123	if self.READ_ACK:	1!
124	get = self.get	×
125	else:
126	get = HsBuilder(self, self.get).buff().end	1✔
127
128	def propagateRequest(frame, indx):	1✔
129	isLastFrame = indx == len(self._frames) - 1	1✔
130	s = [	1✔
131	req.addr(get.data + frame.startBitAddr // 8),
132	req.len(frame.getWordCnt() - 1),
133	self.driveReqRem(req, frame.parts[-1].endOfPart - frame.startBitAddr),
134	req.vld(get.vld),
135	get.rd(req.rd if isLastFrame else 0)
136	]
137
138	ack = StreamNode(masters=[get], slaves=[self.rDatapump.req]).ack()	1✔
139	return s, ack	1✔
140
141	StaticForEach(self, self._frames, propagateRequest)	1✔
142
143	r = self.rDatapump.r	1✔
144	data_sig_to_exclude = []	1✔
145	if self.ID_WIDTH:	1!
146	req.id(self.ID)	×
147	if hasattr(r, "id"):	1!
148	data_sig_to_exclude.append(r.id)	×
149	if hasattr(r, "strb"):	1!
150	data_sig_to_exclude.append(r.strb)	×
151
152	self.parser.dataIn(r, exclude=data_sig_to_exclude)	1✔
153
154	for _, field in self._tmpl.HwIO_walkFlatten():	1✔
155	p = field.getFieldPath()	1✔
156	fieldHwIO = self.dataOut._fieldsToHwIOs[p]	1✔
157	parserHwIO = self.parser.dataOut._fieldsToHwIOs[p]	1✔
158	fieldHwIO(parserHwIO)	1✔
159
160
161	def _example_StructReader():	1✔
162	from hwtLib.types.ctypes import uint16_t, uint32_t, uint64_t	×
163
164	s = HStruct(	×
165	(uint64_t, "item0"), # tuples (type, name) where type has to be instance of Bits type
166	(uint64_t, None), # name = None means this field will be ignored
167	(uint64_t, "item1"),
168	(uint64_t, None),
169	(uint16_t, "item2"),
170	(uint16_t, "item3"),
171	(uint32_t, "item4"),
172
173	(uint32_t, None),
174	(uint64_t, "item5"), # this word is split on two bus words
175	(uint32_t, None),
176
177	(uint64_t, None),
178	(uint64_t, None),
179	(uint64_t, None),
180	(uint64_t, "item6"),
181	(uint64_t, "item7"),
182	)
183
184	m = StructReader(s)	×
185	return m	×
186
187
188	if __name__ == "__main__":
189	from hwt.synth import to_rtl_str
190
191	m = _example_StructReader()
192	print(to_rtl_str(m))

Nic30 / hwtLib / b469f1f6-6a00-4958-bfb0-f9fbf427a589

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