| #!/usr/bin/env python3 | |
| # -*- coding: utf-8 -*- | |
| # | |
| # Copyright (C) 2022 F4PGA Authors | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| # | |
| # SPDX-License-Identifier: Apache-2.0 | |
| import argparse | |
| import pickle | |
| import re | |
| from collections import defaultdict, namedtuple | |
| import fasm | |
| from f4pga.utils.quicklogic.pp3.connections import get_name_and_hop | |
| from pathlib import Path | |
| from f4pga.utils.quicklogic.pp3.data_structs import Loc, SwitchboxPinLoc, PinDirection, ConnectionType | |
| from f4pga.utils.quicklogic.pp3.utils import get_quadrant_for_loc | |
| from f4pga.utils.quicklogic.pp3.verilogmodule import VModule | |
| from quicklogic_fasm.qlfasm import load_quicklogic_database, get_db_dir | |
| from quicklogic_fasm.qlfasm import QL732BAssembler | |
| Feature = namedtuple("Feature", "loc typ signature value") | |
| RouteEntry = namedtuple("RouteEntry", "typ stage_id switch_id mux_id sel_id") | |
| MultiLocCellMapping = namedtuple("MultiLocCellMapping", "typ fromlocset toloc pinnames") | |
| class Fasm2Bels(object): | |
| """Class for parsing FASM file and producing BEL representation. | |
| It takes FASM lines and VPR database and converts the data to Basic | |
| Elements and connections between them. It allows converting this data to | |
| Verilog. | |
| """ | |
| class Fasm2BelsException(Exception): | |
| """Exception for Fasm2Bels errors and unsupported features.""" | |
| def __init__(self, message): | |
| self.message = message | |
| def __str__(self): | |
| return self.message | |
| def __init__(self, phy_db, device_name, package_name): | |
| """Prepares required structures for converting FASM to BELs. | |
| Parameters | |
| ---------- | |
| phy_db: dict | |
| A dictionary containing cell_library, loc_map, vpr_tile_types, | |
| vpr_tile_grid, vpr_switchbox_types, vpr_switchbox_grid, | |
| connections, vpr_package_pinmaps | |
| """ | |
| # load phy_db data | |
| self.quadrants = phy_db["phy_quadrants"] | |
| self.cells_library = phy_db["cells_library"] | |
| self.vpr_tile_types = phy_db["tile_types"] | |
| self.vpr_tile_grid = phy_db["phy_tile_grid"] | |
| self.vpr_switchbox_types = phy_db["switchbox_types"] | |
| self.vpr_switchbox_grid = phy_db["switchbox_grid"] | |
| self.connections = phy_db["connections"] | |
| self.device_name = device_name | |
| self.package_name = package_name | |
| self.io_to_fbio = dict() | |
| if self.package_name not in db["package_pinmaps"]: | |
| raise self.Fasm2BelsException( | |
| "ERROR: '{}' is not a vaild package for device '{}'. Valid ones are: {}".format( | |
| self.package_name, self.device_name, ", ".join(db["package_pinmaps"].keys()) | |
| ) | |
| ) | |
| for name, package in db["package_pinmaps"][self.package_name].items(): | |
| self.io_to_fbio[package[0].loc] = name | |
| # Add ASSP to all locations it covers | |
| # TODO maybe this should be added in original vpr_tile_grid | |
| # set all cels in row 1 and column 2 to ASSP | |
| # In VPR grid, the ASSP tile is located in (1, 1) | |
| assplocs = set() | |
| ramlocs = dict() | |
| multlocs = dict() | |
| for phy_loc, tile in self.vpr_tile_grid.items(): | |
| tile_type = self.vpr_tile_types[tile.type] | |
| if "ASSP" in tile_type.cells: | |
| assplocs.add(phy_loc) | |
| if "RAM" in tile_type.cells: | |
| ramcell = [cell for cell in tile.cells if cell.type == "RAM"] | |
| cellname = ramcell[0].name | |
| if cellname not in ramlocs: | |
| ramlocs[cellname] = set() | |
| ramlocs[cellname].add(phy_loc) | |
| if "MULT" in tile_type.cells: | |
| multcell = [cell for cell in tile.cells if cell.type == "MULT"] | |
| cellname = multcell[0].name | |
| if cellname not in multlocs: | |
| multlocs[cellname] = set() | |
| multlocs[cellname].add(phy_loc) | |
| # this map represents the mapping from input name to its inverter name | |
| self.inversionpins = { | |
| "LOGIC": { | |
| "TA1": "TAS1", | |
| "TA2": "TAS2", | |
| "TB1": "TBS1", | |
| "TB2": "TBS2", | |
| "BA1": "BAS1", | |
| "BA2": "BAS2", | |
| "BB1": "BBS1", | |
| "BB2": "BBS2", | |
| "QCK": "QCKS", | |
| } | |
| } | |
| # prepare helper structure for connections | |
| self.connections_by_loc = defaultdict(list) | |
| for connection in self.connections: | |
| self.connections_by_loc[connection.dst].append(connection) | |
| self.connections_by_loc[connection.src].append(connection) | |
| # a mapping from the type of cell FASM line refers to to its parser | |
| self.featureparsers = { | |
| "LOGIC": self.parse_logic_line, | |
| "QMUX": self.parse_logic_line, | |
| "GMUX": self.parse_logic_line, | |
| "INTERFACE": self.parse_interface_line, | |
| "ROUTING": self.parse_routing_line, | |
| "CAND0": self.parse_colclk_line, | |
| "CAND1": self.parse_colclk_line, | |
| "CAND2": self.parse_colclk_line, | |
| "CAND3": self.parse_colclk_line, | |
| "CAND4": self.parse_colclk_line, | |
| "RAM": self.parse_ram_line, | |
| } | |
| # a mapping from cell type to a set of possible pin names | |
| self.pinnames = defaultdict(set) | |
| for celltype in self.cells_library.values(): | |
| typ = celltype.type | |
| for pin in celltype.pins: | |
| self.pinnames[typ].add(pin.name) | |
| # a mapping from cell types that occupy multiple locations | |
| # to a single location | |
| self.multiloccells = {"ASSP": MultiLocCellMapping("ASSP", assplocs, Loc(1, 1, 0), self.pinnames["ASSP"])} | |
| for ram in ramlocs: | |
| self.multiloccells[ram] = MultiLocCellMapping( | |
| ram, ramlocs[ram], list(ramlocs[ram])[0], self.pinnames["RAM"] | |
| ) | |
| for mult in multlocs: | |
| self.multiloccells[mult] = MultiLocCellMapping( | |
| mult, multlocs[mult], list(multlocs[mult])[1], self.pinnames["MULT"] | |
| ) | |
| # helper routing data | |
| self.routingdata = defaultdict(list) | |
| # a dictionary holding bit settings for BELs | |
| self.belinversions = defaultdict(lambda: defaultdict(list)) | |
| # a dictionary holding bit settings for IOs | |
| self.interfaces = defaultdict(lambda: defaultdict(list)) | |
| # a dictionary holding simplified connections between BELs | |
| self.designconnections = defaultdict(dict) | |
| # a dictionary holding hops from routing | |
| self.designhops = defaultdict(dict) | |
| # Clock column drivers (CAND) data | |
| self.colclk_data = defaultdict(lambda: defaultdict(list)) | |
| # A map of clock wires that connect to switchboxes | |
| self.cand_map = defaultdict(lambda: dict()) | |
| # A map of original (loc, pin) to new (loc, pin). Created during | |
| # aggregation of multi-loc cells. | |
| self.org_loc_map = {} | |
| def parse_logic_line(self, feature: Feature): | |
| """Parses a setting for a BEL. | |
| Parameters | |
| ---------- | |
| feature: Feature | |
| FASM line for BEL | |
| """ | |
| belname, setting = feature.signature.split(".", 1) | |
| if feature.value == 1: | |
| # FIXME handle ZINV pins | |
| if "ZINV." in setting: | |
| setting = setting.replace("ZINV.", "") | |
| elif "INV." in setting: | |
| setting = setting.replace("INV.", "") | |
| self.belinversions[feature.loc][belname].append(setting) | |
| def parse_interface_line(self, feature: Feature): | |
| """Parses a setting for IO. | |
| Parameters | |
| ---------- | |
| feature: Feature | |
| FASM line for BEL | |
| """ | |
| belname, setting = feature.signature.split(".", 1) | |
| if feature.value == 1: | |
| setting = setting.replace("ZINV.", "") | |
| setting = setting.replace("INV.", "") | |
| self.interfaces[feature.loc][belname].append(setting) | |
| def parse_routing_line(self, feature: Feature): | |
| """Parses a routing setting. | |
| Parameters | |
| ---------- | |
| feature: Feature | |
| FASM line for BEL | |
| """ | |
| match = re.match(r"^I_highway\.IM(?P<switch_id>[0-9]+)\.I_pg(?P<sel_id>[0-9]+)$", feature.signature) | |
| if match: | |
| typ = "HIGHWAY" | |
| stage_id = 3 # FIXME: Get HIGHWAY stage id from the switchbox def | |
| switch_id = int(match.group("switch_id")) | |
| mux_id = 0 | |
| sel_id = int(match.group("sel_id")) | |
| match = re.match( | |
| r"^I_street\.Isb(?P<stage_id>[0-9])(?P<switch_id>[0-9])\.I_M(?P<mux_id>[0-9]+)\.I_pg(?P<sel_id>[0-9]+)$", # noqa: E501 | |
| feature.signature, | |
| ) | |
| if match: | |
| typ = "STREET" | |
| stage_id = int(match.group("stage_id")) - 1 | |
| switch_id = int(match.group("switch_id")) - 1 | |
| mux_id = int(match.group("mux_id")) | |
| sel_id = int(match.group("sel_id")) | |
| self.routingdata[feature.loc].append( | |
| RouteEntry(typ=typ, stage_id=stage_id, switch_id=switch_id, mux_id=mux_id, sel_id=sel_id) | |
| ) | |
| def parse_colclk_line(self, feature: Feature): | |
| self.colclk_data[feature.loc][feature.typ].append(feature) | |
| def parse_ram_line(self, feature: Feature): | |
| """Parses a RAM line. | |
| Parameters | |
| ---------- | |
| feature: Feature | |
| FASM line for BEL | |
| """ | |
| raise NotImplementedError("Parsing RAM FASM lines is not supported") | |
| def parse_fasm_lines(self, fasmlines): | |
| """Parses FASM lines. | |
| Parameters | |
| ---------- | |
| fasmlines: list | |
| A list of FasmLine objects | |
| """ | |
| loctyp = re.compile(r"^X(?P<x>[0-9]+)Y(?P<y>[0-9]+)\.(?P<type>[A-Z]+[0-4]?)\.(?P<signature>.*)$") # noqa: E501 | |
| for line in fasmlines: | |
| if not line.set_feature: | |
| continue | |
| match = loctyp.match(line.set_feature.feature) | |
| if not match: | |
| raise self.Fasm2BelsException( | |
| f"FASM features have unsupported format: {line.set_feature}" | |
| ) # noqa: E501 | |
| loc = Loc(x=int(match.group("x")), y=int(match.group("y")), z=0) | |
| typ = match.group("type") | |
| feature = Feature(loc=loc, typ=typ, signature=match.group("signature"), value=line.set_feature.value) | |
| self.featureparsers[typ](feature) | |
| def decode_switchbox(self, switchbox, features): | |
| """Decodes all switchboxes to extract full connections' info. | |
| For every output, this method determines its input in the routing | |
| switchboxes. In this representation, an input and output can be either | |
| directly connected to a BEL, or to a hop wire. | |
| Parameters | |
| ---------- | |
| switchbox: a Switchbox object from vpr_switchbox_types | |
| features: features regarding given switchbox | |
| Returns | |
| ------- | |
| dict: a mapping from output pin to input pin for a given switchbox | |
| """ | |
| # Group switchbox connections by destinationa | |
| conn_by_dst = defaultdict(set) | |
| for c in switchbox.connections: | |
| conn_by_dst[c.dst].add(c) | |
| # Prepare data structure | |
| mux_sel = {} | |
| for stage_id, stage in switchbox.stages.items(): | |
| mux_sel[stage_id] = {} | |
| for switch_id, switch in stage.switches.items(): | |
| mux_sel[stage_id][switch_id] = {} | |
| for mux_id, mux in switch.muxes.items(): | |
| mux_sel[stage_id][switch_id][mux_id] = None | |
| for feature in features: | |
| assert mux_sel[feature.stage_id][feature.switch_id][feature.mux_id] is None, feature # noqa: E501 | |
| mux_sel[feature.stage_id][feature.switch_id][feature.mux_id] = feature.sel_id # noqa: E501 | |
| def expand_mux(out_loc): | |
| """ | |
| Expands a multiplexer output until a switchbox input is reached. | |
| Returns name of the input or None if not found. | |
| Parameters | |
| ---------- | |
| out_loc: the last output location | |
| Returns | |
| ------- | |
| str: None if input name not found, else string | |
| """ | |
| # Get mux selection, If it is set to None then the mux is | |
| # not active | |
| sel = mux_sel[out_loc.stage_id][out_loc.switch_id][out_loc.mux_id] | |
| if sel is None: | |
| return None # TODO can we return None? | |
| stage = switchbox.stages[out_loc.stage_id] | |
| switch = stage.switches[out_loc.switch_id] | |
| mux = switch.muxes[out_loc.mux_id] | |
| pin = mux.inputs[sel] | |
| if pin.name is not None: | |
| return pin.name | |
| inp_loc = SwitchboxPinLoc( | |
| stage_id=out_loc.stage_id, | |
| switch_id=out_loc.switch_id, | |
| mux_id=out_loc.mux_id, | |
| pin_id=sel, | |
| pin_direction=PinDirection.INPUT, | |
| ) | |
| # Expand all "upstream" muxes that connect to the selected | |
| # input pin | |
| assert inp_loc in conn_by_dst, inp_loc | |
| for c in conn_by_dst[inp_loc]: | |
| inp = expand_mux(c.src) | |
| if inp is not None: | |
| return inp | |
| # Nothing found | |
| return None # TODO can we return None? | |
| # For each output pin of a switchbox determine to which input is it | |
| # connected to. | |
| routes = {} | |
| for out_pin in switchbox.outputs.values(): | |
| out_loc = out_pin.locs[0] | |
| routes[out_pin.name] = expand_mux(out_loc) | |
| return routes | |
| def process_switchbox(self, loc, switchbox, features): | |
| """Processes all switchboxes and extract hops from connections. | |
| The function extracts final connections from inputs to outputs, and | |
| hops into separate structures for further processing. | |
| Parameters | |
| ---------- | |
| loc: Loc | |
| location of the current switchbox | |
| switchbox: Switchbox | |
| a switchbox | |
| features: list | |
| list of features regarding given switchbox | |
| """ | |
| routes = self.decode_switchbox(switchbox, features) | |
| for k, v in routes.items(): | |
| if v is not None: | |
| if re.match("[VH][0-9][LRBT][0-9]", k): | |
| self.designhops[Loc(loc.x, loc.y, 0)][k] = v | |
| else: | |
| self.designconnections[loc][k] = v | |
| def resolve_hops(self): | |
| """Resolves remaining hop wires. | |
| It determines the absolute input for the given pin by resolving hop | |
| wires and adds those final connections to the design connections. | |
| """ | |
| for loc, conns in self.designconnections.items(): | |
| for pin, source in conns.items(): | |
| hop = get_name_and_hop(source) | |
| tloc = loc | |
| while hop[1] is not None: | |
| tloc = Loc(tloc[0] + hop[1][0], tloc[1] + hop[1][1], 0) | |
| # in some cases BEL is distanced from a switchbox, in those | |
| # cases the hop will not point to another hop. We should | |
| # simply return the pin here in the correct location | |
| if hop[0] in self.designhops[tloc]: | |
| hop = get_name_and_hop(self.designhops[tloc][hop[0]]) | |
| else: | |
| hop = (hop[0], None) | |
| self.designconnections[loc][pin] = (tloc, hop[0]) | |
| def resolve_connections(self): | |
| """Resolves connections between BELs and IOs.""" | |
| keys = sorted(self.routingdata.keys(), key=lambda loc: (loc.x, loc.y)) | |
| for loc in keys: | |
| routingfeatures = self.routingdata[loc] | |
| if loc in self.vpr_switchbox_grid: | |
| typ = self.vpr_switchbox_grid[loc] | |
| switchbox = self.vpr_switchbox_types[typ] | |
| self.process_switchbox(loc, switchbox, routingfeatures) | |
| self.resolve_hops() | |
| def remap_multiloc_loc(self, loc, pinname=None, celltype=None): | |
| """Unifies coordinates of cells occupying multiple locations. | |
| Some cells, like ASSP, RAM or multipliers occupy multiple locations. | |
| This method groups bits and connections for those cells into a single | |
| artificial location. | |
| Parameters | |
| ---------- | |
| loc: Loc | |
| The current location | |
| pinname: str | |
| The optional name of the pin (used to determine to which cell | |
| pin refers to) | |
| celltype: str | |
| The optional name of the cell type | |
| Returns | |
| ------- | |
| Loc: the new location of the cell | |
| """ | |
| finloc = loc | |
| for multiloc in self.multiloccells.values(): | |
| if pinname is None or pinname in multiloc.pinnames or celltype == multiloc.typ: | |
| if loc in multiloc.fromlocset: | |
| finloc = multiloc.toloc | |
| break | |
| return finloc | |
| def resolve_multiloc_cells(self): | |
| """Groups cells that are scattered around multiple locations.""" | |
| newbelinversions = defaultdict(lambda: defaultdict(list)) | |
| newdesignconnections = defaultdict(dict) | |
| for bellockey, bellocpair in self.belinversions.items(): | |
| for belloctype, belloc in bellocpair.items(): | |
| if belloctype in self.multiloccells: | |
| newbelinversions[self.remap_multiloc_loc(bellockey, celltype=belloctype)][belloctype].extend(belloc) | |
| self.belinversion = newbelinversions | |
| for loc, conns in self.designconnections.items(): | |
| for pin, src in conns.items(): | |
| dstloc = self.remap_multiloc_loc(loc, pinname=pin) | |
| srcloc = self.remap_multiloc_loc(src[0], pinname=src[1]) | |
| if srcloc != src[0]: | |
| k, v = ((srcloc, src[1]), src) | |
| if k in self.org_loc_map: | |
| assert v == self.org_loc_map[k], (k, self.org_loc_map[k], v) | |
| self.org_loc_map[k] = v | |
| if dstloc != loc: | |
| k, v = ((dstloc, pin), (loc, pin)) | |
| if k in self.org_loc_map: | |
| assert v == self.org_loc_map[k], (k, self.org_loc_map[k], v) | |
| self.org_loc_map[k] = v | |
| newdesignconnections[dstloc][pin] = (srcloc, src[1]) | |
| self.designconnections = newdesignconnections | |
| def get_clock_for_gmux(self, gmux, loc): | |
| """Returns location of a CLOCK cell associated with the given GMUX | |
| cell. Returns None if not found | |
| Parameters | |
| ---------- | |
| gmux: str | |
| The GMUX cell name | |
| loc: Loc | |
| The GMUX location | |
| Returns | |
| ------- | |
| Loc: the new location of the cell or None | |
| """ | |
| connections = [ | |
| c for c in self.connections if c.src.type == ConnectionType.TILE and c.dst.type == ConnectionType.TILE | |
| ] | |
| for connection in connections: | |
| # Only to a GMUX at the given location | |
| dst = connection.dst | |
| if dst.loc != loc or "GMUX" not in dst.pin: | |
| continue | |
| # GMUX cells are named "GMUX<index>". | |
| cell, pin = dst.pin.split("_", maxsplit=1) | |
| match = re.match(r"GMUX(?P<idx>[0-9]+)", cell) | |
| if match is None: | |
| continue | |
| # Not the cell that we are looking for | |
| if cell != gmux: | |
| continue | |
| # We are only interested in the IP connection | |
| if pin != "IP": | |
| continue | |
| # Must go from CLOCK<n>.IC pin | |
| cell, pin = connection.src.pin.split("_", maxsplit=1) | |
| if not cell.startswith("CLOCK") or pin != "IC": | |
| continue | |
| # Return the source location | |
| return connection.src.loc | |
| # Not found | |
| return None | |
| def get_gmux_for_qmux(self, qmux, loc): | |
| """Returns a map of the given QMUX selection to driving GMUX cells. | |
| Parameters | |
| ---------- | |
| qmux: str | |
| The QMUX cell name | |
| loc: Loc | |
| The QMUX location | |
| Returns | |
| ------- | |
| Dict: A dict indexed by the selection index holding tuples with format: | |
| (loc, cell, pin) | |
| """ | |
| sel_map = {} | |
| connections = [c for c in self.connections if c.dst.type == ConnectionType.CLOCK] | |
| for connection in connections: | |
| # Only to a QMUX at the given location | |
| dst = connection.dst | |
| if dst.loc != loc or "QMUX" not in dst.pin: | |
| continue | |
| # QMUX cells are named "QMUX_<quad><index>". | |
| cell, pin = dst.pin.split(".", maxsplit=1) | |
| match = re.match(r"QMUX_(?P<quad>[A-Z]+)(?P<idx>[0-9]+)", cell) | |
| if match is None: | |
| continue | |
| # This is not for the given QMUX | |
| qmux_idx = int(match.group("idx")) | |
| if qmux != "QMUX{}".format(qmux_idx): | |
| continue | |
| # Get the QCLKIN pin index. These are named "QCLKIN<index>" | |
| match = re.match(r"QCLKIN(?P<idx>[0-9]+)", pin) | |
| if match is None: | |
| continue | |
| qclkin_idx = int(match.group("idx")) | |
| # Get the source endpoint of the connection | |
| cell, pin = connection.src.pin.split("_", maxsplit=1) | |
| match = re.match(r"GMUX(?P<idx>[0-9]+)", cell) | |
| if match is None: | |
| continue | |
| gmux_idx = int(match.group("idx")) | |
| # Since fasm2bels uses physical database, it is not aware of | |
| # other QMUX inputs that QCLKIN0. Assume that the connection | |
| # found is to the QCLKIN0 pin and add QCLKIN1..2 to the map | |
| # as well. | |
| assert qclkin_idx == 0, connection | |
| # Make map entries | |
| for i in range(3): | |
| # Calculate GMUX index for QCLKIN<i> input of the QMUX | |
| idx = (gmux_idx + i) % 5 | |
| # Add to the map | |
| sel_map[i] = ( | |
| connection.src.loc, | |
| "GMUX{}".format(idx), | |
| pin, | |
| ) | |
| return sel_map | |
| def get_qmux_for_cand(self, cand, loc): | |
| """Returns a QMUX cell and its location that drives the given CAND | |
| cell. | |
| Parameters | |
| ---------- | |
| cand: str | |
| The CAND cell name | |
| loc: Loc | |
| The CAND location | |
| Returns | |
| ------- | |
| Tuple: A tuple holding (loc, cell) | |
| """ | |
| connections = [c for c in self.connections if c.dst.type == ConnectionType.CLOCK] | |
| for connection in connections: | |
| # Only to a CAND at the given location | |
| # Note: Check also the row above. CAND cells are located in two | |
| # rows but with fasm features everything gets aligned to even rows | |
| dst = connection.dst | |
| if (dst.loc != loc and dst.loc != Loc(loc.x, loc.y - 1, loc.z)) or "CAND" not in dst.pin: | |
| continue | |
| # CAND cells are named "CAND<index>_<quad>_<column>". | |
| cell, pin = dst.pin.split(".", maxsplit=1) | |
| match = re.match(r"CAND(?P<idx>[0-9]+)_(?P<quad>[A-Z]+)_(?P<col>[0-9]+)", cell) | |
| if match is None: | |
| continue | |
| # This is not for the given CAND | |
| if cand != "CAND{}".format(match.group("idx")): | |
| continue | |
| # QMUX cells are named "QMUX_<quad><index>". | |
| cell, pin = connection.src.pin.split(".", maxsplit=1) | |
| match = re.match(r"QMUX_(?P<quad>[A-Z]+)(?P<idx>[0-9]+)", cell) | |
| if match is None: | |
| continue | |
| # Return the QMUX and its location | |
| return "QMUX{}".format(match.group("idx")), connection.src.loc | |
| # None found | |
| return None, None | |
| def resolve_gmux(self): | |
| """Resolves GMUX cells, updates the designconnections map. Also creates | |
| connections to CLOCK cells whenever necessary. | |
| Returns | |
| ------- | |
| Dict: A map of GMUX names to their output wires | |
| """ | |
| # Process GMUX | |
| gmux_map = dict() | |
| gmux_locs = [loc for loc, tile in self.vpr_tile_grid.items() if "GMUX" in tile.type] | |
| for loc in gmux_locs: | |
| # Group GMUX input pin connections by GMUX cell names | |
| gmux_connections = defaultdict(lambda: dict()) | |
| for cell_pin, conn in self.designconnections[loc].items(): | |
| if cell_pin.startswith("GMUX"): | |
| cell, pin = cell_pin.split("_", maxsplit=1) | |
| gmux_connections[cell][pin] = conn | |
| # Examine each GMUX config | |
| for gmux, connections in gmux_connections.items(): | |
| # FIXME: Handle IS0 inversion (if any) | |
| # The IS0 pin has to be routed | |
| if "IS0" not in connections: | |
| print("WARNING: Pin '{}.IS0' at '{}' is unrouted!".format(gmux, loc)) | |
| continue | |
| # TODO: For now support only static GMUX settings | |
| if connections["IS0"][1] not in ["GND", "VCC"]: | |
| print("WARNING: Non-static GMUX selection (at '{}') not supported yet!".format(loc)) | |
| continue | |
| # Static selection | |
| sel = int(connections["IS0"][1] == "VCC") | |
| # IP selected | |
| if sel == 0: | |
| # Create a global clock wire for the CLOCK pad | |
| match = re.match(r"GMUX(?P<idx>[0-9]+)", gmux) | |
| assert match is not None, gmux | |
| idx = int(match.group("idx")) | |
| wire = "CLK{}".format(idx) | |
| # Get the clock pad location | |
| clock_loc = self.get_clock_for_gmux(gmux, loc) | |
| assert clock_loc is not None, gmux | |
| # Check if the clock pad is enabled. If not then discard | |
| # the GMUX | |
| bel_features = [] | |
| for bel, features in self.interfaces.get(clock_loc, {}).items(): | |
| for feature in features: | |
| bel_features.append("{}.{}".format(bel, feature)) | |
| if "ASSP.ASSPInvPortAlias" not in bel_features: | |
| continue | |
| # Connect it to the output wire of the GMUX | |
| self.designconnections[clock_loc]["CLOCK0_IC"] = (None, wire) | |
| # The GMUX is implicit. Remove all connections to it | |
| self.designconnections[loc] = { | |
| k: v for k, v in self.designconnections[loc].items() if not k.startswith(gmux) | |
| } | |
| # IC selected | |
| else: | |
| # Check if the IC pin has an active driver. If not then | |
| # discard the mux. | |
| if connections.get("IC", (None, None))[1] in [None, "GND", "VCC"]: | |
| continue | |
| # Create a wire for the GMUX output | |
| wire = "{}_X{}Y{}".format(gmux, loc.x, loc.y) | |
| # Remove the IS0 connection | |
| del self.designconnections[loc]["{}_IS0".format(gmux)] | |
| # Connect the output | |
| self.designconnections[loc]["{}_IZ".format(gmux)] = (None, wire) | |
| # Store the wire | |
| gmux_map[gmux] = wire | |
| return gmux_map | |
| def resolve_qmux(self, gmux_map): | |
| """Resolves QMUX cells, updates the designconnections map. | |
| Parameters | |
| ---------- | |
| gmux_map: Dict | |
| A map of QMUX cells to their GMUX driving wires. | |
| Returns | |
| ------- | |
| Dict: A map of locations and QMUX names to their driving wires | |
| """ | |
| # Process QMUX | |
| qmux_map = defaultdict(lambda: dict()) | |
| qmux_locs = [loc for loc, tile in self.vpr_tile_grid.items() if "QMUX" in tile.type] | |
| for loc in qmux_locs: | |
| # Group QMUX input pin connections by QMUX cell names | |
| qmux_connections = defaultdict(lambda: dict()) | |
| for cell_pin, conn in self.designconnections[loc].items(): | |
| if cell_pin.startswith("QMUX"): | |
| cell, pin = cell_pin.split("_", maxsplit=1) | |
| qmux_connections[cell][pin] = conn | |
| # Examine each QMUX config | |
| for qmux, connections in qmux_connections.items(): | |
| # FIXME: Handle IS0 and IS1 inversion (if any) | |
| # Both IS0 and IS1 must be routed to something | |
| if "IS0" not in connections: | |
| print("WARNING: Pin '{}.IS0' at '{}' is unrouted!".format(qmux, loc)) | |
| if "IS1" not in connections: | |
| print("WARNING: Pin '{}.IS1' at '{}' is unrouted!".format(qmux, loc)) | |
| if "IS0" not in connections or "IS1" not in connections: | |
| continue | |
| # TODO: For now support only static QMUX settings | |
| if connections["IS0"][1] not in ["GND", "VCC"]: | |
| print("WARNING: Non-static QMUX selection (at '{}') not supported yet!".format(loc)) | |
| continue | |
| if connections["IS1"][1] not in ["GND", "VCC"]: | |
| print("WARNING: Non-static QMUX selection (at '{}') not supported yet!".format(loc)) | |
| continue | |
| # Get associated GMUXes | |
| sel_map = self.get_gmux_for_qmux(qmux, loc) | |
| # Static selection | |
| sel = int(connections["IS0"][1] == "VCC") * 2 + int(connections["IS1"][1] == "VCC") | |
| # Input from the routing network selected, create a new wire | |
| if sel == 3: | |
| # Check if the HSCKIN input is connected to an active | |
| # driver. If not then discard the QMUX | |
| if connections.get("HSCKIN", (None, None))[1] in [None, "GND", "VCC"]: | |
| continue | |
| # Create a wire for the QMUX output | |
| wire = "{}_X{}Y{}".format(qmux, loc.x, loc.y) | |
| # Remove IS0 and IS1 from the connection map. | |
| del self.designconnections[loc]["{}_IS0".format(qmux)] | |
| del self.designconnections[loc]["{}_IS1".format(qmux)] | |
| # Connect the output | |
| self.designconnections[loc]["{}_IZ".format(qmux)] = (None, wire) | |
| # Input from a GMUX is selected, assign its wire here | |
| else: | |
| # The GMUX is not active. Discard the QMUX | |
| gmux_loc, gmux_cell, gmux_pin = sel_map[sel] | |
| if gmux_cell not in gmux_map: | |
| continue | |
| # Use the wire of that GMUX | |
| wire = gmux_map[gmux_cell] | |
| # The QMUX is implicit. Remove all connections to it | |
| self.designconnections[loc] = { | |
| k: v for k, v in self.designconnections[loc].items() if not k.startswith(qmux) | |
| } | |
| # Store the wire | |
| qmux_map[loc][qmux] = wire | |
| return dict(qmux_map) | |
| def resolve_cand(self, qmux_map): | |
| """Resolves CAND cells, creates the cand_map map. | |
| Parameters | |
| ---------- | |
| qmux_map: Dict | |
| A map of locations and CAND names to their driving QMUXes. | |
| Returns | |
| ------- | |
| None | |
| """ | |
| # Process CAND | |
| for loc, all_features in self.colclk_data.items(): | |
| for cand, features in all_features.items(): | |
| hilojoint = False | |
| enjoint = False | |
| for feature in features: | |
| if feature.signature == "I_hilojoint": | |
| hilojoint = bool(feature.value) | |
| if feature.signature == "I_enjoint": | |
| enjoint = bool(feature.value) | |
| # TODO: Do not support dynamically enabled CANDs for now. | |
| assert enjoint is False, "Dynamically enabled CANDs are not supported yet" | |
| # Statically disabled, skip this one | |
| if hilojoint is False: | |
| continue | |
| # Find a QMUX driving this CAND cell | |
| qmux_cell, qmux_loc = self.get_qmux_for_cand(cand, loc) | |
| assert qmux_cell is not None, (cand, loc) | |
| # The QMUX is not active, skip this one | |
| if qmux_loc not in qmux_map: | |
| continue | |
| # Get the wire | |
| wire = qmux_map[qmux_loc][qmux_cell] | |
| # Populate the column clock to switchbox connection map | |
| quadrant = get_quadrant_for_loc(loc, self.quadrants) | |
| for y in range(quadrant.y0, quadrant.y1 + 1): | |
| sb_loc = Loc(loc.x, y, 0) | |
| self.cand_map[sb_loc][cand] = wire | |
| def resolve_global_clock_network(self): | |
| """Resolves the global clock network. Creates the cand_map, updates | |
| the designconnections. | |
| Returns | |
| ------- | |
| None | |
| """ | |
| # Resolve GMUXes | |
| gmux_map = self.resolve_gmux() | |
| # Resolve QMUXes | |
| qmux_map = self.resolve_qmux(gmux_map) | |
| # Resolve CANDs | |
| self.resolve_cand(qmux_map) | |
| def produce_verilog(self, pcf_data): | |
| """Produces string containing Verilog module representing FASM. | |
| Returns | |
| ------- | |
| str, str: a Verilog module and PCF | |
| """ | |
| module = VModule( | |
| self.vpr_tile_grid, | |
| self.vpr_tile_types, | |
| self.cells_library, | |
| pcf_data, | |
| self.belinversions, | |
| self.interfaces, | |
| self.designconnections, | |
| self.org_loc_map, | |
| self.cand_map, | |
| self.inversionpins, | |
| self.io_to_fbio, | |
| ) | |
| module.parse_bels() | |
| verilog = module.generate_verilog() | |
| pcf = module.generate_pcf() | |
| qcf = module.generate_qcf() | |
| return verilog, pcf, qcf | |
| def convert_to_verilog(self, fasmlines): | |
| """Runs all methods required to convert FASM lines to Verilog module. | |
| Parameters | |
| ---------- | |
| fasmlines: list | |
| FASM lines to process | |
| Returns | |
| ------- | |
| str: a Verilog module | |
| """ | |
| self.parse_fasm_lines(fasmlines) | |
| self.resolve_connections() | |
| self.resolve_multiloc_cells() | |
| self.resolve_global_clock_network() | |
| verilog, pcf, qcf = self.produce_verilog(pcf_data) | |
| return verilog, pcf, qcf | |
| def parse_pcf(pcf): | |
| pcf_data = {} | |
| with open(pcf, "r") as fp: | |
| for line in fp: | |
| line = line.strip().split() | |
| if len(line) < 3: | |
| continue | |
| if len(line) > 3 and not line[3].startswith("#"): | |
| continue | |
| if line[0] != "set_io": | |
| continue | |
| pcf_data[line[2]] = line[1] | |
| return pcf_data | |
| if __name__ == "__main__": | |
| # Parse arguments | |
| parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) | |
| parser.add_argument("input_file", type=Path, help="Input fasm file") | |
| parser.add_argument("--phy-db", type=str, required=True, help="Physical device database file") | |
| parser.add_argument("--device-name", type=str, required=True, choices=["eos-s3", "pp3e"], help="Device name") | |
| parser.add_argument("--package-name", type=str, required=True, help="Device package name") | |
| parser.add_argument( | |
| "--input-type", | |
| type=str, | |
| choices=["bitstream", "fasm"], | |
| default="fasm", | |
| help="Determines whether the input is a FASM file or bitstream", | |
| ) | |
| parser.add_argument("--output-verilog", type=Path, required=True, help="Output Verilog file") | |
| parser.add_argument( | |
| "--input-pcf", type=Path, required=False, help="Pins constraint file to maintain original pin names" | |
| ) | |
| parser.add_argument("--output-pcf", type=Path, help="Output PCF file") | |
| parser.add_argument("--output-qcf", type=Path, help="Output QCF file") | |
| args = parser.parse_args() | |
| pcf_data = {} | |
| if args.input_pcf is not None: | |
| pcf_data = parse_pcf(args.input_pcf) | |
| # Load data from the database | |
| with open(args.phy_db, "rb") as fp: | |
| db = pickle.load(fp) | |
| # Initialize fasm2bels | |
| f2b = Fasm2Bels(db, args.device_name, args.package_name) | |
| # Disassemble bitstream / load FASM | |
| if args.input_type == "bitstream": | |
| qlfasmdb = load_quicklogic_database(get_db_dir("ql-" + args.device_name)) | |
| if args.device_name == "eos-s3": | |
| assembler = QL732BAssembler(qlfasmdb) | |
| elif args.device_name == "pp3e": | |
| assembler = QL732BAssembler(qlfasmdb) # Workaround: use EOS-S3 assembler for PP3E | |
| else: | |
| assert False, args.device_name | |
| assembler.read_bitstream(args.input_file) | |
| fasmlines = assembler.disassemble() | |
| fasmlines = [line for line in fasm.parse_fasm_string("\n".join(fasmlines))] | |
| else: | |
| fasmlines = [line for line in fasm.parse_fasm_filename(args.input_file)] | |
| # Run fasm2bels | |
| verilog, pcf, qcf = f2b.convert_to_verilog(fasmlines) | |
| # Write output files | |
| with open(args.output_verilog, "w") as outv: | |
| outv.write(verilog) | |
| if args.output_pcf: | |
| with open(args.output_pcf, "w") as outpcf: | |
| outpcf.write(pcf) | |
| if args.output_qcf: | |
| with open(args.output_qcf, "w") as outqcf: | |
| outqcf.write(qcf) |