30545982191

Committed 30 Jul 2026 01:13PM UTC coverage: 63.124%. First build

Build # 30545982191

Build Type

Pull #2682

github

Committed by

web-flow

Commit Message

Merge bb3b53bab into 4cf7d1eca

Pull Request Pull Request #2682: Ruff updates

Coverage Stats

34 of 97 new or added lines in 18 files covered. (35.05%)

1657 of 2625 relevant lines covered (63.12%)

0.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

31.44

/vantage6/vantage6/cli/algorithm/generate_algorithm_json.py

import importlib
import inspect
import json
import os
import sys
from collections import OrderedDict
from collections.abc import Callable
from inspect import getmembers, isfunction, ismodule, signature
from pathlib import Path
from types import ModuleType, UnionType
from typing import Any, get_args, get_origin

import click
import pandas as pd
import questionary as q

from vantage6.common import error, info, warning
from vantage6.common.algorithm_function import (
    get_vantage6_decorator_type,
    is_vantage6_algorithm_func,
)
from vantage6.common.enum import AlgorithmArgumentType, AlgorithmStepType, StrEnumBase

from vantage6.algorithm.client import AlgorithmClient
from vantage6.algorithm.preprocessing.algorithm_json_data import (
    PREPROCESSING_FUNCTIONS_JSON_DATA,
)


class MergePreference:
    """Singleton class to manage global merge preference state"""

    _instance = None
    _prefer_existing = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    @classmethod
    def get_preference(cls) -> bool | None:
        """Get the current merge preference"""
        return cls._prefer_existing

    @classmethod
    def set_preference(cls, prefer_existing: bool) -> None:
        """Set the merge preference globally"""
        cls._prefer_existing = prefer_existing

    @classmethod
    def reset(cls) -> None:
        """Reset the preference to None"""
        cls._prefer_existing = None


class FunctionArgumentType(StrEnumBase):
    """Type of the function argument"""

    PARAMETER = "parameter"
    DATAFRAME = "dataframe"


