f4pga/flows: split commands

Signed-off-by: Unai Martinez-Corral <umartinezcorral@antmicro.com>
This commit is contained in:
Unai Martinez-Corral 2022-08-18 08:15:24 +02:00
parent 7ca33a0ca4
commit 3315f705a3
2 changed files with 324 additions and 297 deletions

View File

@ -37,41 +37,12 @@ This will make F4PGA attempt to create a bitstream for arty_35 platform.
Contains project-specific definitions needed within the flow, such as list of source code files.
"""
from pathlib import Path
from argparse import Namespace
from os import environ
from yaml import load as yaml_load, Loader as yaml_loader
from typing import Iterable
from colorama import Fore, Style
from f4pga.context import FPGA_FAM
from f4pga.flows.common import (
F4PGAException,
ResolutionEnv,
fatal,
scan_modules,
set_verbosity_level,
sfprint,
sub as common_sub
)
from f4pga.flows.cache import F4Cache
from f4pga.flows.flow_config import (
ProjectFlowConfig,
FlowConfig,
FlowDefinition,
open_project_flow_cfg,
verify_platform_name
)
from f4pga.flows.flow import Flow
from f4pga.flows.inspector import get_module_info
from f4pga.flows.stage import Stage
from f4pga.flows.argparser import setup_argparser, get_cli_flow_config
from f4pga.flows.common import bin_dir_path, share_dir_path
F4CACHEPATH = '.f4cache'
ROOT = Path(__file__).resolve().parent
from f4pga.flows.common import set_verbosity_level, sfprint
from f4pga.flows.argparser import setup_argparser
from f4pga.flows.commands import cmd_build, cmd_show_dependencies, f4pga_done
def platform_stages(platform_flow, r_env):
@ -93,271 +64,6 @@ def prepare_stage_io_input(stage: Stage):
return { 'params': stage.params } if stage.params is not None else {}
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:
l = len(out.name)
if l > longest_out_name_len:
longest_out_name_len = l
desc_indent = longest_out_name_len + 7
nl_indentstr = '\n'
for _ in range(0, desc_indent):
nl_indentstr += ' '
for stage in stages:
for out in stage.produces:
pname = Style.BRIGHT + out.name + Style.RESET_ALL
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}')
def display_stage_info(stage: Stage):
if stage is None:
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:')
mod_info = get_module_info(stage.module)
mod_info = '\n '.join(mod_info.split('\n'))
sfprint(0, f' {mod_info}')
f4pga_done_str = Style.BRIGHT + Fore.GREEN + 'DONE'
def f4pga_fail():
global f4pga_done_str
f4pga_done_str = Style.BRIGHT + Fore.RED + 'FAILED'
def f4pga_done():
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. """
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'
def _generate_values():
"""
Generate initial values, available in configs.
"""
conf = {
'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', '')
return conf
r_env.add_values(_generate_values())
return r_env
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')
return flow_cfg
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.')
return False
if part not in flow_cfg.part():
sfprint(0, f'Platform `{part}`` is not in project.')
return False
return True
def get_platform_name_for_part(part_name: str):
"""
Gets a name that identifies the platform setup required for a specific chip.
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:
for key, val in yaml_load(rfptr, yaml_loader).items():
if part_name.upper() in val:
return key
raise Exception(f"Unknown part name <{part_name}>!")
def make_flow_config(project_flow_cfg: ProjectFlowConfig, part_name: str) -> FlowConfig:
""" 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.'
)
if part_name not in project_flow_cfg.parts():
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()})
scan_modules(str(ROOT))
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!')
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')
return flow_cfg
def cmd_build(args: Namespace):
""" `build` command implementation """
project_flow_cfg: ProjectFlowConfig = None
part_name = args.part
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.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.')
flow_cfg = make_flow_config(project_flow_cfg, part_name)
if args.info:
display_dep_info(flow_cfg.stages.values())
f4pga_done()
if args.stageinfo:
display_stage_info(flow_cfg.stages.get(args.stageinfo[0]))
f4pga_done()
target = args.target
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.')
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:')
flow.print_resolved_dependencies(dep_print_verbosity)
sfprint(dep_print_verbosity, '')
if args.pretend:
f4pga_done()
try:
flow.execute()
except AssertionError as e:
raise e
except Exception as e:
sfprint(0, f'{e}')
f4pga_fail()
if flow.f4cache:
flow.f4cache.save()
def cmd_show_dependencies(args: Namespace):
""" `showd` command implementation """
flow_cfg = open_project_flow_config(args.flow)
if not verify_part_stage_params(flow_cfg, args.part):
f4pga_fail()
return
platform_overrides: 'set | None' = None
if args.platform is not None:
platform_overrides = \
set(flow_cfg.get_dependency_platform_overrides(args.part).keys())
display_list = []
raw_deps = flow_cfg.get_dependencies_raw(args.platform)
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}'
else:
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])
for _, prstr in display_list:
sfprint(0, prstr)
set_verbosity_level(-1)
def main():
parser = setup_argparser()
args = parser.parse_args()

