148 lines
5.0 KiB
Python
148 lines
5.0 KiB
Python
# Here are the things necessary to write a symbiflow Module
|
|
|
|
import abc
|
|
from types import SimpleNamespace
|
|
from f4pga.common import *
|
|
from colorama import Fore, Style
|
|
|
|
class Module:
|
|
"""
|
|
A `Module` is a wrapper for whatever tool is used in a flow.
|
|
Modules can request dependencies, values and are guranteed to have all the
|
|
required ones present when entering `exec` mode.
|
|
They also have to specify what dependencies they produce and create the files
|
|
for these dependencies.
|
|
"""
|
|
|
|
no_of_phases: int
|
|
name: str
|
|
takes: 'list[str]'
|
|
produces: 'list[str]'
|
|
values: 'list[str]'
|
|
prod_meta: 'dict[str, str]'
|
|
|
|
@abc.abstractmethod
|
|
def execute(self, ctx):
|
|
"""
|
|
Executes module. Use yield to print a message informing about current
|
|
execution phase.
|
|
`ctx` is `ModuleContext`.
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def map_io(self, ctx) -> 'dict[str, ]':
|
|
"""
|
|
Returns paths for outputs derived from given inputs.
|
|
`ctx` is `ModuleContext`.
|
|
"""
|
|
pass
|
|
|
|
def __init__(self, params: 'dict[str, ]'):
|
|
self.no_of_phases = 0
|
|
self.current_phase = 0
|
|
self.name = '<BASE STAGE>'
|
|
self.prod_meta = {}
|
|
|
|
class ModuleContext:
|
|
"""
|
|
A class for object holding mappings for dependencies and values as well as
|
|
other information needed during modules execution.
|
|
"""
|
|
|
|
share: str # Absolute path to Symbiflow's share directory
|
|
bin: str # Absolute path to Symbiflow's bin directory
|
|
takes: SimpleNamespace # Maps symbolic dependency names to relative
|
|
# paths.
|
|
produces: SimpleNamespace # Contains mappings for explicitely specified
|
|
# dependencies. Useful mostly for checking for
|
|
# on-demand optional outputs (such as logs)
|
|
# with `is_output_explicit` method.
|
|
outputs: SimpleNamespace # Contains mappings for all available outputs.
|
|
values: SimpleNamespace # Contains all available requested values.
|
|
r_env: ResolutionEnv # `ResolutionEnvironmet` object holding mappings
|
|
# for current scope.
|
|
module_name: str # Name of the module.
|
|
|
|
def is_output_explicit(self, name: str):
|
|
""" True if user has explicitely specified output's path. """
|
|
o = getattr(self.produces, name)
|
|
return o is not None
|
|
|
|
def _getreqmaybe(self, obj, deps: 'list[str]', deps_cfg: 'dict[str, ]'):
|
|
"""
|
|
Add attribute for a dependency or panic if a required dependency has not
|
|
been given to the module on its input.
|
|
"""
|
|
|
|
for name in deps:
|
|
name, spec = decompose_depname(name)
|
|
value = deps_cfg.get(name)
|
|
if value is None and spec == 'req':
|
|
fatal(-1, f'Dependency `{name}` is required by module '
|
|
f'`{self.module_name}` but wasn\'t provided')
|
|
setattr(obj, name, self.r_env.resolve(value))
|
|
|
|
# `config` should be a dictionary given as modules input.
|
|
def __init__(self, module: Module, config: 'dict[str, ]',
|
|
r_env: ResolutionEnv, share: str, bin: str):
|
|
self.module_name = module.name
|
|
self.takes = SimpleNamespace()
|
|
self.produces = SimpleNamespace()
|
|
self.values = SimpleNamespace()
|
|
self.outputs = SimpleNamespace()
|
|
self.r_env = r_env
|
|
self.share = share
|
|
self.bin = bin
|
|
|
|
self._getreqmaybe(self.takes, module.takes, config['takes'])
|
|
self._getreqmaybe(self.values, module.values, config['values'])
|
|
|
|
produces_resolved = self.r_env.resolve(config['produces'])
|
|
for name, value in produces_resolved.items():
|
|
setattr(self.produces, name, value)
|
|
|
|
outputs = module.map_io(self)
|
|
outputs.update(produces_resolved)
|
|
|
|
self._getreqmaybe(self.outputs, module.produces, outputs)
|
|
|
|
def shallow_copy(self):
|
|
cls = type(self)
|
|
mycopy = cls.__new__(cls)
|
|
|
|
mycopy.module_name = self.module_name
|
|
mycopy.takes = self.takes
|
|
mycopy.produces = self.produces
|
|
mycopy.values = self.values
|
|
mycopy.outputs = self.outputs
|
|
mycopy.r_env = self.r_env
|
|
mycopy.share = self.share
|
|
mycopy.bin = self.bin
|
|
|
|
return mycopy
|
|
|
|
class ModuleRuntimeException(Exception):
|
|
info: str
|
|
|
|
def __init__(self, info: str):
|
|
self.info = info
|
|
|
|
def __str___(self):
|
|
return self.info
|
|
|
|
def get_mod_metadata(module: Module):
|
|
""" Get descriptions for produced dependencies. """
|
|
|
|
meta = {}
|
|
has_meta = hasattr(module, 'prod_meta')
|
|
for prod in module.produces:
|
|
prod = prod.replace('?', '')
|
|
prod = prod.replace('!', '')
|
|
if not has_meta:
|
|
meta[prod] = '<no descritption>'
|
|
continue
|
|
prod_meta = module.prod_meta.get(prod)
|
|
meta[prod] = prod_meta if prod_meta else '<no description>'
|
|
return meta
|