class Function:
    """Class to handle a function and its JSON representation"""

    def __init__(self, func: Callable):
        self.func = func
        self.name = func.__name__
        self.signature = signature(func)
        self.docstring = func.__doc__
        self.json = None
        self.step_type = None

    def prepare_json(self) -> None:
        """Convert the function to a JSON format"""
        self.step_type = self._get_step_type()
        function_json = {
            "name": self.name,
            "display_name": self._pretty_print_name(self.name),
            "standalone": True,
            "description": self._extract_headline_of_docstring(),
            "step_type": self.step_type.value if self.step_type else None,
            "ui_visualizations": [],
            "arguments": [],
            "databases": [],
        }

        parameters = OrderedDict(self.signature.parameters)

        # if the function is a data extraction function, the first argument is a dict
        # with database connection details. This argument should not be added to the
        # function json. Instead, a database should be added to the function json.
        if self.step_type == AlgorithmStepType.DATA_EXTRACTION:
            function_json["databases"].append(
                {
                    "name": "Database",
                    "description": "Database to extract data from",
                }
            )
            # remove database connection details from the signature
            parameters.popitem(last=False)

        # add the arguments to the function json
        for name, param in parameters.items():
            arg_json, arg_type = self._get_argument_json(name, param)
            if arg_json is None:
                continue
            elif arg_type == FunctionArgumentType.DATAFRAME:
                function_json["databases"].append(arg_json)
            else:
                function_json["arguments"].append(arg_json)
        self.json = function_json

    def merge_with_template_json_data(self) -> None:
        """
        Merge the function jsons with the json data from the algorithm_json_data module
        """
        # Only merge the function jsons with template json data if it is an
        # infrastructure-defined function
        if (
            not self._is_func_defined_in_vantage6()
            or self.json["name"] not in PREPROCESSING_FUNCTIONS_JSON_DATA
        ):
            return

        # get the template json data for the function
        template_json = PREPROCESSING_FUNCTIONS_JSON_DATA[self.json["name"]]
        # merge the dicts, with the template dict taking precedence
        for argument in self.json["arguments"]:
            if argument["name"] in template_json["arguments"]:
                argument.update(template_json["arguments"][argument["name"]])
        self._expand_frontend_arguments(template_json)

    def merge_with_existing_json(self, existing_json: dict) -> None:
        """Merge the function json with the existing json data"""
        self._expand_frontend_arguments(existing_json)
        existing_without_frontend = {
            key: value
            for key, value in existing_json.items()
            if key != "frontend_arguments"
        }
        self._merge_dicts(self.json, existing_without_frontend)

    def _expand_frontend_arguments(self, source_json: dict) -> None:
        """
        Expand ``frontend_arguments`` into the ``arguments`` list.

        Used for built-in preprocessing templates and legacy algorithm_store.json
        files that still store frontend-only arguments separately.
        """
        if "frontend_arguments" not in source_json:
            return
        for frontend_argument in source_json["frontend_arguments"]:
            self._add_frontend_argument(source_json, frontend_argument)

    def _is_func_defined_in_vantage6(self) -> bool:
        """Check if the function is defined in the vantage6 package"""
        return self.func.__module__.startswith("vantage6.algorithm.")

    def _merge_dicts(self, target: dict, source: dict) -> None:
        """
        Recursively merge source dict into target dict, with source taking precedence
        """
        for key, value in source.items():
            if key in target:
                if isinstance(value, dict) and isinstance(target[key], dict):
                    # Recursively merge nested dictionaries
                    self._merge_dicts(target[key], value)
                else:
                    self._replace_target_with_source(target, key, value)
            else:
                target[key] = value

    def _replace_target_with_source(self, target: dict, key: str, value: Any) -> None:
        """Replace the value in target with the one from source"""
        if key not in target:
            target[key] = value
            return
        if target[key] == value:
            return

        prefer_existing = MergePreference.get_preference()
        if prefer_existing:
            target[key] = value
        elif prefer_existing is None:
            info(
                f"Different values for the same key '{key}' in function '{self.name}' "
                "were found."
            )
            info(f"Value from function itself: {target[key]}")
            info(f"Value from algorithm.json: {value}")
            result = q.select(
                "Please select the value to keep:",
                choices=[
                    "function itself",
                    "algorithm.json",
                    "function itself (also for all other conflicts)",
                    "algorithm.json (also for all other conflicts)",
                ],
            ).unsafe_ask()
            if result == "algorithm.json":
                target[key] = value
            elif result == "function itself":
                pass  # do nothing
            elif result == "function itself (also for all other conflicts)":
                MergePreference.set_preference(False)
            elif result == "algorithm.json (also for all other conflicts)":
                MergePreference.set_preference(True)
                target[key] = value

    def _get_argument_json(
        self, name: str, param: inspect.Parameter, warn_if_unsupported_arg: bool = True
    ) -> tuple[dict | None, FunctionArgumentType | None]:
        """Get the argument JSON"""

        if param.annotation is None:
            error(f"Function {self.name} has no annotation for argument {name}")
            info(f"Please add a type annotation to the argument {name}")
            info(f"For example, for string arguments: 'def {self.name}({name}: str)'")
            sys.exit(1)

        if param.annotation is AlgorithmClient:
            # Algorithm client arguments do not have to be provided by the user
            return None, None
        elif param.annotation is pd.DataFrame:
            # this is an argument that requires the user to supply a dataframe. That
            # only requires a name and description.
            return {
                "name": name if name != "df" else "Data to use",
                "description": self._extract_parameter_description(name),
            }, FunctionArgumentType.DATAFRAME
        else:
            # This is a regular function parameter
            type_ = self._get_argument_type(
                param, name, warn_if_unsupported=not self._is_func_defined_in_vantage6()
            )
            arg_json = {
                "name": name,
                "display_name": self._pretty_print_name(name),
                "description": self._extract_parameter_description(name),
                "type": type_.value if type_ else None,
                "has_default_value": param.default != inspect.Parameter.empty,
                "is_frontend_only": False,
            }
            if param.default != inspect.Parameter.empty:
                arg_json["default_value"] = param.default

            return arg_json, FunctionArgumentType.PARAMETER

    def _add_frontend_argument(
        self, template_json: dict, frontend_argument: str
    ) -> None:
        """Add a frontend argument to the function json"""
        frontend_argument_json: dict = template_json["frontend_arguments"][
            frontend_argument
        ]
        before_arg_name = frontend_argument_json.pop("before_argument")

        try:
            before_arg_idx = next(
                idx
                for idx, arg in enumerate(self.json["arguments"])
                if arg["name"] == before_arg_name
            )
            self.json["arguments"].insert(before_arg_idx, frontend_argument_json)
        except StopIteration:
            warning(
                f"Could not find argument {before_arg_name} in function "
                f"{self.json['name']}. Frontend argument {frontend_argument} "
                "will not be added."
            )

    def _get_argument_type(
        self, param: inspect.Parameter, name: str, warn_if_unsupported: bool = True
    ) -> AlgorithmArgumentType | None:
        """Get the type of the argument"""
        if isinstance(param.annotation, UnionType):
            # Arguments with default values may have type 'str | None'. If that is the
            # case, we want to use the type of the first element in the union.
            if len(param.annotation.__args__) > 2:
                # if there are more than 2 elements in the union, don't handle
                if warn_if_unsupported:
                    warning(
                        f"Unsupported argument type: {param.annotation} for argument "
                        f"{name} in function {self.name}"
                    )
                return None
            elif len(param.annotation.__args__) == 2:
                # if there are two, we want to use the first one if the second is None
                if param.annotation.__args__[1] is type(None):
                    type_ = param.annotation.__args__[0]
                else:
                    if warn_if_unsupported:
                        warning(
                            f"Unsupported argument type: {param.annotation} for "
                            f"argument '{name}' in function '{self.name}'"
                        )
                    return None
            else:
                # normally, unions have 2+ elements. If there is only one, use that
                type_ = param.annotation.__args__[0]
        else:
            type_ = param.annotation

        if type_ is str:
            return AlgorithmArgumentType.STRING
        elif type_ is dict:
            return AlgorithmArgumentType.JSON
        elif type_ is int:
            return AlgorithmArgumentType.INTEGER
        elif type_ is float:
            return AlgorithmArgumentType.FLOAT
        elif type_ is bool:
            return AlgorithmArgumentType.BOOLEAN
        elif type_ is list:
            return AlgorithmArgumentType.STRINGS
        elif get_origin(type_) is list:
            # Handle generic list types like list[str], list[int], list[float]
            args = get_args(type_)
            if len(args) == 1:
                inner_type = args[0]
                if inner_type is str:
                    return AlgorithmArgumentType.STRINGS
                elif inner_type is int:
                    return AlgorithmArgumentType.INTEGERS
                elif inner_type is float:
                    return AlgorithmArgumentType.FLOATS
            # Fallback: if list has no args or multiple args, default to STRINGS
            return AlgorithmArgumentType.STRINGS
        else:
            if warn_if_unsupported:
                warning(
                    f"Unsupported argument type: {param.annotation} for argument "
                    f"'{name}' in function '{self.name}'"
                )
            return None

    def _pretty_print_name(self, name: str) -> str:
        """Pretty print the name of the function"""
        pretty = name.replace("_", " ")
        if len(pretty):
            pretty = pretty[0].upper() + pretty[1:]
        return pretty

    def _extract_headline_of_docstring(self) -> str:
        """Extract the headline of the docstring"""
        if not self.docstring:
            return ""

        # Split by double newlines to get the first paragraph
        paragraphs = self.docstring.split("\n\n")
        first_paragraph = paragraphs[0]

        # Split by single newlines and join the lines with spaces
        lines = first_paragraph.split("\n")
        header = " ".join(line.strip() for line in lines if line.strip() != "")
        return header

    def _get_step_type(self) -> AlgorithmStepType | None:
        """Get the step type of the function"""
        decorator_type = get_vantage6_decorator_type(self.func)
        if decorator_type in AlgorithmStepType.list():
            return decorator_type
        else:
            warning(
                f"Unsupported decorator type: {decorator_type} for function {self.name}"
            )
            return None

    def _extract_parameter_description(self, name: str) -> str:
        """Extract the description of the parameter"""
        if not self.docstring:
            return ""

        # Try both patterns: "{name}:" and "{name} :"
        patterns = [f"{name}:", f"{name} :"]

        for pattern in patterns:
            if pattern in self.docstring:
                return self.docstring.split(pattern)[1].split("\n")[1].strip()

        return ""