321
f4pga/flows/commands.py Normal file
View File

@ -0,0 +1,321 @@
#!/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 typing import Iterable
from pathlib import Path
from os import environ
from argparse import Namespace
from colorama import Fore, Style
from yaml import load as yaml_load, Loader as yaml_loader
from f4pga.context import FPGA_FAM
from f4pga.flows.common import (
bin_dir_path,
share_dir_path,
F4PGAException,
ResolutionEnv,
fatal,
scan_modules,
set_verbosity_level,
sfprint,
sub as common_sub
)
from f4pga.flows.argparser import get_cli_flow_config
from f4pga.flows.cache import F4Cache
from f4pga.flows.flow_config import (
ProjectFlowConfig,
FlowConfig,
FlowDefinition,
open_project_flow_cfg,
verify_platform_name
)
from f4pga.flows.flow import Flow
from f4pga.flows.stage import Stage
from f4pga.flows.inspector import get_module_info
ROOT = Path(__file__).resolve().parent
F4CACHEPATH = '.f4cache'
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:
l = len(out.name)
if l > longest_out_name_len:
longest_out_name_len = l
desc_indent = longest_out_name_len + 7
nl_indentstr = '\n'
for _ in range(0, desc_indent):
nl_indentstr += ' '
for stage in stages:
for out in stage.produces:
pname = Style.BRIGHT + out.name + Style.RESET_ALL
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}')
def display_stage_info(stage: Stage):
if stage is None:
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:')
mod_info = get_module_info(stage.module)
mod_info = '\n '.join(mod_info.split('\n'))
sfprint(0, f' {mod_info}')
f4pga_done_str = Style.BRIGHT + Fore.GREEN + 'DONE'
def f4pga_fail():
global f4pga_done_str
f4pga_done_str = Style.BRIGHT + Fore.RED + 'FAILED'
def f4pga_done():
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. """
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'
def _generate_values():
"""
Generate initial values, available in configs.
"""
conf = {
'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', '')
return conf
r_env.add_values(_generate_values())
return r_env
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')
return flow_cfg
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.')
return False
if part not in flow_cfg.part():
sfprint(0, f'Platform `{part}`` is not in project.')
return False
return True
def get_platform_name_for_part(part_name: str):
"""
Gets a name that identifies the platform setup required for a specific chip.
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:
for key, val in yaml_load(rfptr, yaml_loader).items():
if part_name.upper() in val:
return key
raise Exception(f"Unknown part name <{part_name}>!")
def make_flow_config(project_flow_cfg: ProjectFlowConfig, part_name: str) -> FlowConfig:
""" 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.'
)
if part_name not in project_flow_cfg.parts():
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()})
scan_modules(str(ROOT))
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!')
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')
return flow_cfg
def cmd_build(args: Namespace):
""" `build` command implementation """
project_flow_cfg: ProjectFlowConfig = None
part_name = args.part
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.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.')
flow_cfg = make_flow_config(project_flow_cfg, part_name)
if args.info:
display_dep_info(flow_cfg.stages.values())
f4pga_done()
if args.stageinfo:
display_stage_info(flow_cfg.stages.get(args.stageinfo[0]))
f4pga_done()
target = args.target
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.')
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:')
flow.print_resolved_dependencies(dep_print_verbosity)
sfprint(dep_print_verbosity, '')
if args.pretend:
f4pga_done()
try:
flow.execute()
except AssertionError as e:
raise e
except Exception as e:
sfprint(0, f'{e}')
f4pga_fail()
if flow.f4cache:
flow.f4cache.save()
def cmd_show_dependencies(args: Namespace):
""" `showd` command implementation """
flow_cfg = open_project_flow_config(args.flow)
if not verify_part_stage_params(flow_cfg, args.part):
f4pga_fail()
return
platform_overrides: 'set | None' = None
if args.platform is not None:
platform_overrides = \
set(flow_cfg.get_dependency_platform_overrides(args.part).keys())
display_list = []
raw_deps = flow_cfg.get_dependencies_raw(args.platform)
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}'
else:
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])
for _, prstr in display_list:
sfprint(0, prstr)
set_verbosity_level(-1)