Central, switchable logging across CLI, run(), and generated backends

tvbo/__init__.py

@@ -35,7 +35,23 @@
 tempdir = os.path.join(tempfile.gettempdir(), "tvbo")
 os.makedirs(tempdir, exist_ok=True)
 
-logging.disable(logging.CRITICAL)
+# Central logging. Importing tvbo installs only a NullHandler on the ``tvbo``
+# logger, so the package is silent as a library; entry points (``tvbo run``,
+# ``SimulationExperiment.run``) surface progress via ``configure_logging`` and
+# everything is controlled by one switch (``TVBO_LOG_LEVEL`` / ``set_log_level``).
+# This replaces a former process-wide ``logging.disable(CRITICAL)`` that muted
+# every logger — see :mod:`tvbo.log`.
+from tvbo.log import (  # noqa: E402,F401
+    configure_logging,
+    ensure_configured,
+    get_log_level,
+    log_level,
+    set_log_level,
+    silence,
+)
+
+if os.environ.get("TVBO_LOG_LEVEL"):
+    configure_logging()
 
 __authors__ = [
     "Leon K. Martin",

tvbo/adapters/julia.py

@@ -8,9 +8,12 @@
 will install Julia (if needed) and all declared packages.
 """
 
+import logging
 import re
 from typing import Any, Optional
 
+logger = logging.getLogger(__name__)
+
 _julia_main = None
 _installed_packages = set()
 
@@ -60,16 +63,16 @@ def install_julia_package(package_name: str, Main: Optional[Any] = None, update:
     if package_name in _installed_packages and not update:
         return
 
-    print(f"{'Updating' if update else 'Installing'} Julia package: {package_name}...")
+    logger.info("%s Julia package: %s...", "Updating" if update else "Installing", package_name)
     try:
         if update:
             Main.seval(f'import Pkg; Pkg.update("{package_name}")')
         else:
             Main.seval(f'import Pkg; Pkg.add("{package_name}")')
         _installed_packages.add(package_name)
-        print(f"Successfully {'updated' if update else 'installed'} {package_name}")
+        logger.info("Successfully %s %s", "updated" if update else "installed", package_name)
     except Exception as e:
-        print(f"Warning: Failed to {'update' if update else 'install'} {package_name}: {e}")
+        logger.warning("Failed to %s %s: %s", "update" if update else "install", package_name, e)
         raise
 
 
@@ -91,7 +94,7 @@ def eval_with_auto_install(code, max_retries=3):
                     # Retry after installing
                     continue
                 except Exception:
-                    print(f"Failed to auto-install {package_name}, re-raising original error")
+                    logger.warning("Failed to auto-install %s, re-raising original error", package_name)
                     raise e
             else:
                 # Not a package error or max retries reached

tvbo/classes/atlas.py

@@ -6,6 +6,7 @@
 produce ranked (relabelled) parcellation volumes from FreeSurfer segmentations.
 """
 
+import logging
 import os
 
 try:
@@ -63,6 +64,7 @@ def tqdm(x, **kwargs):
 ]
 
 available_atlases = bids_utils.get_unique_entity_values(atlas_data, "atlas")
+logger = logging.getLogger(__name__)
 
 
 class Atlas(tvbo_datamodel.BrainAtlas):
@@ -258,7 +260,7 @@ def centers(self):
             except ImportError:
                 centers = []
                 lookup_labels = []
-                print(
+                logger.warning(
                     "nilearn is required to compute atlas region centers. "
                     "Setting to empty. Install nilearn or provide atlas metadata."
                 )

tvbo/classes/dynamics.py

@@ -16,6 +16,7 @@ class that augments the generated LinkML `Dynamics` datamodel with model
 together with the `Model` and `Dynamics` convenience subclasses.
 """
 
+import logging
 import copy as _copy
 import functools
 import os
@@ -46,6 +47,8 @@ class that augments the generated LinkML `Dynamics` datamodel with model
 from tvbo.parse.expression import parse_eq
 from tvbo.utils import report
 
+logger = logging.getLogger(__name__)
+
 TEMPLATES = templates.root
 
 @functools.cache
@@ -425,7 +428,7 @@ def update_parameters(metadata, ontoclass, verbose=0, only_used=True, **kwargs):
             synonym in metadata.parameters for synonym in k.synonym + k.symbol
         ):
             if verbose > 0:
-                print(f"using parameter {label} from the ontology")
+                logger.debug("using parameter %s from the ontology", label)
             metadata.parameters.update(
                 {
                     label: tvbo_datamodel.Parameter(

tvbo/classes/equation.py

@@ -10,6 +10,7 @@
 # Handling Equations and Expressions
 """
 
+import logging
 import re
 from collections import deque
 
@@ -55,6 +56,7 @@
 coupling_variables = ["lrc", "short_range_coupling", "coupling", "lc_0", "c_0", "lc_1"]
 lambda_symbol = sp.symbols("lambda")
 E = sp.symbols("E")  # TODO: not used
+logger = logging.getLogger(__name__)
 
 
 def add_spaces_around_operators(expression):
@@ -262,8 +264,7 @@ def sympify_value(v, acronym="", evaluate=False):
             evaluate=False,
         )
     except Exception as e:
