From f5dd824cdc5d545cded78ca1b08397562ead006e Mon Sep 17 00:00:00 2001 From: Jon Clucas Date: Thu, 18 Jul 2024 12:41:45 -0400 Subject: [PATCH] :art: Standardize docstring format across changes. --- CPAC/pipeline/engine/nodeblock.py | 115 +++++++++++++++++---------- CPAC/pipeline/engine/resource.py | 126 ++++++++++++++++-------------- CPAC/pipeline/schema.py | 53 +++---------- CPAC/pipeline/test/test_engine.py | 14 ++-- CPAC/pipeline/utils.py | 30 ++----- 5 files changed, 162 insertions(+), 176 deletions(-) diff --git a/CPAC/pipeline/engine/nodeblock.py b/CPAC/pipeline/engine/nodeblock.py index 5db1c6c4cd..e1b1437d3c 100644 --- a/CPAC/pipeline/engine/nodeblock.py +++ b/CPAC/pipeline/engine/nodeblock.py @@ -14,7 +14,7 @@ # You should have received a copy of the GNU Lesser General Public # License along with C-PAC. If not, see . -"""Class and decorator for NodeBlock functions.""" +"""Classes and decorator for :py:class:`NodeBlock`\u200bs and :py:class:`NodeBlockFunction`\u200bs.""" from typing import Any, Callable, Optional, TYPE_CHECKING @@ -50,32 +50,37 @@ def __init__( outputs: Optional[NODEBLOCK_OUTPUTS] = None, ) -> None: self.func = func - """Nodeblock function reference.""" + """`Nodeblock` function reference.""" self.name: str = name - """Used in the graph and logging to identify the NodeBlock and its component nodes.""" + """Used in the graph and logging to identify the :py:class:`NodeBlock` and its component :py:class:`~nipype.pipeline.engine.Node`\u200bs.""" self.config: Optional[list[str]] = config """ - Indicates the nested keys in a C-PAC pipeline configuration should configure a NodeBlock built from this - function. If config is set to ``None``, then all other configuration-related entities must be specified from the + Indicates the nested keys in a C-PAC pipeline :py:class:`Configuration` + should configure a `NodeBlock` built from this function. If `config` is set to + `None`, then all other configuration-related entities must be specified from the root of the configuration. """ self.switch: Optional[list[str] | list[list[str]]] = switch """ - Indicates any keys that should evaluate to True for this NodeBlock to be active. A list of lists of strings - indicates multiple switches that must all be True to run, and is currently only an option if config is set to - ``None``. + Indicates any keys that should evaluate to `True` for this :py:class:`NodeBlock` + to be active. A list of lists of strings indicates multiple `switch`\u200bes + that must all be `True` to run, and is currently only an option if `config` is + set to `None`. """ self.option_key: Optional[str | list[str]] = option_key """ - Indicates the nested keys (starting at the nested key indicated by config) that should configure this NodeBlock. + Indicates the nested keys (starting at the nested key indicated by `config`) + that should configure this :py:class:`NodeBlock`. """ self.option_val: Optional[str | list[str]] = option_val - """Indicates values for which this NodeBlock should be active.""" + """Indicates values for which this :py:class:`NodeBlock` should be active.""" self.inputs: list[str | list | tuple] = inputs if inputs else [] - """ResourcePool keys indicating resources needed for the NodeBlock's functionality.""" + """:py:class:`~CPAC.pipeline.engine.resource.ResourcePool` keys indicating + resources needed for the :py:class:`NodeBlock`\u200b's functionality.""" self.outputs: list[str] | dict[str, Any] = outputs if outputs else [] """ - ResourcePool keys indicating resources generated or updated by the NodeBlock, optionally including metadata + :py:class:`~CPAC.pipeline.engine.resource.ResourcePool` keys indicating + resources generated or updated by the `NodeBlock`, optionally including metadata for the outputs' respective sidecars. """ @@ -101,14 +106,14 @@ def __call__( pipe_num: Optional[int | str], opt: Optional[str] = None, ) -> tuple[Workflow, dict[str, "ResourceData"]]: - """Call a NodeBlockFunction. + """Call a `NodeBlockFunction`. - All node block functions have the same signature. + All `NodeBlockFunction`\u200bs have the same signature. """ return self.func(wf, cfg, strat_pool, pipe_num, opt) def legacy_nodeblock_dict(self): - """Return nodeblock metadata as a dictionary. + """Return :py:class:`NodeBlock` metadata as a dictionary. Helper for compatibility reasons. """ @@ -123,7 +128,7 @@ def legacy_nodeblock_dict(self): } def __repr__(self) -> str: - """Return reproducible string representation of a NodeBlockFunction.""" + """Return reproducible string representation of a `NodeBlockFunction`.""" return ( f"NodeBlockFunction({self.func.__module__}." f'{self.func.__name__}, "{self.name}", ' @@ -134,19 +139,19 @@ def __repr__(self) -> str: ) def __str__(self) -> str: - """Return string representation of a NodeBlockFunction.""" + """Return string representation of a `NodeBlockFunction`.""" return f"NodeBlockFunction({self.name})" class NodeBlock: - """A worflow subgraph composed of :py:class:`NodeBlockFunction`s.""" + """A :py:class:`Workflow` subgraph composed of :py:class:`NodeBlockFunction`\u200bs.""" def __init__( self, node_block_functions: NodeBlockFunction | PIPELINE_BLOCKS, debug: bool = False, ) -> None: - """Create a ``NodeBlock`` from a list of py:class:`~CPAC.pipeline.engine.nodeblock.NodeBlockFunction`s.""" + """Create a `NodeBlock` from a list of py:class:`NodeBlockFunction`\u200bs.""" if not isinstance(node_block_functions, list): node_block_functions = [node_block_functions] @@ -218,9 +223,12 @@ def __init__( logging.update_logging(config) def check_output(self, outputs: NODEBLOCK_OUTPUTS, label: str, name: str) -> None: - """Check if a label is listed in a NodeBlock's ``outputs``. + """Check if a label is listed in a `NodeBlock`\u200b's `outputs`. - Raises ``NameError`` if a mismatch is found. + Raises + ------ + NameError + If a mismatch is found. """ if label not in outputs: msg = ( @@ -234,13 +242,20 @@ def check_output(self, outputs: NODEBLOCK_OUTPUTS, label: str, name: str) -> Non def list_blocks( pipeline_blocks: PIPELINE_BLOCKS, indent: Optional[int] = None ) -> str: - """List node blocks line by line. + """List :py:class:`NodeBlockFunction`\u200bs line by line. Parameters ---------- - pipeline_blocks: list of :py:class:`NodeBlockFunction`s + pipeline_blocks + list of :py:class:`NodeBlockFunction`\u200bs - indent: number of spaces after a tab indent + indent + number of spaces after a tab indent + + Returns + ------- + str + formatted list of :py:class:`NodeBlockFunction`\u200bs """ blockstring = yaml.dump( [ @@ -277,26 +292,46 @@ def nodeblock( inputs: Optional[NODEBLOCK_INPUTS] = None, outputs: Optional[list[str] | dict[str, Any]] = None, ): - """ - Define a node block. + """Define a :py:class:`NodeBlockFunction`\u200b. - Connections to the pipeline configuration and to other node blocks. + Connections to the pipeline :py:class:`Configuration` and to other :py:class:`NodeBlockFunction`\u200bs. Parameters ---------- - name: Used in the graph and logging to identify the NodeBlock and its component nodes. Function's ``.__name__`` is used if ``name`` is not provided. - - config: Indicates the nested keys in a C-PAC pipeline configuration should configure a NodeBlock built from this function. If config is set to ``None``, then all other configuration-related entities must be specified from the root of the configuration. - - switch: Indicates any keys that should evaluate to True for this NodeBlock to be active. A list of lists of strings indicates multiple switches that must all be True to run, and is currently only an option if config is set to ``None``. - - option_key: Indicates the nested keys (starting at the nested key indicated by config) that should configure this NodeBlock. - - option_val: Indicates values for which this NodeBlock should be active. - - inputs: ResourcePool keys indicating files needed for the NodeBlock's functionality. - - outputs: ResourcePool keys indicating files generated or updated by the NodeBlock, optionally including metadata for the outputs' respective sidecars. + name + Used in the graph and logging to identify the :py:class:`NodeBlock` and its + component :py:class:`~nipype.pipeline.engine.Node`\u200bs. + The :py:class:`NodeBlockFunction`\u200b's `.__name__` is used if `name` is not + provided. + + config + Indicates the nested keys in a C-PAC pipeline :py:class:`Configuration` should + configure a :py:class:`NodeBlock` built from this + :py:class:`NodeBlockFunction`\u200b. If `config` is set to `None`, then all other + :py:class:`Configuration`\u200b-related entities must be specified from the root + of the :py:class:`Configuration`\u200b. + + switch + Indicates any keys that should evaluate to `True` for this :py:class:`NodeBlock` + to be active. A list of lists of strings indicates multiple switches that must + all be `True` to run, and is currently only an option if config is set to + `None`. + + option_key + Indicates the nested keys (starting at the nested key indicated by `config`) + that should configure this :py:class:`NodeBlock`\u200b. + + option_val + Indicates values for which this :py:class:`NodeBlock` should be active. + + inputs + ResourcePool keys indicating files needed for the :py:class:`NodeBlock`\u200b's + functionality. + + outputs + :py:class:`~CPAC.pipeline.engine.resource.ResourcePool` keys indicating files + generated or updated by the :py:class:`NodeBlock`, optionally including metadata + for the outputs' respective sidecars. """ return lambda func: NodeBlockFunction( func, diff --git a/CPAC/pipeline/engine/resource.py b/CPAC/pipeline/engine/resource.py index 6d46e14237..3e669444f2 100644 --- a/CPAC/pipeline/engine/resource.py +++ b/CPAC/pipeline/engine/resource.py @@ -14,7 +14,7 @@ # You should have received a copy of the GNU Lesser General Public # License along with C-PAC. If not, see . -"""Resources and ResourcePools for C-PAC.""" +""":py:class:`Resource`\u200bs and :py:class:`ResourcePool`\u200bs for C-PAC.""" import ast from collections.abc import KeysView @@ -30,13 +30,13 @@ from nipype.interfaces import utility as util # type: ignore [import-untyped] from nipype.interfaces.utility import Rename # type: ignore [import-untyped] +from nipype.pipeline import engine as pe from CPAC.image_utils.spatial_smoothing import spatial_smoothing from CPAC.image_utils.statistical_transforms import ( fisher_z_score_standardize, z_score_standardize, ) -from CPAC.pipeline import nipype_pipeline_engine as pe from CPAC.pipeline.check_outputs import ExpectedOutputs from CPAC.pipeline.engine.nodeblock import ( NodeBlock, @@ -92,7 +92,7 @@ class DataPaths: def __init__( self, *, data_paths: Optional[dict] = None, part_id: Optional[str] = "" ) -> None: - """Initialize a ``DataPaths`` instance.""" + """Initialize a `DataPaths` instance.""" if not data_paths: data_paths = {} if part_id and "part_id" in data_paths and part_id != data_paths["part_id"]: @@ -117,17 +117,17 @@ def __init__( self.derivatives_dir: Optional[str] = data_paths.get("derivatives_dir") def __repr__(self) -> str: - """Return reproducible string representation of ``DataPaths`` instance.""" + """Return reproducible string representation of `DataPaths` instance.""" return f"DataPaths(data_paths={self.as_dict()})" def __str__(self) -> str: - """Return string representation of a ``DataPaths`` instance.""" + """Return string representation of a `DataPaths` instance.""" return f"" def as_dict(self) -> dict: - """Return ``data_paths`` dictionary. + """Return a `data_paths` dictionary. - data_paths format:: + `data_paths` format:: {"anat": {"T1w": "{T1w path}", "T2w": "{T2w path}"}, "creds_path": {None OR path to credentials CSV}, @@ -178,7 +178,7 @@ def set_iterables( def strip_template(data_label: str) -> tuple[str, dict[str, str]]: - """Strip a template name from a data label to use as a Resource key.""" + """Strip a template name from a data label to use as a :py:class:`Resource` key.""" json = {} # rename to template for prefix in ["space-", "from-", "to-"]: @@ -199,21 +199,21 @@ def strip_template(data_label: str) -> tuple[str, dict[str, str]]: class ResourceData(NamedTuple): - """Attribute and tuple access for ResourceData.""" + """Attribute and tuple access for `ResourceData`.""" node: pe.Node - """Resource Node.""" + """Resource :py:class:`~pe.Node`.""" out: str """Output key.""" class Resource: - """A single Resource and its methods.""" + """A single `Resource` and its methods.""" def __init__(self, data: tuple[pe.Node, str], json: dict) -> None: - """Initialize a Resource.""" + """Initialize a `Resource`.""" self.data = ResourceData(*data) - """Tuple of source Node and output key.""" + """Tuple of source :py:class:`~pe.Node` and output key.""" self._json = json """Metadata.""" self._keys = {"data", "json"} @@ -224,7 +224,7 @@ def keys(self) -> list[str]: return list(self._keys) def __contains__(self, item: Any) -> bool: - """Return True if item in self.keys(), False otherwise.""" + """Return `True` if `item` in `self.keys()`, `False` otherwise.""" return item in self.keys() def __getitem__(self, name: str) -> Any: @@ -235,7 +235,7 @@ def __getitem__(self, name: str) -> Any: raise KeyError(msg) def __repr__(self) -> str: - """Return reproducible string for Resource.""" + """Return reproducible string for `Resource`.""" positional = f"Resource(data={self.data}, json={self.json}" kw = ", ".join( f"{key}={getattr(self, key)}" @@ -245,17 +245,17 @@ def __repr__(self) -> str: return f"{positional}{kw})" def __setitem__(self, name: str, value: Any) -> None: - """Provide legacy dict-style set access.""" + """Provide legacy dict-style set access for `Resource`.""" setattr(self, name, value) if name not in self.keys(): self._keys.add(name) def __str__(self) -> str: - """Return string representation of Resource.""" + """Return string representation of `Resource`.""" return f"{self.data[0]}" def get_json(self) -> dict[str | tuple, Any]: - """Return a deep copy of Resource JSON.""" + """Return a deep copy of `Resource` JSON.""" UTLOGGER.debug( "%s is a deep copy of the attached JSON. Assign it to a variable before modifying or the changes will be ephemeral.", self.__class__.__name__, @@ -263,14 +263,14 @@ def get_json(self) -> dict[str | tuple, Any]: return json.loads(json.dumps(self._json)) def set_json(self, value=dict) -> None: - """Update Resource JSON.""" + """Update `Resource` JSON.""" self._json.update(value) json = property(get_json, set_json, doc=get_json.__doc__) @property def cpac_provenance(self) -> list: - """Get CpacProvenance of a Resource.""" + """Get "CpacProvenance" of a `Resource`.""" return self.json["CpacProvenance"] @@ -278,7 +278,7 @@ class _Pool: """All Resources.""" def __init__(self) -> None: - """Initialize a ResourcePool or StratPool.""" + """Initialize a :py:class:`ResourcePool` or :py:class:`StratPool`\u200b.""" self.ants_interp: str self.cfg: Configuration self.creds_paths: Optional[str] @@ -305,7 +305,7 @@ def __init__(self) -> None: self.wf: pe.Workflow def __repr__(self) -> str: - """Return reproducible _Pool string.""" + """Return reproducible `_Pool` string.""" params = [ f"{param}={getattr(self, param)}" for param in ["rpool", "name", "cfg", "pipe_list"] @@ -314,7 +314,7 @@ def __repr__(self) -> str: return f'{self.__class__.__name__}({", ".join(params)})' def __str__(self) -> str: - """Return string representation of a _Pool.""" + """Return string representation of a `_Pool`.""" if self.name: return f"{self.__class__.__name__}({self.name}): {list(self.rpool)}" return f"{self.__class__.__name__}: {list(self.rpool)}" @@ -336,7 +336,7 @@ def generate_prov_string(prov: LIST_OF_LIST_OF_STR | tuple) -> tuple[str, str]: return (resource, str(prov)) def check_rpool(self, resource: list[str] | str) -> bool: - """Check if a resource is present in the _Pool.""" + """Check if a `resource` is present in the `_Pool`.""" if not isinstance(resource, list): resource = [resource] for name in resource: @@ -345,11 +345,11 @@ def check_rpool(self, resource: list[str] | str) -> bool: return False def keys(self) -> KeysView: - """Return rpool's keys.""" + """Return `rpool`'s keys.""" return self.rpool.keys() def __contains__(self, key) -> bool: - """Return True if key in Pool, False otherwise.""" + """Return `True` if key in `_Pool`, `False` otherwise.""" return key in self.keys() @staticmethod @@ -359,7 +359,7 @@ def get_resource_from_prov(prov: LIST_OF_LIST_OF_STR) -> Optional[str]: Each resource (i.e. "desc-cleaned_bold" AKA nuisance-regressed BOLD data) has its own provenance list. the name of the resource, and the node that produced it, is always the last item in the provenance - list, with the two separated by a colon : + list, with the two separated by a colon (`:`) """ if not len(prov): return None @@ -382,7 +382,7 @@ def set_data( fork: bool = False, inject: bool = False, ) -> None: - """Plug a Resource into a _Pool.""" + """Plug a :py:class:`Resource` into a `_Pool`.""" json_info = json_info.copy() cpac_prov: LIST_OF_LIST_OF_STR = [] if "CpacProvenance" in json_info: @@ -449,7 +449,7 @@ def get( Optional[Resource | STRAT_DICT | dict] | tuple[Optional[Resource | STRAT_DICT], Optional[str]] ): - """Return a dictionary of strats or a single Resource.""" + """Return a dictionary of strats or a single :py:class:`Resource`\u200b.""" if not isinstance(resource, list): resource = [resource] # if a list of potential inputs are given, pick the first one found @@ -483,7 +483,7 @@ def get( class ResourcePool(_Pool): - """A pool of Resources.""" + """A pool of :py:class:`Resource`\u200bs.""" from CPAC.pipeline.engine.nodeblock import ( NODEBLOCK_INPUTS, @@ -502,7 +502,7 @@ def __init__( pipeline_name: str = "", wf: Optional[pe.Workflow] = None, ) -> None: - """Initialize a ResourcePool.""" + """Initialize a `ResourcePool`.""" self.name = name super().__init__() if isinstance(data_paths, dict): @@ -617,9 +617,9 @@ def __init__( self.ingress_pipeconfig_paths() def back_propogate_template_name( - self, resource_idx: str, json_info: dict, id_string: "pe.Node" + self, resource_idx: str, json_info: dict, id_string: pe.Node ) -> None: - """Find and apply the template name from a resource's provenance.""" + """Find and apply the template name from a :py:class:`Resource`\u200b's provenance.""" if "template" in resource_idx and self.check_rpool("derivatives-dir"): if self.check_rpool("template"): node, out = self.get_data("template") @@ -1038,7 +1038,7 @@ def get_data( report_fetched=False, quick_single=False, ): - """Get ResourceData from ResourcePool.""" + """Get :py:class:`ResourceData` from `ResourcePool`.""" _resource = self.get(resource, pipe_idx=pipe_idx, report_fetched=report_fetched) if report_fetched: if pipe_idx: @@ -1053,7 +1053,7 @@ def get_data( return _resource.data def get_json(self, resource: str, strat: str | tuple) -> dict: - """Get JSON metadata from a Resource in a strategy.""" + """Get JSON metadata from a :py:class:`Resource` in a strategy.""" return self.get(resource, pipe_idx=strat).json def get_json_info(self, resource: str, key: str) -> Any: @@ -1077,7 +1077,7 @@ def get_raw_label(resource: str) -> str: def get_strats( # noqa: PLR0912,PLR0915 self, resources: NODEBLOCK_INPUTS, debug: bool = False ) -> dict[str | tuple, "StratPool"]: - """Get a dictionary of StratPools.""" + """Get a dictionary of :py:class:`StratPool`\u200bs.""" # TODO: NOTE: NOT COMPATIBLE WITH SUB-RPOOL/STRAT_POOLS # TODO: (and it doesn't have to be) import itertools @@ -1316,7 +1316,7 @@ def get_strats( # noqa: PLR0912,PLR0915 return new_strats def initialize_nipype_wf(self, name: str = "") -> None: - """Initialize a new nipype workflow.""" + """Initialize a new nipype :py:class:`~pe.Workflow`\u200b.""" if name: name = f"_{name}" workflow_name = f"cpac{name}_{self.unique_id}" @@ -1430,7 +1430,7 @@ def ingress_freesurfer(self) -> None: return def ingress_output_dir(self) -> None: - """Ingress an output directory into a ResourcePool.""" + """Ingress an output directory into a `ResourcePool`.""" dir_path = self.data_paths.derivatives_dir assert dir_path is not None WFLOGGER.info("\nPulling outputs from %s.\n", dir_path) @@ -1982,7 +1982,7 @@ def ingress_pipeconfig_paths(self): def create_func_datasource( self, rest_dict: dict, wf_name="func_datasource" ) -> pe.Workflow: - """Create a workflow to gather timeseries data. + """Create a :py:class:`~pe.Workflow` to gather timeseries data. Return the functional timeseries-related file paths for each series/scan from the dictionary of functional files described in the data configuration (sublist) YAML @@ -2345,7 +2345,7 @@ def ingress_raw_anat_data(self) -> None: self.ingress_freesurfer() def connect_block(self, wf: pe.Workflow, block: NodeBlock) -> pe.Workflow: # noqa: PLR0912,PLR0915 - """Connect a NodeBlock via the ResourcePool.""" + """Connect a :py:class:`NodeBlock` via the `ResourcePool`.""" debug = bool(self.cfg.pipeline_setup["Debugging"]["verbose"]) # type: ignore [attr-defined] all_opts: list[str] = [] @@ -2938,9 +2938,9 @@ def post_process( def get_resource_strats_from_prov(prov: list | str) -> dict[str, list | str]: """Return all entries that led to this provenance. - If you provide the provenance of a resource pool output, this will - return a dictionary of all the preceding resource pool entries that - led to that one specific output: + If you provide the provenance of a `ResourcePool` output, this will + return a dictionary of all the preceding `ResourcePool` entries that + led to that one specific output:: {rpool entry}: {that entry's provenance} {rpool entry}: {that entry's provenance} """ @@ -2961,18 +2961,21 @@ def get_resource_strats_from_prov(prov: list | str) -> dict[str, list | str]: def _config_lookup( self, keylist: str | list[str], fallback_type: type = NoneType ) -> Any: - """Lookup a config key, return None if not found.""" + """Lookup a :py:class:`Configuration` key, return `None` if not found.""" try: return self.cfg[keylist] except (AttributeError, KeyError): return fallback_type() def _get_pipe_number(self, pipe_idx: str | tuple) -> int: - """Return the index of a strategy in ``self.pipe_list``.""" + """Return the index of a strategy in `self.pipe_list`.""" return self.pipe_list.index(pipe_idx) def _get_unlabelled(self, resource: str) -> set[str]: - """Get unlabelled resources (that need integer suffixes to differentiate).""" + """Get unlabelled :py:class:`Resource`\u200bs. + + These :py:class:`Resource`\u200bs need integer suffixes to differentiate. + """ from CPAC.func_preproc.func_motion import motion_estimate_filter all_jsons = [ @@ -3019,7 +3022,7 @@ def _get_unlabelled(self, resource: str) -> set[str]: class StratPool(_Pool): - """A pool of ResourcePools keyed by strategy.""" + """A pool of :py:class:`ResourcePool`s keyed by strategy.""" def __init__( self, @@ -3028,7 +3031,7 @@ def __init__( rpool: Optional[dict] = None, name: str | list[str] = "", ) -> None: - """Initialize a StratPool.""" + """Initialize a `StratPool`.""" super().__init__() if not rpool: self.rpool = STRAT_DICT({}) @@ -3042,7 +3045,7 @@ def __init__( self._regressor_dct: dict def append_name(self, name: str) -> None: - """Append a name to the StratPool.""" + """Append a name to the `StratPool`.""" self.name.append(name) @overload @@ -3112,7 +3115,7 @@ def get( report_fetched: bool = False, optional: bool = False, ): - """Return a Resource.""" + """Return a :py:class:`Resource`\u200b.""" return super().get(resource, pipe_idx, report_fetched, optional) @overload @@ -3124,7 +3127,7 @@ def get_data( self, resource: list[str] | str, report_fetched: Literal[False] = False ) -> ResourceData: ... def get_data(self, resource, report_fetched=False): - """Get ResourceData from a StratPool.""" + """Get :py:class:`ResourceData` from a `StratPool`.""" _resource = self.get(resource, report_fetched=report_fetched) if report_fetched: assert isinstance(_resource, tuple) @@ -3135,17 +3138,17 @@ def get_data(self, resource, report_fetched=False): return _resource.data def get_json(self, resource: str) -> dict: - """Get JSON metadata from a Resource in a StratPool.""" + """Get JSON metadata from a :py:class:`Resource` in a `StratPool`.""" return self.get(resource).json json = property( fget=Resource.get_json, fset=Resource.set_json, - doc="""Return a deep copy of full-StratPool-strategy-specific JSON.""", + doc="""Return a deep copy of full-`StratPool`-strategy-specific JSON.""", ) def get_cpac_provenance(self, resource: list[str] | str) -> list: - """Get CpacProvenance for a given Resource.""" + """Get "CpacProvenance" for a given :py:class:`Resource`\u200b.""" # NOTE: strat_resource has to be entered properly by the developer # it has to either be rpool[resource][strat] or strat_pool[resource] if isinstance(resource, list): @@ -3157,7 +3160,7 @@ def get_cpac_provenance(self, resource: list[str] | str) -> list: return self.get(resource).cpac_provenance def copy_resource(self, resource: str, new_name: str): - """Copy a resource within a StratPool.""" + """Copy a :py:class:`Resource` within a `StratPool`.""" try: self.rpool[new_name] = self.rpool[resource] except KeyError: @@ -3168,7 +3171,7 @@ def filter_name(self, cfg: Configuration) -> str: """ Return the name of the filter for this strategy. - In a strat_pool with filtered movement parameters. + In a `StratPool` with filtered movement parameters. """ motion_filters = cfg[ "functional_preproc", @@ -3197,7 +3200,7 @@ def filter_name(self, cfg: Configuration) -> str: return "none" def preserve_json_info(self, resource: str, strat_resource: Resource) -> None: - """Preserve JSON info when updating a StratPool.""" + """Preserve JSON info when updating a `StratPool`.""" data_type = resource.split("_")[-1] if data_type not in self._json["subjson"]: self._json["subjson"][data_type] = {} @@ -3207,7 +3210,10 @@ def preserve_json_info(self, resource: str, strat_resource: Resource) -> None: def regressor_dct(self) -> dict: """Return the regressor dictionary for the current strategy if one exists. - Raises KeyError otherwise. + Raises + ------ + KeyError + If regressor dictionary does not exist in current strategy. """ # pylint: disable=attribute-defined-outside-init if hasattr(self, "_regressor_dct"): # memoized @@ -3243,7 +3249,7 @@ def regressor_dct(self) -> dict: @property def filtered_movement(self) -> bool: - """Check if the movement parameters have been filtered in this StratPool.""" + """Check if the movement parameters have been filtered in this `StratPool`.""" try: return "motion_estimate_filter" in str( self.get_cpac_provenance("desc-movementParameters_motion") @@ -3254,7 +3260,7 @@ def filtered_movement(self) -> bool: def _check_null(val: Any) -> Any: - """Return ``None`` if ``val`` == "none" (case-insensitive).""" + """Return `None` if `val` == "none" (case-insensitive).""" if isinstance(val, str): val = None if val.lower() == "none" else val return val diff --git a/CPAC/pipeline/schema.py b/CPAC/pipeline/schema.py index 915cb47045..8f9e2ffc58 100644 --- a/CPAC/pipeline/schema.py +++ b/CPAC/pipeline/schema.py @@ -63,18 +63,12 @@ Number = Any(float, int, All(str, Match(SCIENTIFIC_NOTATION_STR_REGEX))) -def str_to_bool1_1(x): # pylint: disable=invalid-name - """Convert strings to Booleans for YAML1.1 syntax. +def str_to_bool1_1(x: Any) -> bool: # pylint: disable=invalid-name + """Convert strings to Booleans for YAML1.1 syntax[1]_. - Ref https://yaml.org/type/bool.html - - Parameters + References ---------- - x : any - - Returns - ------- - bool + .. [1] 2005-01-18. Oren Ben-Kiki, Clark Evans & Brian Ingerson. `"Boolean Language-Independent Type for YAML™ Version 1.1" [Working Draft] `_. Copyright © 2001-2005 Oren Ben-Kiki, Clark Evans, Brian Ingerson. """ if isinstance(x, str): try: @@ -316,19 +310,9 @@ def str_to_bool1_1(x): # pylint: disable=invalid-name ) -def name_motion_filter(mfilter, mfilters=None): +def name_motion_filter(mfilter: dict, mfilters: Optional[list] = None) -> str: """Given a motion filter, create a short string for the filename. - Parameters - ---------- - mfilter : dict - - mfliters : list or None - - Returns - ------- - str - Examples -------- >>> name_motion_filter({'filter_type': 'notch', 'filter_order': 2, @@ -385,19 +369,8 @@ def name_motion_filter(mfilter, mfilters=None): return name -def permutation_message(key, options): - """Give a human-readable error message for keys that accept permutation values. - - Parameters - ---------- - key: str - - options: list or set - - Returns - ------- - msg: str - """ +def permutation_message(key: str, options: list | set) -> str: + """Give a human-readable error message for keys that accept permutation values.""" return f""" \'{key}\' takes a dictionary with paths to region-of-interest (ROI) @@ -412,7 +385,7 @@ def permutation_message(key, options): """ -def sanitize(filename): +def sanitize(filename: str) -> str: """Sanitize a filename and replace whitespaces with underscores.""" return re.sub(r"\s+", "_", sanitize_filename(filename)) @@ -1253,20 +1226,12 @@ def sanitize(filename): ) -def schema(config_dict): +def schema(config_dict: dict) -> dict: """Validate a participant-analysis pipeline configuration. Validate against the latest validation schema by first applying backwards- compatibility patches, then applying Voluptuous validation, then handling complex configuration interaction checks before returning validated config_dict. - - Parameters - ---------- - config_dict : dict - - Returns - ------- - dict """ from CPAC.utils.utils import _changes_1_8_0_to_1_8_1 diff --git a/CPAC/pipeline/test/test_engine.py b/CPAC/pipeline/test/test_engine.py index 01283042fb..e23741d2e8 100644 --- a/CPAC/pipeline/test/test_engine.py +++ b/CPAC/pipeline/test/test_engine.py @@ -32,21 +32,21 @@ def _set_up_test( bids_examples: Path, preconfig: str, tmp_path: Path ) -> tuple[Configuration, dict]: - """Set up ``cfg`` and ``sub_data`` for engine tests.""" + """Set up `cfg` and `sub_data` for engine tests.""" bids_dir = str(bids_examples / "ds051") sub_data = create_cpac_data_config(bids_dir, skip_bids_validator=True)[0] cfg = Preconfiguration(preconfig) cfg.pipeline_setup["output_directory"]["path"] = str(tmp_path / "out") cfg.pipeline_setup["working_directory"]["path"] = str(tmp_path / "work") cfg.pipeline_setup["log_directory"]["path"] = str(tmp_path / "logs") - return (cfg, sub_data) + return cfg, sub_data @pytest.mark.parametrize("preconfig", ["default"]) def test_ingress_func_raw_data( bids_examples: Path, preconfig: str, tmp_path: Path ) -> None: - """Test :py:method:~`CPAC.pipeline.engine.resource.ResourcePool.ingress_raw_func_data`.""" + """Test :py:method:`ResourcePool.ingress_raw_func_data`\u200b.""" cfg, sub_data_dct = _set_up_test(bids_examples, preconfig, tmp_path) rpool = ResourcePool(cfg=cfg, data_paths=sub_data_dct) rpool.gather_pipes(rpool.wf, cfg, all_types=True) @@ -56,7 +56,7 @@ def test_ingress_func_raw_data( def test_ingress_anat_raw_data( bids_examples: Path, preconfig: str, tmp_path: Path ) -> None: - """Test :py:method:~`CPAC.pipeline.engine.resource.ResourcePool.ingress_raw_anat_data`.""" + """Test :py:method:`ResourcePool.ingress_raw_anat_data`\u200b.""" cfg, sub_data_dct = _set_up_test(bids_examples, preconfig, tmp_path) rpool = ResourcePool( cfg=cfg, @@ -70,7 +70,7 @@ def test_ingress_anat_raw_data( def test_ingress_pipeconfig_data( bids_examples: Path, preconfig: str, tmp_path: Path ) -> None: - """Test :py:method:~`CPAC.pipeline.engine.resource.ResourcePool.ingress_pipeconfig_paths`.""" + """Test :py:method:`ResourcePool.ingress_pipeconfig_paths`\u200b.""" cfg, sub_data_dct = _set_up_test(bids_examples, preconfig, tmp_path) rpool = ResourcePool( cfg=cfg, @@ -83,7 +83,7 @@ def test_ingress_pipeconfig_data( def test_build_anat_preproc_stack( bids_examples: Path, preconfig: str, tmp_path: Path ) -> None: - """Test :py:func:~`CPAC.pipeline.cpac_pipeline.build_anat_preproc_stack`.""" + """Test :py:func:`~build_anat_preproc_stack`\u200b.""" cfg, sub_data_dct = _set_up_test(bids_examples, preconfig, tmp_path) rpool = ResourcePool(cfg=cfg, data_paths=sub_data_dct) @@ -94,7 +94,7 @@ def test_build_anat_preproc_stack( @pytest.mark.parametrize("preconfig", ["default"]) def test_build_workflow(bids_examples: Path, preconfig: str, tmp_path: Path) -> None: - """Test :py:func:~`CPAC.pipeline.cpac_pipeline.build_workflow`.""" + """Test :py:func:`~build_workflow`\u200b.""" cfg, sub_data_dct = _set_up_test(bids_examples, preconfig, tmp_path) rpool = ResourcePool(cfg=cfg, data_paths=sub_data_dct) wf = build_workflow(sub_data_dct["subject_id"], sub_data_dct, cfg) diff --git a/CPAC/pipeline/utils.py b/CPAC/pipeline/utils.py index 7b1dbaffff..6f6953fef2 100644 --- a/CPAC/pipeline/utils.py +++ b/CPAC/pipeline/utils.py @@ -19,27 +19,13 @@ from itertools import chain from CPAC.utils.bids_utils import insert_entity +from CPAC.utils.configuration.configuration import Configuration -def name_fork(resource_idx, cfg, json_info, out_dct): - """Create and insert entities for forkpoints. - - Parameters - ---------- - resource_idx : str - - cfg : CPAC.utils.configuration.Configuration - - json_info : dict - - out_dct : dict - - Returns - ------- - resource_idx : str - - out_dct : dict - """ +def name_fork( + resource_idx: str, cfg: Configuration, json_info: dict, out_dct: dict +) -> tuple[str, dict]: + """Create and insert entities for forkpoints.""" from CPAC.func_preproc.func_motion import motion_estimate_filter if cfg.switch_is_on( @@ -104,12 +90,6 @@ def present_outputs(outputs: dict, keys: list) -> dict: NodeBlocks that differ only by configuration options and relevant output keys. - Parameters - ---------- - outputs : dict - - keys : list of str - Returns ------- dict