Use typing.NamedTuple instead of collections.namedtuple to add __bytes__ cast
This commit is contained in:
@@ -23,7 +23,8 @@ THE SOFTWARE.
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
from collections import namedtuple, Counter
|
from collections import Counter
|
||||||
|
from typing import List, NamedTuple, Union
|
||||||
|
|
||||||
import cocotb
|
import cocotb
|
||||||
from cocotb.queue import Queue
|
from cocotb.queue import Queue
|
||||||
@@ -34,19 +35,75 @@ from .constants import AxiBurstType, AxiLockType, AxiProt, AxiResp
|
|||||||
from .axi_channels import AxiAWSource, AxiWSource, AxiBSink, AxiARSource, AxiRSink
|
from .axi_channels import AxiAWSource, AxiWSource, AxiBSink, AxiARSource, AxiRSink
|
||||||
from .reset import Reset
|
from .reset import Reset
|
||||||
|
|
||||||
|
|
||||||
# AXI master write helper objects
|
# AXI master write helper objects
|
||||||
AxiWriteCmd = namedtuple("AxiWriteCmd", ["address", "data", "awid", "burst", "size",
|
class AxiWriteCmd(NamedTuple):
|
||||||
"lock", "cache", "prot", "qos", "region", "user", "wuser", "event"])
|
address: int
|
||||||
AxiWriteRespCmd = namedtuple("AxiWriteRespCmd", ["address", "length", "size", "cycles",
|
data: bytes
|
||||||
"prot", "burst_list", "event"])
|
awid: int
|
||||||
AxiWriteResp = namedtuple("AxiWriteResp", ["address", "length", "resp", "user"])
|
burst: AxiBurstType
|
||||||
|
size: int
|
||||||
|
lock: AxiLockType
|
||||||
|
cache: int
|
||||||
|
prot: AxiProt
|
||||||
|
qos: int
|
||||||
|
region: int
|
||||||
|
user: int
|
||||||
|
wuser: Union[list, int, None]
|
||||||
|
event: Event
|
||||||
|
|
||||||
|
|
||||||
|
class AxiWriteRespCmd(NamedTuple):
|
||||||
|
address: int
|
||||||
|
length: int
|
||||||
|
size: int
|
||||||
|
cycles: int
|
||||||
|
prot: AxiProt
|
||||||
|
burst_list: list[int]
|
||||||
|
event: Event
|
||||||
|
|
||||||
|
|
||||||
|
class AxiWriteResp(NamedTuple):
|
||||||
|
address: int
|
||||||
|
length: int
|
||||||
|
resp: AxiResp
|
||||||
|
user: Union[list, None]
|
||||||
|
|
||||||
|
|
||||||
# AXI master read helper objects
|
# AXI master read helper objects
|
||||||
AxiReadCmd = namedtuple("AxiReadCmd", ["address", "length", "arid", "burst", "size",
|
class AxiReadCmd(NamedTuple):
|
||||||
"lock", "cache", "prot", "qos", "region", "user", "event"])
|
address: int
|
||||||
AxiReadRespCmd = namedtuple("AxiReadRespCmd", ["address", "length", "size", "cycles",
|
length: int
|
||||||
"prot", "burst_list", "event"])
|
arid: int
|
||||||
AxiReadResp = namedtuple("AxiReadResp", ["address", "data", "resp", "user"])
|
burst: AxiBurstType
|
||||||
|
size: int
|
||||||
|
lock: AxiLockType
|
||||||
|
cache: int
|
||||||
|
prot: AxiProt
|
||||||
|
qos: int
|
||||||
|
region: int
|
||||||
|
user: int
|
||||||
|
event: Event
|
||||||
|
|
||||||
|
|
||||||
|
class AxiReadRespCmd(NamedTuple):
|
||||||
|
address: int
|
||||||
|
length: int
|
||||||
|
size: int
|
||||||
|
cycles: int
|
||||||
|
prot: AxiProt
|
||||||
|
burst_list: list[int]
|
||||||
|
event: Event
|
||||||
|
|
||||||
|
|
||||||
|
class AxiReadResp(NamedTuple):
|
||||||
|
address: int
|
||||||
|
data: bytes
|
||||||
|
resp: AxiResp
|
||||||
|
user: Union[list, None]
|
||||||
|
|
||||||
|
def __bytes__(self):
|
||||||
|
return self.data
|
||||||
|
|
||||||
|
|
||||||
class TagContext:
|
class TagContext:
|
||||||
@@ -889,7 +946,7 @@ class AxiMasterRead(Reset):
|
|||||||
self.log.info("Read complete addr: 0x%08x prot: %s resp: %s data: %s",
|
self.log.info("Read complete addr: 0x%08x prot: %s resp: %s data: %s",
|
||||||
cmd.address, cmd.prot, resp, ' '.join((f'{c:02x}' for c in data)))
|
cmd.address, cmd.prot, resp, ' '.join((f'{c:02x}' for c in data)))
|
||||||
|
|
||||||
read_resp = AxiReadResp(cmd.address, data, resp, user)
|
read_resp = AxiReadResp(cmd.address, bytes(data), resp, user)
|
||||||
|
|
||||||
cmd.event.set(read_resp)
|
cmd.event.set(read_resp)
|
||||||
|
|
||||||
|
|||||||
@@ -23,7 +23,7 @@ THE SOFTWARE.
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
from collections import namedtuple
|
from typing import NamedTuple
|
||||||
|
|
||||||
import cocotb
|
import cocotb
|
||||||
from cocotb.queue import Queue
|
from cocotb.queue import Queue
|
||||||
@@ -34,15 +34,52 @@ from .constants import AxiProt, AxiResp
|
|||||||
from .axil_channels import AxiLiteAWSource, AxiLiteWSource, AxiLiteBSink, AxiLiteARSource, AxiLiteRSink
|
from .axil_channels import AxiLiteAWSource, AxiLiteWSource, AxiLiteBSink, AxiLiteARSource, AxiLiteRSink
|
||||||
from .reset import Reset
|
from .reset import Reset
|
||||||
|
|
||||||
# AXI lite master write
|
|
||||||
AxiLiteWriteCmd = namedtuple("AxiLiteWriteCmd", ["address", "data", "prot", "event"])
|
|
||||||
AxiLiteWriteRespCmd = namedtuple("AxiLiteWriteRespCmd", ["address", "length", "cycles", "prot", "event"])
|
|
||||||
AxiLiteWriteResp = namedtuple("AxiLiteWriteResp", ["address", "length", "resp"])
|
|
||||||
|
|
||||||
# AXI lite master read
|
# AXI lite master write helper objects
|
||||||
AxiLiteReadCmd = namedtuple("AxiLiteReadCmd", ["address", "length", "prot", "event"])
|
class AxiLiteWriteCmd(NamedTuple):
|
||||||
AxiLiteReadRespCmd = namedtuple("AxiLiteReadRespCmd", ["address", "length", "cycles", "prot", "event"])
|
address: int
|
||||||
AxiLiteReadResp = namedtuple("AxiLiteReadResp", ["address", "data", "resp"])
|
data: bytes
|
||||||
|
prot: AxiProt
|
||||||
|
event: Event
|
||||||
|
|
||||||
|
|
||||||
|
class AxiLiteWriteRespCmd(NamedTuple):
|
||||||
|
address: int
|
||||||
|
length: int
|
||||||
|
cycles: int
|
||||||
|
prot: AxiProt
|
||||||
|
event: Event
|
||||||
|
|
||||||
|
|
||||||
|
class AxiLiteWriteResp(NamedTuple):
|
||||||
|
address: int
|
||||||
|
length: int
|
||||||
|
resp: AxiResp
|
||||||
|
|
||||||
|
|
||||||
|
# AXI lite master read helper objects
|
||||||
|
class AxiLiteReadCmd(NamedTuple):
|
||||||
|
address: int
|
||||||
|
length: int
|
||||||
|
prot: AxiProt
|
||||||
|
event: Event
|
||||||
|
|
||||||
|
|
||||||
|
class AxiLiteReadRespCmd(NamedTuple):
|
||||||
|
address: int
|
||||||
|
length: int
|
||||||
|
cycles: int
|
||||||
|
prot: AxiProt
|
||||||
|
event: Event
|
||||||
|
|
||||||
|
|
||||||
|
class AxiLiteReadResp(NamedTuple):
|
||||||
|
address: int
|
||||||
|
data: bytes
|
||||||
|
resp: AxiResp
|
||||||
|
|
||||||
|
def __bytes__(self):
|
||||||
|
return self.data
|
||||||
|
|
||||||
|
|
||||||
class AxiLiteMasterWrite(Reset):
|
class AxiLiteMasterWrite(Reset):
|
||||||
@@ -498,7 +535,7 @@ class AxiLiteMasterRead(Reset):
|
|||||||
self.log.info("Read complete addr: 0x%08x prot: %s resp: %s data: %s",
|
self.log.info("Read complete addr: 0x%08x prot: %s resp: %s data: %s",
|
||||||
cmd.address, cmd.prot, resp, ' '.join((f'{c:02x}' for c in data)))
|
cmd.address, cmd.prot, resp, ' '.join((f'{c:02x}' for c in data)))
|
||||||
|
|
||||||
read_resp = AxiLiteReadResp(cmd.address, data, resp)
|
read_resp = AxiLiteReadResp(cmd.address, bytes(data), resp)
|
||||||
|
|
||||||
cmd.event.set(read_resp)
|
cmd.event.set(read_resp)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user