-        print(f"Error parsing equation: {eq}")
-        print(f"Error message: {e}")
+        logger.debug("Error parsing equation %r: %s", eq, e)
         raise ValueError(f"Failed to parse equation: {eq}. Ensure the equation is in a valid format.")
 
     return eq
@@ -808,7 +809,7 @@ def get_latex_equation(model, func_dict="all", mul_symbol="dot"):
                 if len(search) == 1:
                     c_rhs = search[0]
                 else:
-                    print(search)
+                    logger.debug("ambiguous search for %s in %s: %s", s, model, search)
                     raise ValueError(f"Could not find {s} in {model}")
 
             symb = c_rhs.symbol.first()
@@ -909,17 +910,17 @@ class and the equation's class in the ontology. Equations that are `None` or
     for k, v in symbolic_model_equations(model).items():
         k_cls = ontology.find_variables(k, model)
         if k_cls is None:
-            print('did not find "{}" in "{}"'.format(k, model))
+            logger.warning('did not find "%s" in "%s"', k, model)
             continue
 
         # Handle cases where v might be None (e.g., for discrete maps with conditionals)
         if v is None:
-            print(f'Skipping "{k}" in "{model}" - equation is None')
+            logger.debug('Skipping "%s" in "%s" - equation is None', k, model)
             continue
 
         # Check if v has free_symbols attribute (valid SymPy expression)
         if not hasattr(v, "free_symbols"):
-            print(f'Skipping "{k}" in "{model}" - not a valid SymPy expression')
+            logger.debug('Skipping "%s" in "%s" - not a valid SymPy expression', k, model)
             continue
 
         for symbol in v.free_symbols:

tvbo/classes/experiment.py

@@ -56,6 +56,9 @@
 from tvbo.utils import traverse_metadata
 from tvbo.utils import Bunch
 from tvbo.utils import as_list
+from tvbo.log import ensure_configured
+
+logger = logging.getLogger(__name__)
 
 sessionid = 1
 
@@ -1120,20 +1123,25 @@ def from_bids(
         # 6. Verify reproducibility if requested
         # =====================================================================
         if run_to_verify and timeseries is not None:
-            print("Running simulation to verify reproducibility...")
+            logger.info("Running simulation to verify reproducibility...")
             ts_rerun = experiment.run(format="jax")
 
             # Compare shapes
             if ts_rerun.shape != timeseries.shape:
-                print(f"WARNING: Shape mismatch! Loaded: {timeseries.shape}, Rerun: {ts_rerun.shape}")
+                logger.warning(
+                    "Shape mismatch! Loaded: %s, Rerun: %s", timeseries.shape, ts_rerun.shape
+                )
             else:
                 # Compare data
                 max_diff = np.max(np.abs(np.asarray(ts_rerun.data) - np.asarray(timeseries.data)))
                 if max_diff < 1e-6:
-                    print(f"✓ Verification passed! Max difference: {max_diff:.2e}")
+                    logger.info("Verification passed! Max difference: %.2e", max_diff)
                 else:
-                    print(f"⚠ Data differs. Max difference: {max_diff:.2e}")
-                    print("  This may be expected if noise was used or parameters differ.")
+                    logger.warning(
+                        "Data differs. Max difference: %.2e "
+                        "(may be expected if noise was used or parameters differ).",
+                        max_diff,
+                    )
 
         return experiment, timeseries
 
@@ -1279,8 +1287,7 @@ def noise_sigma_array(self) -> np.ndarray:
                 try:
                     sigma = float(sv.noise.parameters["sigma"].value)
                 except Exception as e:
-                    print(f"Error retrieving sigma for state variable {sv.name}: {e}")
-                    pass
+                    logger.debug("Error retrieving sigma for state variable %s: %s", sv.name, e)
 
             if sigma == 0.0:
                 try:
@@ -1295,8 +1302,7 @@ def noise_sigma_array(self) -> np.ndarray:
                     ):
                         sigma = float(inparams["sigma"].value)
                 except Exception as e:
-                    print(f"Error retrieving integration-level sigma: {e}")
-                    pass
+                    logger.debug("Error retrieving integration-level sigma: %s", e)
 
             sigmas.append(float(sigma))
 