@click.command()
@click.option(
    "--algo-function-file",
    default=None,
    type=str,
    help="Path to the file containing or importing the algorithm functions",
)
@click.option(
    "--current-json",
    default=None,
    type=str,
    help="Path to the current algorithm.json file",
)
@click.option(
    "--output-file",
    default="new-algorithm.json",
    type=str,
    help="Path to the output file",
)
def cli_algorithm_generate_json(
    algo_function_file: str, current_json: str, output_file: str
) -> dict:
    """
    Generate an updated algorithm.json file to submit to the algorithm store.

    You should provide the path to the file where the algorithm functions are
    defined.

    Note that if you do asterisk ('from x import *') imports, all functions from the
    imported module will be added to the algorithm.json file.
    """
    algo_function_file = _get_algo_function_file_location(algo_function_file)

    current_json = _get_current_json_location(current_json)

    # read the current algorithm.json file
    with open(current_json, "r", encoding="utf-8") as f:
        current_json_data = json.load(f)

    # get the functions from the file
    info(f"Importing functions from {algo_function_file}...")
    functions = _get_functions_from_file(algo_function_file)
    function_objs = [Function(f) for f in functions]

    info("Converting functions to JSON...")
    for function in function_objs:
        function.prepare_json()
        function.merge_with_template_json_data()

        # merge the function jsons with the existing json data
        current_json_func = [
            f for f in current_json_data["functions"] if f["name"] == function.name
        ]
        if current_json_func:
            function.merge_with_existing_json(current_json_func[0])

    # write the new algorithm.json file
    info(f"Writing new algorithm.json file to {output_file}...")
    current_json_data["functions"] = [f.json for f in function_objs]
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(current_json_data, f, indent=2)

    info(f"New algorithm.json file written to: {output_file}")

    warning("-" * 60)
    warning(f"Check the generated '{output_file}' file before ")
    warning("submitting it to the algorithm store!")
    warning("-" * 60)


