| #!/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 |
| |
| """ |
| This module is intended for wrapping simple scripts without rewriting them as |
| an sfbuild module. This is mostly to maintain compatibility with workflows |
| that do not use sfbuild and instead rely on legacy scripts. |
| |
| Accepted module parameters: |
| * `stage_name` (string, optional): Name describing the stage |
| * `script` (string, mandatory): Path to the script to be executed |
| * `interpreter` (string, optional): Interpreter for the script |
| * `cwd` (string, optional): Current Working Directory for the script |
| * `outputs` (dict[string -> dict[string -> string]], mandatory): |
| A dict with output descriptions (dicts). |
| Keys name output dependencies. |
| * `mode` (string, mandatory): "file" or "stdout". |
| Describes how the output is grabbed from the script. |
| * `file` (string, required if `mode` is "file"): Name of the file generated by the script. |
| * `target` (string, required): Default name of the file of the generated dependency. |
| You can use all values available durng map_io stage. |
| Each input dependency alsogets two extra values associated with it: |
| `:dependency_name[noext]`, which contains the path to the dependency the extension with anything after last "." |
| removed and `:dependency_name[dir]` which contains directory paths of the dependency. |
| This is useful for deriving an output name from the input. |
| * `meta` (string, optional): Description of the output dependency. |
| * `inputs` (dict[string -> string | bool], mandatory): |
| A dict with input descriptions. |
| Key is either a name of a named argument or a position of unnamed argument prefaced with "#" (eg. "#1"). |
| Positions are indexed from 1, as it's a convention that 0th argument is the path of the executed program. |
| Values are strings that can contains references to variables to be resolved after the project flow configuration is |
| loaded (that means they can reference values and dependencies which are to be set by the user). |
| All of modules inputs will be determined by the references used. |
| Thus dependency and value definitions are implicit. |
| If the value of the resolved string is empty and is associated with a named argument, the argument in question will be |
| skipped entirely. |
| This allows using optional dependencies. |
| To use a named argument as a flag instead, set it to `true`. |
| """ |
| |
| # TODO: `environment` input kind |
| |
| from pathlib import Path |
| from re import match as re_match, finditer as re_finditer |
| |
| from f4pga.flows.common import decompose_depname, deep, get_verbosity_level, sub |
| 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") |
| 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:] |
| elif param_def[0] == "-": |
| return "char", param_def[1:] |
| return "named", param_def |
| |
| |
| class InputReferences: |
| dependencies: "set[str]" |
| values: "set[str]" |
| |
| def merge(self, other): |
| self.dependencies.update(other.dependencies) |
| self.values.update(other.values) |
| |
| def __init__(self): |
| self.dependencies = set() |
| self.values = set() |
| |
| |
| def _get_input_references(input: str) -> InputReferences: |
| refs = InputReferences() |
| if type(input) is not str: |
| return refs |
| for match in re_finditer("\$\{([^${}]*)\}", input): |
| match_str = match.group(1) |
| 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)) |
| return refs |
| |
| |
| def _make_noop1(): |
| def noop(_): |
| return |
| |
| return noop |
| |
| |
| 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" |
| |
| @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) |
| |
| def map_io(self, ctx: ModuleContext): |
| self._add_extra_values_to_env(ctx) |
| |
| outputs = {} |
| for dep, _, out_path in self.file_outputs: |
| out_path_resolved = ctx.r_env.resolve(out_path, final=True) |
| outputs[dep] = out_path_resolved |
| |
| if self.stdout_target: |
| out_path_resolved = ctx.r_env.resolve(self.stdout_target[1], final=True) |
| outputs[self.stdout_target[0]] = out_path_resolved |
| |
| return outputs |
| |
| def execute(self, ctx: ModuleContext): |
| self._add_extra_values_to_env(ctx) |
| |
| cwd = ctx.r_env.resolve(self.cwd) |
| |
| sub_args = ( |
| ([ctx.r_env.resolve(self.interpreter, final=True)] if self.interpreter else []) |
| + ( |
| self.script_path |
| if isinstance(self.script_path, list) |
| else [ctx.r_env.resolve(self.script_path, final=True)] |
| ) |
| + self.get_args(ctx) |
| ) |
| |
| sub_env = self.get_env(ctx) |
| |
| # XXX: This may produce incorrect string if arguments contains whitespace |
| # characters |
| cmd = " ".join(sub_args) |
| |
| if get_verbosity_level() >= 2: |
| yield f"Running script...\n {cmd}" |
| else: |
| yield f"Running an externel script..." |
| |
| data = sub(*sub_args, cwd=cwd, env=sub_env) |
| |
| yield "Writing outputs..." |
| if self.stdout_target: |
| target = ctx.r_env.resolve(self.stdout_target[1], final=True) |
| with open(target, "wb") as f: |
| f.write(data) |
| |
| for dep, file, _ in self.file_outputs: |
| file = ctx.r_env.resolve(file, final=True) |
| target = ctx.r_env.resolve(getattr(ctx.outputs, dep), final=True) |
| print(f"file is: {file}, target is: {target}") |
| if target != file: |
| Path(file).rename(target) |
| |
| 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") |
| if meta is str: |
| self.prod_meta[dname] = meta |
| |
| mode = output_def.get("mode") |
| if type(mode) is not str: |
| raise Exception(f"Output mode for `{dep_name}` is not specified") |
| |
| target = output_def.get("target") |
| if type(target) is not str: |
| raise Exception("`target` field is not specified") |
| |
| if mode == "file": |
| file = output_def.get("file") |
| if type(file) is not str: |
| raise Exception("Output file is not specified") |
| self.file_outputs.append((dname, file, target)) |
| elif mode == "stdout": |
| if self.stdout_target is not None: |
| raise Exception("stdout output is already specified") |
| self.stdout_target = dname, target |
| |
| # A very functional approach |
| def _init_inputs(self, input_defs): |
| positional_args = [] |
| named_args = [] |
| env_vars = {} |
| refs = InputReferences() |
| |
| get_args = _make_noop1() |
| get_env = _make_noop1() |
| |
| for arg_code, input in input_defs.items(): |
| param_kind, param = _parse_param_def(arg_code) |
| |
| push = None |
| push_env = None |
| 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}") |
| else: |
| named_args += [f"--{param}", str(val)] |
| |
| push = push_named |
| elif param_kind == "char": |
| |
| def push_char(val: "str | bool | int", param=param): |
| nonlocal named_args |
| if type(val) is bool: |
| named_args.append(f"-{param}") |
| else: |
| named_args += [f"-{param}", str(val)] |
| |
| push = push_char |
| 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 != "": |
| 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 != "": |
| push_env(val) |
| |
| get_env = _tailcall1(get_env, push_q) |
| |
| def get_all_args(ctx: ModuleContext): |
| nonlocal get_args, positional_args, named_args |
| |
| get_args(ctx) |
| |
| positional_args.sort(key=lambda t: t[0]) |
| pos = [a for _, a in positional_args] |
| |
| return named_args + pos |
| |
| def get_all_env(ctx: ModuleContext): |
| nonlocal get_env, env_vars |
| get_env(ctx) |
| if len(env_vars.items()) == 0: |
| return None |
| return env_vars |
| |
| setattr(self, "get_args", get_all_args) |
| setattr(self, "get_env", get_all_env) |
| |
| for dep in refs.dependencies: |
| self.takes.append(dep) |
| for val in refs.values: |
| self.values.append(val) |
| |
| def __init__(self, params): |
| 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.takes = [] |
| self.produces = [] |
| self.values = [] |
| self.prod_meta = {} |
| |
| self._init_outputs(_get_param(params, "outputs")) |
| self._init_inputs(_get_param(params, "inputs")) |
| |
| |
| ModuleClass = GenericScriptWrapperModule |
