Skip to content
Permalink

Comparing changes

Choose two branches to see what’s changed or to start a new pull request. If you need to, you can also or learn more about diff comparisons.

Open a pull request

Create a new pull request by comparing changes across two branches. If you need to, you can also . Learn more about diff comparisons here.
base repository: m-labs/nmigen
Failed to load repositories. Confirm that selected base ref is valid, then try again.
Loading
base: c7d13b78e5c4
Choose a base ref
...
head repository: m-labs/nmigen
Failed to load repositories. Confirm that selected head ref is valid, then try again.
Loading
compare: d393c5ec644b
Choose a head ref
  • 1 commit
  • 2 files changed
  • 1 contributor

Commits on May 26, 2019

  1. build.res: add ConstraintManager.

    Jean-François Nguyen authored and whitequark committed May 26, 2019
    Copy the full SHA
    d393c5e View commit details
Showing with 357 additions and 0 deletions.
  1. +172 −0 nmigen/build/res.py
  2. +185 −0 nmigen/test/test_build_res.py
172 changes: 172 additions & 0 deletions nmigen/build/res.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,172 @@
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)
185 changes: 185 additions & 0 deletions nmigen/test/test_build_res.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,185 @@
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)