def _get_functions_from_file(file_path: str) -> None:
    """Get the functions from the file

    Parameters
    ----------
    file_path : str
        Path to the file containing or importing the algorithm functions
    """
    # Convert path to absolute path
    file_path = str(Path(file_path).resolve())

    # Get the package root directory (two levels up from the file)
    package_root = str(Path(file_path).parent.parent)
    if package_root not in sys.path:
        sys.path.insert(0, package_root)

    # Get the module name from the file path, including the package name
    package_name = Path(file_path).parent.name
    module_name = f"{package_name}.{Path(file_path).stem}"

    # Import the module
    try:
        module = importlib.import_module(module_name)
    except ImportError as e:
        raise ImportError(f"Could not import module {module_name}: {e!s}") from e

    def get_members_from_module(module: ModuleType) -> list:
        """Get the functions from the module"""
        return [
            member for name, member in getmembers(module) if not name.startswith("_")
        ]

    # get the functions from the algorithm module
    import_members = get_members_from_module(module)
    import_functions = [
        m for m in import_members if isfunction(m) and is_vantage6_algorithm_func(m)
    ]
    import_modules = [m for m in import_members if ismodule(m)]

    # add the functions from the imported modules (only 1 level deep). This is so that
    # if you do e.g. 'from vantage6.algorithm.preprocessing import *', all functions
    # from within those modules are also imported.
    for import_module in import_modules:
        second_level_import_members = get_members_from_module(import_module)
        import_functions.extend(
            [
                m
                for m in second_level_import_members
                if isfunction(m) and is_vantage6_algorithm_func(m)
            ]
        )

    return import_functions


def _get_algo_function_file_location(algo_function_file: str | None) -> None:
    """Get user input for the algorithm creation

    Parameters
    ----------
    algo_function_file : str
        Path to the file containing or importingthe algorithm functions
    """
    if not algo_function_file:
        default_dir = str(Path(os.getcwd()) / "__init__.py")
        algo_function_file = q.text(
            "Path to the file containing or importing the algorithm functions:",
            default=default_dir,
        ).unsafe_ask()

    # Convert to absolute path using pathlib
    algo_function_file = str(Path(algo_function_file).resolve())

    # check if the file exists
    if not Path(algo_function_file).exists():
        raise FileNotFoundError(f"File {algo_function_file} does not exist")

    return algo_function_file


def _get_current_json_location(current_json: str) -> None:
    """Get user input for the current algorithm.json file

    Parameters
    ----------
    current_json : str
        Path to the current algorithm.json file
    """
    if not current_json:
        default_dir = str(Path(os.getcwd()) / "algorithm_store.json")
        current_json = q.text(
            "Path to the current algorithm.json file:",
            default=default_dir,
        ).unsafe_ask()

    # Convert to absolute path using pathlib
    current_json = str(Path(current_json).resolve())

    # check if the file exists
    if not Path(current_json).exists():
        raise FileNotFoundError(f"File {current_json} does not exist")

    return current_json

