add wip fetcher

src/groovylight/fetcher.py

@@ -0,0 +1,82 @@
+# Holds various implementations of a "fetcher", which is given a display string
+# to know its location.
+# during operation, it is given a row index, and responds with the data.
+
+
+from amaranth import Module, Cat, Mux, ShapeLike, Signal, Assert, Array, unsigned
+from amaranth.build import Platform
+from amaranth.lib import wiring, data
+from amaranth.lib.wiring import In, Out
+from amaranth.lib.memory import Memory, ReadPort, WritePort
+from amaranth.lib import stream
+from amaranth.utils import ceil_log2
+import logging
+
+from .common import Rgb666Layout, Hub75Stream, Hub75Ctrl, Hub75Data, Rgb888Layout
+from .geom import DisplayRotation, DisplayString
+
+
+logger = logging.getLogger(__name__)
+
+
+CoordLayout = data.StructLayout({"x": unsigned(32), "y": unsigned(32)})
+
+
+class AddressGenerator(wiring.Component):
+    """Generates (x,y) sequences corresponding to a display row."""
+
+    def __init__(self, geom: DisplayString, *, src_loc_at=0):
+        self.geom = geom
+        super().__init__(
+            {
+                "coordstream": Out(stream.Signature(CoordLayout)),
+                "start": In(1),
+                "done": Out(1),
+                "addr": In(geom.dimensions.addr_bits),
+            },
+            src_loc_at=src_loc_at,
+        )
+
+    def elaborate(self, platform: Platform) -> Module:
+        m = Module()
+
+        counter = Signal(self.geom.dimensions.length)
+
+        # based on the geometry we generate x,y pairs.
+
+        with m.FSM():
+            with m.State("init"):
+                m.d.comb += self.done.eq(0)
+                m.d.sync += counter.eq(0)
+                with m.If(self.start):
+                    m.next = "run"
+
+            with m.State("run"):
+                if self.geom.rotation == DisplayRotation.LEFTRIGHT:
+                    # default case, +x.
+                    pass
+                elif self.geom.rotation == DisplayRotation.UPDOWN:
+                    # r 90, +y
+                    pass
+            with m.State("done"):
+                m.d.comb += self.done.eq(1)
+                m.next = "init"
+
+        return m
+
+
+class BasicFetcher(wiring.Component):
+    """A generic fetcher. Takes a function of the form f(x,y: int) -> RGB."""
+
+    def __init__(
+        self, geom: DisplayString, dfunc, data_shape=Rgb888Layout, *, src_loc_at=0
+    ):
+        self.dfunc = dfunc
+        super().__init__(
+            {
+                "pixstream": Out(stream.Signature(Rgb888Layout)),
+                "start": In(1),
+                "addr": In(geom.dimensions.addr_bits),
+            },
+            src_loc_at=src_loc_at,
+        )