Move a helper function to prevent circular dependency (#120)

* Move a helper function to prevent circular dependency

* perun/logic: Add missing typing hints

* Fix typing issue in new mypy version.

* Update codecov

---------

Co-authored-by: JiriPavela <[email protected]>
Co-authored-by: Tomas Fiedor <[email protected]>

codecov.yml

@@ -2,4 +2,5 @@ ignore:
   - "perun/collect/trace"
   - "perun/thirdparty"
   - "tests/test_tracer"
+  - "tests/sources"
 

perun/collect/trace/optimizations/optimization.py

@@ -3,8 +3,7 @@
 
 import collections
 
-from perun.profile.helpers import sanitize_filepart
-from perun.utils.helpers import SuppressedExceptions
+from perun.utils.helpers import (SuppressedExceptions, sanitize_filepart)
 from perun.collect.trace.optimizations.structs import Optimizations, Pipeline, Parameters, \
     CallGraphTypes, ParametersManager, CGShapingMode
 import perun.collect.trace.optimizations.resources.manager as resources

perun/logic/runner.py

@@ -6,7 +6,7 @@
 import signal
 import types
 
-from typing import Any, Iterable, Optional, TYPE_CHECKING, cast, Callable
+from typing import Any, Iterable, Optional, TYPE_CHECKING, cast, Callable, Type
 
 import distutils.util as dutils
 
@@ -28,6 +28,7 @@
 
 if TYPE_CHECKING:
     from perun.profile.factory import Profile
+    from perun.workload.generator import WorkloadGenerator
 
 from perun.utils import get_module
 from perun.utils.structs import GeneratorSpec, Unit, Executable, RunnerReport, \
@@ -459,7 +460,7 @@ def run_prephase_commands(phase: str, phase_colour: str = 'white') -> None:
 @decorators.phase_function('batch job run')
 def generate_jobs_on_current_working_dir(
         job_matrix: dict[str, dict[str, list[Job]]], number_of_jobs: int
-) -> Iterable[tuple[CollectStatus, dict[str, Any], Job]]:
+) -> Iterable[tuple[CollectStatus, Profile, Job]]:
     """Runs the batch of jobs on current state of the VCS.
 
     This function expects no changes not commited in the repo, it excepts correct version
@@ -469,7 +470,7 @@ def generate_jobs_on_current_working_dir(
     :param int number_of_jobs: number of jobs that will be run
     :return: status, generated profile, and associated job
     """
-    workload_generators_specs = workloads.load_generator_specifications()
+    workload_generators_specs: dict[str, GeneratorSpec] = workloads.load_generator_specifications()
 
     log.print_job_progress.current_job = 1
     collective_status = CollectStatus.OK
@@ -479,6 +480,8 @@ def generate_jobs_on_current_working_dir(
         for workload, jobs_per_workload in workloads_per_cmd.items():
             log.print_current_phase(" = processing generator {}", workload, COLLECT_PHASE_WORKLOAD)
             # Prepare the specification
+            generator: Type[WorkloadGenerator]
+            params: dict[str, Any]
             generator, params = workload_generators_specs.get(
                 workload, GeneratorSpec(SingletonGenerator, {'value': workload})
             )
@@ -510,7 +513,7 @@ def generate_jobs_on_current_working_dir(
 @decorators.phase_function('overall profiling')
 def generate_jobs(
         minor_version_list: list[MinorVersion], job_matrix: dict[str, dict[str, list[Job]]], number_of_jobs: int
-) -> Iterable[tuple[CollectStatus, dict[str, Any], Job]]:
+) -> Iterable[tuple[CollectStatus, Profile, Job]]:
     """
     :param list minor_version_list: list of MinorVersion info
     :param dict job_matrix: dictionary with jobs that will be run

perun/profile/helpers.py

@@ -38,6 +38,7 @@
 from perun.utils.exceptions import (
     InvalidParameterException, MissingConfigSectionException, TagOutOfRangeException
 )
+from perun.utils.helpers import sanitize_filepart
 from perun.utils.structs import Unit, Executable, Job
 
 
@@ -58,16 +59,6 @@ def lookup_value(container: dict[str, str] | profiles.Profile, key: str, missing
     return str(container.get(key, missing)) or missing
 
 
-def sanitize_filepart(part: str) -> str:
-    """Helper function for sanitization of part of the filenames
-
-    :param part: part of the filename, that needs to be sanitized, i.e. we are removing invalid characters
-    :return: sanitized string representation of the part
-    """
-    invalid_characters = r"# %&{}\<>*?/ $!'\":@"
-    return "".join('_' if c in invalid_characters else c for c in str(part))
-
-
 def lookup_param(profile: profiles.Profile, unit: str, param: str) -> str:
     """Helper function for looking up the unit in the profile (can be either collector or
     postprocessor and finds the value of the param in it

perun/utils/helpers.py

@@ -438,3 +438,13 @@ def safe_match(pattern: re.Pattern[str], searched_string: str, default: str) ->
     """
     match = pattern.search(searched_string)
     return match.group() if match else default
+
+
+def sanitize_filepart(part: str) -> str:
+    """Helper function for sanitization of part of the filenames
+
+    :param part: part of the filename, that needs to be sanitized, i.e. we are removing invalid characters
+    :return: sanitized string representation of the part
+    """
+    invalid_characters = r"# %&{}\<>*?/ $!'\":@"
+    return "".join("_" if c in invalid_characters else c for c in str(part))

perun/view/scatter/run.py

@@ -5,7 +5,7 @@
 
 from typing import Any
 
-import perun.profile.helpers as profiles
+from perun.utils.helpers import sanitize_filepart
 import perun.view.scatter.factory as scatter_factory
 from perun.profile.factory import pass_profile, Profile
 from perun.utils import cli_helpers, view_helpers
@@ -105,5 +105,5 @@ def scatter(profile: Profile, filename: str, view_in_browser: bool, **kwargs: An
     # Temporary solution for plotting multiple graphs from one command
     graphs = scatter_factory.create_from_params(profile, **kwargs)
     for uid, graph in graphs:
-        filename_uid = f"{filename}_{profiles.sanitize_filepart(uid)}.html"
+        filename_uid = f"{filename}_{sanitize_filepart(uid)}.html"
         view_helpers.save_view_graph(graph, filename_uid, view_in_browser)