dasharo/f4pga
0
0
Fork 0
mirror of https://github.com/Dasharo/f4pga.git synced 2026-06-13 19:16:35 -07:00
Code Issues Packages Projects Releases Wiki Activity

use 'black' for Python formatting (#616)

Browse Source
This commit is contained in:
Unai Martinez-Corral
2022-08-19 19:15:11 +02:00
committed by GitHub
parent 30eb91feed 5fdea4bd11
commit 2d8d85706e
32 changed files with 932 additions and 1095 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/Pipeline.yml
+25 -1
View File
@@ -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'
f4pga/__init__.py
+24 -24
View File
@@ -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()
f4pga/context.py
+7 -7
View File
@@ -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"))
f4pga/flows/__init__.py
+8 -8
View File
@@ -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()
f4pga/flows/argparser.py
+65 -121
View File
@@ -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}
f4pga/flows/cache.py
+29 -20
View File
@@ -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)
f4pga/flows/commands.py
+66 -86
View File
@@ -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)
f4pga/flows/common.py
+76 -66
View File
@@ -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:
f4pga/flows/common_modules/analysis.py
+15 -34
View File
@@ -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
f4pga/flows/common_modules/fasm.py
+19 -36
View File
@@ -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
f4pga/flows/common_modules/generic_script_wrapper.py
+66 -56
View File
@@ -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"{'<unknown>' 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
f4pga/flows/common_modules/io_rename.py
+10 -9
View File
@@ -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
f4pga/flows/common_modules/mkdirs.py
+3 -3
View File
@@ -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 []
f4pga/flows/common_modules/pack.py
+14 -30
View File
@@ -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
f4pga/flows/common_modules/place.py
+19 -28
View File
@@ -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
f4pga/flows/common_modules/place_constraints.py
+28 -30
View File
@@ -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
f4pga/flows/common_modules/route.py
+12 -29
View File
@@ -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
f4pga/flows/common_modules/synth.py
+73 -103
View File
@@ -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
f4pga/flows/flow.py
+68 -75
View File
@@ -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:
f4pga/flows/flow_config.py
+32 -47
View File
@@ -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

Some files were not shown because too many files have changed in this diff Show More