diff --git a/.github/workflows/Pipeline.yml b/.github/workflows/Pipeline.yml index 83f266a..4bb47bf 100644 --- a/.github/workflows/Pipeline.yml +++ b/.github/workflows/Pipeline.yml @@ -28,7 +28,8 @@ jobs: steps: - - uses: actions/checkout@v3 + - name: 🧰 Checkout + uses: actions/checkout@v3 with: submodules: recursive @@ -36,6 +37,29 @@ jobs: uses: SymbiFlow/actions/checks@main + Format: + name: '🐍 Format' + runs-on: ubuntu-latest + + steps: + + - name: 🧰 Checkout + uses: actions/checkout@v3 + with: + submodules: recursive + + - name: 🐍 Setup Python + uses: actions/setup-python@v2 + with: + python-version: '3.10' + + - name: 🔧 Install dependencies + run: python -m pip install -r test/requirements.txt + + - name: 🚦 Check if Python sources follow code formatting standards + run: python -m black --check f4pga + + Docs: runs-on: ubuntu-latest name: '📓 Docs' diff --git a/f4pga/__init__.py b/f4pga/__init__.py index c38f804..8b75bd9 100644 --- a/f4pga/__init__.py +++ b/f4pga/__init__.py @@ -1,24 +1,24 @@ -#!/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 - -from f4pga.flows import main - - -if __name__ == '__main__': - main() +#!/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 + +from f4pga.flows import main + + +if __name__ == "__main__": + main() diff --git a/f4pga/context.py b/f4pga/context.py index 0140346..bea88fe 100644 --- a/f4pga/context.py +++ b/f4pga/context.py @@ -21,15 +21,15 @@ from pathlib import Path from os import environ -FPGA_FAM = environ.get('FPGA_FAM', 'xc7') -if FPGA_FAM not in ['xc7', 'eos-s3', 'qlf_k4n8']: - raise(Exception(f"Unsupported FPGA_FAM <{FPGA_FAM}>!")) +FPGA_FAM = environ.get("FPGA_FAM", "xc7") +if FPGA_FAM not in ["xc7", "eos-s3", "qlf_k4n8"]: + raise (Exception(f"Unsupported FPGA_FAM <{FPGA_FAM}>!")) -F4PGA_DEBUG = environ.get('F4PGA_DEBUG') +F4PGA_DEBUG = environ.get("F4PGA_DEBUG") -install_dir = environ.get('F4PGA_INSTALL_DIR') +install_dir = environ.get("F4PGA_INSTALL_DIR") if install_dir is None: - default_install_dir = Path('/usr/local') + default_install_dir = Path("/usr/local") if F4PGA_DEBUG is not None: print("Environment variable F4PGA_INSTALL_DIR is undefined!") print(f"Using default {default_install_dir}") @@ -37,4 +37,4 @@ if install_dir is None: else: F4PGA_INSTALL_DIR = Path(install_dir) -F4PGA_SHARE_DIR = Path(environ.get('F4PGA_SHARE_DIR', F4PGA_INSTALL_DIR / FPGA_FAM / 'share/f4pga')) +F4PGA_SHARE_DIR = Path(environ.get("F4PGA_SHARE_DIR", F4PGA_INSTALL_DIR / FPGA_FAM / "share/f4pga")) diff --git a/f4pga/flows/__init__.py b/f4pga/flows/__init__.py index bac8620..5663a9d 100755 --- a/f4pga/flows/__init__.py +++ b/f4pga/flows/__init__.py @@ -46,10 +46,10 @@ from f4pga.flows.commands import cmd_build, cmd_show_dependencies, f4pga_done def platform_stages(platform_flow, r_env): - """ Iterates over all stages available in a given flow. """ + """Iterates over all stages available in a given flow.""" - stage_options = platform_flow.get('stage_options') - for stage_name, modulestr in platform_flow['stages'].items(): + stage_options = platform_flow.get("stage_options") + for stage_name, modulestr in platform_flow["stages"].items(): mod_opts = stage_options.get(stage_name) if stage_options else None yield Stage(stage_name, modulestr, mod_opts, r_env) @@ -61,7 +61,7 @@ def get_stage_values_override(og_values: dict, stage: Stage): def prepare_stage_io_input(stage: Stage): - return { 'params': stage.params } if stage.params is not None else {} + return {"params": stage.params} if stage.params is not None else {} def main(): @@ -70,17 +70,17 @@ def main(): set_verbosity_level(args.verbose - (1 if args.silent else 0)) - if args.command == 'build': + if args.command == "build": cmd_build(args) f4pga_done() - if args.command == 'showd': + if args.command == "showd": cmd_show_dependencies(args) f4pga_done() - sfprint(0, 'Please use a command.\nUse `--help` flag to learn more.') + sfprint(0, "Please use a command.\nUse `--help` flag to learn more.") f4pga_done() -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/f4pga/flows/argparser.py b/f4pga/flows/argparser.py index c73f0cd..40cb2de 100644 --- a/f4pga/flows/argparser.py +++ b/f4pga/flows/argparser.py @@ -22,91 +22,46 @@ from re import finditer as re_finditer def _add_flow_arg(parser: ArgumentParser): - parser.add_argument( - '-f', - '--flow', - metavar='flow_path', - type=str, - help='Path to flow definition file' - ) + parser.add_argument("-f", "--flow", metavar="flow_path", type=str, help="Path to flow definition file") def _setup_build_parser(parser: ArgumentParser): _add_flow_arg(parser) parser.add_argument( - '-t', - '--target', - metavar='target_name', - type=str, - help='Perform stages necessary to acquire target' + "-t", "--target", metavar="target_name", type=str, help="Perform stages necessary to acquire target" ) parser.add_argument( - '-P', - '--pretend', - action='store_true', - help='Show dependency resolution without executing flow' + "-P", "--pretend", action="store_true", help="Show dependency resolution without executing flow" ) - parser.add_argument( - '-i', - '--info', - action='store_true', - help='Display info about available targets' - ) + parser.add_argument("-i", "--info", action="store_true", help="Display info about available targets") parser.add_argument( - '-c', - '--nocache', - action='store_true', - help='Ignore caching and rebuild everything up to the target.' + "-c", "--nocache", action="store_true", help="Ignore caching and rebuild everything up to the target." ) - parser.add_argument( - '-S', - '--stageinfo', - nargs=1, - metavar='stage_name', - help='Display info about stage' - ) + parser.add_argument("-S", "--stageinfo", nargs=1, metavar="stage_name", help="Display info about stage") - parser.add_argument( - '-p', - '--part', - metavar='part_name', - help='Name of the target chip' - ) + parser.add_argument("-p", "--part", metavar="part_name", help="Name of the target chip") - parser.add_argument( - '--dep', - '-D', - action='append', - default=[] - ) + parser.add_argument("--dep", "-D", action="append", default=[]) + + parser.add_argument("--val", "-V", action="append", default=[]) - parser.add_argument( - '--val', - '-V', - action='append', - default=[] - ) def _setup_show_dep_parser(parser: ArgumentParser): parser.add_argument( - '-p', - '--part', - metavar='part_name', - type=str, - help='Name of the part (use to display part-specific values.)' + "-p", "--part", metavar="part_name", type=str, help="Name of the part (use to display part-specific values.)" ) parser.add_argument( - '-s', - '--stage', - metavar='stage_name', + "-s", + "--stage", + metavar="stage_name", type=str, - help='Name of the stage (use if you want to set the value only for that stage). Requires `-p`.' + help="Name of the stage (use if you want to set the value only for that stage). Requires `-p`.", ) _add_flow_arg(parser) @@ -116,24 +71,15 @@ def setup_argparser(): """ Set up argument parser for the program. """ - parser = ArgumentParser(description='F4PGA Build System') + parser = ArgumentParser(description="F4PGA Build System") - parser.add_argument( - '-v', - '--verbose', - action='count', - default=0 - ) + parser.add_argument("-v", "--verbose", action="count", default=0) - parser.add_argument( - '-s', - '--silent', - action='store_true' - ) + parser.add_argument("-s", "--silent", action="store_true") - subparsers = parser.add_subparsers(dest='command') - _setup_build_parser(subparsers.add_parser('build')) - show_dep = subparsers.add_parser('showd', description='Show the value(s) assigned to a dependency') + subparsers = parser.add_subparsers(dest="command") + _setup_build_parser(subparsers.add_parser("build")) + show_dep = subparsers.add_parser("showd", description="Show the value(s) assigned to a dependency") _setup_show_dep_parser(show_dep) return parser @@ -146,37 +92,37 @@ def _parse_depval(depvalstr: str): See `_parse_cli_value` for detail on how to pass different kinds of values. """ - d = { 'name': None, 'stage': None, 'value': None } + d = {"name": None, "stage": None, "value": None} - splitted = list(_unescaped_separated('=', depvalstr)) + splitted = list(_unescaped_separated("=", depvalstr)) if len(splitted) != 2: - raise Exception('Too many components') + raise Exception("Too many components") pathstr = splitted[0] valstr = splitted[1] - path_components = pathstr.split('.') + path_components = pathstr.split(".") if len(path_components) < 1: - raise Exception('Missing value') - d['name'] = path_components.pop(len(path_components) - 1) + raise Exception("Missing value") + d["name"] = path_components.pop(len(path_components) - 1) if len(path_components) > 0: - d['stage'] = path_components.pop(0) + d["stage"] = path_components.pop(0) if len(path_components) > 0: - raise Exception('Too many path components') + raise Exception("Too many path components") - d['value'] = _parse_cli_value(valstr) + d["value"] = _parse_cli_value(valstr) return d -def _unescaped_matches(regexp: str, s: str, escape_chr='\\'): +def _unescaped_matches(regexp: str, s: str, escape_chr="\\"): """ Find all occurences of a pattern in a string that contains escape sequences. Yields pairs of starting and ending indices of the pattern. """ - noescapes = '' + noescapes = "" # We remove all escape sequnces from a string, so it will match only with # unescaped characters, but to map the results back to the string containing the @@ -187,7 +133,7 @@ def _unescaped_matches(regexp: str, s: str, escape_chr='\\'): for sl in s.split(escape_chr): if len(sl) <= 1: continue - noescape = sl[(1 if offset != 0 else 0):] + noescape = sl[(1 if offset != 0 else 0) :] for _ in noescape: offsets.append(offset) offset += 2 @@ -203,7 +149,7 @@ def _unescaped_matches(regexp: str, s: str, escape_chr='\\'): yield off1, off2 -def _unescaped_separated(regexp: str, s: str, escape_chr='\\'): +def _unescaped_separated(regexp: str, s: str, escape_chr="\\"): """ Yields substrings of a string that contains escape sequences. """ @@ -215,7 +161,7 @@ def _unescaped_separated(regexp: str, s: str, escape_chr='\\'): if last_end < len(s): yield s[last_end:] else: - yield '' + yield "" def _parse_cli_value(s: str): @@ -242,73 +188,71 @@ def _parse_cli_value(s: str): """ if len(s) == 0: - return '' + return "" # List - if s[0] == '[': - if len(s) < 2 or s[len(s)-1] != ']': - raise Exception('Missing \']\' delimiter') - inner = s[1:(len(s)-1)] - if inner == '': + if s[0] == "[": + if len(s) < 2 or s[len(s) - 1] != "]": + raise Exception("Missing ']' delimiter") + inner = s[1 : (len(s) - 1)] + if inner == "": return [] - return [_parse_cli_value(v) for v in _unescaped_separated(',', inner)] + return [_parse_cli_value(v) for v in _unescaped_separated(",", inner)] # Dictionary - if s[0] == '{': - if len(s) < 2 or s[len(s)-1] != '}': - raise Exception('Missing \'}\' delimiter') + if s[0] == "{": + if len(s) < 2 or s[len(s) - 1] != "}": + raise Exception("Missing '}' delimiter") d = {} - inner = s[1:(len(s)-1)] - if inner == '': + inner = s[1 : (len(s) - 1)] + if inner == "": return {} - for kv in _unescaped_separated(',', inner): - k_v = list(_unescaped_separated(':', kv)) + for kv in _unescaped_separated(",", inner): + k_v = list(_unescaped_separated(":", kv)) if len(k_v) < 2: - raise Exception('Missing value in dictionary entry') + raise Exception("Missing value in dictionary entry") if len(k_v) > 2: - raise Exception('Unexpected \':\' token') + raise Exception("Unexpected ':' token") key = k_v[0] - value = _parse_cli_value(k_v[1]) + value = _parse_cli_value(k_v[1]) d[key] = value return d # Bool hack - if s == '\\True': + if s == "\\True": return True - if s == '\\False': + if s == "\\False": return False # Number hack - if len(s) >= 3 and s[0:1] == '\\N': + if len(s) >= 3 and s[0:1] == "\\N": return int(s[2:]) # String - return s.replace('\\', '') + return s.replace("\\", "") def get_cli_flow_config(args: Namespace, part: str): def create_defdict(): return { - 'dependencies': {}, - 'values': {}, + "dependencies": {}, + "values": {}, } part_flow_config = create_defdict() - def add_entries(arglist: 'list[str]', dict_name: str): + def add_entries(arglist: "list[str]", dict_name: str): for value_def in (_parse_depval(cliv) for cliv in arglist): - stage = value_def['stage'] + stage = value_def["stage"] if stage is None: - part_flow_config[dict_name][value_def['name']] = \ - value_def['value'] + part_flow_config[dict_name][value_def["name"]] = value_def["value"] else: if part_flow_config.get(stage) is None: part_flow_config[stage] = create_defdict() - part_flow_config[stage][dict_name][value_def['name']] = \ - value_def['value'] + part_flow_config[stage][dict_name][value_def["name"]] = value_def["value"] - add_entries(args.dep, 'dependencies') - add_entries(args.val, 'values') + add_entries(args.dep, "dependencies") + add_entries(args.val, "values") - return { part: part_flow_config } + return {part: part_flow_config} diff --git a/f4pga/flows/cache.py b/f4pga/flows/cache.py index ec2ec21..690e8db 100755 --- a/f4pga/flows/cache.py +++ b/f4pga/flows/cache.py @@ -23,11 +23,13 @@ from json import dump as json_dump, load as json_load, JSONDecodeError from f4pga.flows.common import sfprint + def _get_hash(path: Path): if not path.is_dir(): - with path.open('rb') as rfptr: + with path.open("rb") as rfptr: return zlib_adler32(rfptr.read()) - return 0 # Directories always get '0' hash. + return 0 # Directories always get '0' hash. + class F4Cache: """ @@ -36,9 +38,9 @@ class F4Cache: If file's checksum differs from the one saved in a file, that means, the file has changed. """ - hashes: 'dict[str, dict[str, str]]' - current_hashes: 'dict[str, str]' - status: 'dict[str, str]' + hashes: "dict[str, dict[str, str]]" + current_hashes: "dict[str, str]" + status: "dict[str, str]" cachefile_path: str def __init__(self, cachefile_path): @@ -65,19 +67,20 @@ class F4Cache: if not self.hashes.get(path): self.hashes[path] = {} self.hashes[path][consumer] = hash + def _try_push_consumer_status(self, path: str, consumer: str, status): if not self.status.get(path): self.status[path] = {} self.status[path][consumer] = status def process_file(self, path: Path): - """ Process file for tracking with f4cache. """ + """Process file for tracking with f4cache.""" hash = _get_hash(path) self.current_hashes[path.as_posix()] = hash def update(self, path: Path, consumer: str): - """ Add/remove a file to.from the tracked files, update checksum if necessary and calculate status. + """Add/remove a file to.from the tracked files, update checksum if necessary and calculate status. Multiple hashes are stored per file, one for each consumer module. "__target" is used as a convention for a "fake" consumer in case the file is requested as a target and not used @@ -97,14 +100,14 @@ class F4Cache: last_hash = None if last_hashes is None else last_hashes.get(consumer) if hash != last_hash: - self._try_push_consumer_status(posix_path, consumer, 'changed') + self._try_push_consumer_status(posix_path, consumer, "changed") self._try_push_consumer_hash(posix_path, consumer, hash) return True - self._try_push_consumer_status(posix_path, consumer, 'same') + self._try_push_consumer_status(posix_path, consumer, "same") return False def get_status(self, path: str, consumer: str): - """ Get status for a file with a given path. + """Get status for a file with a given path. returns 'untracked' if the file is not tracked. """ @@ -117,30 +120,36 @@ class F4Cache: last_hash = hashes.get(consumer) if last_hash is not None: if self.current_hashes[path] != last_hash: - return 'changed' - return 'same' - return 'untracked' + return "changed" + return "same" + return "untracked" status = statuses.get(consumer) if not status: - return 'untracked' + return "untracked" return status def load(self): """Loads cache's state from the persistent storage""" try: - with Path(self.cachefile_path).open('r') as rfptr: + with Path(self.cachefile_path).open("r") as rfptr: self.hashes = json_load(rfptr) except JSONDecodeError: - sfprint(0, f'WARNING: `{self.cachefile_path}` f4cache is corrupted!\n' - 'This will cause flow to re-execute from the beginning.') + sfprint( + 0, + f"WARNING: `{self.cachefile_path}` f4cache is corrupted!\n" + "This will cause flow to re-execute from the beginning.", + ) self.hashes = {} except FileNotFoundError: - sfprint(0, f'Couldn\'t open `{self.cachefile_path}` cache file.\n' - 'This will cause flow to re-execute from the beginning.') + sfprint( + 0, + f"Couldn't open `{self.cachefile_path}` cache file.\n" + "This will cause flow to re-execute from the beginning.", + ) self.hashes = {} def save(self): """Saves cache's state to the persistent storage.""" - with Path(self.cachefile_path).open('w') as wfptr: + with Path(self.cachefile_path).open("w") as wfptr: json_dump(self.hashes, wfptr, indent=4) diff --git a/f4pga/flows/commands.py b/f4pga/flows/commands.py index a3085a5..0f78563 100644 --- a/f4pga/flows/commands.py +++ b/f4pga/flows/commands.py @@ -35,7 +35,7 @@ from f4pga.flows.common import ( scan_modules, set_verbosity_level, sfprint, - sub as common_sub + sub as common_sub, ) from f4pga.flows.argparser import get_cli_flow_config from f4pga.flows.cache import F4Cache @@ -44,7 +44,7 @@ from f4pga.flows.flow_config import ( FlowConfig, FlowDefinition, open_project_flow_cfg, - verify_platform_name + verify_platform_name, ) from f4pga.flows.flow import Flow from f4pga.flows.stage import Stage @@ -53,11 +53,11 @@ from f4pga.flows.inspector import get_module_info ROOT = Path(__file__).resolve().parent -F4CACHEPATH = '.f4cache' +F4CACHEPATH = ".f4cache" -def display_dep_info(stages: 'Iterable[Stage]'): - sfprint(0, 'Platform dependencies/targets:') +def display_dep_info(stages: "Iterable[Stage]"): + sfprint(0, "Platform dependencies/targets:") longest_out_name_len = 0 for stage in stages: for out in stage.produces: @@ -66,85 +66,79 @@ def display_dep_info(stages: 'Iterable[Stage]'): longest_out_name_len = l desc_indent = longest_out_name_len + 7 - nl_indentstr = '\n' + nl_indentstr = "\n" for _ in range(0, desc_indent): - nl_indentstr += ' ' + nl_indentstr += " " for stage in stages: for out in stage.produces: pname = Style.BRIGHT + out.name + Style.RESET_ALL - indent = '' + indent = "" for _ in range(0, desc_indent - len(pname) + 3): - indent += ' ' - specstr = '???' - if out.spec == 'req': - specstr = f'{Fore.BLUE}guaranteed{Fore.RESET}' - elif out.spec == 'maybe': - specstr = f'{Fore.YELLOW}not guaranteed{Fore.RESET}' - elif out.spec == 'demand': - specstr = f'{Fore.RED}on-demand{Fore.RESET}' - pgen = f'{Style.DIM}stage: `{stage.name}`, '\ - f'spec: {specstr}{Style.RESET_ALL}' - pdesc = stage.meta[out.name].replace('\n', nl_indentstr) - sfprint(0, f' {Style.BRIGHT + out.name + Style.RESET_ALL}:' - f'{indent}{pdesc}{nl_indentstr}{pgen}') + indent += " " + specstr = "???" + if out.spec == "req": + specstr = f"{Fore.BLUE}guaranteed{Fore.RESET}" + elif out.spec == "maybe": + specstr = f"{Fore.YELLOW}not guaranteed{Fore.RESET}" + elif out.spec == "demand": + specstr = f"{Fore.RED}on-demand{Fore.RESET}" + pgen = f"{Style.DIM}stage: `{stage.name}`, " f"spec: {specstr}{Style.RESET_ALL}" + pdesc = stage.meta[out.name].replace("\n", nl_indentstr) + sfprint(0, f" {Style.BRIGHT + out.name + Style.RESET_ALL}:" f"{indent}{pdesc}{nl_indentstr}{pgen}") def display_stage_info(stage: Stage): if stage is None: - sfprint(0, f'Stage does not exist') + sfprint(0, f"Stage does not exist") f4pga_fail() return - sfprint(0, f'Stage `{Style.BRIGHT}{stage.name}{Style.RESET_ALL}`:') - sfprint(0, f' Module: `{Style.BRIGHT}{stage.module.name}{Style.RESET_ALL}`') - sfprint(0, f' Module info:') + sfprint(0, f"Stage `{Style.BRIGHT}{stage.name}{Style.RESET_ALL}`:") + sfprint(0, f" Module: `{Style.BRIGHT}{stage.module.name}{Style.RESET_ALL}`") + sfprint(0, f" Module info:") mod_info = get_module_info(stage.module) - mod_info = '\n '.join(mod_info.split('\n')) + mod_info = "\n ".join(mod_info.split("\n")) - sfprint(0, f' {mod_info}') + sfprint(0, f" {mod_info}") -f4pga_done_str = Style.BRIGHT + Fore.GREEN + 'DONE' +f4pga_done_str = Style.BRIGHT + Fore.GREEN + "DONE" def f4pga_fail(): global f4pga_done_str - f4pga_done_str = Style.BRIGHT + Fore.RED + 'FAILED' + f4pga_done_str = Style.BRIGHT + Fore.RED + "FAILED" def f4pga_done(): - sfprint(1, f'f4pga: {f4pga_done_str}' - f'{Style.RESET_ALL + Fore.RESET}') + sfprint(1, f"f4pga: {f4pga_done_str}" f"{Style.RESET_ALL + Fore.RESET}") exit(0) def setup_resolution_env(): - """ Sets up a ResolutionEnv with default built-ins. """ + """Sets up a ResolutionEnv with default built-ins.""" - r_env = ResolutionEnv({ - 'shareDir': share_dir_path, - 'binDir': bin_dir_path - }) + r_env = ResolutionEnv({"shareDir": share_dir_path, "binDir": bin_dir_path}) def _noisy_warnings(): """ Emit some noisy warnings. """ - environ['OUR_NOISY_WARNINGS'] = 'noisy_warnings.log' - return 'noisy_warnings.log' + environ["OUR_NOISY_WARNINGS"] = "noisy_warnings.log" + return "noisy_warnings.log" def _generate_values(): """ Generate initial values, available in configs. """ conf = { - 'python3': common_sub('which', 'python3').decode().replace('\n', ''), - 'noisyWarnings': _noisy_warnings() + "python3": common_sub("which", "python3").decode().replace("\n", ""), + "noisyWarnings": _noisy_warnings(), } - if (FPGA_FAM == 'xc7'): - conf['prjxray_db'] = common_sub('prjxray-config').decode().replace('\n', '') + if FPGA_FAM == "xc7": + conf["prjxray_db"] = common_sub("prjxray-config").decode().replace("\n", "") return conf @@ -156,19 +150,18 @@ def open_project_flow_config(path: str) -> ProjectFlowConfig: try: flow_cfg = open_project_flow_cfg(path) except FileNotFoundError as _: - fatal(-1, 'The provided flow configuration file does not exist') + fatal(-1, "The provided flow configuration file does not exist") return flow_cfg -def verify_part_stage_params(flow_cfg: FlowConfig, - part: 'str | None' = None): +def verify_part_stage_params(flow_cfg: FlowConfig, part: "str | None" = None): if part: platform_name = get_platform_name_for_part(part) if not verify_platform_name(platform_name, str(ROOT)): - sfprint(0, f'Platform `{part}`` is unsupported.') + sfprint(0, f"Platform `{part}`` is unsupported.") return False if part not in flow_cfg.part(): - sfprint(0, f'Platform `{part}`` is not in project.') + sfprint(0, f"Platform `{part}`` is not in project.") return False return True @@ -180,7 +173,7 @@ def get_platform_name_for_part(part_name: str): The reason for such distinction is that plenty of chips with different names differ only in a type of package they use. """ - with (ROOT / 'part_db.yml').open('r') as rfptr: + with (ROOT / "part_db.yml").open("r") as rfptr: for key, val in yaml_load(rfptr, yaml_loader).items(): if part_name.upper() in val: return key @@ -188,43 +181,35 @@ def get_platform_name_for_part(part_name: str): def make_flow_config(project_flow_cfg: ProjectFlowConfig, part_name: str) -> FlowConfig: - """ Create `FlowConfig` from given project flow configuration and part name """ + """Create `FlowConfig` from given project flow configuration and part name""" platform = get_platform_name_for_part(part_name) if platform is None: - raise F4PGAException( - message='You have to specify a part name or configure a default part.' - ) + raise F4PGAException(message="You have to specify a part name or configure a default part.") if part_name not in project_flow_cfg.parts(): - raise F4PGAException( - message='Project flow configuration does not support requested part.' - ) + raise F4PGAException(message="Project flow configuration does not support requested part.") r_env = setup_resolution_env() - r_env.add_values({'part_name': part_name.lower()}) + r_env.add_values({"part_name": part_name.lower()}) scan_modules(str(ROOT)) - with (ROOT / 'platforms.yml').open('r') as rfptr: + with (ROOT / "platforms.yml").open("r") as rfptr: platforms = yaml_load(rfptr, yaml_loader) if platform not in platforms: - raise F4PGAException(message=f'Flow definition for platform <{platform}> cannot be found!') + raise F4PGAException(message=f"Flow definition for platform <{platform}> cannot be found!") - flow_cfg = FlowConfig( - project_flow_cfg, - FlowDefinition(platforms[platform], r_env), - part_name - ) + flow_cfg = FlowConfig(project_flow_cfg, FlowDefinition(platforms[platform], r_env), part_name) if len(flow_cfg.stages) == 0: - raise F4PGAException(message = 'Platform flow does not define any stage') + raise F4PGAException(message="Platform flow does not define any stage") return flow_cfg def cmd_build(args: Namespace): - """ `build` command implementation """ + """`build` command implementation""" project_flow_cfg: ProjectFlowConfig = None @@ -233,14 +218,13 @@ def cmd_build(args: Namespace): if args.flow: project_flow_cfg = open_project_flow_config(args.flow) elif part_name is not None: - project_flow_cfg = ProjectFlowConfig('.temp.flow.json') + project_flow_cfg = ProjectFlowConfig(".temp.flow.json") project_flow_cfg.flow_cfg = get_cli_flow_config(args, part_name) if part_name is None and project_flow_cfg is not None: part_name = project_flow_cfg.get_default_part() if project_flow_cfg is None: - fatal(-1, 'No configuration was provided. Use `--flow`, and/or ' - '`--part` to configure flow.') + fatal(-1, "No configuration was provided. Use `--flow`, and/or " "`--part` to configure flow.") flow_cfg = make_flow_config(project_flow_cfg, part_name) @@ -256,19 +240,14 @@ def cmd_build(args: Namespace): if target is None: target = project_flow_cfg.get_default_target(part_name) if target is None: - fatal(-1, 'Please specify desired target using `--target` option ' - 'or configure a default target.') + fatal(-1, "Please specify desired target using `--target` option " "or configure a default target.") - flow = Flow( - target=target, - cfg=flow_cfg, - f4cache=F4Cache(F4CACHEPATH) if not args.nocache else None - ) + flow = Flow(target=target, cfg=flow_cfg, f4cache=F4Cache(F4CACHEPATH) if not args.nocache else None) dep_print_verbosity = 0 if args.pretend else 2 - sfprint(dep_print_verbosity, '\nProject status:') + sfprint(dep_print_verbosity, "\nProject status:") flow.print_resolved_dependencies(dep_print_verbosity) - sfprint(dep_print_verbosity, '') + sfprint(dep_print_verbosity, "") if args.pretend: f4pga_done() @@ -278,7 +257,7 @@ def cmd_build(args: Namespace): except AssertionError as e: raise e except Exception as e: - sfprint(0, f'{e}') + sfprint(0, f"{e}") f4pga_fail() if flow.f4cache: @@ -286,7 +265,7 @@ def cmd_build(args: Namespace): def cmd_show_dependencies(args: Namespace): - """ `showd` command implementation """ + """`showd` command implementation""" flow_cfg = open_project_flow_config(args.flow) @@ -294,10 +273,9 @@ def cmd_show_dependencies(args: Namespace): f4pga_fail() return - platform_overrides: 'set | None' = None + platform_overrides: "set | None" = None if args.platform is not None: - platform_overrides = \ - set(flow_cfg.get_dependency_platform_overrides(args.part).keys()) + platform_overrides = set(flow_cfg.get_dependency_platform_overrides(args.part).keys()) display_list = [] @@ -306,14 +284,16 @@ def cmd_show_dependencies(args: Namespace): for dep_name, dep_paths in raw_deps.items(): prstr: str if (platform_overrides is not None) and (dep_name in platform_overrides): - prstr = f'{Style.DIM}({args.platform}){Style.RESET_ALL} ' \ - f'{Style.BRIGHT + dep_name + Style.RESET_ALL}: {dep_paths}' + prstr = ( + f"{Style.DIM}({args.platform}){Style.RESET_ALL} " + f"{Style.BRIGHT + dep_name + Style.RESET_ALL}: {dep_paths}" + ) else: - prstr = f'{Style.BRIGHT + dep_name + Style.RESET_ALL}: {dep_paths}' + prstr = f"{Style.BRIGHT + dep_name + Style.RESET_ALL}: {dep_paths}" display_list.append((dep_name, prstr)) - display_list.sort(key = lambda p: p[0]) + display_list.sort(key=lambda p: p[0]) for _, prstr in display_list: sfprint(0, prstr) diff --git a/f4pga/flows/common.py b/f4pga/flows/common.py index 0b660f9..f2efe8f 100644 --- a/f4pga/flows/common.py +++ b/f4pga/flows/common.py @@ -33,35 +33,35 @@ share_dir_path = str(F4PGA_SHARE_DIR) class F4PGAException(Exception): - def __init__(self, message = 'unknown exception'): + def __init__(self, message="unknown exception"): self.message = message def __repr__(self): - return f'F4PGAException(message = \'{self.message}\')' + return f"F4PGAException(message = '{self.message}')" def __str__(self): return self.message def decompose_depname(name: str): - spec = 'req' + spec = "req" specchar = name[len(name) - 1] - if specchar == '?': - spec = 'maybe' - elif specchar == '!': - spec = 'demand' - if spec != 'req': - name = name[:len(name) - 1] + if specchar == "?": + spec = "maybe" + elif specchar == "!": + spec = "demand" + if spec != "req": + name = name[: len(name) - 1] return name, spec def with_qualifier(name: str, q: str) -> str: - if q == 'req': + if q == "req": return decompose_depname(name)[0] - if q == 'maybe': - return decompose_depname(name)[0] + '?' - if q == 'demand': - return decompose_depname(name)[0] + '!' + if q == "maybe": + return decompose_depname(name)[0] + "?" + if q == "demand": + return decompose_depname(name)[0] + "!" _sfbuild_module_collection_name_to_path = {} @@ -71,8 +71,8 @@ def scan_modules(mypath: str): global _sfbuild_module_collection_name_to_path sfbuild_home = mypath _sfbuild_module_collection_name_to_path = { - re_match('(.*)_modules$', moddir).groups()[0]: str(Path(sfbuild_home) / moddir) - for moddir in [dir for dir in os_listdir(sfbuild_home) if re_match('.*_modules$', dir)] + re_match("(.*)_modules$", moddir).groups()[0]: str(Path(sfbuild_home) / moddir) + for moddir in [dir for dir in os_listdir(sfbuild_home) if re_match(".*_modules$", dir)] } @@ -80,17 +80,17 @@ def resolve_modstr(modstr: str): """ Resolves module location from modulestr. """ - sl = modstr.split(':') + sl = modstr.split(":") if len(sl) > 2: - raise Exception('Incorrect module sysntax. Expected one \':\' or one \'::\'') + raise Exception("Incorrect module sysntax. Expected one ':' or one '::'") if len(sl) < 2: return modstr collection_name = sl[0] - module_filename = sl[1] + '.py' + module_filename = sl[1] + ".py" col_path = _sfbuild_module_collection_name_to_path.get(collection_name) if not col_path: - fatal(-1, f'Module collection {collection_name} does not exist') + fatal(-1, f"Module collection {collection_name} does not exist") return str(Path(col_path) / module_filename) @@ -98,18 +98,20 @@ def deep(fun, allow_none=False): """ Create a recursive string transform function for 'str | list | dict', i.e a dependency. """ + def d(paths, *args, **kwargs): nonlocal allow_none if type(paths) is str: return fun(paths, *args, **kwargs) elif type(paths) is list: - return [d(p, *args, **kwargs) for p in paths]; + return [d(p, *args, **kwargs) for p in paths] elif type(paths) is dict: return dict([(k, d(p, *args, **kwargs)) for k, p in paths.items()]) elif allow_none and (paths is None): return paths else: - raise RuntimeError(f'paths is of type {type(paths)}') + raise RuntimeError(f"paths is of type {type(paths)}") + return d @@ -127,10 +129,10 @@ class VprArgs: eblif: str optional: list - def __init__(self, share: str, eblif, values: Namespace, - sdc_file: 'str | None' = None, - vpr_extra_opts: 'list | None' = None): - self.arch_dir = str(Path(share) / 'arch') + def __init__( + self, share: str, eblif, values: Namespace, sdc_file: "str | None" = None, vpr_extra_opts: "list | None" = None + ): + self.arch_dir = str(Path(share) / "arch") self.arch_def = values.arch_def self.lookahead = values.rr_graph_lookahead_bin self.rr_graph = values.rr_graph_real_bin @@ -144,7 +146,7 @@ class VprArgs: if vpr_extra_opts is not None: self.optional += vpr_extra_opts if sdc_file is not None: - self.optional += ['--sdc_file', sdc_file] + self.optional += ["--sdc_file", sdc_file] class SubprocessException(Exception): @@ -158,9 +160,7 @@ def sub(*args, env=None, cwd=None): out = run(args, capture_output=True, env=env, cwd=cwd) if out.returncode != 0: - print(f'[ERROR]: {args[0]} non-zero return code.\n' - f'stderr:\n{out.stderr.decode()}\n\n' - ) + print(f"[ERROR]: {args[0]} non-zero return code.\n" f"stderr:\n{out.stderr.decode()}\n\n") exit(out.returncode) return out.stdout @@ -171,34 +171,44 @@ def vpr(mode: str, vprargs: VprArgs, cwd=None): """ modeargs = [] - if mode == 'pack': - modeargs = ['--pack'] - elif mode == 'place': - modeargs = ['--place'] - elif mode == 'route': - modeargs = ['--route'] - elif mode == 'analysis': - modeargs = ['--analysis'] - - return sub(*([ - 'vpr', - vprargs.arch_def, - vprargs.eblif, - '--device', vprargs.device_name, - '--read_rr_graph', vprargs.rr_graph, - '--read_router_lookahead', vprargs.lookahead, - '--read_placement_delay_lookup', vprargs.place_delay - ] + modeargs + vprargs.optional), cwd=str(cwd)) + if mode == "pack": + modeargs = ["--pack"] + elif mode == "place": + modeargs = ["--place"] + elif mode == "route": + modeargs = ["--route"] + elif mode == "analysis": + modeargs = ["--analysis"] + return sub( + *( + [ + "vpr", + vprargs.arch_def, + vprargs.eblif, + "--device", + vprargs.device_name, + "--read_rr_graph", + vprargs.rr_graph, + "--read_router_lookahead", + vprargs.lookahead, + "--read_placement_delay_lookup", + vprargs.place_delay, + ] + + modeargs + + vprargs.optional + ), + cwd=cwd, + ) _vpr_specific_values = [ - 'arch_def', - 'rr_graph_lookahead_bin', - 'rr_graph_real_bin', - 'vpr_place_delay', - 'vpr_grid_layout_name', - 'vpr_options?' + "arch_def", + "rr_graph_lookahead_bin", + "rr_graph_real_bin", + "vpr_place_delay", + "vpr_grid_layout_name", + "vpr_options?", ] @@ -215,8 +225,8 @@ def options_dict_to_list(opt_dict: dict): opts = [] for key, val in opt_dict.items(): - opts.append(f'--{key}') - if not(type(val) is list and val == []): + opts.append(f"--{key}") + if not (type(val) is list and val == []): opts.append(str(val)) return opts @@ -225,7 +235,7 @@ def noisy_warnings(device): """ Emit some noisy warnings. """ - environ['OUR_NOISY_WARNINGS'] = f'noisy_warnings-{device}_pack.log' + environ["OUR_NOISY_WARNINGS"] = f"noisy_warnings-{device}_pack.log" def my_path(): @@ -235,18 +245,18 @@ def my_path(): return str(Path(sys_argv[0]).resolve().parent) -def save_vpr_log(filename, build_dir=''): +def save_vpr_log(filename, build_dir=""): """ Save VPR logic (moves the default output file into a desired path). """ - sh_mv(str(Path(build_dir) / 'vpr_stdout.log'), filename) + sh_mv(str(Path(build_dir) / "vpr_stdout.log"), filename) def fatal(code, message): """ Print a message informing about an error that has occured and terminate program with a given return code. """ - raise(Exception(f'[FATAL ERROR]: {message}')) + raise (Exception(f"[FATAL ERROR]: {message}")) exit(code) @@ -276,23 +286,23 @@ class ResolutionEnv: """ if type(s) is str: - match_list = list(re_finditer('\$\{([^${}]*)\}', s)) + match_list = list(re_finditer("\$\{([^${}]*)\}", s)) # Assumption: re_finditer finds matches in a left-to-right order match_list.reverse() for match in match_list: match_str = match.group(1) - match_str = match_str.replace('?', '') + match_str = match_str.replace("?", "") v = self.values.get(match_str) if not v: if final: - v = '' + v = "" else: continue span = match.span() if type(v) is str: - s = s[:span[0]] + v + s[span[1]:] - elif type(v) is list: # Assume it's a list of strings - ns = list([s[:span[0]] + ve + s[span[1]:] for ve in v]) + s = s[: span[0]] + v + s[span[1] :] + elif type(v) is list: # Assume it's a list of strings + ns = list([s[: span[0]] + ve + s[span[1] :] for ve in v]) s = ns elif type(s) is list: diff --git a/f4pga/flows/common_modules/analysis.py b/f4pga/flows/common_modules/analysis.py index 45ba326..398a554 100644 --- a/f4pga/flows/common_modules/analysis.py +++ b/f4pga/flows/common_modules/analysis.py @@ -24,18 +24,18 @@ from f4pga.flows.module import Module, ModuleContext def analysis_merged_post_implementation_file(ctx: ModuleContext): - return str(Path(ctx.takes.eblif).with_suffix('')) + '_merged_post_implementation.v' + return str(Path(ctx.takes.eblif).with_suffix("")) + "_merged_post_implementation.v" def analysis_post_implementation_file(ctx: ModuleContext): - return str(Path(ctx.takes.eblif).with_suffix('')) + '_post_synthesis.v' + return str(Path(ctx.takes.eblif).with_suffix("")) + "_post_synthesis.v" class analysisModule(Module): def map_io(self, ctx: ModuleContext): return { - 'merged_post_implementation_v': analysis_merged_post_implementation_file(ctx), - 'post_implementation_v': analysis_post_implementation_file(ctx) + "merged_post_implementation_v": analysis_merged_post_implementation_file(ctx), + "post_implementation_v": analysis_post_implementation_file(ctx), } def execute(self, ctx: ModuleContext): @@ -43,43 +43,24 @@ class analysisModule(Module): vpr_options = options_dict_to_list(ctx.values.vpr_options) if ctx.values.vpr_options else [] - yield 'Analysis with VPR...' - common_vpr( - 'analysis', - VprArgs( - ctx.share, - ctx.takes.eblif, - ctx.values, - sdc_file=ctx.takes.sdc - ), - cwd=build_dir - ) + yield "Analysis with VPR..." + common_vpr("analysis", VprArgs(ctx.share, ctx.takes.eblif, ctx.values, sdc_file=ctx.takes.sdc), cwd=build_dir) - if ctx.is_output_explicit('merged_post_implementation_v'): + if ctx.is_output_explicit("merged_post_implementation_v"): Path(analysis_merged_post_implementation_file(ctx)).rename(ctx.outputs.merged_post_implementation_v) - if ctx.is_output_explicit('post_implementation_v'): + if ctx.is_output_explicit("post_implementation_v"): Path(analysis_post_implementation_file(ctx)).rename(ctx.outputs.post_implementation_v) - yield 'Saving log...' - save_vpr_log('analysis.log', build_dir=build_dir) + yield "Saving log..." + save_vpr_log("analysis.log", build_dir=build_dir) def __init__(self, _): - self.name = 'analysis' + self.name = "analysis" self.no_of_phases = 2 - self.takes = [ - 'eblif', - 'route', - 'sdc?' - ] - self.produces = [ - 'merged_post_implementation_v', - 'post_implementation_v', - 'analysis_log' - ] - self.values = [ - 'device', - 'vpr_options?' - ] + vpr_specific_values() + self.takes = ["eblif", "route", "sdc?"] + self.produces = ["merged_post_implementation_v", "post_implementation_v", "analysis_log"] + self.values = ["device", "vpr_options?"] + vpr_specific_values() + ModuleClass = analysisModule diff --git a/f4pga/flows/common_modules/fasm.py b/f4pga/flows/common_modules/fasm.py index 1ba36e0..925bf22 100644 --- a/f4pga/flows/common_modules/fasm.py +++ b/f4pga/flows/common_modules/fasm.py @@ -24,12 +24,9 @@ from f4pga.flows.module import Module, ModuleContext class FasmModule(Module): - def map_io(self, ctx: ModuleContext): build_dir = str(Path(ctx.takes.eblif).parent) - return { - 'fasm': f'{(Path(build_dir)/ctx.values.top)!s}.fasm' - } + return {"fasm": f"{(Path(build_dir)/ctx.values.top)!s}.fasm"} def execute(self, ctx: ModuleContext): build_dir = str(Path(ctx.takes.eblif).parent) @@ -38,59 +35,45 @@ class FasmModule(Module): optional = [] if ctx.values.pnr_corner is not None: - optional += ['--pnr_corner', ctx.values.pnr_corner] + optional += ["--pnr_corner", ctx.values.pnr_corner] if ctx.takes.sdc: - optional += ['--sdc', ctx.takes.sdc] + optional += ["--sdc", ctx.takes.sdc] s = [ - 'genfasm', + "genfasm", vprargs.arch_def, str(Path(ctx.takes.eblif).resolve()), - '--device', + "--device", vprargs.device_name, - '--read_rr_graph', - vprargs.rr_graph + "--read_rr_graph", + vprargs.rr_graph, ] + vprargs.optional if get_verbosity_level() >= 2: - yield 'Generating FASM...\n ' + ' '.join(s) + yield "Generating FASM...\n " + " ".join(s) else: - yield 'Generating FASM...' + yield "Generating FASM..." common_sub(*s, cwd=build_dir) - default_fasm_output_name = Path(build_dir)/ f'{ctx.values.top}.fasm' + default_fasm_output_name = Path(build_dir) / f"{ctx.values.top}.fasm" if str(default_fasm_output_name) != ctx.outputs.fasm: default_fasm_output_name.rename(ctx.outputs.fasm) if ctx.takes.fasm_extra: - yield 'Appending extra FASM...' - with \ - open(ctx.outputs.fasm, 'a') as fasm_file, \ - open(ctx.takes.fasm_extra, 'r') as fasm_extra_file: + yield "Appending extra FASM..." + with open(ctx.outputs.fasm, "a") as fasm_file, open(ctx.takes.fasm_extra, "r") as fasm_extra_file: fasm_file.write(f"\n{fasm_extra_file.read()}") else: - yield 'No extra FASM to append' + yield "No extra FASM to append" def __init__(self, _): - self.name = 'fasm' + self.name = "fasm" self.no_of_phases = 2 - self.takes = [ - 'eblif', - 'net', - 'place', - 'route', - 'fasm_extra?', - 'sdc?' - ] - self.produces = [ 'fasm' ] - self.values = [ - 'device', - 'top', - 'pnr_corner?' - ] + vpr_specific_values() - self.prod_meta = { - 'fasm': 'FPGA assembly file' - } + self.takes = ["eblif", "net", "place", "route", "fasm_extra?", "sdc?"] + self.produces = ["fasm"] + self.values = ["device", "top", "pnr_corner?"] + vpr_specific_values() + self.prod_meta = {"fasm": "FPGA assembly file"} + ModuleClass = FasmModule diff --git a/f4pga/flows/common_modules/generic_script_wrapper.py b/f4pga/flows/common_modules/generic_script_wrapper.py index 572d8e9..ad860e8 100644 --- a/f4pga/flows/common_modules/generic_script_wrapper.py +++ b/f4pga/flows/common_modules/generic_script_wrapper.py @@ -66,22 +66,21 @@ from f4pga.flows.module import Module, ModuleContext def _get_param(params, name: str): param = params.get(name) if not param: - raise Exception(f'generic module wrapper parameters ' - f'missing `{name}` field') + raise Exception(f"generic module wrapper parameters " f"missing `{name}` field") return param def _parse_param_def(param_def: str): - if param_def[0] == '#': - return 'positional', int(param_def[1:]) - elif param_def[0] == '$': - return 'environmental', param_def[1:] - return 'named', param_def + if param_def[0] == "#": + return "positional", int(param_def[1:]) + elif param_def[0] == "$": + return "environmental", param_def[1:] + return "named", param_def class InputReferences: - dependencies: 'set[str]' - values: 'set[str]' + dependencies: "set[str]" + values: "set[str]" def merge(self, other): self.dependencies.update(other.dependencies) @@ -96,20 +95,21 @@ def _get_input_references(input: str) -> InputReferences: refs = InputReferences() if type(input) is not str: return refs - for match in re_finditer('\$\{([^${}]*)\}', input): + for match in re_finditer("\$\{([^${}]*)\}", input): match_str = match.group(1) - if match_str[0] != ':': + if match_str[0] != ":": refs.values.add(match_str) continue if len(match_str) < 2: - raise Exception('Dependency name must be at least 1 character long') - refs.dependencies.add(re_match('([^\\[\\]]*)', match_str[1:]).group(1)) + raise Exception("Dependency name must be at least 1 character long") + refs.dependencies.add(re_match("([^\\[\\]]*)", match_str[1:]).group(1)) return refs def _make_noop1(): def noop(_): return + return noop @@ -117,22 +117,23 @@ def _tailcall1(self, fun): def newself(arg, self=self, fun=fun): fun(arg) self(arg) + return newself class GenericScriptWrapperModule(Module): script_path: str - stdout_target: 'None | tuple[str, str]' - file_outputs: 'list[tuple[str, str, str]]' - interpreter: 'None | str' - cwd: 'None | str' + stdout_target: "None | tuple[str, str]" + file_outputs: "list[tuple[str, str, str]]" + interpreter: "None | str" + cwd: "None | str" @staticmethod def _add_extra_values_to_env(ctx: ModuleContext): for take_name, take_path in vars(ctx.takes).items(): if take_path is not None: - ctx.r_env.values[f':{take_name}[noext]'] = deep(lambda p: str(Path(p).with_suffix('')))(take_path) - ctx.r_env.values[f':{take_name}[dir]'] = deep(lambda p: str(Path(p).parent.resolve()))(take_path) + ctx.r_env.values[f":{take_name}[noext]"] = deep(lambda p: str(Path(p).with_suffix("")))(take_path) + ctx.r_env.values[f":{take_name}[dir]"] = deep(lambda p: str(Path(p).parent.resolve()))(take_path) def map_io(self, ctx: ModuleContext): self._add_extra_values_to_env(ctx) @@ -143,8 +144,7 @@ class GenericScriptWrapperModule(Module): outputs[dep] = out_path_resolved if self.stdout_target: - out_path_resolved = \ - ctx.r_env.resolve(self.stdout_target[1], final=True) + out_path_resolved = ctx.r_env.resolve(self.stdout_target[1], final=True) outputs[self.stdout_target[0]] = out_path_resolved return outputs @@ -154,8 +154,7 @@ class GenericScriptWrapperModule(Module): cwd = ctx.r_env.resolve(self.cwd) - sub_args = [ctx.r_env.resolve(self.script_path, final=True)] \ - + self.get_args(ctx) + sub_args = [ctx.r_env.resolve(self.script_path, final=True)] + self.get_args(ctx) if self.interpreter: sub_args = [ctx.r_env.resolve(self.interpreter, final=True)] + sub_args @@ -163,19 +162,19 @@ class GenericScriptWrapperModule(Module): # XXX: This may produce incorrect string if arguments contains whitespace # characters - cmd = ' '.join(sub_args) + cmd = " ".join(sub_args) if get_verbosity_level() >= 2: - yield f'Running script...\n {cmd}' + yield f"Running script...\n {cmd}" else: - yield f'Running an externel script...' + yield f"Running an externel script..." data = sub(*sub_args, cwd=cwd, env=sub_env) - yield 'Writing outputs...' + yield "Writing outputs..." if self.stdout_target: target = ctx.r_env.resolve(self.stdout_target[1], final=True) - with open(target, 'wb') as f: + with open(target, "wb") as f: f.write(data) for _, file, target in self.file_outputs: @@ -184,33 +183,33 @@ class GenericScriptWrapperModule(Module): if target != file: Path(file).rename(target) - def _init_outputs(self, output_defs: 'dict[str, dict[str, str]]'): + def _init_outputs(self, output_defs: "dict[str, dict[str, str]]"): self.stdout_target = None self.file_outputs = [] for dep_name, output_def in output_defs.items(): dname, _ = decompose_depname(dep_name) self.produces.append(dep_name) - meta = output_def.get('meta') + meta = output_def.get("meta") if meta is str: self.prod_meta[dname] = meta - mode = output_def.get('mode') + mode = output_def.get("mode") if type(mode) is not str: - raise Exception(f'Output mode for `{dep_name}` is not specified') + raise Exception(f"Output mode for `{dep_name}` is not specified") - target = output_def.get('target') + target = output_def.get("target") if type(target) is not str: - raise Exception('`target` field is not specified') + raise Exception("`target` field is not specified") - if mode == 'file': - file = output_def.get('file') + if mode == "file": + file = output_def.get("file") if type(file) is not str: - raise Exception('Output file is not specified') + raise Exception("Output file is not specified") self.file_outputs.append((dname, file, target)) - elif mode == 'stdout': + elif mode == "stdout": if self.stdout_target is not None: - raise Exception('stdout output is already specified') + raise Exception("stdout output is already specified") self.stdout_target = dname, target # A very functional approach @@ -228,39 +227,49 @@ class GenericScriptWrapperModule(Module): push = None push_env = None - if param_kind == 'named': - def push_named(val: 'str | bool | int', param=param): + if param_kind == "named": + + def push_named(val: "str | bool | int", param=param): nonlocal named_args if type(val) is bool: - named_args.append(f'--{param}') + named_args.append(f"--{param}") else: - named_args += [f'--{param}', str(val)] + named_args += [f"--{param}", str(val)] + push = push_named - elif param_kind == 'environmental': - def push_environ(val: 'str | bool | int', param=param): + elif param_kind == "environmental": + + def push_environ(val: "str | bool | int", param=param): nonlocal env_vars env_vars[param] = val + push_env = push_environ else: + def push_positional(val: str, param=param): nonlocal positional_args positional_args.append((param, val)) + push = push_positional input_refs = _get_input_references(input) refs.merge(input_refs) if push is not None: + def push_q(ctx: ModuleContext, push=push, input=input): val = ctx.r_env.resolve(input, final=True) - if val != '': + if val != "": push(val) + get_args = _tailcall1(get_args, push_q) else: + def push_q(ctx: ModuleContext, push_env=push_env, input=input): val = ctx.r_env.resolve(input, final=True) - if val != '': + if val != "": push_env(val) + get_env = _tailcall1(get_env, push_q) def get_all_args(ctx: ModuleContext): @@ -269,7 +278,7 @@ class GenericScriptWrapperModule(Module): get_args(ctx) positional_args.sort(key=lambda t: t[0]) - pos = [ a for _, a in positional_args] + pos = [a for _, a in positional_args] return named_args + pos @@ -280,8 +289,8 @@ class GenericScriptWrapperModule(Module): return None return env_vars - setattr(self, 'get_args', get_all_args) - setattr(self, 'get_env', get_all_env) + setattr(self, "get_args", get_all_args) + setattr(self, "get_env", get_all_env) for dep in refs.dependencies: self.takes.append(dep) @@ -289,18 +298,19 @@ class GenericScriptWrapperModule(Module): self.values.append(val) def __init__(self, params): - stage_name = params.get('stage_name') + stage_name = params.get("stage_name") self.name = f"{'' if stage_name is None else stage_name}-generic" self.no_of_phases = 2 - self.script_path = params.get('script') - self.interpreter = params.get('interpreter') - self.cwd = params.get('cwd') + self.script_path = params.get("script") + self.interpreter = params.get("interpreter") + self.cwd = params.get("cwd") self.takes = [] self.produces = [] self.values = [] self.prod_meta = {} - self._init_outputs(_get_param(params, 'outputs')) - self._init_inputs(_get_param(params, 'inputs')) + self._init_outputs(_get_param(params, "outputs")) + self._init_inputs(_get_param(params, "inputs")) + ModuleClass = GenericScriptWrapperModule diff --git a/f4pga/flows/common_modules/io_rename.py b/f4pga/flows/common_modules/io_rename.py index bd00b8f..c22c728 100644 --- a/f4pga/flows/common_modules/io_rename.py +++ b/f4pga/flows/common_modules/io_rename.py @@ -44,7 +44,7 @@ from f4pga.flows.module import Module, ModuleContext from f4pga.flows.runner import get_module -def _switch_keys(d: 'dict[str, ]', renames: 'dict[str, str]') -> 'dict[str, ]': +def _switch_keys(d: "dict[str, ]", renames: "dict[str, str]") -> "dict[str, ]": newd = {} for k, v in d.items(): r = renames.get(k) @@ -55,7 +55,7 @@ def _switch_keys(d: 'dict[str, ]', renames: 'dict[str, str]') -> 'dict[str, ]': return newd -def _switchback_attrs(d: Namespace, renames: 'dict[str, str]') -> SimpleNamespace: +def _switchback_attrs(d: Namespace, renames: "dict[str, str]") -> SimpleNamespace: newn = SimpleNamespace() for k, v in vars(d).items(): setattr(newn, k, v) @@ -67,7 +67,7 @@ def _switchback_attrs(d: Namespace, renames: 'dict[str, str]') -> SimpleNamespac return newn -def _switch_entries(l: 'list[str]', renames: 'dict[str, str]') -> 'list[str]': +def _switch_entries(l: "list[str]", renames: "dict[str, str]") -> "list[str]": newl = [] for e in l: r = renames.get(e) @@ -79,15 +79,15 @@ def _switch_entries(l: 'list[str]', renames: 'dict[str, str]') -> 'list[str]': return newl -def _or_empty_dict(d: 'dict | None'): +def _or_empty_dict(d: "dict | None"): return d if d is not None else {} class IORenameModule(Module): module: Module - rename_takes: 'dict[str, str]' - rename_produces: 'dict[str, str]' - rename_values: 'dict[str, str]' + rename_takes: "dict[str, str]" + rename_produces: "dict[str, str]" + rename_values: "dict[str, str]" def map_io(self, ctx: ModuleContext): newctx = ctx.shallow_copy() @@ -114,12 +114,13 @@ class IORenameModule(Module): self.rename_values = _or_empty_dict(params.get("rename_values")) self.module = module - self.name = f'{module.name}-io_renamed' + self.name = f"{module.name}-io_renamed" self.no_of_phases = module.no_of_phases self.takes = _switch_entries(module.takes, self.rename_takes) self.produces = _switch_entries(module.produces, self.rename_produces) self.values = _switch_entries(module.values, self.rename_values) - if hasattr(module, 'prod_meta'): + if hasattr(module, "prod_meta"): self.prod_meta = _switch_keys(module.prod_meta, self.rename_produces) + ModuleClass = IORenameModule diff --git a/f4pga/flows/common_modules/mkdirs.py b/f4pga/flows/common_modules/mkdirs.py index b9c8859..be5811a 100644 --- a/f4pga/flows/common_modules/mkdirs.py +++ b/f4pga/flows/common_modules/mkdirs.py @@ -31,7 +31,7 @@ from f4pga.flows.module import Module, ModuleContext class MkDirsModule(Module): - deps_to_produce: 'dict[str, str]' + deps_to_produce: "dict[str, str]" def map_io(self, ctx: ModuleContext): return ctx.r_env.resolve(self.deps_to_produce) @@ -39,11 +39,11 @@ class MkDirsModule(Module): def execute(self, ctx: ModuleContext): outputs = vars(ctx.outputs) for _, path in outputs.items(): - yield f'Creating directory {path}...' + yield f"Creating directory {path}..." Path(path).mkdir(parents=True, exist_ok=True) def __init__(self, params): - self.name = 'mkdirs' + self.name = "mkdirs" self.no_of_phases = len(params) if params else 0 self.takes = [] self.produces = list(params.keys()) if params else [] diff --git a/f4pga/flows/common_modules/pack.py b/f4pga/flows/common_modules/pack.py index 462be54..6a98549 100644 --- a/f4pga/flows/common_modules/pack.py +++ b/f4pga/flows/common_modules/pack.py @@ -23,8 +23,8 @@ from f4pga.flows.common import vpr_specific_values, noisy_warnings, vpr as commo from f4pga.flows.module import Module, ModuleContext -DEFAULT_TIMING_RPT = 'pre_pack.report_timing.setup.rpt' -DEFAULT_UTIL_RPT = 'packing_pin_util.rpt' +DEFAULT_TIMING_RPT = "pre_pack.report_timing.setup.rpt" +DEFAULT_UTIL_RPT = "packing_pin_util.rpt" class PackModule(Module): @@ -32,30 +32,21 @@ class PackModule(Module): epath = Path(ctx.takes.eblif) build_dir = epath.parent return { - 'net': str(epath.with_suffix('.net')), - 'util_rpt': str(build_dir / DEFAULT_UTIL_RPT), - 'timing_rpt': str(build_dir / DEFAULT_TIMING_RPT) + "net": str(epath.with_suffix(".net")), + "util_rpt": str(build_dir / DEFAULT_UTIL_RPT), + "timing_rpt": str(build_dir / DEFAULT_TIMING_RPT), } def execute(self, ctx: ModuleContext): noisy_warnings(ctx.values.device) build_dir = Path(ctx.outputs.net).parent - yield 'Packing with VPR...' - common_vpr( - 'pack', - VprArgs( - ctx.share, - ctx.takes.eblif, - ctx.values, - sdc_file=ctx.takes.sdc - ), - cwd=build_dir - ) + yield "Packing with VPR..." + common_vpr("pack", VprArgs(ctx.share, ctx.takes.eblif, ctx.values, sdc_file=ctx.takes.sdc), cwd=build_dir) - og_log = build_dir / 'vpr_stdout.log' + og_log = build_dir / "vpr_stdout.log" - yield 'Moving/deleting files...' + yield "Moving/deleting files..." if ctx.outputs.pack_log: og_log.rename(ctx.outputs.pack_log) else: @@ -68,20 +59,13 @@ class PackModule(Module): (build_dir / DEFAULT_UTIL_RPT).rename(ctx.outputs.util_rpt) def __init__(self, _): - self.name = 'pack' + self.name = "pack" self.no_of_phases = 2 - self.takes = [ - 'eblif', - 'sdc?' - ] - self.produces = [ - 'net', - 'util_rpt', - 'timing_rpt', - 'pack_log!' - ] + self.takes = ["eblif", "sdc?"] + self.produces = ["net", "util_rpt", "timing_rpt", "pack_log!"] self.values = [ - 'device', + "device", ] + vpr_specific_values() + ModuleClass = PackModule diff --git a/f4pga/flows/common_modules/place.py b/f4pga/flows/common_modules/place.py index 4f87b50..3014a24 100644 --- a/f4pga/flows/common_modules/place.py +++ b/f4pga/flows/common_modules/place.py @@ -25,10 +25,10 @@ from f4pga.flows.module import Module, ModuleContext def default_output_name(place_constraints): - m = re_match('(.*)\\.[^.]*$', place_constraints) + m = re_match("(.*)\\.[^.]*$", place_constraints) if m: - return m.groups()[0] + '.place' - return f'{place_constraints}.place' + return m.groups()[0] + ".place" + return f"{place_constraints}.place" def place_constraints_file(ctx: ModuleContext): @@ -36,36 +36,34 @@ def place_constraints_file(ctx: ModuleContext): return ctx.takes.place_constraints, False if ctx.takes.io_place: return ctx.takes.io_place, False - return f'{Path(ctx.takes.eblif).stem}.place', True + return f"{Path(ctx.takes.eblif).stem}.place", True class PlaceModule(Module): def map_io(self, ctx: ModuleContext): p, _ = place_constraints_file(ctx) - return { - 'place': default_output_name(p) - } + return {"place": default_output_name(p)} def execute(self, ctx: ModuleContext): place_constraints, dummy = place_constraints_file(ctx) place_constraints = Path(place_constraints).resolve() if dummy: - with place_constraints.open('wb') as wfptr: - wfptr.write(b'') + with place_constraints.open("wb") as wfptr: + wfptr.write(b"") build_dir = Path(ctx.takes.eblif).parent - yield 'Running VPR...' + yield "Running VPR..." common_vpr( - 'place', + "place", VprArgs( ctx.share, ctx.takes.eblif, ctx.values, sdc_file=ctx.takes.sdc, - vpr_extra_opts=['--fix_clusters', place_constraints] + vpr_extra_opts=["--fix_clusters", place_constraints], ), - cwd=build_dir + cwd=build_dir, ) # VPR names output on its own. If user requested another name, the @@ -75,25 +73,18 @@ class PlaceModule(Module): # when the problem gets tackled, we should keep in mind that VPR-based # modules may produce some temporary files with names that differ from # the ones in flow configuration. - if ctx.is_output_explicit('place'): + if ctx.is_output_explicit("place"): Path(default_output_name(str(place_constraints))).rename(ctx.outputs.place) - yield 'Saving log...' - save_vpr_log('place.log', build_dir=build_dir) + yield "Saving log..." + save_vpr_log("place.log", build_dir=build_dir) def __init__(self, _): - self.name = 'place' + self.name = "place" self.no_of_phases = 2 - self.takes = [ - 'eblif', - 'sdc?', - 'place_constraints?', - 'io_place?' - ] - self.produces = [ 'place' ] - self.values = [ - 'device', - 'vpr_options?' - ] + vpr_specific_values() + self.takes = ["eblif", "sdc?", "place_constraints?", "io_place?"] + self.produces = ["place"] + self.values = ["device", "vpr_options?"] + vpr_specific_values() + ModuleClass = PlaceModule diff --git a/f4pga/flows/common_modules/place_constraints.py b/f4pga/flows/common_modules/place_constraints.py index 2d0e3ab..1eababf 100644 --- a/f4pga/flows/common_modules/place_constraints.py +++ b/f4pga/flows/common_modules/place_constraints.py @@ -25,43 +25,41 @@ from f4pga.flows.module import Module, ModuleContext class PlaceConstraintsModule(Module): def map_io(self, ctx: ModuleContext): - return { - 'place_constraints': f'{Path(ctx.takes.net).stem!s}.preplace' - } + return {"place_constraints": f"{Path(ctx.takes.net).stem!s}.preplace"} def execute(self, ctx: ModuleContext): - yield 'Saving place constraint data...' - with Path(ctx.outputs.place_constraints).open('wb') as wfptr: + yield "Saving place constraint data..." + with Path(ctx.outputs.place_constraints).open("wb") as wfptr: wfptr.write( - common_sub(*( - [ - 'python3', ctx.values.script, - '--net', ctx.takes.net, - '--arch', str(Path(ctx.share) / 'arch' / ctx.values.device / 'arch.timing.xml'), - '--blif', ctx.takes.eblif, - '--input', ctx.takes.io_place, - '--db_root', common_sub('prjxray-config').decode().replace('\n', ''), - '--part', ctx.values.part_name - ] + ( - options_dict_to_list(ctx.values.extra_opts) if ctx.values.extra_opts else [] + common_sub( + *( + [ + "python3", + ctx.values.script, + "--net", + ctx.takes.net, + "--arch", + str(Path(ctx.share) / "arch" / ctx.values.device / "arch.timing.xml"), + "--blif", + ctx.takes.eblif, + "--input", + ctx.takes.io_place, + "--db_root", + common_sub("prjxray-config").decode().replace("\n", ""), + "--part", + ctx.values.part_name, + ] + + (options_dict_to_list(ctx.values.extra_opts) if ctx.values.extra_opts else []) ) - )) + ) ) def __init__(self, _): - self.name = 'place_constraints' + self.name = "place_constraints" self.no_of_phases = 2 - self.takes = [ - 'eblif', - 'net', - 'io_place' - ] - self.produces = [ 'place_constraints' ] - self.values = [ - 'device', - 'part_name', - 'script', - 'extra_opts?' - ] + self.takes = ["eblif", "net", "io_place"] + self.produces = ["place_constraints"] + self.values = ["device", "part_name", "script", "extra_opts?"] + ModuleClass = PlaceConstraintsModule diff --git a/f4pga/flows/common_modules/route.py b/f4pga/flows/common_modules/route.py index 6e14071..dd4a614 100644 --- a/f4pga/flows/common_modules/route.py +++ b/f4pga/flows/common_modules/route.py @@ -24,50 +24,33 @@ from f4pga.flows.module import Module, ModuleContext def route_place_file(ctx: ModuleContext): - return Path(ctx.takes.eblif).with_suffix('.route') + return Path(ctx.takes.eblif).with_suffix(".route") class RouteModule(Module): def map_io(self, ctx: ModuleContext): - return { - 'route': str(route_place_file(ctx)) - } + return {"route": str(route_place_file(ctx))} def execute(self, ctx: ModuleContext): build_dir = Path(ctx.takes.eblif).parent vpr_options = options_dict_to_list(ctx.values.vpr_options) if ctx.values.vpr_options else [] - yield 'Routing with VPR...' - common_vpr( - 'route', - VprArgs( - ctx.share, - ctx.takes.eblif, - ctx.values, - sdc_file=ctx.takes.sdc - ), - cwd=build_dir - ) + yield "Routing with VPR..." + common_vpr("route", VprArgs(ctx.share, ctx.takes.eblif, ctx.values, sdc_file=ctx.takes.sdc), cwd=build_dir) - if ctx.is_output_explicit('route'): + if ctx.is_output_explicit("route"): route_place_file(ctx).rename(ctx.outputs.route) - yield 'Saving log...' - save_vpr_log('route.log', build_dir=build_dir) + yield "Saving log..." + save_vpr_log("route.log", build_dir=build_dir) def __init__(self, _): - self.name = 'route' + self.name = "route" self.no_of_phases = 2 - self.takes = [ - 'eblif', - 'place', - 'sdc?' - ] - self.produces = [ 'route' ] - self.values = [ - 'device', - 'vpr_options?' - ] + vpr_specific_values() + self.takes = ["eblif", "place", "sdc?"] + self.produces = ["route"] + self.values = ["device", "vpr_options?"] + vpr_specific_values() + ModuleClass = RouteModule diff --git a/f4pga/flows/common_modules/synth.py b/f4pga/flows/common_modules/synth.py index e014ab4..46ff941 100755 --- a/f4pga/flows/common_modules/synth.py +++ b/f4pga/flows/common_modules/synth.py @@ -25,145 +25,115 @@ from f4pga.flows.module import Module, ModuleContext from f4pga.wrappers.tcl import get_script_path as get_tcl_wrapper_path -def yosys_setup_tcl_env(tcl_env_def): - """ - Setup environmental variables for YOSYS TCL scripts. - """ - return { - key: (' '.join(val) if type(val) is list else val) - for key, val in tcl_env_def.items() - if val is not None - } - - -def yosys_synth(tcl, tcl_env, verilog_files=[], read_verilog_args=None, log=None): - tcl = f'tcl {tcl}' - # Use append read_verilog commands to the scripts for more sophisticated - # input if arguments are specified. Omit direct input throught `yosys` command. - if read_verilog_args: - args_str = ' '.join(read_verilog_args) - for verilog in verilog_files: - tcl = f'read_verilog {args_str} {verilog}; {tcl}' - verilog_files = [] - - # Set up environment for TCL weirdness - env = environ.copy() - env.update(tcl_env) - # Execute YOSYS command - return common_sub(*(['yosys', '-p', tcl] + (['-l', log] if log else []) + verilog_files), env=env) - - -def yosys_conv(tcl, tcl_env, synth_json): - # Set up environment for TCL weirdness - env = environ.copy() - env.update(tcl_env) - return common_sub('yosys', '-p', f'read_json {synth_json}; tcl {tcl}', env=env) - - class SynthModule(Module): - extra_products: 'list[str]' + extra_products: "list[str]" def map_io(self, ctx: ModuleContext): - mapping = {} + top = Path(ctx.takes.build_dir) / ctx.values.top if ctx.takes.build_dir else Path(ctx.values.top) - top = ctx.values.top - if ctx.takes.build_dir: - top = str(Path(ctx.takes.build_dir) / top) - mapping['eblif'] = top + '.eblif' - mapping['fasm_extra'] = top + '_fasm_extra.fasm' - mapping['json'] = top + '.json' - mapping['synth_json'] = top + '_io.json' + mapping = { + "eblif": f"{top!s}.eblif", + "fasm_extra": f"{top!s}_fasm_extra.fasm", + "json": f"{top!s}.json", + "synth_json": f"{top!s}_io.json", + } for extra in self.extra_products: name, spec = decompose_depname(extra) - if spec == 'maybe': + if spec == "maybe": raise ModuleRuntimeException( - f'Yosys synth extra products can\'t use \'maybe\ ' - f'(?) specifier. Product causing this error: `{extra}`.' + f"Yosys synth extra products can't use 'maybe\ " + f"(?) specifier. Product causing this error: `{extra}`." ) - elif spec == 'req': - mapping[name] = str(Path(top).parent / f'{ctx.values.device}_{name}.{name}') + elif spec == "req": + mapping[name] = str(top.parent / f"{ctx.values.device}_{name}.{name}") return mapping def execute(self, ctx: ModuleContext): - tcl_env = yosys_setup_tcl_env(ctx.values.yosys_tcl_env) \ - if ctx.values.yosys_tcl_env else {} - split_inouts = Path(tcl_env["UTILS_PATH"]) / 'split_inouts.py' - - if get_verbosity_level() >= 2: - yield f'Synthesizing sources: {ctx.takes.sources}...' - else: - yield f'Synthesizing sources...' - - yosys_synth( - str(get_tcl_wrapper_path('synth')), - tcl_env, - ctx.takes.sources, - ctx.values.read_verilog_args, - ctx.outputs.synth_log + # Setup environmental variables for YOSYS TCL scripts. + tcl_env = ( + { + key: (" ".join(val) if type(val) is list else val) + for key, val in ctx.values.yosys_tcl_env.items() + if val is not None + } + if ctx.values.yosys_tcl_env + else {} ) - yield f'Splitting in/outs...' - common_sub('python3', str(split_inouts), '-i', ctx.outputs.json, '-o', - ctx.outputs.synth_json) + yield f"Synthesizing sources{f': {ctx.takes.sources}...' if get_verbosity_level() >= 2 else f'...'}" + tcl = f'tcl {str(get_tcl_wrapper_path("synth"))}' + verilog_files = [] + # Use append read_verilog commands to the scripts for more sophisticated + # input if arguments are specified. Omit direct input throught `yosys` command. + if ctx.values.read_verilog_args: + args_str = " ".join(ctx.values.read_verilog_args) + for vfile in ctx.takes.sources: + tcl = f"read_verilog {args_str} {vfile}; {tcl}" + else: + verilog_files = ctx.takes.sources + # Set up environment for TCL weirdness + env = environ.copy() + env.update(tcl_env) + # Execute YOSYS command + common_sub( + *(["yosys", "-p", tcl] + (["-l", ctx.outputs.synth_log] if ctx.outputs.synth_log else []) + verilog_files), + env=env, + ) + + yield f"Splitting in/outs..." + common_sub( + "python3", + str(Path(tcl_env["UTILS_PATH"]) / "split_inouts.py"), + "-i", + ctx.outputs.json, + "-o", + ctx.outputs.synth_json, + ) if not Path(ctx.produces.fasm_extra).is_file(): - with Path(ctx.produces.fasm_extra).open('w') as wfptr: - wfptr.write('') + with Path(ctx.produces.fasm_extra).open("w") as wfptr: + wfptr.write("") - yield f'Converting...' - yosys_conv( - str(get_tcl_wrapper_path('conv')), - tcl_env, - ctx.outputs.synth_json + yield f"Converting..." + # Set up environment for TCL weirdness + env = environ.copy() + env.update(tcl_env) + common_sub( + "yosys", "-p", f'read_json {ctx.outputs.synth_json}; tcl {str(get_tcl_wrapper_path("conv"))}', env=env ) def __init__(self, params): - self.name = 'synthesize' + self.name = "synthesize" self.no_of_phases = 3 - self.takes = [ - 'sources', - 'build_dir?' - ] + self.takes = ["sources", "build_dir?"] # Extra takes for use with TCL scripts - extra_takes = params.get('takes') + extra_takes = params.get("takes") if extra_takes: self.takes += extra_takes - self.produces = [ - 'eblif', - 'fasm_extra', - 'json', - 'synth_json', - 'synth_log!' - ] + self.produces = ["eblif", "fasm_extra", "json", "synth_json", "synth_log!"] # Extra products for use with TCL scripts - extra_products = params.get('produces') + extra_products = params.get("produces") if extra_products: self.produces += extra_products self.extra_products = extra_products else: self.extra_products = [] - self.values = [ - 'top', - 'device', - 'tcl_scripts', - 'yosys_tcl_env?', - 'read_verilog_args?' - ] + self.values = ["top", "device", "tcl_scripts", "yosys_tcl_env?", "read_verilog_args?"] self.prod_meta = { - 'eblif': 'Extended BLIF hierarchical sequential designs file\n' - 'generated by YOSYS', - 'json': 'JSON file containing a design generated by YOSYS', - 'synth_log': 'YOSYS synthesis log', - 'fasm_extra': 'Extra FASM generated during sythesis stage. Needed in ' - 'some designs.\nIn case it\'s not necessary, the file ' - 'will be empty.' + "eblif": "Extended BLIF hierarchical sequential designs file\n" "generated by YOSYS", + "json": "JSON file containing a design generated by YOSYS", + "synth_log": "YOSYS synthesis log", + "fasm_extra": "Extra FASM generated during sythesis stage. Needed in " + "some designs.\nIn case it's not necessary, the file " + "will be empty.", } - extra_meta = params.get('prod_meta') + extra_meta = params.get("prod_meta") if extra_meta: self.prod_meta.update(extra_meta) + ModuleClass = SynthModule diff --git a/f4pga/flows/flow.py b/f4pga/flows/flow.py index 8e5d98c..811fd10 100644 --- a/f4pga/flows/flow.py +++ b/f4pga/flows/flow.py @@ -29,47 +29,48 @@ from f4pga.flows.stage import Stage class Flow: - """ Describes a complete, configured flow, ready for execution. """ + """Describes a complete, configured flow, ready for execution.""" # Dependendecy to build target: str # Values in global scope cfg: FlowConfig # dependency-producer map - os_map: 'dict[str, Stage]' + os_map: "dict[str, Stage]" # Paths resolved for dependencies - dep_paths: 'dict[str, str | list[str]]' + dep_paths: "dict[str, str | list[str]]" # Explicit configs for dependency paths # config_paths: 'dict[str, str | list[str]]' # Stages that need to be run - run_stages: 'set[str]' + run_stages: "set[str]" # Number of stages that relied on outdated version of a (checked) dependency - deps_rebuilds: 'dict[str, int]' - f4cache: 'F4Cache | None' + deps_rebuilds: "dict[str, int]" + f4cache: "F4Cache | None" flow_cfg: FlowConfig - def __init__(self, target: str, cfg: FlowConfig, - f4cache: 'F4Cache | None'): + def __init__(self, target: str, cfg: FlowConfig, f4cache: "F4Cache | None"): self.target = target # Associate a stage with every possible output. # This is commonly refferef to as `os_map` (output-stage-map) through the code. - os_map: 'dict[str, Stage]' = {} # Output-Stage map + os_map: "dict[str, Stage]" = {} # Output-Stage map for stage in cfg.stages.values(): for output in stage.produces: if not os_map.get(output.name): os_map[output.name] = stage elif os_map[output.name] != stage: - raise Exception(f'Dependency `{output.name}` is generated by ' - f'stage `{os_map[output.name].name}` and ' - f'`{stage.name}`. Dependencies can have only one ' - 'provider at most.') + raise Exception( + f"Dependency `{output.name}` is generated by " + f"stage `{os_map[output.name].name}` and " + f"`{stage.name}`. Dependencies can have only one " + "provider at most." + ) self.os_map = os_map self.dep_paths = { n: p for n, p in cfg.get_dependency_overrides().items() - if p_req_exists(p) # and not p_dep_differ(p, f4cache) + if p_req_exists(p) # and not p_dep_differ(p, f4cache) } if f4cache is not None: for dep in self.dep_paths.values(): @@ -85,14 +86,14 @@ class Flow: @staticmethod def _config_mod_runctx( stage: Stage, - values: 'dict[str, ]', - dep_paths: 'dict[str, str | list[str]]', - config_paths: 'dict[str, str | list[str]]' + values: "dict[str, ]", + dep_paths: "dict[str, str | list[str]]", + config_paths: "dict[str, str | list[str]]", ): takes = {} for take in stage.takes: paths = dep_paths.get(take.name) - if paths: # Some takes may be not required + if paths: # Some takes may be not required takes[take.name] = paths produces = {} @@ -102,20 +103,13 @@ class Flow: elif config_paths.get(prod.name): produces[prod.name] = config_paths[prod.name] - return ModRunCtx( - share_dir_path, - bin_dir_path, - { - 'takes': takes, - 'produces': produces, - 'values': values - } - ) + return ModRunCtx(share_dir_path, bin_dir_path, {"takes": takes, "produces": produces, "values": values}) @staticmethod def _cache_deps(path: str, f4cache: F4Cache): def _process_dep_path(path: str, f4cache: F4Cache): f4cache.process_file(Path(path)) + deep(_process_dep_path)(path, f4cache) def _dep_will_differ(self, dep: str, paths, consumer: str): @@ -123,15 +117,14 @@ class Flow: Check if a dependency or any of the dependencies it depends on differ from their last versions. """ - if not self.f4cache: # Handle --nocache mode + if not self.f4cache: # Handle --nocache mode return True provider = self.os_map.get(dep) if provider and (provider.name in self.run_stages): return True return p_dep_differ(paths, consumer, self.f4cache) - def _resolve_dependencies(self, dep: str, stages_checked: 'set[str]', - skip_dep_warnings: 'set[str]' = None): + def _resolve_dependencies(self, dep: str, stages_checked: "set[str]", skip_dep_warnings: "set[str]" = None): if skip_dep_warnings is None: skip_dep_warnings = set() @@ -156,12 +149,13 @@ class Flow: # provider stage cannot be run take_paths = self.dep_paths.get(take.name) # Add input path to values (dirty hack) - provider.value_overrides[f':{take.name}'] = take_paths + provider.value_overrides[f":{take.name}"] = take_paths - if not take_paths and take.spec == 'req': - sfprint(0, - f' Stage `{Style.BRIGHT + provider.name + Style.RESET_ALL}` is ' - f'unreachable due to unmet dependency `{Style.BRIGHT + take.name + Style.RESET_ALL}`' + if not take_paths and take.spec == "req": + sfprint( + 0, + f" Stage `{Style.BRIGHT + provider.name + Style.RESET_ALL}` is " + f"unreachable due to unmet dependency `{Style.BRIGHT + take.name + Style.RESET_ALL}`", ) return @@ -176,8 +170,11 @@ class Flow: if will_differ: if take.name not in skip_dep_warnings: - sfprint(2, f'{Style.BRIGHT}{take.name}{Style.RESET_ALL} is causing ' - f'rebuild for `{Style.BRIGHT}{provider.name}{Style.RESET_ALL}`') + sfprint( + 2, + f"{Style.BRIGHT}{take.name}{Style.RESET_ALL} is causing " + f"rebuild for `{Style.BRIGHT}{provider.name}{Style.RESET_ALL}`", + ) skip_dep_warnings.add(take.name) self.run_stages.add(provider.name) self.deps_rebuilds[take.name] += 1 @@ -185,11 +182,8 @@ class Flow: outputs = module_map( provider.module, self._config_mod_runctx( - provider, - self.cfg.get_r_env(provider.name).values, - self.dep_paths, - self.cfg.get_dependency_overrides() - ) + provider, self.cfg.get_r_env(provider.name).values, self.dep_paths, self.cfg.get_dependency_overrides() + ), ) for output_paths in outputs.values(): if output_paths is not None: @@ -207,10 +201,8 @@ class Flow: outs = outputs.keys() for o in provider.produces: if o.name not in outs: - if o.spec == 'req' or (o.spec == 'demand' and \ - o.name in self.cfg.get_dependency_overrides().keys()): - fatal(-1, f'Module {provider.name} did not produce a mapping ' - f'for a required output `{o.name}`') + if o.spec == "req" or (o.spec == "demand" and o.name in self.cfg.get_dependency_overrides().keys()): + fatal(-1, f"Module {provider.name} did not produce a mapping " f"for a required output `{o.name}`") else: # Remove an on-demand/optional output that is not produced # from os_map. @@ -219,43 +211,40 @@ class Flow: o_path = outputs.get(o.name) if o_path is not None: - provider.value_overrides[f':{o.name}'] = \ - outputs.get(o.name) - + provider.value_overrides[f":{o.name}"] = outputs.get(o.name) def print_resolved_dependencies(self, verbosity: int): deps = list(self.deps_rebuilds.keys()) deps.sort() for dep in deps: - status = Fore.RED + '[X]' + Fore.RESET - source = Fore.YELLOW + 'MISSING' + Fore.RESET + status = Fore.RED + "[X]" + Fore.RESET + source = Fore.YELLOW + "MISSING" + Fore.RESET paths = self.dep_paths.get(dep) if paths: exists = p_req_exists(paths) provider = self.os_map.get(dep) if provider and provider.name in self.run_stages: - status = Fore.YELLOW + ('[R]' if exists else '[S]') + Fore.RESET - source = f'{Fore.BLUE + self.os_map[dep].name + Fore.RESET} -> {paths}' + status = Fore.YELLOW + ("[R]" if exists else "[S]") + Fore.RESET + source = f"{Fore.BLUE + self.os_map[dep].name + Fore.RESET} -> {paths}" elif exists: - status = Fore.GREEN + ('[N]' if self.deps_rebuilds[dep] > 0 else '[O]') + Fore.RESET + status = Fore.GREEN + ("[N]" if self.deps_rebuilds[dep] > 0 else "[O]") + Fore.RESET source = paths elif self.os_map.get(dep): - status = Fore.RED + '[U]' + Fore.RESET - source = \ - f'{Fore.BLUE + self.os_map[dep].name + Fore.RESET} -> ???' + status = Fore.RED + "[U]" + Fore.RESET + source = f"{Fore.BLUE + self.os_map[dep].name + Fore.RESET} -> ???" - sfprint(verbosity, f' {Style.BRIGHT + status} ' - f'{dep + Style.RESET_ALL}: {source}') + sfprint(verbosity, f" {Style.BRIGHT + status} " f"{dep + Style.RESET_ALL}: {source}") def _build_dep(self, dep): paths = self.dep_paths.get(dep) + if not paths: + sfprint(2, f"Dependency {dep} is unresolved.") + return False + provider = self.os_map.get(dep) run = (provider.name in self.run_stages) if provider else False - if not paths: - sfprint(2, f'Dependency {dep} is unresolved.') - return False if p_req_exists(paths) and not run: return True @@ -265,7 +254,7 @@ class Flow: any_dep_differ = False if (self.f4cache is not None) else True for p_dep in provider.takes: if not self._build_dep(p_dep.name): - assert (p_dep.spec != 'req') + assert p_dep.spec != "req" continue if self.f4cache is not None: any_dep_differ |= p_update_dep_statuses(self.dep_paths[p_dep.name], provider.name, self.f4cache) @@ -277,9 +266,12 @@ class Flow: # will reamin the same, thus making it unnecessary to continue the # rebuild process. if (not any_dep_differ) and p_req_exists(paths): - sfprint(2, f'Skipping rebuild of `' - f'{Style.BRIGHT + dep + Style.RESET_ALL}` because all ' - f'of it\'s dependencies remained unchanged') + sfprint( + 2, + f"Skipping rebuild of `" + f"{Style.BRIGHT + dep + Style.RESET_ALL}` because all " + f"of it's dependencies remained unchanged", + ) return True module_exec( @@ -287,14 +279,15 @@ class Flow: self._config_mod_runctx( provider, self.cfg.get_r_env(provider.name).values, - self.dep_paths,self.cfg.get_dependency_overrides() - ) + self.dep_paths, + self.cfg.get_dependency_overrides(), + ), ) self.run_stages.discard(provider.name) for product in provider.produces: - if (product.spec == 'req') and not p_req_exists(paths): + if (product.spec == "req") and not p_req_exists(paths): raise DependencyNotProducedException(dep, provider.name) prod_paths = self.dep_paths[product.name] if (prod_paths is not None) and p_req_exists(paths) and self.f4cache: @@ -306,8 +299,8 @@ class Flow: self._build_dep(self.target) if self.f4cache: self._cache_deps(self.dep_paths[self.target], self.f4cache) - p_update_dep_statuses(self.dep_paths[self.target], '__target', self.f4cache) - sfprint(0, f'Target {Style.BRIGHT + self.target + Style.RESET_ALL} -> {self.dep_paths[self.target]}') + p_update_dep_statuses(self.dep_paths[self.target], "__target", self.f4cache) + sfprint(0, f"Target {Style.BRIGHT + self.target + Style.RESET_ALL} -> {self.dep_paths[self.target]}") class DependencyNotProducedException(F4PGAException): @@ -317,7 +310,7 @@ class DependencyNotProducedException(F4PGAException): def __init__(self, dep_name: str, provider: str): self.dep_name = dep_name self.provider = provider - self.message = f'Stage `{self.provider}` did not produce promised dependency `{self.dep_name}`' + self.message = f"Stage `{self.provider}` did not produce promised dependency `{self.dep_name}`" def p_req_exists(r): @@ -330,7 +323,7 @@ def p_req_exists(r): elif type(r) is list: return not (False in map(p_req_exists, r)) else: - raise Exception(f'Requirements can be currently checked only for single paths, or path lists (reason: {r})') + raise Exception(f"Requirements can be currently checked only for single paths, or path lists (reason: {r})") return True @@ -343,7 +336,7 @@ def p_update_dep_statuses(paths, consumer: str, f4cache: F4Cache): elif type(paths) is dict: for _, p in paths.items(): return p_update_dep_statuses(p, consumer, f4cache) - fatal(-1, 'WRONG PATHS TYPE') + fatal(-1, "WRONG PATHS TYPE") def p_dep_differ(paths, consumer: str, f4cache: F4Cache): @@ -353,7 +346,7 @@ def p_dep_differ(paths, consumer: str, f4cache: F4Cache): if type(paths) is str: if not Path(paths).exists(): return True - return f4cache.get_status(paths, consumer) != 'same' + return f4cache.get_status(paths, consumer) != "same" elif type(paths) is list: return True in [p_dep_differ(p, consumer, f4cache) for p in paths] elif type(paths) is dict: diff --git a/f4pga/flows/flow_config.py b/f4pga/flows/flow_config.py index ae2998c..29ffaf7 100644 --- a/f4pga/flows/flow_config.py +++ b/f4pga/flows/flow_config.py @@ -27,15 +27,11 @@ from f4pga.flows.stage import Stage def open_flow_cfg(path: str) -> dict: - with Path(path).open('r') as rfptr: + with Path(path).open("r") as rfptr: return json_load(rfptr) -def _get_ovs_raw( - dict_name: str, - flow_cfg, - part: 'str | None', - stage: 'str | None' -): + +def _get_ovs_raw(dict_name: str, flow_cfg, part: "str | None", stage: "str | None"): vals = flow_cfg.get(dict_name) if vals is None: vals = {} @@ -50,8 +46,9 @@ def _get_ovs_raw( return vals + def verify_platform_name(platform: str, mypath: str): - for plat_def_filename in os_listdir(str(Path(mypath) / 'platforms')): + for plat_def_filename in os_listdir(str(Path(mypath) / "platforms")): platform_name = str(Path(plat_def_filename).stem) if platform == platform_name: return True @@ -59,17 +56,12 @@ def verify_platform_name(platform: str, mypath: str): def _is_kword(w: str): - kwords = { - 'dependencies', - 'values', - 'default_platform', - 'default_target' - } + kwords = {"dependencies", "values", "default_platform", "default_target"} return w in kwords class FlowDefinition: - stages: 'dict[str, Stage]' # stage name -> module path mapping + stages: "dict[str, Stage]" # stage name -> module path mapping r_env: ResolutionEnv def __init__(self, flow_def: dict, r_env: ResolutionEnv): @@ -77,18 +69,17 @@ class FlowDefinition: self.r_env = r_env self.stages = {} - global_vals = flow_def.get('values') + global_vals = flow_def.get("values") if global_vals is not None: self.r_env.add_values(global_vals) - stages_d = flow_def['stages'] - - for stage_name, stage_def in stages_d.items(): + for stage_name, stage_def in flow_def["stages"].items(): self.stages[stage_name] = Stage(stage_name, stage_def) def stage_names(self): return self.stages.keys() + class ProjectFlowConfig: flow_cfg: dict path: str @@ -102,44 +93,40 @@ class ProjectFlowConfig: if not _is_kword(part): yield part - def get_default_part(self) -> 'str | None': - return self.flow_cfg.get('default_part') + def get_default_part(self) -> "str | None": + return self.flow_cfg.get("default_part") - def get_default_target(self, part: str) -> 'str | None': - return self.flow_cfg[part].get('default_target') + def get_default_target(self, part: str) -> "str | None": + return self.flow_cfg[part].get("default_target") - def get_dependencies_raw(self, part: 'str | None' = None): + def get_dependencies_raw(self, part: "str | None" = None): """ Get dependencies without value resolution applied. """ - return _get_ovs_raw('dependencies', self.flow_cfg, part, None) + return _get_ovs_raw("dependencies", self.flow_cfg, part, None) - def get_values_raw( - self, - part: 'str | None' = None, - stage: 'str | None' = None - ): + def get_values_raw(self, part: "str | None" = None, stage: "str | None" = None): """ Get values without value resolution applied. """ - return _get_ovs_raw('values', self.flow_cfg, part, stage) + return _get_ovs_raw("values", self.flow_cfg, part, stage) def get_stage_value_overrides(self, part: str, stage: str): stage_vals_ovds = {} - vals = self.flow_cfg.get('values') + vals = self.flow_cfg.get("values") if vals is not None: stage_vals_ovds.update(vals) stage_cfg = self.flow_cfg[part].get(stage) if stage_cfg is not None: - vals = stage_cfg.get('values') + vals = stage_cfg.get("values") if vals is not None: stage_vals_ovds.update(vals) return stage_vals_ovds def get_dependency_platform_overrides(self, part: str): - platform_ovds = self.flow_cfg[part].get('dependencies') + platform_ovds = self.flow_cfg[part].get("dependencies") if platform_ovds is None: return {} return platform_ovds @@ -148,24 +135,21 @@ class ProjectFlowConfig: class FlowConfig: part: str r_env: ResolutionEnv - dependencies_explicit: 'dict[str, ]' - stages: 'dict[str, Stage]' + dependencies_explicit: "dict[str, ]" + stages: "dict[str, Stage]" - def __init__(self, project_config: ProjectFlowConfig, - platform_def: FlowDefinition, part: str): + def __init__(self, project_config: ProjectFlowConfig, platform_def: FlowDefinition, part: str): self.r_env = platform_def.r_env - platform_vals = project_config.get_values_raw(part) - self.r_env.add_values(platform_vals) + self.r_env.add_values(project_config.get_values_raw(part)) self.stages = platform_def.stages self.part = part - raw_project_deps = project_config.get_dependencies_raw(part) - - self.dependencies_explicit = deep(lambda p: str(Path(p).resolve()))(self.r_env.resolve(raw_project_deps)) + self.dependencies_explicit = deep(lambda p: str(Path(p).resolve()))( + self.r_env.resolve(project_config.get_dependencies_raw(part)) + ) for stage_name, stage in platform_def.stages.items(): - project_val_ovds = \ - project_config.get_stage_value_overrides(part, stage_name) + project_val_ovds = project_config.get_stage_value_overrides(part, stage_name) stage.value_overrides.update(project_val_ovds) def get_dependency_overrides(self): @@ -181,6 +165,7 @@ class FlowConfig: def get_stage(self, stage_name: str) -> Stage: return self.stages[stage_name] + class FlowConfigException(Exception): path: str message: str @@ -190,11 +175,11 @@ class FlowConfigException(Exception): self.message = message def __str__(self) -> str: - return f'Error in config `{self.path}: {self.message}' + return f"Error in config `{self.path}: {self.message}" def open_project_flow_cfg(path: str) -> ProjectFlowConfig: cfg = ProjectFlowConfig(path) - with Path(path).open('r') as rfptr: + with Path(path).open("r") as rfptr: cfg.flow_cfg = json_load(rfptr) return cfg diff --git a/f4pga/flows/inspector.py b/f4pga/flows/inspector.py index a78d532..7ed4ea6 100644 --- a/f4pga/flows/inspector.py +++ b/f4pga/flows/inspector.py @@ -23,37 +23,39 @@ from f4pga.flows.module import Module from f4pga.flows.common import decompose_depname -def _get_if_qualifier(deplist: 'list[str]', qualifier: str): +def _get_if_qualifier(deplist: "list[str]", qualifier: str): for dep_name in deplist: name, q = decompose_depname(dep_name) if q == qualifier: - yield f'● {Style.BRIGHT}{name}{Style.RESET_ALL}' + yield f"● {Style.BRIGHT}{name}{Style.RESET_ALL}" -def _list_if_qualifier(deplist: 'list[str]', qualifier: str, indent: int = 4): - indent_str = ''.join([' ' for _ in range(0, indent)]) - r = '' + +def _list_if_qualifier(deplist: "list[str]", qualifier: str, indent: int = 4): + indent_str = "".join([" " for _ in range(0, indent)]) + r = "" for line in _get_if_qualifier(deplist, qualifier): - r += indent_str + line + '\n' + r += indent_str + line + "\n" return r + def get_module_info(module: Module) -> str: - r= '' - r += f'Module `{Style.BRIGHT}{module.name}{Style.RESET_ALL}`:\n' - r += 'Inputs:\n Required:\n Dependencies\n' - r += _list_if_qualifier(module.takes, 'req', indent=6) - r += ' Values:\n' - r += _list_if_qualifier(module.values, 'req', indent=6) - r += ' Optional:\n Dependencies:\n' - r += _list_if_qualifier(module.takes, 'maybe', indent=6) - r += ' Values:\n' - r += _list_if_qualifier(module.values, 'maybe', indent=6) - r += 'Outputs:\n Guaranteed:\n' - r += _list_if_qualifier(module.produces, 'req', indent=4) - r += ' On-demand:\n' - r += _list_if_qualifier(module.produces, 'demand', indent=4) - r += ' Not guaranteed:\n' - r += _list_if_qualifier(module.produces, 'maybe', indent= 4) + r = "" + r += f"Module `{Style.BRIGHT}{module.name}{Style.RESET_ALL}`:\n" + r += "Inputs:\n Required:\n Dependencies\n" + r += _list_if_qualifier(module.takes, "req", indent=6) + r += " Values:\n" + r += _list_if_qualifier(module.values, "req", indent=6) + r += " Optional:\n Dependencies:\n" + r += _list_if_qualifier(module.takes, "maybe", indent=6) + r += " Values:\n" + r += _list_if_qualifier(module.values, "maybe", indent=6) + r += "Outputs:\n Guaranteed:\n" + r += _list_if_qualifier(module.produces, "req", indent=4) + r += " On-demand:\n" + r += _list_if_qualifier(module.produces, "demand", indent=4) + r += " Not guaranteed:\n" + r += _list_if_qualifier(module.produces, "maybe", indent=4) return r diff --git a/f4pga/flows/module.py b/f4pga/flows/module.py index 350d184..0e25b7b 100644 --- a/f4pga/flows/module.py +++ b/f4pga/flows/module.py @@ -24,11 +24,7 @@ Here are the things necessary to write an F4PGA Module. from types import SimpleNamespace from abc import abstractmethod -from f4pga.flows.common import ( - decompose_depname, - ResolutionEnv, - fatal -) +from f4pga.flows.common import decompose_depname, ResolutionEnv, fatal class Module: @@ -41,10 +37,10 @@ class Module: no_of_phases: int name: str - takes: 'list[str]' - produces: 'list[str]' - values: 'list[str]' - prod_meta: 'dict[str, str]' + takes: "list[str]" + produces: "list[str]" + values: "list[str]" + prod_meta: "dict[str, str]" @abstractmethod def execute(self, ctx): @@ -56,17 +52,17 @@ class Module: pass @abstractmethod - def map_io(self, ctx) -> 'dict[str, ]': + def map_io(self, ctx) -> "dict[str, ]": """ Returns paths for outputs derived from given inputs. `ctx` is `ModuleContext`. """ pass - def __init__(self, params: 'dict[str, ]'): + def __init__(self, params: "dict[str, ]"): self.no_of_phases = 0 self.current_phase = 0 - self.name = '' + self.name = "" self.prod_meta = {} @@ -76,16 +72,16 @@ class ModuleContext: execution. """ - share: str # Absolute path to F4PGA's share directory - bin: str # Absolute path to F4PGA's bin directory - takes: SimpleNamespace # Maps symbolic dependency names to relative paths. + share: str # Absolute path to F4PGA's share directory + bin: str # Absolute path to F4PGA's bin directory + takes: SimpleNamespace # Maps symbolic dependency names to relative paths. produces: SimpleNamespace # Contains mappings for explicitely specified dependencies. - # Useful mostly for checking for on-demand optional outputs (such as logs) with - # `is_output_explicit` method. - outputs: SimpleNamespace # Contains mappings for all available outputs. - values: SimpleNamespace # Contains all available requested values. - r_env: ResolutionEnv # `ResolutionEnvironmet` object holding mappings for current scope. - module_name: str # Name of the module. + # Useful mostly for checking for on-demand optional outputs (such as logs) with + # `is_output_explicit` method. + outputs: SimpleNamespace # Contains mappings for all available outputs. + values: SimpleNamespace # Contains all available requested values. + r_env: ResolutionEnv # `ResolutionEnvironmet` object holding mappings for current scope. + module_name: str # Name of the module. def is_output_explicit(self, name: str): """ @@ -93,26 +89,19 @@ class ModuleContext: """ return getattr(self.produces, name) is not None - def _getreqmaybe(self, obj, deps: 'list[str]', deps_cfg: 'dict[str, ]'): + def _getreqmaybe(self, obj, deps: "list[str]", deps_cfg: "dict[str, ]"): """ Add attribute for a dependency or panic if a required dependency has not been given to the module on its input. """ for name in deps: name, spec = decompose_depname(name) value = deps_cfg.get(name) - if value is None and spec == 'req': - fatal(-1, f'Dependency `{name}` is required by module `{self.module_name}` but wasn\'t provided') + if value is None and spec == "req": + fatal(-1, f"Dependency `{name}` is required by module `{self.module_name}` but wasn't provided") setattr(obj, name, self.r_env.resolve(value)) # `config` should be a dictionary given as modules input. - def __init__( - self, - module: Module, - config: 'dict[str, ]', - r_env: ResolutionEnv, - share: str, - bin: str - ): + def __init__(self, module: Module, config: "dict[str, ]", r_env: ResolutionEnv, share: str, bin: str): self.module_name = module.name self.takes = SimpleNamespace() self.produces = SimpleNamespace() @@ -122,10 +111,10 @@ class ModuleContext: self.share = share self.bin = bin - self._getreqmaybe(self.takes, module.takes, config['takes']) - self._getreqmaybe(self.values, module.values, config['values']) + self._getreqmaybe(self.takes, module.takes, config["takes"]) + self._getreqmaybe(self.values, module.values, config["values"]) - produces_resolved = self.r_env.resolve(config['produces']) + produces_resolved = self.r_env.resolve(config["produces"]) for name, value in produces_resolved.items(): setattr(self.produces, name, value) @@ -165,12 +154,12 @@ def get_mod_metadata(module: Module): Get descriptions for produced dependencies. """ meta = {} - has_meta = hasattr(module, 'prod_meta') + has_meta = hasattr(module, "prod_meta") for prod in module.produces: - prod = prod.replace('?', '').replace('!', '') + prod = prod.replace("?", "").replace("!", "") if not has_meta: - meta[prod] = '' + meta[prod] = "" continue prod_meta = module.prod_meta.get(prod) - meta[prod] = prod_meta if prod_meta else '' + meta[prod] = prod_meta if prod_meta else "" return meta diff --git a/f4pga/flows/runner.py b/f4pga/flows/runner.py index 5255637..92711cb 100644 --- a/f4pga/flows/runner.py +++ b/f4pga/flows/runner.py @@ -34,6 +34,7 @@ from f4pga.flows.common import ResolutionEnv, deep, sfprint @contextmanager def _add_to_sys_path(path: str): import sys + old_syspath = sys.path sys.path = [path] + sys.path try: @@ -72,15 +73,15 @@ def get_module(path: str): class ModRunCtx: share: str bin: str - config: 'dict[str, ]' + config: "dict[str, ]" - def __init__(self, share: str, bin: str, config: 'dict[str, ]'): + def __init__(self, share: str, bin: str, config: "dict[str, ]"): self.share = share self.bin = bin self.config = config def make_r_env(self): - return ResolutionEnv(self.config['values']) + return ResolutionEnv(self.config["values"]) class ModuleFailException(Exception): @@ -101,50 +102,34 @@ class ModuleFailException(Exception): def module_io(module: Module): - return { - 'name': module.name, - 'takes': module.takes, - 'produces': module.produces, - 'meta': get_mod_metadata(module) - } + return {"name": module.name, "takes": module.takes, "produces": module.produces, "meta": get_mod_metadata(module)} _deep_resolve = deep(lambda p: str(Path(p).resolve()), allow_none=True) + def module_map(module: Module, ctx: ModRunCtx): try: - mod_ctx = ModuleContext( - module, - ctx.config, - ctx.make_r_env(), - ctx.share, - ctx.bin - ) + mod_ctx = ModuleContext(module, ctx.config, ctx.make_r_env(), ctx.share, ctx.bin) except Exception as e: - raise ModuleFailException(module.name, 'map', e) + raise ModuleFailException(module.name, "map", e) return _deep_resolve(vars(mod_ctx.outputs)) def module_exec(module: Module, ctx: ModRunCtx): try: - mod_ctx = ModuleContext( - module, - ctx.config, - ctx.make_r_env(), - ctx.share, - ctx.bin - ) + mod_ctx = ModuleContext(module, ctx.config, ctx.make_r_env(), ctx.share, ctx.bin) except Exception as e: - raise ModuleFailException(module.name, 'exec', e) + raise ModuleFailException(module.name, "exec", e) - sfprint(1, f'Executing module `{Style.BRIGHT + module.name + Style.RESET_ALL}`:') + sfprint(1, f"Executing module `{Style.BRIGHT + module.name + Style.RESET_ALL}`:") current_phase = 1 try: for phase_msg in module.execute(mod_ctx): - sfprint(1, f' {Style.BRIGHT}[{current_phase}/{module.no_of_phases}] {Style.RESET_ALL}: {phase_msg}') + sfprint(1, f" {Style.BRIGHT}[{current_phase}/{module.no_of_phases}] {Style.RESET_ALL}: {phase_msg}") current_phase += 1 except Exception as e: - raise ModuleFailException(module.name, 'exec', e) + raise ModuleFailException(module.name, "exec", e) - sfprint(1, f'Module `{Style.BRIGHT + module.name + Style.RESET_ALL}` has finished its work!') + sfprint(1, f"Module `{Style.BRIGHT + module.name + Style.RESET_ALL}` has finished its work!") diff --git a/f4pga/flows/stage.py b/f4pga/flows/stage.py index bb1ebc8..a5cf402 100644 --- a/f4pga/flows/stage.py +++ b/f4pga/flows/stage.py @@ -21,14 +21,14 @@ from f4pga.flows.common import decompose_depname, resolve_modstr from f4pga.flows.module import Module from f4pga.flows.runner import get_module, module_io + class StageIO: """ Stage dependency input/output. - TODO: Solve the inconsistecy between usage of that and usage of - `decompose_depname` with an unprocessed string. + TODO: Solve the inconsistecy between usage of that and usage of `decompose_depname` with an unprocessed string. """ - name: str # A symbolic name given to the dependency + name: str # A symbolic name given to the dependency spec: str def __init__(self, encoded_name: str): @@ -41,59 +41,53 @@ class StageIO: self.name, self.spec = decompose_depname(encoded_name) def __repr__(self) -> str: - return 'StageIO { name: \'' + self.name + '\', spec: ' + \ - self.spec + '}' + return "StageIO { name: '" + self.name + "', spec: " + self.spec + "}" + class Stage: """ - Represents a single stage in a flow. I.e an instance of a module with a - local set of values. + Represents a single stage in a flow. + I.e an instance of a module with a local set of values. """ - name: str # Name of the stage (module's name) - takes: 'list[StageIO]' # List of symbolic names of dependencies used by - # the stage - produces: 'list[StageIO]' # List of symbolic names of dependencies - # produced by the stage - value_overrides: 'dict[str, ]' # Stage-specific values module: Module - meta: 'dict[str, str]' # Stage's metadata extracted from module's - # output. - def __init__(self, name: str, stage_def: 'dict[str, ]'): + # Name of the stage (module's name) + name: str + + # List of symbolic names of dependencies used by the stage + takes: "list[StageIO]" + + # List of symbolic names of dependencies produced by the stage + produces: "list[StageIO]" + + # Stage-specific values + value_overrides: "dict[str, ]" + + # Stage's metadata extracted from module's output. + meta: "dict[str, str]" + + def __init__(self, name: str, stage_def: "dict[str, ]"): + self.name = name + if stage_def is None: stage_def = {} - modstr = stage_def['module'] + self.module = get_module(resolve_modstr(stage_def["module"]))(stage_def.get("params")) - module_path = resolve_modstr(modstr) - ModuleClass = get_module(module_path) - self.module = ModuleClass(stage_def.get('params')) - - values = stage_def.get('values') - if values is not None: - self.value_overrides = values - else: - self.value_overrides = {} + values = stage_def.get("values") + self.value_overrides = values if values is not None else {} mod_io = module_io(self.module) - self.name = name - - self.takes = [] - for input in mod_io['takes']: - io = StageIO(input) - self.takes.append(io) - - self.produces = [] - for input in mod_io['produces']: - io = StageIO(input) - self.produces.append(io) - - self.meta = mod_io['meta'] + self.takes = [StageIO(input) for input in mod_io["takes"]] + self.produces = [StageIO(input) for input in mod_io["produces"]] + self.meta = mod_io["meta"] def __repr__(self) -> str: - return 'Stage \'' + self.name + '\' {' \ - f' value_overrides: {self.value_ovds},' \ - f' args: {self.args},' \ - f' takes: {self.takes},' \ - f' produces: {self.produces} ' + '}' + return ( + "Stage '" + self.name + "' {" + f" value_overrides: {self.value_ovds}," + f" args: {self.args}," + f" takes: {self.takes}," + f" produces: {self.produces} " + "}" + ) diff --git a/f4pga/pyproject.toml b/f4pga/pyproject.toml new file mode 100644 index 0000000..0788a5e --- /dev/null +++ b/f4pga/pyproject.toml @@ -0,0 +1,2 @@ +[tool.black] +line-length = 120 diff --git a/f4pga/setup.py b/f4pga/setup.py index af9256a..f6e2c84 100644 --- a/f4pga/setup.py +++ b/f4pga/setup.py @@ -52,16 +52,16 @@ def get_requirements(file: Path) -> List[str]: semver = "0.0.0" version = None -with (packagePath.parent / '.gitcommit').open("r") as rptr: +with (packagePath.parent / ".gitcommit").open("r") as rptr: sha = rptr.read().strip() - if sha != '$Format:%h$': - version = f'{semver}-{sha}' + if sha != "$Format:%h$": + version = f"{semver}+{sha}" -git = which('git') +git = which("git") if git is not None: - proc = run(['git', 'rev-parse', 'HEAD'], capture_output=True) + proc = run(["git", "rev-parse", "HEAD"], capture_output=True) if proc.returncode == 0: - version = f'{semver}-{proc.stdout.decode("utf8")[0:8]}' + version = f'{semver}+{proc.stdout.decode("utf8")[0:8]}' if version is None: version = semver @@ -78,39 +78,32 @@ setuptools_setup( author="F4PGA Authors", description="F4PGA.", url="https://github.com/chipsalliance/f4pga", - packages=[ - "f4pga", - "f4pga.flows", - "f4pga.flows.common_modules", - "f4pga.wrappers.sh", - "f4pga.wrappers.tcl" - ], + packages=["f4pga", "f4pga.flows", "f4pga.flows.common_modules", "f4pga.wrappers.sh", "f4pga.wrappers.tcl"], package_dir={"f4pga": "."}, package_data={ - 'f4pga.flows': [ - '*.yml', + "f4pga.flows": [ + "*.yml", ], - 'f4pga.wrappers.sh': [ - 'xc7/*.f4pga.sh', - 'quicklogic/*.f4pga.sh' + "f4pga.wrappers.sh": ["xc7/*.f4pga.sh", "quicklogic/*.f4pga.sh"], + "f4pga.wrappers.tcl": [ + "xc7/*.f4pga.tcl", + "eos-s3/*.f4pga.tcl", + "qlf_k4n8/*.f4pga.tcl", + "ice40/*.f4pga.tcl", ], - 'f4pga.wrappers.tcl': [ - 'xc7/*.f4pga.tcl', - 'eos-s3/*.f4pga.tcl', - 'qlf_k4n8/*.f4pga.tcl', - 'ice40/*.f4pga.tcl', - ] }, classifiers=[], - python_requires='>=3.6', + python_requires=">=3.6", install_requires=list(set(get_requirements(requirementsFile))), entry_points={ "console_scripts": [ "f4pga = f4pga.__init__:main", # QuickLogic only f"ql_{sf} = {shwrappers}:ql", - ] + [ - f"{sf}_{script} = {shwrappers}:{script}" for script in [ + ] + + [ + f"{sf}_{script} = {shwrappers}:{script}" + for script in [ "pack", "place", "route", diff --git a/f4pga/wrappers/sh/__init__.py b/f4pga/wrappers/sh/__init__.py index 7378eb6..c4eee89 100644 --- a/f4pga/wrappers/sh/__init__.py +++ b/f4pga/wrappers/sh/__init__.py @@ -27,15 +27,15 @@ from subprocess import check_call from f4pga.context import FPGA_FAM, F4PGA_SHARE_DIR -python3 = which('python3') +python3 = which("python3") ROOT = Path(__file__).resolve().parent -isQuickLogic = FPGA_FAM != 'xc7' -SH_SUBDIR = 'quicklogic' if isQuickLogic else FPGA_FAM +isQuickLogic = FPGA_FAM != "xc7" +SH_SUBDIR = "quicklogic" if isQuickLogic else FPGA_FAM f4pga_environ = environ.copy() -f4pga_environ['F4PGA_SHARE_DIR'] = f4pga_environ.get('F4PGA_SHARE_DIR', F4PGA_SHARE_DIR) +f4pga_environ["F4PGA_SHARE_DIR"] = f4pga_environ.get("F4PGA_SHARE_DIR", F4PGA_SHARE_DIR) # Helper functions @@ -44,87 +44,89 @@ f4pga_environ['F4PGA_SHARE_DIR'] = f4pga_environ.get('F4PGA_SHARE_DIR', F4PGA_SH def p_run_sh_script(script): stdout.flush() stderr.flush() - check_call([str(script)]+sys_argv[1:], env=f4pga_environ) + check_call([str(script)] + sys_argv[1:], env=f4pga_environ) def p_run_bash_cmds(cmds): stdout.flush() stderr.flush() - check_call(cmds, env=f4pga_environ, shell=True, executable='/bin/bash') + check_call(cmds, env=f4pga_environ, shell=True, executable="/bin/bash") def p_run_pym(module): stdout.flush() stderr.flush() - check_call([python3, '-m' , module]+sys_argv[1:], env=f4pga_environ) + check_call([python3, "-m", module] + sys_argv[1:], env=f4pga_environ) -def p_vpr_common_cmds(log_suffix = None): +def p_vpr_common_cmds(log_suffix=None): return f""" set -e source {ROOT / SH_SUBDIR}/vpr_common.f4pga.sh {' '.join([f"'{arg}'" for arg in sys_argv[1:]])} -""" + (f""" +""" + ( + f""" export OUT_NOISY_WARNINGS=noisy_warnings-${{DEVICE}}_{log_suffix}.log -""" if log_suffix is not None else '') +""" + if log_suffix is not None + else "" + ) def p_args_str2list(args): - return [arg for arg in args.strip().split() if arg != ''] + return [arg for arg in args.strip().split() if arg != ""] def p_vpr_run(): print("[F4PGA] Running (deprecated) vpr run") - arg_arch_def = f4pga_environ.get('ARCH_DEF') + arg_arch_def = f4pga_environ.get("ARCH_DEF") if arg_arch_def is None: - raise(Exception('[F4PGA] vpr run: envvar ARCH_DEF cannot be unset/empty!')) + raise (Exception("[F4PGA] vpr run: envvar ARCH_DEF cannot be unset/empty!")) - arg_eblif = f4pga_environ.get('EBLIF') + arg_eblif = f4pga_environ.get("EBLIF") if arg_eblif is None: - raise(Exception('[F4PGA] vpr run: envvar EBLIF cannot be unset/empty!')) + raise (Exception("[F4PGA] vpr run: envvar EBLIF cannot be unset/empty!")) - arg_vpr_options = f4pga_environ.get('VPR_OPTIONS') + arg_vpr_options = f4pga_environ.get("VPR_OPTIONS") if arg_vpr_options is None: - raise(Exception('[F4PGA] vpr run: envvar VPR_OPTIONS cannot be unset/empty!')) + raise (Exception("[F4PGA] vpr run: envvar VPR_OPTIONS cannot be unset/empty!")) - arg_device_name = f4pga_environ.get('DEVICE_NAME') + arg_device_name = f4pga_environ.get("DEVICE_NAME") if arg_device_name is None: - raise(Exception('[F4PGA] vpr run: envvar DEVICE_NAME cannot be unset/empty!')) + raise (Exception("[F4PGA] vpr run: envvar DEVICE_NAME cannot be unset/empty!")) - arg_rr_graph = f4pga_environ.get('RR_GRAPH') + arg_rr_graph = f4pga_environ.get("RR_GRAPH") if arg_rr_graph is None: - raise(Exception('[F4PGA] vpr run: envvar RR_GRAPH cannot be unset/empty!')) + raise (Exception("[F4PGA] vpr run: envvar RR_GRAPH cannot be unset/empty!")) - arg_lookahead = f4pga_environ.get('LOOKAHEAD') + arg_lookahead = f4pga_environ.get("LOOKAHEAD") if arg_lookahead is None: - raise(Exception('[F4PGA] vpr run: envvar LOOKAHEAD cannot be unset/empty!')) + raise (Exception("[F4PGA] vpr run: envvar LOOKAHEAD cannot be unset/empty!")) - arg_place_delay = f4pga_environ.get('PLACE_DELAY') + arg_place_delay = f4pga_environ.get("PLACE_DELAY") if arg_place_delay is None: - raise(Exception('[F4PGA] vpr run: envvar PLACE_DELAY cannot be unset/empty!')) + raise (Exception("[F4PGA] vpr run: envvar PLACE_DELAY cannot be unset/empty!")) - sdc = f4pga_environ.get('SDC') - if sdc == '': + sdc = f4pga_environ.get("SDC") + if sdc == "": sdc = None check_call( - [ - which('vpr'), - arg_arch_def, - arg_eblif - ] + p_args_str2list(arg_vpr_options) + [ - '--device', + [which("vpr"), arg_arch_def, arg_eblif] + + p_args_str2list(arg_vpr_options) + + [ + "--device", arg_device_name, - '--read_rr_graph', + "--read_rr_graph", arg_rr_graph, - '--read_router_lookahead', + "--read_router_lookahead", arg_lookahead, - '--read_placement_delay_lookup', - arg_place_delay - ] + ( - ['--sdc_file', sdc] if sdc is not None else [] - ) + sys_argv[1:], - env=f4pga_environ + "--read_placement_delay_lookup", + arg_place_delay, + ] + + (["--sdc_file", sdc] if sdc is not None else []) + + sys_argv[1:], + env=f4pga_environ, ) @@ -136,10 +138,11 @@ def generate_constraints(): if isQuickLogic: (pcf, eblif, net, part, device, arch_def, corner) = sys_argv[1:8] place_file_prefix = Path(eblif).stem - share_dir = Path(f4pga_environ['F4PGA_SHARE_DIR']) - scripts_dir = share_dir / 'scripts' - archs_dir = share_dir / 'arch' - p_run_bash_cmds(f""" + share_dir = Path(f4pga_environ["F4PGA_SHARE_DIR"]) + scripts_dir = share_dir / "scripts" + archs_dir = share_dir / "arch" + p_run_bash_cmds( + f""" set -e if [[ '{device}' =~ ^(qlf_.*)$ ]]; then @@ -194,27 +197,36 @@ elif [[ '{device}' =~ ^(ql-.*)$ ]]; then # EOS-S3 IOMUX configuration if [[ '{device}' =~ ^(ql-eos-s3)$ ]]; then -""" + '\n'.join([f""" +""" + + "\n".join( + [ + f""" '{python3}' '{scripts_dir}/pp3_eos_s3_iomux_config.py' \ --eblif '{eblif}' \ --pcf '{pcf}' \ --map "$PINMAP" \ --output-format={fmt[0]} \ > '{place_file_prefix}_iomux.{fmt[1]}' -""" for fmt in [['jlink', 'jlink'], ['openocd', 'openocd'], ['binary', 'bin']]]) + f""" +""" + for fmt in [["jlink", "jlink"], ["openocd", "openocd"], ["binary", "bin"]] + ] + ) + + f""" fi else echo "FIXME: Unsupported device '{device}'" exit -1 fi -""") +""" + ) else: (eblif, net, part, device, arch_def) = sys_argv[1:6] - pcf_opts = f"'--pcf' '{sys_argv[6]}'" if len(sys_argv) > 6 else '' - ioplace_file = f'{Path(eblif).stem}.ioplace' - share_dir = f4pga_environ['F4PGA_SHARE_DIR'] - p_run_bash_cmds(f""" + pcf_opts = f"'--pcf' '{sys_argv[6]}'" if len(sys_argv) > 6 else "" + ioplace_file = f"{Path(eblif).stem}.ioplace" + share_dir = f4pga_environ["F4PGA_SHARE_DIR"] + p_run_bash_cmds( + f""" set -e python3 '{share_dir}/scripts/prjxray_create_ioplace.py' \ --blif '{eblif}' \ @@ -230,14 +242,15 @@ python3 '{share_dir}'/scripts/prjxray_create_place_constraints.py \ --input '{ioplace_file}' \ --db_root "${{DATABASE_DIR:-$(prjxray-config)}}" \ > constraints.place -""") +""" + ) def pack(): print("[F4PGA] Running (deprecated) pack") - extra_args = ['--write_block_usage', 'block_usage.json'] if isQuickLogic else [] - p_run_bash_cmds(p_vpr_common_cmds('pack')+f"python3 -m f4pga.wrappers.sh.vpr_run --pack {' '.join(extra_args)}") - Path('vpr_stdout.log').rename('pack.log') + extra_args = ["--write_block_usage", "block_usage.json"] if isQuickLogic else [] + p_run_bash_cmds(p_vpr_common_cmds("pack") + f"python3 -m f4pga.wrappers.sh.vpr_run --pack {' '.join(extra_args)}") + Path("vpr_stdout.log").rename("pack.log") def place(): @@ -267,15 +280,15 @@ python3 -m f4pga.wrappers.sh.generate_constraints $EBLIF $NET $PART $DEVICE $ARC VPR_PLACE_FILE='constraints.place' """ place_cmds += 'python3 -m f4pga.wrappers.sh.vpr_run --fix_clusters "${VPR_PLACE_FILE}" --place' - p_run_bash_cmds(p_vpr_common_cmds('place')+place_cmds) - Path('vpr_stdout.log').rename('place.log') + p_run_bash_cmds(p_vpr_common_cmds("place") + place_cmds) + Path("vpr_stdout.log").rename("place.log") def route(): print("[F4PGA] Running (deprecated) route") - extra_args = ['--write_timing_summary', 'timing_summary.json'] if isQuickLogic else [] - p_run_bash_cmds(p_vpr_common_cmds('pack')+f"python3 -m f4pga.wrappers.sh.vpr_run --route {' '.join(extra_args)}") - Path('vpr_stdout.log').rename('route.log') + extra_args = ["--write_timing_summary", "timing_summary.json"] if isQuickLogic else [] + p_run_bash_cmds(p_vpr_common_cmds("pack") + f"python3 -m f4pga.wrappers.sh.vpr_run --route {' '.join(extra_args)}") + Path("vpr_stdout.log").rename("route.log") def synth(): @@ -283,22 +296,26 @@ def synth(): p_run_sh_script(ROOT / SH_SUBDIR / "synth.f4pga.sh") -def write_fasm(genfasm_extra_args = None): +def write_fasm(genfasm_extra_args=None): print("[F4PGA] Running (deprecated) write fasm") - p_run_bash_cmds(p_vpr_common_cmds('fasm')+f""" + p_run_bash_cmds( + p_vpr_common_cmds("fasm") + + f""" '{which('genfasm')}' \ ${{ARCH_DEF}} ${{EBLIF}} --device ${{DEVICE_NAME}} \ ${{VPR_OPTIONS}} \ --read_rr_graph ${{RR_GRAPH}} {' '.join(genfasm_extra_args) if genfasm_extra_args is not None else ''} -""" + """ +""" + + """ TOP="${EBLIF%.*}" FASM_EXTRA="${TOP}_fasm_extra.fasm" if [ -f $FASM_EXTRA ]; then echo "writing final fasm (extra: $FASM_EXTRA)" cat $FASM_EXTRA >> ${TOP}.fasm fi -""") - Path('vpr_stdout.log').rename('fasm.log') +""" + ) + Path("vpr_stdout.log").rename("fasm.log") # Xilinx only @@ -306,19 +323,22 @@ fi def write_bitstream(): print("[F4PGA] Running (deprecated) write bitstream") - p_run_bash_cmds(""" + p_run_bash_cmds( + """ set -e echo "Writing bitstream ..." FRM2BIT="" if [ ! -z ${FRAMES2BIT} ]; then FRM2BIT="--frm2bit ${FRAMES2BIT}"; fi -""" + f""" +""" + + f""" eval set -- $( getopt \ --options=d:f:b:p: \ --longoptions=device:,fasm:,bit:,part: \ --name $0 -- {' '.join(sys_argv[1:])} ) -""" + """ +""" + + """ DEVICE="" FASM="" BIT="" @@ -354,7 +374,8 @@ xcfasm \ --emit_pudc_b_pullup \ --fn_in ${FASM} \ --bit_out ${BIT} ${FRM2BIT} -""") +""" + ) # QuickLogic only @@ -362,7 +383,9 @@ xcfasm \ def analysis(): print("[F4PGA] Running (deprecated) analysis") - p_run_bash_cmds(p_vpr_common_cmds('analysis')+""" + p_run_bash_cmds( + p_vpr_common_cmds("analysis") + + """ python3 -m f4pga.wrappers.sh.vpr_run \ --analysis \ --gen_post_synthesis_netlist on \ @@ -370,17 +393,21 @@ python3 -m f4pga.wrappers.sh.vpr_run \ --post_synth_netlist_unconn_inputs nets \ --post_synth_netlist_unconn_outputs nets \ --verify_file_digests off -""") - Path('vpr_stdout.log').rename('analysis.log') +""" + ) + Path("vpr_stdout.log").rename("analysis.log") def repack(): print("[F4PGA] Running (deprecated) repack") - p_run_bash_cmds(p_vpr_common_cmds()+""" + p_run_bash_cmds( + p_vpr_common_cmds() + + """ DESIGN=${EBLIF/.eblif/} [ ! -z "${JSON}" ] && JSON_ARGS="--json-constraints ${JSON}" || JSON_ARGS= [ ! -z "${PCF_PATH}" ] && PCF_ARGS="--pcf-constraints ${PCF_PATH}" || PCF_ARGS= -""" + f""" +""" + + f""" PYTHONPATH=$F4PGA_SHARE_DIR/scripts:$PYTHONPATH \ '{python3}' "$F4PGA_SHARE_DIR"/scripts/repacker/repack.py \ --vpr-arch ${{ARCH_DEF}} \ @@ -395,12 +422,14 @@ PYTHONPATH=$F4PGA_SHARE_DIR/scripts:$PYTHONPATH \ --place-out ${{DESIGN}}.repacked.place \ --absorb_buffer_luts on \ > repack.log 2>&1 -""") +""" + ) def generate_bitstream(): print("[F4PGA] Running (deprecated) generate_bitstream") - p_run_bash_cmds(f""" + p_run_bash_cmds( + f""" set -e eval set -- "$( getopt \ @@ -408,7 +437,8 @@ eval set -- "$( --longoptions=device:,fasm:,format:,bit:,part: \ --name $0 -- {' '.join(sys_argv[1:])} )" -""" + """ +""" + + """ DEVICE="" FASM="" BIT_FORMAT="4byte" @@ -428,7 +458,8 @@ if [ -z $DEVICE ]; then echo "Please provide device name"; exit 1; fi if [ -z $FASM ]; then echo "Please provide an input FASM file name"; exit 1; fi if [ ! -f "$FASM" ]; then echo "File <$FASM> does not exist!"; exit 1; fi if [ -z $BIT ]; then echo "Please provide an output bistream file name"; exit 1; fi -""" + f""" +""" + + f""" if [[ "$DEVICE" =~ ^(qlf_k4n8.*)$ ]]; then '{which('qlf_fasm')}' \ --db-root "${{F4PGA_SHARE_DIR}}/fasm_database/${{DEVICE}}" \ @@ -446,13 +477,15 @@ else echo "ERROR: Unsupported device '${{DEVICE}}' for bitstream generation" exit -1 fi -""") +""" + ) def generate_libfile(): print("[F4PGA] Running (deprecated) generate_libfile") (part, device, corner) = sys_argv[1:4] - p_run_bash_cmds(f""" + p_run_bash_cmds( + f""" set -e if [[ '{device}' =~ ^(qlf_k4n8_qlf_k4n8)$ ]];then DEVICE_1="qlf_k4n8-qlf_k4n8_umc22_{corner}" @@ -462,10 +495,12 @@ if [[ '{device}' =~ ^(qlf_k4n8_qlf_k4n8)$ ]];then else DEVICE_1={device} fi -""" + """ +""" + + """ ARCH_DIR="${F4PGA_SHARE_DIR}/arch/${DEVICE_1}_${DEVICE_1}" PINMAP_XML=${ARCH_DIR}/${PINMAPXML} -""" + f""" +""" + + f""" '{python3}' "$F4PGA_SHARE_DIR"/scripts/create_lib.py \ -n "${{DEV}}_0P72_SSM40" \ -m fpga_top \ @@ -473,7 +508,8 @@ PINMAP_XML=${ARCH_DIR}/${PINMAPXML} -x "${{ARCH_DIR}}/lib/${{INTERFACEXML}}" \ -l "${{DEV}}_0P72_SSM40.lib" \ -t "${{ARCH_DIR}}/lib" -""") +""" + ) def ql(): @@ -483,7 +519,8 @@ def ql(): def fasm2bels(): print("[F4PGA] Running (deprecated) fasm2bels") - p_run_bash_cmds(f""" + p_run_bash_cmds( + f""" set -e eval set -- "$( getopt \ @@ -491,7 +528,8 @@ eval set -- "$( --longoptions=device:,part:,pcf:,bit:,out-verilog:,out-pcf:,out-qcf:, \ --name $0 -- {' '.join(sys_argv[1:])} )" -""" + """ +""" + + """ DEVICE="" PART="" PCF="" @@ -517,7 +555,8 @@ if [ -z $BIT ]; then echo "Please provide an input bistream file name"; exit 1; if ! [[ "$DEVICE" =~ ^(ql-eos-s3|ql-pp3e)$ ]]; then echo "ERROR: Unsupported device '${DEVICE}' for fasm2bels"; exit -1; fi if [ -z "{PCF}" ]; then PCF_ARGS=""; else PCF_ARGS="--input-pcf ${PCF}"; fi echo "Running fasm2bels" -""" + f""" +""" + + f""" '{python3}' "${{F4PGA_SHARE_DIR}}"/scripts/fasm2bels.py "${{BIT}}" \ --phy-db "${{F4PGA_SHARE_DIR}}/arch/${{DEVICE}}_wlcsp/db_phy.pickle" \ --device-name "${{DEVICE/ql-/}}" \ @@ -527,25 +566,29 @@ echo "Running fasm2bels" ${{PCF_ARGS}} \ --output-pcf "${{OUT_PCF:-$BIT.v.pcf}}" \ --output-qcf "${{OUT_QCF:-$BIT.v.qcf}}" -""") +""" + ) def write_bitheader(): print("[F4PGA] Running (deprecated) write bitheader") print("Converting bitstream to C Header") - p_run_pym('quicklogic_fasm.bitstream_to_header') + p_run_pym("quicklogic_fasm.bitstream_to_header") + def write_binary(): print("[F4PGA] Running (deprecated) write binary") print("Converting bitstream to flashable binary format") - p_run_pym('quicklogic_fasm.bitstream_to_binary') + p_run_pym("quicklogic_fasm.bitstream_to_binary") + def write_jlink(): print("[F4PGA] Running (deprecated) write jlink") print("Converting bitstream to JLink script") - p_run_pym('quicklogic_fasm.bitstream_to_jlink') + p_run_pym("quicklogic_fasm.bitstream_to_jlink") + def write_openocd(): print("[F4PGA] Running (deprecated) write openocd") print("Converting bitstream to OpenOCD script") - p_run_pym('quicklogic_fasm.bitstream_to_openocd') + p_run_pym("quicklogic_fasm.bitstream_to_openocd") diff --git a/f4pga/wrappers/sh/generate_constraints.py b/f4pga/wrappers/sh/generate_constraints.py index 58e6a90..7ab63dd 100644 --- a/f4pga/wrappers/sh/generate_constraints.py +++ b/f4pga/wrappers/sh/generate_constraints.py @@ -19,5 +19,5 @@ from f4pga.wrappers.sh import generate_constraints -if __name__ == '__main__': +if __name__ == "__main__": generate_constraints() diff --git a/f4pga/wrappers/sh/vpr_run.py b/f4pga/wrappers/sh/vpr_run.py index 0d325a8..1fef4fd 100644 --- a/f4pga/wrappers/sh/vpr_run.py +++ b/f4pga/wrappers/sh/vpr_run.py @@ -19,5 +19,5 @@ from f4pga.wrappers.sh import p_vpr_run -if __name__ == '__main__': +if __name__ == "__main__": p_vpr_run() diff --git a/f4pga/wrappers/tcl/__init__.py b/f4pga/wrappers/tcl/__init__.py index 138e3d6..c333e5c 100644 --- a/f4pga/wrappers/tcl/__init__.py +++ b/f4pga/wrappers/tcl/__init__.py @@ -25,36 +25,18 @@ from f4pga.context import FPGA_FAM ROOT = Path(__file__).resolve().parent -ARCHS = { - 'xc7': [ - 'artix7', - 'artix7_100t', - 'artix7_200t', - 'zynq7', - 'zynq7_z020', - 'spartan7' - ], - 'eos-s3': [ - 'ql-s3', - 'pp3' - ] -} +ARCHS = {"xc7": ["artix7", "artix7_100t", "artix7_200t", "zynq7", "zynq7_z020", "spartan7"], "eos-s3": ["ql-s3", "pp3"]} -def get_script_path(arg, arch = None): +def get_script_path(arg, arch=None): if arch is None: arch = FPGA_FAM for key, val in ARCHS.items(): if arch in val: arch = key break - if arch not in [ - 'xc7', - 'eos-s3', - 'qlf_k4n8', - 'ice40' - ]: - raise(Exception(f"Unsupported arch <{arch}>!")) - if arg not in ['synth', 'conv']: - raise Exception(f'Unknown tcl wrapper <{arg}>!') - return ROOT / arch / f'{arg}.f4pga.tcl' + if arch not in ["xc7", "eos-s3", "qlf_k4n8", "ice40"]: + raise (Exception(f"Unsupported arch <{arch}>!")) + if arg not in ["synth", "conv"]: + raise Exception(f"Unknown tcl wrapper <{arg}>!") + return ROOT / arch / f"{arg}.f4pga.tcl" diff --git a/f4pga/wrappers/tcl/__main__.py b/f4pga/wrappers/tcl/__main__.py index 727b8f2..57c71d9 100644 --- a/f4pga/wrappers/tcl/__main__.py +++ b/f4pga/wrappers/tcl/__main__.py @@ -23,5 +23,5 @@ from sys import argv as sys_argv from f4pga.wrappers.tcl import get_script_path -if __name__ == '__main__': - print(get_script_path(sys_argv[1], sys_argv[2]) if len(sys_argv)>2 else get_script_path(sys_argv[1])) +if __name__ == "__main__": + print(get_script_path(sys_argv[1], sys_argv[2]) if len(sys_argv) > 2 else get_script_path(sys_argv[1])) diff --git a/test/requirements.txt b/test/requirements.txt index e079f8a..6e9e9dd 100644 --- a/test/requirements.txt +++ b/test/requirements.txt @@ -1 +1,2 @@ +black pytest