Hardware
/
litex
mirror of https://github.com/enjoy-digital/litex.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

interconnect/csr/EventManager: simpifly/cleanup code that documents CSRs and always enable documentation. 

A read_only mode has been added to CSRStatus to allow enabling writes on pending register and get the written
data used to clear events.
This commit is contained in:
Florent Kermarrec 2020-11-18 13:06:55 +01:00
parent 30b2f187f4
commit 470b6873ca
4 changed files with 78 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
litex/soc/cores/i2s.py
View File

@ -198,7 +198,7 @@ class S7I2S(Module, AutoCSR, AutoDoc):
]
# Interrupts
self.submodules.ev = EventManager(document_fields=document_interrupts)
self.submodules.ev = EventManager()
if hasattr(pads, 'rx'):
self.ev.rx_ready = EventSourcePulse(description="Indicates FIFO is ready to read") # Rising edge triggered
self.ev.rx_error = EventSourcePulse(description="Indicates an Rx error has happened (over/underflow)")

2
litex/soc/integration/export.py
View File

@ -219,7 +219,7 @@ def get_csr_header(regions, constants, csr_base=None, with_access_functions=True
for csr in region.obj:
nr = (csr.size + region.busword - 1)//region.busword
r += _get_rw_functions_c(name + "_" + csr.name, origin, nr, region.busword, alignment,
isinstance(csr, CSRStatus), with_access_functions)
getattr(csr, "read_only", False), with_access_functions)
origin += alignment//8*nr
if hasattr(csr, "fields"):
for field in csr.fields.fields:

11
litex/soc/interconnect/csr.py
View File

