fix lint

Author: nastaran78

neps/plot/plotting.py

@@ -139,11 +139,11 @@ def _interpolate_time(
 
     if x_range is not None:
         min_b, max_b = x_range
-        new_entry = {c: np.nan for c in df.columns}
+        new_entry = dict.fromkeys(df.columns, np.nan)
         _df = pd.DataFrame.from_dict(new_entry, orient="index").T
         _df.index = [min_b]
         df = pd.concat((df, _df)).sort_index()
-        new_entry = {c: np.nan for c in df.columns}
+        new_entry = dict.fromkeys(df.columns, np.nan)
         _df = pd.DataFrame.from_dict(new_entry, orient="index").T
         _df.index = [max_b]
         df = pd.concat((df, _df)).sort_index()
@@ -157,7 +157,7 @@ def _interpolate_time(
 
 def _df_to_x_range(df: pd.DataFrame, x_range: tuple | None = None) -> pd.DataFrame:
     x_max = np.inf if x_range is None else int(x_range[-1])
-    new_entry = {c: np.nan for c in df.columns}
+    new_entry = dict.fromkeys(df.columns, np.nan)
     _df = pd.DataFrame.from_dict(new_entry, orient="index").T
     _df.index = [x_max]
     df = pd.concat((df, _df)).sort_index()

neps/plot/tensorboard_eval.py

@@ -1,20 +1,19 @@
-"""The tblogger module provides a simplified interface for logging NePS trials to TensorBoard."""
+"""The tblogger module provides a simplified interface for logging to TensorBoard."""
 
 from __future__ import annotations
 
 import logging
 import time
-from collections.abc import Mapping
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, ClassVar
+from typing import TYPE_CHECKING, ClassVar
 
 from torch.utils.tensorboard.writer import SummaryWriter
 
 from neps.runtime import (
     get_in_progress_trial,
     get_workers_neps_state,
-    register_notify_trial_end,
     is_in_progress_trial_set,
+    register_notify_trial_end,
 )
 from neps.status.status import status
 from neps.utils.common import get_initial_directory
@@ -40,9 +39,9 @@ class tblogger:  # noqa: N801
     @classmethod
     def initiate_internal_configurations(
         cls,
-        root_directory: Path | str | None = None,
-        pipeline_directory: Path | str | None = None,
-        previous_pipeline_directory: Path | str | None = None,
+        root_directory: Path | None = None,
+        pipeline_directory: Path | None = None,
+        previous_pipeline_directory: Path | None = None,
     ) -> None:
         """Initialize internal directories and configuration for TensorBoard logging.
 
@@ -54,13 +53,12 @@ def initiate_internal_configurations(
             pipeline_directory (Path | str | None): Current trial directory.
             previous_pipeline_directory (Path | str | None): Previous trial directory.
         """
-        if not is_in_progress_trial_set():
-            if not (root_directory and pipeline_directory):
-                raise RuntimeError(
-                    "Cannot determine directories for TensorBoard logging. "
-                    "Provide `root_directory`, `pipeline_directory`, and optionally "
-                    "`previous_pipeline_directory`."
-                )
+        if not is_in_progress_trial_set() and not (root_directory and pipeline_directory):
+            raise RuntimeError(
+                "Cannot determine directories for TensorBoard logging. "
+                "Provide `root_directory`, `pipeline_directory`, and optionally "
+                "`previous_pipeline_directory`."
+            )
 
         if is_in_progress_trial_set():
             trial = get_in_progress_trial()
@@ -73,18 +71,6 @@ def initiate_internal_configurations(
                 if trial.metadata.previous_trial_location
                 else None
             )
-        else:
-            # Convert str paths to Path objects
-            root_directory = Path(root_directory) if isinstance(root_directory, str) else root_directory
-            pipeline_directory = Path(pipeline_directory) if isinstance(pipeline_directory, str) else pipeline_directory
-            previous_pipeline_directory = (
-                Path(previous_pipeline_directory)
-                if isinstance(previous_pipeline_directory, str)
-                else previous_pipeline_directory
-            )
-
-            if previous_pipeline_directory and not previous_pipeline_directory.exists():
-                previous_pipeline_directory = None
 
         register_notify_trial_end("NEPS_TBLOGGER", cls.end_of_config)
 
@@ -93,13 +79,13 @@ def initiate_internal_configurations(
         cls.optimizer_dir = root_directory
 
     @classmethod
-    def write_incumbent(cls) -> None:
+    def WriteIncumbent(cls) -> None:  # noqa: N802
         """Enable logging of the incumbent (best) configuration for the current search."""
         cls.initiate_internal_configurations()
         cls.write_incumbent = True
 
     @classmethod
-    def config_writer(
+    def ConfigWriter(  # noqa: N802
         cls,
         *,
         write_summary_incumbent: bool = True,
@@ -120,14 +106,23 @@ def config_writer(
             or None if a writer cannot be initialized.
         """
         cls.write_incumbent = write_summary_incumbent
-        cls.initiate_internal_configurations(root_directory, pipeline_directory, previous_pipeline_directory)
-
-        if cls.config_previous_directory is None and cls.config_working_directory is not None:
+        cls.initiate_internal_configurations(
+            root_directory,
+            pipeline_directory,
+            previous_pipeline_directory,
+        )
+
+        if (
+            cls.config_previous_directory is None
+            and cls.config_working_directory is not None
+        ):
             cls.config_writer = SummaryWriter(cls.config_working_directory / "tbevents")
             return cls.config_writer
 
         if cls.config_working_directory is not None:
-            init_dir = get_initial_directory(pipeline_directory=cls.config_working_directory)
+            init_dir = get_initial_directory(
+                pipeline_directory=cls.config_working_directory,
+            )
             if (init_dir / "tbevents").exists():
                 cls.config_writer = SummaryWriter(init_dir / "tbevents")
                 return cls.config_writer
@@ -139,7 +134,7 @@ def config_writer(
         return None
 
     @classmethod
-    def end_of_config(cls, trial: Trial) -> None:  # noqa: ARG004
+    def end_of_config(cls, _: Trial) -> None:
         """Close the TensorBoard writer at the end of a configuration."""
         if cls.config_writer:
             cls.config_writer.close()
@@ -176,7 +171,7 @@ def _tracking_incumbent_api(cls) -> None:
             cls.summary_writer.close()
             time.sleep(0.5)
 
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             logger.warning(
                 "Incumbent tracking for TensorBoard failed and is now disabled: %s", e
             )

neps/space/domain.py

@@ -135,7 +135,7 @@ def __post_init__(self) -> None:
 
         mid = self.from_unit(torch.tensor(0.5)).item()
         if is_int:
-            mid = int(round(mid))
+            mid = round(mid)
 
         object.__setattr__(self, "bounds", (self.lower, self.upper))
 
@@ -193,8 +193,8 @@ def integer(
             A domain for a range of integer values.
         """
         return Domain(
-            lower=int(round(lower)),
-            upper=int(round(upper)),
+            lower=round(lower),
+            upper=round(upper),
             log_bounds=(math.log(lower), math.log(upper)) if log else None,
             round=True,
             bins=bins,

neps_examples/convenience/async_evaluation/run_pipeline.py

@@ -42,7 +42,7 @@
             batch_size=1024,
         )
 
-        writer = tblogger.config_writer(write_summary_incumbent=True, 
+        writer = tblogger.ConfigWriter(write_summary_incumbent=True, 
                                        root_directory=args.root_directory,
                                        pipeline_directory=args.pipeline_directory,
                                        previous_pipeline_directory=args.previous_pipeline_directory)

neps_examples/convenience/neps_tblogger_tutorial.py

@@ -253,7 +253,7 @@ def evaluate_pipeline(lr, optim, weight_decay):
 
     # Substitute the Tensorboard SummaryWriter with ConfigWriter from NePS
     # writer = SummaryWriter()
-    writer = tblogger.config_writer(write_summary_incumbent=True)
+    writer = tblogger.ConfigWriter(write_summary_incumbent=True)
 
     for i in range(max_epochs):
         objective_to_minimize = training(