@@ -1943,6 +1949,12 @@ def run(self, format="tvboptim", initial_conditions=None, results_root=None, **k
             An `ExperimentResult` holding the integrated time series and any
             observations, with `source` set to this experiment.
         """
+        # Make tvbo progress visible for interactive/script runs without
+        # clobbering an app's own logging — the same switch the CLI uses, so
+        # ``exp.run(...)`` and ``tvbo run`` log identically. No-op if the
+        # embedding application already configured logging.
+        ensure_configured()
+
         if "duration" in kwargs:
             self.integration.duration = kwargs.pop("duration")
 

tvbo/classes/observation.py

@@ -8,6 +8,7 @@
 callables and curated ontology instances into the underlying datamodel shape.
 """
 
+import logging
 import importlib
 import inspect
 from types import FunctionType
@@ -34,6 +35,7 @@
 from tvbo.codegen.code import render_expression
 from tvbo.ontology import owl as ontology
 from tvbo.plot.ontology import draw_custom_nodes
+logger = logging.getLogger(__name__)
 
 
 def expand_to_4d(array):
@@ -845,7 +847,7 @@ def apply(self, timeseries, mode=0):
                 if "Input" in self.graph.predecessors(node_label):
                     node.update({"data": self.current_data})
                 elif node["data"] is None:
-                    print("Node", node_label, "has no data")
+                    logger.warning("Node %s has no data", node_label)
                 continue
 
             time = (

tvbo/classes/perturbation.py

@@ -6,6 +6,7 @@
 convert ontology classes to metadata and to replay audio files as stimuli.
 """
 
+import logging
 import os
 
 import matplotlib.pyplot as plt
@@ -15,6 +16,7 @@
     import librosa
 except ImportError:
     librosa = None
+logger = logging.getLogger(__name__)
 
 
 def _require_librosa():
@@ -210,7 +212,7 @@ class directly.
             if not ontoclasses:
                 raise ValueError(f"No stimulus class found for label '{ontoclass}'")
             if len(ontoclasses) > 1:
-                print(f"Multiple stimulus classes found: {ontoclasses}")
+                logger.warning("Multiple stimulus classes found: %s", ontoclasses)
             ontoclass = ontoclasses[0]
         metadata = class2metadata(ontoclass)
         return cls(**metadata._as_dict)
@@ -313,7 +315,7 @@ def execute(
                 )
             if self.weighting:
                 weighting = np.array(self.weighting)
-                print(weighting)
+                logger.debug("stimulus weighting: %s", weighting)
             elif weighting is None and connectivity:
                 weighting = np.zeros(connectivity.number_of_regions)
                 weighting[region_indices] = 1

tvbo/cli/__init__.py

@@ -70,4 +70,36 @@
 app.add_typer(_skills_cmd.app, name="skills", help="Render skills for Claude Code / Copilot / Cursor; install user skills locally.")
 
 
+@app.callback()
+def _configure(
+    log_level: str = typer.Option(
+        None, "--log-level", "-L", metavar="LEVEL",
+        help="tvbo log level (DEBUG|INFO|WARNING|ERROR|OFF); overrides TVBO_LOG_LEVEL.",
+    ),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output (DEBUG)."),
+    quiet: bool = typer.Option(False, "--quiet", "-q", help="Errors only — suppress progress."),
+) -> None:
+    """Configure tvbo logging once for every verb.
+
+    Progress and status flow through the central ``tvbo`` logger (see
+    :mod:`tvbo.log`), so ``tvbo run`` and the in-process ``.run()`` API behave
+    identically. ``--log-level`` wins over ``--quiet``/``--verbose``; with none
+    set the level falls back to ``TVBO_LOG_LEVEL`` and then INFO.
+    """
+    from tvbo.log import configure_logging
+
+    level = None
+    if verbose:
+        level = "DEBUG"
+    if quiet:
+        level = "ERROR"
+    if log_level:
+        level = log_level
+    # CLI output is user-facing: keep it bare (no "LEVEL [name]" diagnostic
+    # prefix), matching the plain lines the CLI printed before. force=True so
+    # this format wins even if ``import tvbo`` already installed the default
+    # (diagnostic) handler because ``TVBO_LOG_LEVEL`` was set in the environment.
+    configure_logging(level, fmt="%(message)s", force=True)
+
+
 __all__ = ["app"]

tvbo/cli/_common.py

@@ -5,12 +5,15 @@
 from __future__ import annotations
 
 import json as _json
+import logging
 import sys
 from pathlib import Path
 from typing import Any
 
 import typer
 
+logger = logging.getLogger("tvbo.cli")
+
 
 # ---------------------------------------------------------------------------
 # SPEC resolution
@@ -181,10 +184,23 @@ def emit_json(payload: Any) -> None:
 
 
 def info(msg: str) -> None:
-    """Human-facing log line — always to stderr."""
-    typer.echo(msg, err=True)
+    """Human-facing progress line, routed through the central ``tvbo`` logger.
+
+    Emits at INFO on ``tvbo.cli`` (stderr by default), so ``--quiet`` /
+    ``TVBO_LOG_LEVEL`` govern it exactly as they govern in-process ``.run()``.
+    """
+    logger.info(msg)
 
 
 def die(msg: str, code: int = 1) -> None:
-    typer.echo(f"error: {msg}", err=True)
+    """Log *msg* at ERROR and abort the CLI with *code*.
+
+    A fatal abort must always explain itself: when the configured level would
+    suppress ERROR (e.g. ``--log-level OFF`` / ``TVBO_LOG_LEVEL=OFF``) the reason
+    still goes to stderr, so the CLI never exits non-zero in silence.
+    """
+    if logger.isEnabledFor(logging.ERROR):
+        logger.error(msg)
+    else:
+        typer.echo(f"error: {msg}", err=True)
     raise typer.Exit(code)