@ -286,16 +286,19 @@ class CSRStatus(_CompoundCSR):
The value of the CSRStatus register.
"""
def __init__(self, size=1, reset=0, fields=[], name=None, description=None):
def __init__(self, size=1, reset=0, fields=[], name=None, description=None, read_only=True):
if fields != []:
self.fields = CSRFieldAggregate(fields, CSRAccess.ReadOnly)
size = self.fields.get_size()
reset = self.fields.get_reset()
_CompoundCSR.__init__(self, size, name)
self.description = description
self.read_only = read_only
self.status = Signal(self.size, reset=reset)
self.we = Signal()
self.re = Signal()
if not read_only:
self.r = Signal(self.size)
for field in fields:
self.comb += self.status[field.offset:field.offset + field.size].eq(getattr(self.fields, field.name))
@ -306,8 +309,12 @@ class CSRStatus(_CompoundCSR):
sc = CSR(nbits, self.name + str(i) if nwords > 1 else self.name)
self.comb += sc.w.eq(self.status[i*busword:i*busword+nbits])
self.simple_csrs.append(sc)
if not self.read_only:
lo = i*busword
hi = lo+nbits
self.sync += If(sc.re, self.r[lo:hi].eq(sc.r))
self.comb += self.we.eq(sc.we)
self.comb += self.re.eq(sc.re)
self.sync += self.re.eq(sc.re)
def read(self):
"""Read method for simulation."""

147
litex/soc/interconnect/csr_eventmanager.py
View File

@ -135,116 +135,87 @@ class EventManager(Module, AutoCSR):
asserted.
"""
def __init__(self, document_fields=False):
def __init__(self):
self.irq = Signal()
self.document_fields = document_fields
def do_finalize(self):
def source_description(src):
if hasattr(src, "name") and src.name is not None:
base_text = "`1` if a `{}` event occurred. ".format(src.name)
else:
base_text = "`1` if a this particular event occurred. "
if hasattr(src, "description") and src.description is not None:
return src.description
elif isinstance(src, EventSourceLevel):
return base_text + "This Event is **level triggered** when the signal is **high**."
elif isinstance(src, EventSourcePulse):
return base_text + "This Event is triggered on a **rising** edge."
elif isinstance(src, EventSourceProcess):
return base_text + "This Event is triggered on a **falling** edge."
else:
return base_text + "This Event uses an unknown method of triggering."
sources_u = [v for k, v in xdir(self, True) if isinstance(v, _EventSource)]
sources = sorted(sources_u, key=lambda x: x.duid)
n = len(sources)
if self.document_fields:
# annotate status
# Common
def get_source_name(src, i):
r = getattr(src, "name", None)
if r is None:
r = f"event{i}"
return r
def get_pending_source_description(src):
r = ""
if getattr(src, "name", None) is not None:
r += "`1` if a `{}` event occurred. ".format(src.name)
else:
r += "`1` if a this particular event occurred. "
if getattr(src, "description", None) is not None:
return src.description
elif isinstance(src, EventSourceLevel):
return r + "This Event is **level triggered** when the signal is **high**."
elif isinstance(src, EventSourcePulse):
return r + "This Event is triggered on a **rising** edge."
elif isinstance(src, EventSourceProcess):
return r + "This Event is triggered on a **falling** edge."
else:
return r + "This Event uses an unknown method of triggering."
# Status register
fields = []
for i, source in enumerate(sources):
if source.description == None:
desc = "This register contains the current raw level of the {} event trigger. Writes to this register have no effect.".format(str(source.name))
else:
desc = source.description
if hasattr(source, "name") and source.name is not None:
fields.append(CSRField(
name=source.name,
size=1,
description="Level of the ``{}`` event".format(source.name)))
else:
fields.append(CSRField(
name="event{}".format(i),
size=1,
description="Level of the ``event{}`` event".format(i)))
# Get source name and use default if None.
name = get_source_name(source, i)
# Get description and use default description if None.
desc = getattr(source, "description", None)
if desc is None:
desc = "This register contains the current raw level of the {} event trigger. Writes to this register have no effect.".format(str(name))
# Add CSRField
fields.append(CSRField(name=name, size=1, description=f"Level of the ``{name}`` event"))
self.status = CSRStatus(n, description=desc, fields=fields)
# annotate pending
# Pending Register
fields = []
for i, source in enumerate(sources):
if source.description is None:
desc = "When a {} event occurs, the corresponding bit will be set in this register. To clear the Event, set the corresponding bit in this register.".format(str(source.name))
else:
desc = source.description
# Get source name and use default if None.
name = get_source_name(source, i)
# Get description and use default description if None.
desc = getattr(source, "description", None)
if desc is None:
desc = "When a {} event occurs, the corresponding bit will be set in this register. To clear the Event, set the corresponding bit in this register.".format(str(name))
# Add CSRField
fields.append(CSRField(name=name, size=1, description=get_pending_source_description(source)))
self.pending = CSRStatus(n, description=desc, fields=fields, read_only=False)
if hasattr(source, "name") and source.name is not None:
fields.append(CSRField(
name=source.name,
size=1,
description=source_description(source)))
else:
fields.append(CSRField(
name="event{}".format(i),
size=1,
description=source_description(source)))
self.pending = CSRStatus(n, description=desc, fields=fields)
# annotate enable
# Enable Register
fields = []
for i, source in enumerate(sources):
if source.description is None:
desc = "This register enables the corresponding {} events. Write a ``0`` to this register to disable individual events.".format(str(source.name))
else:
desc = source.description
if hasattr(source, "name") and source.name is not None:
fields.append(CSRField(
name=source.name,
offset=i,
description="Write a ``1`` to enable the ``{}`` Event".format(source.name)))
else:
fields.append(CSRField(
name="event{}".format(i),
offset=i,
description="Write a ``1`` to enable the ``{}`` Event".format(i)))
# Get source name and use default if None.
name = get_source_name(source, i)
# Get description and use default description if None.
desc = getattr(source, "description", None)
if desc is None:
desc = "This register enables the corresponding {} events. Write a ``0`` to this register to disable individual events.".format(str(name))
# Add CSRField
fields.append(CSRField(name=name, offset=i, description=f"Write a ``1`` to enable the ``{name}`` Event"))
self.enable = CSRStorage(n, description=desc, fields=fields)
# Connect Events/Fields
for i, source in enumerate(sources):
if source.name == None:
src_name = "event{}".format(i)
else:
src_name = source.name
# Get source name and use default if None.
name = get_source_name(source, i)
self.comb += [
getattr(self.status.fields, src_name).eq(source.status),
getattr(self.pending.fields, src_name).eq(source.pending),
If(self.pending.re & getattr(self.pending.fields, src_name), source.clear.eq(1)),
]
irqs = [self.pending.status[i] & self.enable.storage[i] for i in range(n)]
else:
self.status = CSR(n)
self.pending = CSR(n)
self.enable = CSRStorage(n)
for i, source in enumerate(sources):
self.comb += [
self.status.w[i].eq(source.status),
getattr(self.status.fields, name).eq(source.status),
getattr(self.pending.fields, name).eq(source.pending),
If(self.pending.re & self.pending.r[i], source.clear.eq(1)),
self.pending.w[i].eq(source.pending)
]
irqs = [self.pending.w[i] & self.enable.storage[i] for i in range(n)]
irqs = [self.pending.status[i] & self.enable.storage[i] for i in range(n)]
self.comb += self.irq.eq(reduce(or_, irqs))
def __setattr__(self, name, value):