luke
/
manta
mirror of https://github.com/fischermoseley/manta.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

finish playback module pipelining

This commit is contained in:
Fischer Moseley 2023-12-28 19:20:06 -08:00
parent 9867e369cb
commit 90a5dba665
2 changed files with 13 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pyproject.toml
View File

@ -11,7 +11,6 @@ dependencies = [
"PyYAML", "PyYAML",
"pyserial", "pyserial",
"pyvcd", "pyvcd",
"scapy"
] ]
requires-python = ">=3.7" requires-python = ">=3.7"

42
src/manta/logic_analyzer_core.py
View File

@ -509,12 +509,7 @@ class LogicAnalyzerPlayback(Elaboratable):
# State Machine # State Machine
self.start = Signal(1) self.start = Signal(1)
self.done = Signal(1) self.idle = Signal(1, reset=1)
self.states = {
"IDLE" : 0,
"RUN" : 1,
"DONE" : 2
}
# Top-Level Probe signals # Top-Level Probe signals
self.top_level_probes = {} self.top_level_probes = {}
@ -528,37 +523,24 @@ class LogicAnalyzerPlayback(Elaboratable):
init=self.data, init=self.data,
) )
self.read_port = self.mem.read_port() self.read_port = self.mem.read_port(domain="comb")
def elaborate(self, platform): def elaborate(self, platform):
m = Module() m = Module()
m.submodules["mem"] = self.mem m.submodules["mem"] = self.mem
m.d.comb += self.read_port.en.eq(1)
state = Signal(range(len(self.states)))
addr = self.read_port.addr
# Run state machine # Run state machine
with m.If(state == self.states["IDLE"]): with m.If(self.idle):
with m.If(self.start): with m.If(self.start):
m.d.sync += state.eq(self.states["RUN"]) m.d.sync += self.idle.eq(0)
m.d.sync += addr.eq(0)
with m.Elif(state == self.states["RUN"]): with m.Else():
with m.If(addr < self.config["sample_depth"]): with m.If(self.read_port.addr == self.config["sample_depth"] - 1):
m.d.sync += addr.eq(addr + 1) m.d.sync += self.idle.eq(1)
m.d.sync += self.read_port.addr.eq(0)
with m.Else(): with m.Else():
m.d.sync += state.eq(self.states["DONE"]) m.d.sync += self.read_port.addr.eq(self.read_port.addr + 1)
m.d.sync += self.done.eq(1)
with m.Else(state == self.states["DONE"]):
with m.If(self.start):
m.d.sync += self.done.eq(0)
m.d.sync += state.eq(self.states["RUN"])
m.d.sync += addr.eq(0)
# Assign the probe values by part-selecting from the data port # Assign the probe values by part-selecting from the data port
lower = 0 lower = 0
@ -566,7 +548,7 @@ class LogicAnalyzerPlayback(Elaboratable):
signal = self.top_level_probes[name] signal = self.top_level_probes[name]
# Set output probe to zero if we're not # Set output probe to zero if we're not
with m.If(~self.done): with m.If(~self.idle):
m.d.comb += signal.eq(self.read_port.data[lower : lower + width]) m.d.comb += signal.eq(self.read_port.data[lower : lower + width])
with m.Else(): with m.Else():
@ -574,5 +556,7 @@ class LogicAnalyzerPlayback(Elaboratable):
lower += width lower += width
return m
def get_top_level_ports(self): def get_top_level_ports(self):
return [self.enable, self.done] + list(self.top_level_probes.values()) return [self.start, self.idle] + list(self.top_level_probes.values())