--- /dev/null
+from collections import OrderedDict
+
+from .. import *
+from ..hdl.rec import *
+from ..lib.io import *
+
+from .dsl import *
+
+
+__all__ = ["ConstraintError", "ConstraintManager"]
+
+
+class ConstraintError(Exception):
+ pass
+
+
+class ConstraintManager:
+ def __init__(self, resources):
+ self.resources = OrderedDict()
+ self.requested = OrderedDict()
+ self.clocks = OrderedDict()
+
+ self._ports = []
+ self._tristates = []
+ self._diffpairs = []
+
+ self.add_resources(resources)
+
+ def add_resources(self, resources):
+ for r in resources:
+ if not isinstance(r, Resource):
+ raise TypeError("Object {!r} is not a Resource".format(r))
+ if (r.name, r.number) in self.resources:
+ raise NameError("Trying to add {!r}, but {!r} has the same name and number"
+ .format(r, self.resources[r.name, r.number]))
+ self.resources[r.name, r.number] = r
+
+ def add_clock(self, name, number, frequency):
+ resource = self.lookup(name, number)
+ if isinstance(resource.io[0], Subsignal):
+ raise ConstraintError("Cannot constrain frequency of resource {}#{} because it has "
+ "subsignals"
+ .format(resource.name, resource.number, frequency))
+ if (resource.name, resource.number) in self.clocks:
+ other = self.clocks[resource.name, resource.number]
+ raise ConstraintError("Resource {}#{} is already constrained to a frequency of "
+ "{:f} MHz"
+ .format(resource.name, resource.number, other / 1e6))
+ self.clocks[resource.name, resource.number] = frequency
+
+ def lookup(self, name, number):
+ if (name, number) not in self.resources:
+ raise NameError("Resource {}#{} does not exist"
+ .format(name, number))
+ return self.resources[name, number]
+
+ def request(self, name, number, dir=None, xdr=None):
+ resource = self.lookup(name, number)
+ if (resource.name, resource.number) in self.requested:
+ raise ConstraintError("Resource {}#{} has already been requested"
+ .format(name, number))
+
+ def resolve_dir_xdr(subsignal, dir, xdr):
+ if isinstance(subsignal.io[0], Subsignal):
+ if dir is None:
+ dir = dict()
+ if xdr is None:
+ xdr = dict()
+ if not isinstance(dir, dict):
+ raise TypeError("Directions must be a dict, not {!r}, because {!r} "
+ "has subsignals"
+ .format(dir, subsignal))
+ if not isinstance(xdr, dict):
+ raise TypeError("Data rate must be a dict, not {!r}, because {!r} "
+ "has subsignals"
+ .format(xdr, subsignal))
+ for sub in subsignal.io:
+ sub_dir = dir.get(sub.name, None)
+ sub_xdr = xdr.get(sub.name, None)
+ dir[sub.name], xdr[sub.name] = resolve_dir_xdr(sub, sub_dir, sub_xdr)
+ else:
+ if dir is None:
+ dir = subsignal.io[0].dir
+ if xdr is None:
+ xdr = 1
+ if dir not in ("i", "o", "io"):
+ raise TypeError("Direction must be one of \"i\", \"o\" or \"io\", not {!r}"
+ .format(dir))
+ if subsignal.io[0].dir != "io" and dir != subsignal.io[0].dir:
+ raise ValueError("Direction of {!r} cannot be changed from \"{}\" to \"{}\"; "
+ "direction can be changed from \"io\" to \"i\" or from \"io\""
+ "to \"o\""
+ .format(subsignal.io[0], subsignal.io[0].dir, dir))
+ if not isinstance(xdr, int) or xdr < 1:
+ raise ValueError("Data rate of {!r} must be a positive integer, not {!r}"
+ .format(subsignal.io[0], xdr))
+ return dir, xdr
+
+ dir, xdr = resolve_dir_xdr(resource, dir, xdr)
+
+ def get_value(subsignal, dir, xdr, name):
+ if isinstance(subsignal.io[0], Subsignal):
+ fields = OrderedDict()
+ for sub in subsignal.io:
+ fields[sub.name] = get_value(sub, dir[sub.name], xdr[sub.name],
+ "{}__{}".format(name, sub.name))
+ rec = Record([
+ (f_name, f.layout) for (f_name, f) in fields.items()
+ ], fields=fields, name=name)
+ return rec
+ elif isinstance(subsignal.io[0], DiffPairs):
+ pairs = subsignal.io[0]
+ return Pin(len(pairs), dir, xdr, name=name)
+ elif isinstance(subsignal.io[0], Pins):
+ pins = subsignal.io[0]
+ return Pin(len(pins), dir, xdr, name=name)
+ else:
+ assert False # :nocov:
+
+ value_name = "{}_{}".format(resource.name, resource.number)
+ value = get_value(resource, dir, xdr, value_name)
+
+ def match_constraints(value, subsignal):
+ if isinstance(subsignal.io[0], Subsignal):
+ for sub in subsignal.io:
+ yield from match_constraints(value[sub.name], sub)
+ else:
+ assert isinstance(value, Pin)
+ yield (value, subsignal.io[0], subsignal.extras)
+
+ for (pin, io, extras) in match_constraints(value, resource):
+ if isinstance(io, DiffPairs):
+ p = Signal(pin.width, name="{}_p".format(pin.name))
+ n = Signal(pin.width, name="{}_n".format(pin.name))
+ self._diffpairs.append((pin, p, n))
+ self._ports.append((p, io.p.names, extras))
+ self._ports.append((n, io.n.names, extras))
+ elif isinstance(io, Pins):
+ if pin.dir == "io":
+ port = Signal(pin.width, name="{}_io".format(pin.name))
+ self._tristates.append((pin, port))
+ else:
+ port = getattr(pin, pin.dir)
+ self._ports.append((port, io.names, extras))
+ else:
+ assert False # :nocov:
+
+ self.requested[resource.name, resource.number] = value
+ return value
+
+ def iter_ports(self):
+ for port, pins, extras in self._ports:
+ yield port
+
+ def iter_port_constraints(self):
+ for port, pins, extras in self._ports:
+ yield (port.name, pins, extras)
+
+ def iter_clock_constraints(self):
+ for name, number in self.clocks.keys() & self.requested.keys():
+ resource = self.resources[name, number]
+ pin = self.requested[name, number]
+ period = self.clocks[name, number]
+ if pin.dir == "io":
+ raise ConstraintError("Cannot constrain frequency of resource {}#{} because "
+ "it has been requested as a tristate buffer"
+ .format(name, number))
+ if isinstance(resource.io[0], DiffPairs):
+ port_name = "{}_p".format(pin.name)
+ else:
+ port_name = getattr(pin, pin.dir).name
+ yield (port_name, period)
--- /dev/null
+from .. import *
+from ..hdl.rec import *
+from ..lib.io import *
+from ..build.dsl import *
+from ..build.res import *
+from .tools import *
+
+
+class ConstraintManagerTestCase(FHDLTestCase):
+ def setUp(self):
+ self.resources = [
+ Resource("clk100", 0, DiffPairs("H1", "H2", dir="i")),
+ Resource("clk50", 0, Pins("K1")),
+ Resource("user_led", 0, Pins("A0", dir="o")),
+ Resource("i2c", 0,
+ Subsignal("scl", Pins("N10", dir="o")),
+ Subsignal("sda", Pins("N11"))
+ )
+ ]
+ self.cm = ConstraintManager(self.resources)
+
+ def test_basic(self):
+ self.assertEqual(self.cm.resources, {
+ ("clk100", 0): self.resources[0],
+ ("clk50", 0): self.resources[1],
+ ("user_led", 0): self.resources[2],
+ ("i2c", 0): self.resources[3]
+ })
+
+ def test_add_resources(self):
+ new_resources = [
+ Resource("user_led", 1, Pins("A1", dir="o"))
+ ]
+ self.cm.add_resources(new_resources)
+ self.assertEqual(self.cm.resources, {
+ ("clk100", 0): self.resources[0],
+ ("clk50", 0): self.resources[1],
+ ("user_led", 0): self.resources[2],
+ ("i2c", 0): self.resources[3],
+ ("user_led", 1): new_resources[0]
+ })
+
+ def test_lookup(self):
+ r = self.cm.lookup("user_led", 0)
+ self.assertIs(r, self.cm.resources["user_led", 0])
+
+ def test_request_basic(self):
+ r = self.cm.lookup("user_led", 0)
+ user_led = self.cm.request("user_led", 0)
+
+ self.assertIsInstance(user_led, Pin)
+ self.assertEqual(user_led.name, "user_led_0")
+ self.assertEqual(user_led.width, 1)
+ self.assertEqual(user_led.dir, "o")
+
+ ports = list(self.cm.iter_ports())
+ self.assertEqual(len(ports), 1)
+ self.assertIs(user_led.o, ports[0])
+
+ self.assertEqual(list(self.cm.iter_port_constraints()), [
+ ("user_led_0__o", ["A0"], [])
+ ])
+
+ def test_request_with_dir(self):
+ i2c = self.cm.request("i2c", 0, dir={"sda": "o"})
+ self.assertIsInstance(i2c, Record)
+ self.assertIsInstance(i2c.sda, Pin)
+ self.assertEqual(i2c.sda.dir, "o")
+
+ def test_request_tristate(self):
+ i2c = self.cm.request("i2c", 0)
+ self.assertEqual(i2c.sda.dir, "io")
+
+ ports = list(self.cm.iter_ports())
+ self.assertEqual(len(ports), 2)
+ self.assertIs(i2c.scl.o, ports[0]),
+ self.assertEqual(ports[1].name, "i2c_0__sda_io")
+ self.assertEqual(ports[1].nbits, 1)
+
+ self.assertEqual(self.cm._tristates, [(i2c.sda, ports[1])])
+ self.assertEqual(list(self.cm.iter_port_constraints()), [
+ ("i2c_0__scl__o", ["N10"], []),
+ ("i2c_0__sda_io", ["N11"], [])
+ ])
+
+ def test_request_diffpairs(self):
+ clk100 = self.cm.request("clk100", 0)
+ self.assertIsInstance(clk100, Pin)
+ self.assertEqual(clk100.dir, "i")
+ self.assertEqual(clk100.width, 1)
+
+ ports = list(self.cm.iter_ports())
+ self.assertEqual(len(ports), 2)
+ p, n = ports
+ self.assertEqual(p.name, "clk100_0_p")
+ self.assertEqual(p.nbits, clk100.width)
+ self.assertEqual(n.name, "clk100_0_n")
+ self.assertEqual(n.nbits, clk100.width)
+
+ self.assertEqual(self.cm._diffpairs, [(clk100, p, n)])
+ self.assertEqual(list(self.cm.iter_port_constraints()), [
+ ("clk100_0_p", ["H1"], []),
+ ("clk100_0_n", ["H2"], [])
+ ])
+
+ def test_add_clock(self):
+ self.cm.add_clock("clk100", 0, 10e6)
+ self.assertEqual(self.cm.clocks["clk100", 0], 10e6)
+ self.cm.add_clock("clk50", 0, 5e6)
+
+ clk100 = self.cm.request("clk100", 0)
+ clk50 = self.cm.request("clk50", 0, dir="i")
+ self.assertEqual(list(sorted(self.cm.iter_clock_constraints())), [
+ ("clk100_0_p", 10e6),
+ ("clk50_0__i", 5e6)
+ ])
+
+ def test_wrong_resources(self):
+ with self.assertRaises(TypeError, msg="Object 'wrong' is not a Resource"):
+ self.cm.add_resources(['wrong'])
+
+ def test_wrong_resources_duplicate(self):
+ with self.assertRaises(NameError,
+ msg="Trying to add (resource user_led 0 (pins o A1) ), but "
+ "(resource user_led 0 (pins o A0) ) has the same name and number"):
+ self.cm.add_resources([Resource("user_led", 0, Pins("A1", dir="o"))])
+
+ def test_wrong_lookup(self):
+ with self.assertRaises(NameError,
+ msg="Resource user_led#1 does not exist"):
+ r = self.cm.lookup("user_led", 1)
+
+ def test_wrong_frequency_subsignals(self):
+ with self.assertRaises(ConstraintError,
+ msg="Cannot constrain frequency of resource i2c#0 because "
+ "it has subsignals"):
+ self.cm.add_clock("i2c", 0, 10e6)
+
+ def test_wrong_frequency_tristate(self):
+ with self.assertRaises(ConstraintError,
+ msg="Cannot constrain frequency of resource clk50#0 because "
+ "it has been requested as a tristate buffer"):
+ self.cm.add_clock("clk50", 0, 20e6)
+ clk50 = self.cm.request("clk50", 0)
+ list(self.cm.iter_clock_constraints())
+
+ def test_wrong_frequency_duplicate(self):
+ with self.assertRaises(ConstraintError,
+ msg="Resource clk100#0 is already constrained to a frequency of 10.000000 MHz"):
+ self.cm.add_clock("clk100", 0, 10e6)
+ self.cm.add_clock("clk100", 0, 5e6)
+
+ def test_wrong_request_duplicate(self):
+ with self.assertRaises(ConstraintError,
+ msg="Resource user_led#0 has already been requested"):
+ self.cm.request("user_led", 0)
+ self.cm.request("user_led", 0)
+
+ def test_wrong_request_with_dir(self):
+ with self.assertRaises(TypeError,
+ msg="Direction must be one of \"i\", \"o\" or \"io\", not 'wrong'"):
+ user_led = self.cm.request("user_led", 0, dir="wrong")
+
+ def test_wrong_request_with_dir_io(self):
+ with self.assertRaises(ValueError,
+ msg="Direction of (pins o A0) cannot be changed from \"o\" to \"i\"; direction "
+ "can be changed from \"io\" to \"i\" or from \"io\"to \"o\""):
+ user_led = self.cm.request("user_led", 0, dir="i")
+
+ def test_wrong_request_with_dir_dict(self):
+ with self.assertRaises(TypeError,
+ msg="Directions must be a dict, not 'i', because (resource i2c 0 (subsignal scl "
+ "(pins o N10) ) (subsignal sda (pins io N11) ) ) has subsignals"):
+ i2c = self.cm.request("i2c", 0, dir="i")
+
+ def test_wrong_request_with_wrong_xdr(self):
+ with self.assertRaises(ValueError,
+ msg="Data rate of (pins o A0) must be a positive integer, not 0"):
+ user_led = self.cm.request("user_led", 0, xdr=0)
+
+ def test_wrong_request_with_xdr_dict(self):
+ with self.assertRaises(TypeError,
+ msg="Data rate must be a dict, not 2, because (resource i2c 0 (subsignal scl "
+ "(pins o N10) ) (subsignal sda (pins io N11) ) ) has subsignals"):
+ i2c = self.cm.request("i2c", 0, xdr=2)
projects / nmigen.git / commitdiff