add wip fetcher

.gitignore

@@ -207,3 +207,4 @@ cython_debug/
 #.idea/
 
 result
+.aider*

example-config.toml

@@ -31,22 +31,22 @@ ip = "dhcp" # Can also be e.g. "192.168.0.123"
 strict = true # allows some wacky configurations, like panels that overlap.
 
 [[display.strings]]
-position = { x = 31, y = 0 }
+position = { x = 64, y = 0 }
-dimensions = { width = 256, height = 64 }
+dimensions = { length = 256, height = 64 }
 rotation = "UPDOWN"
 [[display.strings]]
-position = { x = 32, y = 0 }
+position = { x = 65, y = 0 }
-dimensions = { width = 256, height = 64 }
+dimensions = { length = 256, height = 64 }
 rotation = "LEFTRIGHT"
 [[display.strings]]
-position = { x = 32, y = 64 }
+position = { x = 65, y = 65 }
-dimensions = { width = 256, height = 64 }
+dimensions = { length = 256, height = 64 }
 rotation = "LEFTRIGHT"
 [[display.strings]]
-position = { x = 32, y = 128 }
+position = { x = 65, y = 130 }
-dimensions = { width = 256, height = 64 }
+dimensions = { length = 256, height = 64 }
 rotation = "LEFTRIGHT"
 [[display.strings]]
-position = { x = 32, y = 192 }
+position = { x = 65, y = 195 }
-dimensions = { width = 256, height = 64 }
+dimensions = { length = 256, height = 64 }
 rotation = "LEFTRIGHT"

src/groovylight/common.py

@@ -24,9 +24,11 @@ Rgb111Layout = RGBLayout(1, 1, 1)
 
 
 class RGBView(data.View):
+    def bit_depth(self) -> int:
+        return max(
+            self.red.shape().width, self.green.shape().width, self.blue.shape().width
+        )
 
-    def channel_size(self) -> int:
-        return self.red.shape()
     def channel_slice(self, bit: int) -> Rgb111Layout:
         """Select bits from each channel and use it to form an Rgb111Layout.
         This is useful for BCM stuff, since the bits are sliced to form a bitplane.

src/groovylight/config.py

@@ -1,5 +1,4 @@
 import tomllib
-from pathlib import Path
 
 from groovylight.geom import (
     Coord,
@@ -11,15 +10,15 @@ from groovylight.geom import (
 
 
 class Config:
-    def __init__(self, file: Path) -> None:
+    def __init__(self, file) -> None:
-        with file.open("rb") as f:
+        self.conf = tomllib.load(file)
-            self.conf = tomllib.load(f)
+        print 
 
-        self.geom = DisplayGeometry(self.conf.display.strict)
+        self.geom = DisplayGeometry(strict=self.conf["display"]["strict"])
-        for string in self.conf.display.strings:
+        for string in self.conf["display"]["strings"]:
-            pos = Coord(**string.position)
+            pos = Coord(**string['position'])
-            dims = DisplayDimensions(**string.dimensions)
+            dims = DisplayDimensions(**string["dimensions"])
-            rot = DisplayRotation[string.rotation]
+            rot = DisplayRotation[string["rotation"]]
             disp = DisplayString(pos, dims, rot)
             self.geom.add_string(disp)
-        self.network = self.conf.network  # FIXME: parse this properly.
+        self.network = self.conf['network']  # FIXME: parse this properly.

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,
+        )

src/groovylight/geom.py

@@ -192,6 +192,6 @@ class DisplayGeometry:
 
         if self.strict:
             for e in self._strings:
-                if e.intersects(s):
+                if e.intersects(s.bbox):
                     raise RuntimeError(f"node {e} intersects with {s}")
         self._strings.append(s)

src/groovylight/hub75.py

@@ -1,13 +1,17 @@
-from amaranth import Module, Cat, Mux, Print, ShapeLike, Signal, Assert, Array
+from amaranth import Module, Cat, Mux, ShapeLike, Signal, Assert, Array
 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.utils import ceil_log2
+import logging
 
 from .common import Rgb666Layout, Hub75Stream, Hub75Ctrl, Hub75Data, Rgb888Layout
 
 
+logger = logging.getLogger(__name__)
+
+
 class SwapBuffer(wiring.Component):
     """A pair of BRAMs for holdling line data that are swapped between using an external signal.
 
@@ -25,7 +29,8 @@ class SwapBuffer(wiring.Component):
 
     """
 
-    def __init__(self, shape: ShapeLike, depth: int):
+    def __init__(self, shape: ShapeLike, depth: int, init=[]):
+        self.mem_init = init
         super().__init__(
             {
                 "selector": In(1),
@@ -44,10 +49,10 @@ class SwapBuffer(wiring.Component):
         m = Module()
 
         m.submodules.bram0 = self.bram0 = Memory(
-            shape=self.data_shape, depth=self.depth, init=[]
+            shape=self.data_shape, depth=self.depth, init=self.mem_init
         )
         m.submodules.bram1 = self.bram1 = Memory(
-            shape=self.data_shape, depth=self.depth, init=[]
+            shape=self.data_shape, depth=self.depth, init=self.mem_init
         )
 
         rd0 = self.bram0.read_port()
@@ -93,6 +98,7 @@ class DisplayClock(wiring.Component):
         if startup_delay is None:
             self.startup_delay = 4 if double_fetch else 1  # FIXME: choose right values.
         else:
+            logger.warn("Using custom startup delay. This is not tested")
             self.startup_delay = startup_delay
 
         super().__init__(
@@ -138,6 +144,65 @@ class DisplayClock(wiring.Component):
         return m
 
 
+class DisplayStrober(wiring.Component):
+    """Generates ~OE/blank strobes to expose the latched-in data."""
+
+    def __init__(
+        self,
+        n_bits: int = 8,
+        base_expose_time: int = None,
+        *,
+        periods=None,
+        src_loc_at=0,
+    ):
+        self.n_bits = n_bits
+        self.base_expose_time = base_expose_time
+        if periods is None:
+            # compute durations
+            self.periods = Array([base_expose_time << x for x in range(n_bits)])
+        else:
+            assert len(periods) == n_bits, "must have one period for each bit"
+            if base_expose_time is not None:
+                logger.warn(
+                    "Received both base_expose_time and predifined periods. Ignoring base_expose_time"
+                )
+            self.periods = Array(periods)
+        logger.info("Strobe Clock Periods = %s", self.periods)
+        super().__init__(
+            {
+                "start": In(1),
+                "done": Out(1),
+                "blank": Out(1),
+                "bcm": In(ceil_log2(n_bits - 1)),
+            },
+            src_loc_at=src_loc_at,
+        )
+
+    def elaborate(self, platform: Platform) -> Module:
+        m = Module()
+
+        counter = Signal(32)
+        _bcm = Signal(self.bcm)  # we save this value so that the coordinator can
+        # start the next line while this is running.
+
+        with m.FSM():
+            with m.State("init"):
+                m.d.sync += [counter.eq(0), self.done.eq(0), _bcm.eq(self.bcm)]
+                m.d.comb += self.blank.eq(1)
+                with m.If(self.start):
+                    m.next = "run"
+            with m.State("run"):
+                m.d.comb += self.blank.eq(0)
+                m.d.sync += counter.eq(counter + 1)
+                with m.If(counter == self.periods[_bcm]):
+                    m.next = "done"
+            with m.State("done"):
+                m.d.sync += self.done.eq(1)
+                m.d.comb += self.blank.eq(1)
+                m.next = "init"
+        return m
+
+
 class Hub75DataDriver(wiring.Component):
     def __init__(
         self,
@@ -248,7 +313,6 @@ class Hub75DataDriver(wiring.Component):
 
         return m
 
-
 class Hub75Coordinator(wiring.Component):
     """A shared-control hub75 driver"""
 

src/groovylight/main.py

@@ -2,6 +2,7 @@
 
 import logging
 import argparse
+from groovylight.config import Config
 
 logger = logging.getLogger(__loader__.name)
 
@@ -10,7 +11,10 @@ def setup_logger(args):
     root_logger = logging.getLogger()
     handler = logging.StreamHandler()
 
-    formatter = logging.Formatter(style="{", fmt="{levelname:s}: {name:s}: {message:s}")
+    formatter = logging.Formatter(
+        style=r"{", fmt=r"{levelname:s}: {name:s}: {message:s}"
+    )
+
     handler.setFormatter(formatter)
 
     root_logger.addHandler(handler)
@@ -38,10 +42,21 @@ def main():
         metavar="FILE",
     )
 
+    parser.add_argument(
+        "config",
+        help="Configuration file",
+        type=argparse.FileType("rb"),
+        metavar="FILE",
+    )
+
     args = parser.parse_args()
 
     setup_logger(args)
 
+    conf = Config(args.config)
+    print(conf)
+
+
 
 if __name__ == "__main__":
     main()

src/groovylight/tests/test_common.py

@@ -7,14 +7,14 @@ from groovylight.common import Rgb888Layout, Rgb666Layout, RGBView
 def test_rgbview():
     rgb = Rgb888Layout(0xAABBCC)
 
-    assert rgb.channel_size() == unsigned(8)
+    assert rgb.bit_depth() == 8
 
     rgb18 = Rgb666Layout(0x2DEFD)
 
     slice = rgb.channel_slice(1)
     assert isinstance(slice, RGBView), "channel_slice should return another rgbview"
 
-    assert slice.channel_size() == unsigned(1), "channel_slice channel size should be 1"
+    assert slice.bit_depth() == 1, "channel_slice channel size should be 1"
     assert isinstance(
         rgb18.channel_slice(5), RGBView
     ), "channel_slice should return another rgbview"

src/groovylight/tests/test_hub75.py

@@ -90,7 +90,6 @@ def test_datadriver(bcm):
             assert counter >= 0, "should not do more than 128 clocks"
             e0 = ctx.get(mem.data[counter << 1])
             e1 = ctx.get(mem.data[(counter << 1) + 1])
-            print(counter)
             for r, e in [(rgb0, e0), (rgb1, e1)]:
                 assert r.red == (e.red >> bcm) & 1
                 assert r.green == (e.green >> bcm) & 1
@@ -133,7 +132,6 @@ def test_datadriver_single(bcm):
         ):
             assert counter >= 0, "should not do more than 128 clocks"
             e0, e1 = ctx.get(mem.data[counter])
-            print(counter)
             for r, e in [(rgb0, e0), (rgb1, e1)]:
                 assert r.red == (e.red >> bcm) & 1
                 assert r.green == (e.green >> bcm) & 1