1	import importlib	1✔
2	import inspect	1✔
3	import json	1✔
4	import os	1✔
5	import sys	1✔
6	from collections import OrderedDict	1✔
7	from collections.abc import Callable	1✔
8	from inspect import getmembers, isfunction, ismodule, signature	1✔
9	from pathlib import Path	1✔
10	from types import ModuleType, UnionType	1✔
11	from typing import Any, get_args, get_origin	1✔
12
13	import click	1✔
14	import pandas as pd	1✔
15	import questionary as q	1✔
16
17	from vantage6.common import error, info, warning	1✔
18	from vantage6.common.algorithm_function import (	1✔
19	get_vantage6_decorator_type,
20	is_vantage6_algorithm_func,
21	)
22	from vantage6.common.enum import AlgorithmArgumentType, AlgorithmStepType, StrEnumBase	1✔
23
24	from vantage6.algorithm.client import AlgorithmClient	1✔
25	from vantage6.algorithm.preprocessing.algorithm_json_data import (	1✔
26	PREPROCESSING_FUNCTIONS_JSON_DATA,
27	)
28
29
30	class MergePreference:	1✔
31	"""Singleton class to manage global merge preference state"""
32
33	_instance = None	1✔
34	_prefer_existing = None	1✔
35
36	def __new__(cls):	1✔
37	if cls._instance is None:	×
NEW 38	cls._instance = super().__new__(cls)	×
39	return cls._instance	×
40
41	@classmethod	1✔
42	def get_preference(cls) -> bool \| None:	1✔
43	"""Get the current merge preference"""
44	return cls._prefer_existing	×
45
46	@classmethod	1✔
47	def set_preference(cls, prefer_existing: bool) -> None:	1✔
48	"""Set the merge preference globally"""
49	cls._prefer_existing = prefer_existing	×
50
51	@classmethod	1✔
52	def reset(cls) -> None:	1✔
53	"""Reset the preference to None"""
54	cls._prefer_existing = None	1✔
55
56
57	class FunctionArgumentType(StrEnumBase):	1✔
58	"""Type of the function argument"""
59
60	PARAMETER = "parameter"	1✔
61	DATAFRAME = "dataframe"	1✔
62
63
64	class Function:	1✔
65	"""Class to handle a function and its JSON representation"""
66
67	def __init__(self, func: Callable):	1✔
68	self.func = func	1✔
69	self.name = func.__name__	1✔
70	self.signature = signature(func)	1✔
71	self.docstring = func.__doc__	1✔
72	self.json = None	1✔
73	self.step_type = None	1✔
74
75	def prepare_json(self) -> None:	1✔
76	"""Convert the function to a JSON format"""
77	self.step_type = self._get_step_type()	×
78	function_json = {	×
79	"name": self.name,
80	"display_name": self._pretty_print_name(self.name),
81	"standalone": True,
82	"description": self._extract_headline_of_docstring(),
83	"step_type": self.step_type.value if self.step_type else None,
84	"ui_visualizations": [],
85	"arguments": [],
86	"databases": [],
87	}
88
89	parameters = OrderedDict(self.signature.parameters)	×
90
91	# if the function is a data extraction function, the first argument is a dict
92	# with database connection details. This argument should not be added to the
93	# function json. Instead, a database should be added to the function json.
94	if self.step_type == AlgorithmStepType.DATA_EXTRACTION:	×
95	function_json["databases"].append(	×
96	{
97	"name": "Database",
98	"description": "Database to extract data from",
99	}
100	)
101	# remove database connection details from the signature
102	parameters.popitem(last=False)	×
103
104	# add the arguments to the function json
105	for name, param in parameters.items():	×
106	arg_json, arg_type = self._get_argument_json(name, param)	×
107	if arg_json is None:	×
108	continue	×
109	elif arg_type == FunctionArgumentType.DATAFRAME:	×
110	function_json["databases"].append(arg_json)	×
111	else:
112	function_json["arguments"].append(arg_json)	×
113	self.json = function_json	×
114
115	def merge_with_template_json_data(self) -> None:	1✔
116	"""
117	Merge the function jsons with the json data from the algorithm_json_data module
118	"""
119	# Only merge the function jsons with template json data if it is an
120	# infrastructure-defined function
121	if (	×
122	not self._is_func_defined_in_vantage6()
123	or self.json["name"] not in PREPROCESSING_FUNCTIONS_JSON_DATA
124	):
125	return	×
126
127	# get the template json data for the function
128	template_json = PREPROCESSING_FUNCTIONS_JSON_DATA[self.json["name"]]	×
129	# merge the dicts, with the template dict taking precedence
130	for argument in self.json["arguments"]:	×
131	if argument["name"] in template_json["arguments"]:	×
132	argument.update(template_json["arguments"][argument["name"]])	×
133	self._expand_frontend_arguments(template_json)	×
134
135	def merge_with_existing_json(self, existing_json: dict) -> None:	1✔
136	"""Merge the function json with the existing json data"""
137	self._expand_frontend_arguments(existing_json)	1✔
138	existing_without_frontend = {	1✔
139	key: value
140	for key, value in existing_json.items()
141	if key != "frontend_arguments"
142	}
143	self._merge_dicts(self.json, existing_without_frontend)	1✔
144
145	def _expand_frontend_arguments(self, source_json: dict) -> None:	1✔
146	"""
147	Expand ``frontend_arguments`` into the ``arguments`` list.
148
149	Used for built-in preprocessing templates and legacy algorithm_store.json
150	files that still store frontend-only arguments separately.
151	"""
152	if "frontend_arguments" not in source_json:	1✔
153	return	1✔
154	for frontend_argument in source_json["frontend_arguments"]:	1✔
155	self._add_frontend_argument(source_json, frontend_argument)	1✔
156
157	def _is_func_defined_in_vantage6(self) -> bool:	1✔
158	"""Check if the function is defined in the vantage6 package"""
159	return self.func.__module__.startswith("vantage6.algorithm.")	×
160
161	def _merge_dicts(self, target: dict, source: dict) -> None:	1✔
162	"""
163	Recursively merge source dict into target dict, with source taking precedence
164	"""
165	for key, value in source.items():	1✔
166	if key in target:	1✔
167	if isinstance(value, dict) and isinstance(target[key], dict):	1✔
168	# Recursively merge nested dictionaries
169	self._merge_dicts(target[key], value)	×
170	else:
171	self._replace_target_with_source(target, key, value)	1✔
172	else:
173	target[key] = value	1✔
174
175	def _replace_target_with_source(self, target: dict, key: str, value: Any) -> None:	1✔
176	"""Replace the value in target with the one from source"""
177	if key not in target:	1✔
178	target[key] = value	×
179	return	×
180	if target[key] == value:	1✔
181	return	1✔
182
183	prefer_existing = MergePreference.get_preference()	×
184	if prefer_existing:	×
185	target[key] = value	×
186	elif prefer_existing is None:	×
187	info(	×
188	f"Different values for the same key '{key}' in function '{self.name}' "
189	"were found."
190	)
191	info(f"Value from function itself: {target[key]}")	×
192	info(f"Value from algorithm.json: {value}")	×
193	result = q.select(	×
194	"Please select the value to keep:",
195	choices=[
196	"function itself",
197	"algorithm.json",
198	"function itself (also for all other conflicts)",
199	"algorithm.json (also for all other conflicts)",
200	],
201	).unsafe_ask()
202	if result == "algorithm.json":	×
203	target[key] = value	×
204	elif result == "function itself":	×
205	pass # do nothing	×
206	elif result == "function itself (also for all other conflicts)":	×
207	MergePreference.set_preference(False)	×
208	elif result == "algorithm.json (also for all other conflicts)":	×
209	MergePreference.set_preference(True)	×
210	target[key] = value	×
211
212	def _get_argument_json(	1✔
213	self, name: str, param: inspect.Parameter, warn_if_unsupported_arg: bool = True
214	) -> tuple[dict \| None, FunctionArgumentType \| None]:
215	"""Get the argument JSON"""
216
217	if param.annotation is None:	×
218	error(f"Function {self.name} has no annotation for argument {name}")	×
219	info(f"Please add a type annotation to the argument {name}")	×
220	info(f"For example, for string arguments: 'def {self.name}({name}: str)'")	×
NEW 221	sys.exit(1)	×
222
223	if param.annotation is AlgorithmClient:	×
224	# Algorithm client arguments do not have to be provided by the user
225	return None, None	×
226	elif param.annotation is pd.DataFrame:	×
227	# this is an argument that requires the user to supply a dataframe. That
228	# only requires a name and description.
229	return {	×
230	"name": name if name != "df" else "Data to use",
231	"description": self._extract_parameter_description(name),
232	}, FunctionArgumentType.DATAFRAME
233	else:
234	# This is a regular function parameter
235	type_ = self._get_argument_type(	×
236	param, name, warn_if_unsupported=not self._is_func_defined_in_vantage6()
237	)
238	arg_json = {	×
239	"name": name,
240	"display_name": self._pretty_print_name(name),
241	"description": self._extract_parameter_description(name),
242	"type": type_.value if type_ else None,
243	"has_default_value": param.default != inspect.Parameter.empty,
244	"is_frontend_only": False,
245	}
246	if param.default != inspect.Parameter.empty:	×
247	arg_json["default_value"] = param.default	×
248
249	return arg_json, FunctionArgumentType.PARAMETER	×
250
251	def _add_frontend_argument(	1✔
252	self, template_json: dict, frontend_argument: str
253	) -> None:
254	"""Add a frontend argument to the function json"""
255	frontend_argument_json: dict = template_json["frontend_arguments"][	1✔
256	frontend_argument
257	]
258	before_arg_name = frontend_argument_json.pop("before_argument")	1✔
259
260	try:	1✔
261	before_arg_idx = next(	1✔
262	idx
263	for idx, arg in enumerate(self.json["arguments"])
264	if arg["name"] == before_arg_name
265	)
266	self.json["arguments"].insert(before_arg_idx, frontend_argument_json)	1✔
267	except StopIteration:	×
268	warning(	×
269	f"Could not find argument {before_arg_name} in function "
270	f"{self.json['name']}. Frontend argument {frontend_argument} "
271	"will not be added."
272	)
273
274	def _get_argument_type(	1✔
275	self, param: inspect.Parameter, name: str, warn_if_unsupported: bool = True
276	) -> AlgorithmArgumentType \| None:
277	"""Get the type of the argument"""
278	if isinstance(param.annotation, UnionType):	×
279	# Arguments with default values may have type 'str \| None'. If that is the
280	# case, we want to use the type of the first element in the union.
281	if len(param.annotation.__args__) > 2:	×
282	# if there are more than 2 elements in the union, don't handle
283	if warn_if_unsupported:	×
284	warning(	×
285	f"Unsupported argument type: {param.annotation} for argument "
286	f"{name} in function {self.name}"
287	)
288	return None	×
289	elif len(param.annotation.__args__) == 2:	×
290	# if there are two, we want to use the first one if the second is None
291	if param.annotation.__args__[1] is type(None):	×
292	type_ = param.annotation.__args__[0]	×
293	else:
294	if warn_if_unsupported:	×
295	warning(	×
296	f"Unsupported argument type: {param.annotation} for "
297	f"argument '{name}' in function '{self.name}'"
298	)
299	return None	×
300	else:
301	# normally, unions have 2+ elements. If there is only one, use that
302	type_ = param.annotation.__args__[0]	×
303	else:
304	type_ = param.annotation	×
305
306	if type_ is str:	×
307	return AlgorithmArgumentType.STRING	×
308	elif type_ is dict:	×
309	return AlgorithmArgumentType.JSON	×
310	elif type_ is int:	×
311	return AlgorithmArgumentType.INTEGER	×
312	elif type_ is float:	×
313	return AlgorithmArgumentType.FLOAT	×
314	elif type_ is bool:	×
315	return AlgorithmArgumentType.BOOLEAN	×
316	elif type_ is list:	×
317	return AlgorithmArgumentType.STRINGS	×
318	elif get_origin(type_) is list:	×
319	# Handle generic list types like list[str], list[int], list[float]
320	args = get_args(type_)	×
321	if len(args) == 1:	×
322	inner_type = args[0]	×
323	if inner_type is str:	×
324	return AlgorithmArgumentType.STRINGS	×
325	elif inner_type is int:	×
326	return AlgorithmArgumentType.INTEGERS	×
327	elif inner_type is float:	×
328	return AlgorithmArgumentType.FLOATS	×
329	# Fallback: if list has no args or multiple args, default to STRINGS
330	return AlgorithmArgumentType.STRINGS	×
331	else:
332	if warn_if_unsupported:	×
333	warning(	×
334	f"Unsupported argument type: {param.annotation} for argument "
335	f"'{name}' in function '{self.name}'"
336	)
337	return None	×
338
339	def _pretty_print_name(self, name: str) -> str:	1✔
340	"""Pretty print the name of the function"""
341	pretty = name.replace("_", " ")	×
342	if len(pretty):	×
343	pretty = pretty[0].upper() + pretty[1:]	×
344	return pretty	×
345
346	def _extract_headline_of_docstring(self) -> str:	1✔
347	"""Extract the headline of the docstring"""
348	if not self.docstring:	×
349	return ""	×
350
351	# Split by double newlines to get the first paragraph
352	paragraphs = self.docstring.split("\n\n")	×
353	first_paragraph = paragraphs[0]	×
354
355	# Split by single newlines and join the lines with spaces
356	lines = first_paragraph.split("\n")	×
357	header = " ".join(line.strip() for line in lines if line.strip() != "")	×
358	return header	×
359
360	def _get_step_type(self) -> AlgorithmStepType \| None:	1✔
361	"""Get the step type of the function"""
362	decorator_type = get_vantage6_decorator_type(self.func)	×
363	if decorator_type in AlgorithmStepType.list():	×
364	return decorator_type	×
365	else:
366	warning(	×
367	f"Unsupported decorator type: {decorator_type} for function {self.name}"
368	)
369	return None	×
370
371	def _extract_parameter_description(self, name: str) -> str:	1✔
372	"""Extract the description of the parameter"""
373	if not self.docstring:	×
374	return ""	×
375
376	# Try both patterns: "{name}:" and "{name} :"
377	patterns = [f"{name}:", f"{name} :"]	×
378
379	for pattern in patterns:	×
380	if pattern in self.docstring:	×
381	return self.docstring.split(pattern)[1].split("\n")[1].strip()	×
382
383	return ""	×
384
385
386	@click.command()	1✔
387	@click.option(	1✔
388	"--algo-function-file",
389	default=None,
390	type=str,
391	help="Path to the file containing or importing the algorithm functions",
392	)
393	@click.option(	1✔
394	"--current-json",
395	default=None,
396	type=str,
397	help="Path to the current algorithm.json file",
398	)
399	@click.option(	1✔
400	"--output-file",
401	default="new-algorithm.json",
402	type=str,
403	help="Path to the output file",
404	)
405	def cli_algorithm_generate_json(	1✔
406	algo_function_file: str, current_json: str, output_file: str
407	) -> dict:
408	"""
409	Generate an updated algorithm.json file to submit to the algorithm store.
410
411	You should provide the path to the file where the algorithm functions are
412	defined.
413
414	Note that if you do asterisk ('from x import *') imports, all functions from the
415	imported module will be added to the algorithm.json file.
416	"""
417	algo_function_file = _get_algo_function_file_location(algo_function_file)	×
418
419	current_json = _get_current_json_location(current_json)	×
420
421	# read the current algorithm.json file
422	with open(current_json, "r", encoding="utf-8") as f:	×
423	current_json_data = json.load(f)	×
424
425	# get the functions from the file
426	info(f"Importing functions from {algo_function_file}...")	×
427	functions = _get_functions_from_file(algo_function_file)	×
428	function_objs = [Function(f) for f in functions]	×
429
430	info("Converting functions to JSON...")	×
431	for function in function_objs:	×
432	function.prepare_json()	×
433	function.merge_with_template_json_data()	×
434
435	# merge the function jsons with the existing json data
436	current_json_func = [	×
437	f for f in current_json_data["functions"] if f["name"] == function.name
438	]
439	if current_json_func:	×
440	function.merge_with_existing_json(current_json_func[0])	×
441
442	# write the new algorithm.json file
443	info(f"Writing new algorithm.json file to {output_file}...")	×
444	current_json_data["functions"] = [f.json for f in function_objs]	×
445	with open(output_file, "w", encoding="utf-8") as f:	×
446	json.dump(current_json_data, f, indent=2)	×
447
448	info(f"New algorithm.json file written to: {output_file}")	×
449
450	warning("-" * 60)	×
451	warning(f"Check the generated '{output_file}' file before ")	×
452	warning("submitting it to the algorithm store!")	×
453	warning("-" * 60)	×
454
455
456	def _get_functions_from_file(file_path: str) -> None:	1✔
457	"""Get the functions from the file
458
459	Parameters
460	----------
461	file_path : str
462	Path to the file containing or importing the algorithm functions
463	"""
464	# Convert path to absolute path
465	file_path = str(Path(file_path).resolve())	×
466
467	# Get the package root directory (two levels up from the file)
468	package_root = str(Path(file_path).parent.parent)	×
469	if package_root not in sys.path:	×
470	sys.path.insert(0, package_root)	×
471
472	# Get the module name from the file path, including the package name
473	package_name = Path(file_path).parent.name	×
474	module_name = f"{package_name}.{Path(file_path).stem}"	×
475
476	# Import the module
477	try:	×
478	module = importlib.import_module(module_name)	×
479	except ImportError as e:	×
NEW 480	raise ImportError(f"Could not import module {module_name}: {e!s}") from e	×
481
482	def get_members_from_module(module: ModuleType) -> list:	×
483	"""Get the functions from the module"""
484	return [	×
485	member for name, member in getmembers(module) if not name.startswith("_")
486	]
487
488	# get the functions from the algorithm module
489	import_members = get_members_from_module(module)	×
490	import_functions = [	×
491	m for m in import_members if isfunction(m) and is_vantage6_algorithm_func(m)
492	]
493	import_modules = [m for m in import_members if ismodule(m)]	×
494
495	# add the functions from the imported modules (only 1 level deep). This is so that
496	# if you do e.g. 'from vantage6.algorithm.preprocessing import *', all functions
497	# from within those modules are also imported.
498	for import_module in import_modules:	×
499	second_level_import_members = get_members_from_module(import_module)	×
500	import_functions.extend(	×
501	[
502	m
503	for m in second_level_import_members
504	if isfunction(m) and is_vantage6_algorithm_func(m)
505	]
506	)
507
508	return import_functions	×
509
510
511	def _get_algo_function_file_location(algo_function_file: str \| None) -> None:	1✔
512	"""Get user input for the algorithm creation
513
514	Parameters
515	----------
516	algo_function_file : str
517	Path to the file containing or importingthe algorithm functions
518	"""
519	if not algo_function_file:	×
520	default_dir = str(Path(os.getcwd()) / "__init__.py")	×
521	algo_function_file = q.text(	×
522	"Path to the file containing or importing the algorithm functions:",
523	default=default_dir,
524	).unsafe_ask()
525
526	# Convert to absolute path using pathlib
527	algo_function_file = str(Path(algo_function_file).resolve())	×
528
529	# check if the file exists
530	if not Path(algo_function_file).exists():	×
531	raise FileNotFoundError(f"File {algo_function_file} does not exist")	×
532
533	return algo_function_file	×
534
535
536	def _get_current_json_location(current_json: str) -> None:	1✔
537	"""Get user input for the current algorithm.json file
538
539	Parameters
540	----------
541	current_json : str
542	Path to the current algorithm.json file
543	"""
544	if not current_json:	×
545	default_dir = str(Path(os.getcwd()) / "algorithm_store.json")	×
546	current_json = q.text(	×
547	"Path to the current algorithm.json file:",
548	default=default_dir,
549	).unsafe_ask()
550
551	# Convert to absolute path using pathlib
552	current_json = str(Path(current_json).resolve())	×
553
554	# check if the file exists
555	if not Path(current_json).exists():	×
556	raise FileNotFoundError(f"File {current_json} does not exist")	×
557
558	return current_json	×

vantage6 / vantage6 / 30545982191

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous