xiaoxue_admin
/
supervision
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: replace print statements with structured logging (#2154) 

* feat: implement logging system to replace print statements
* refactor: rename get_logger to _get_logger to mark as internal
* test: add logger tests, fix corrupted-file warning test, add docstring example

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: Borda <6035284+Borda@users.noreply.github.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Copilot 2026-02-20 13:14:53 +01:00 committed by GitHub
parent 6ca9555efd
commit c2dedba3db
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 192 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/supervision/annotators/core.py
View File

@ -45,6 +45,9 @@ from supervision.utils.image import (
overlay_image,
scale_image,
)
from supervision.utils.logger import _get_logger
logger = _get_logger(__name__)
@overload
@ -1724,7 +1727,9 @@ class RichLabelAnnotator(_BaseLabelAnnotator):
try:
return ImageFont.truetype(font_path, font_size)
except OSError:
print(f"Font path '{font_path}' not found. Using PIL's default font.")
logger.warning(
"Font path '%s' not found. Using PIL's default font.", font_path
)
return load_default_font(font_size)

9
src/supervision/assets/downloader.py
View File

@ -9,6 +9,9 @@ from requests import get
from tqdm.auto import tqdm
from supervision.assets.list import MEDIA_ASSETS, Assets
from supervision.utils.logger import _get_logger
logger = _get_logger(__name__)
def is_md5_hash_matching(filename: str, original_md5_hash: str) -> bool:
@ -61,7 +64,7 @@ def download_assets(asset_name: Assets | str) -> str:
if filename in MEDIA_ASSETS:
if not Path(filename).exists():
print(f"Downloading {filename} assets \n")
logger.info("Downloading %s assets", filename)
response = get(
MEDIA_ASSETS[filename][0], stream=True, allow_redirects=True, timeout=30
)
@ -78,11 +81,11 @@ def download_assets(asset_name: Assets | str) -> str:
copyfileobj(raw_resp, file)
else:
if not is_md5_hash_matching(filename, MEDIA_ASSETS[filename][1]):
print("File corrupted. Re-downloading... \n")
logger.warning("File corrupted. Re-downloading...")
os.remove(filename)
return download_assets(filename)
print(f"{filename} asset download complete. \n")
logger.info("%s asset download complete.", filename)
else:
valid_assets = ", ".join(filename for filename in MEDIA_ASSETS.keys())
raise ValueError(

10
src/supervision/detection/tools/csv_sink.py
View File

@ -5,6 +5,9 @@ import os
from typing import Any
from supervision.detection.core import Detections
from supervision.utils.logger import _get_logger
logger = _get_logger(__name__)
BASE_HEADER = [
"x_min",
@ -168,9 +171,10 @@ class CSVSink:
self.header_written = True
if field_names != self.field_names:
print(
f"Field names do not match the header. "
f"Expected: {self.field_names}, given: {field_names}"
logger.warning(
"Field names do not match the header. Expected: %s, given: %s",
self.field_names,
field_names,
)
parsed_rows = CSVSink.parse_detection_data(detections, custom_data)

7
src/supervision/metrics/mean_average_precision.py
View File

@ -18,6 +18,9 @@ from supervision.detection.utils.iou_and_nms import box_iou_batch_with_jaccard
from supervision.draw.color import LEGACY_COLOR_PALETTE
from supervision.metrics.core import Metric, MetricTarget
from supervision.metrics.utils.utils import ensure_pandas_installed
from supervision.utils.logger import _get_logger
logger = _get_logger(__name__)
if TYPE_CHECKING:
import pandas as pd
@ -1098,7 +1101,9 @@ class COCOEvaluator:
mean_s = -1.0
else:
mean_s = float(np.mean(s[s > -1]))
print(iStr.format(titleStr, typeStr, iou_str, area_range, max_dets, mean_s))
logger.info(
iStr.format(titleStr, typeStr, iou_str, area_range, max_dets, mean_s)
)
return mean_s
def _summarize_predictions() -> npt.NDArray[np.float64]:

60
src/supervision/utils/logger.py Normal file
View File

@ -0,0 +1,60 @@
from __future__ import annotations
import logging
import os
import sys
def _get_logger(name: str = "supervision", level: int | None = None) -> logging.Logger:
"""Creates and configures a logger with stdout and stderr handlers.
This function creates a logger that sends INFO and DEBUG level logs to stdout,
and WARNING, ERROR, and CRITICAL level logs to stderr. If the logger already
has handlers, it returns the existing logger without adding new handlers.
The log level can be specified directly or through the `LOG_LEVEL` environment
variable.
Args:
name: The name of the logger. Defaults to `"supervision"`.
level: The logging level to set. If `None`, uses the `LOG_LEVEL` environment
variable, defaulting to `INFO` if not set.
Returns:
A configured `logging.Logger` instance.
Example:
```python
from supervision.utils.logger import _get_logger
logger = _get_logger(__name__)
logger.info("Processing started")
logger.warning("File not found, using default")
```
"""
if level is None:
level = getattr(logging, os.getenv("LOG_LEVEL", "INFO").upper(), logging.INFO)
logger = logging.getLogger(name)
logger.setLevel(level)
if not logger.handlers:
formatter = logging.Formatter(
"[%(asctime)s] [%(levelname)s] %(name)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setLevel(logging.DEBUG)
stdout_handler.addFilter(lambda r: r.levelno <= logging.INFO)
stdout_handler.setFormatter(formatter)
stderr_handler = logging.StreamHandler(sys.stderr)
stderr_handler.setLevel(logging.WARNING)
stderr_handler.setFormatter(formatter)
logger.addHandler(stdout_handler)
logger.addHandler(stderr_handler)
logger.propagate = False
return logger

6
src/supervision/utils/video.py
View File

@ -13,6 +13,10 @@ import numpy as np
import numpy.typing as npt
from tqdm.auto import tqdm
from supervision.utils.logger import _get_logger
logger = _get_logger(__name__)
@dataclass
class VideoInfo:
@ -97,7 +101,7 @@ class VideoSink:
try:
self.__fourcc = cv2.VideoWriter_fourcc(*self.__codec)
except TypeError as e:
print(str(e) + ". Defaulting to mp4v...")
logger.warning("%s. Defaulting to mp4v...", str(e))
self.__fourcc = cv2.VideoWriter_fourcc(*"mp4v")
self.__writer = cv2.VideoWriter(
self.target_path,

34
tests/assets/test_downloader.py
View File

@ -36,30 +36,34 @@ class TestMD5HashMatching:
class TestDownloadAssets:
@patch("builtins.print")
@patch("supervision.assets.downloader.logger")
@patch("supervision.assets.downloader.is_md5_hash_matching", return_value=True)
@patch("pathlib.Path.exists", return_value=True)
def test_already_exists_and_valid(self, mock_exists, mock_md5, mock_print):
def test_already_exists_and_valid(self, mock_exists, mock_md5, mock_logger):
"""Test download_assets when file already exists and is valid."""
filename = "vehicles.mp4"
result = download_assets(filename)
assert result == filename
mock_print.assert_called_with(f"{filename} asset download complete. \n")
mock_logger.info.assert_called_with("%s asset download complete.", filename)
@patch("supervision.assets.downloader.download_assets", return_value="vehicles.mp4")
@patch("supervision.assets.downloader.logger")
@patch("os.remove")
@patch("supervision.assets.downloader.is_md5_hash_matching", return_value=False)
@patch(
"supervision.assets.downloader.is_md5_hash_matching",
side_effect=[False, True],
)
@patch("pathlib.Path.exists", return_value=True)
def test_already_exists_but_corrupted(
self, mock_exists, mock_md5, mock_remove, mock_download
self, mock_exists, mock_md5, mock_remove, mock_logger
):
"""Test download_assets when file exists but is corrupted (re-downloads)."""
filename = "vehicles.mp4"
result = download_assets(filename)
assert result == filename
mock_download.assert_called_with(filename)
mock_logger.warning.assert_called_once_with("File corrupted. Re-downloading...")
mock_remove.assert_called_once_with(filename)
@patch("builtins.print")
@patch("supervision.assets.downloader.logger")
@patch("pathlib.Path.open", new_callable=mock_open)
@patch("pathlib.Path.mkdir")
@patch("pathlib.Path.exists", return_value=False)
@ -74,7 +78,7 @@ class TestDownloadAssets:
mock_exists,
mock_mkdir,
mock_open_file,
mock_print,
mock_logger,
):
"""Test download_assets downloading a new file."""
filename = "vehicles.mp4"
@ -92,7 +96,7 @@ class TestDownloadAssets:
result = download_assets(filename)
assert result == filename
mock_print.assert_called_with(f"Downloading {filename} assets \n")
mock_logger.info.assert_called_with("Downloading %s assets", filename)
mock_get.assert_called_once()
mock_response.raise_for_status.assert_called_once_with()
mock_copyfileobj.assert_called_once()
@ -119,7 +123,7 @@ class TestDownloadAssets:
assert "Invalid asset" in str(exc_info.value)
assert "vehicles.mp4" in str(exc_info.value)
@patch("builtins.print")
@patch("supervision.assets.downloader.logger")
@patch("pathlib.Path.open", new_callable=mock_open)
@patch("pathlib.Path.mkdir")
@patch("supervision.assets.downloader.copyfileobj")
@ -134,7 +138,7 @@ class TestDownloadAssets:
mock_copyfileobj,
mock_mkdir,
mock_open_file,
mock_print,
mock_logger,
):
"""Test download_assets with VideoAssets enum."""
asset = VideoAssets.VEHICLES
@ -150,8 +154,9 @@ class TestDownloadAssets:
result = download_assets(asset)
assert result == asset.filename
mock_logger.info.assert_called_with("Downloading %s assets", asset.filename)
@patch("builtins.print")
@patch("supervision.assets.downloader.logger")
@patch("pathlib.Path.open", new_callable=mock_open)
@patch("pathlib.Path.mkdir")
@patch("supervision.assets.downloader.copyfileobj")
@ -166,7 +171,7 @@ class TestDownloadAssets:
mock_copyfileobj,
mock_mkdir,
mock_open_file,
mock_print,
mock_logger,
):
"""Test download_assets with ImageAssets enum."""
asset = ImageAssets.SOCCER
@ -182,3 +187,4 @@ class TestDownloadAssets:
result = download_assets(asset)
assert result == asset.filename
mock_logger.info.assert_called_with("Downloading %s assets", asset.filename)

82
tests/utils/test_logger.py Normal file
View File

@ -0,0 +1,82 @@
from __future__ import annotations
import logging
import sys
from unittest.mock import patch
from supervision.utils.logger import _get_logger
class TestGetLogger:
def test_default_name(self):
"""Logger is created with default name."""
logger = _get_logger()
assert logger.name == "supervision"
def test_custom_name(self):
"""Logger is created with provided name."""
logger = _get_logger("supervision.test_module")
assert logger.name == "supervision.test_module"
def test_default_level_is_info(self):
"""Logger defaults to INFO level when LOG_LEVEL env var is not set."""
with patch.dict("os.environ", {}, clear=True):
# Use a unique name to avoid cached logger state from other tests
logger = _get_logger("supervision.test_default_level")
assert logger.level == logging.INFO
def test_explicit_level(self):
"""Logger uses the explicitly provided level."""
logger = _get_logger("supervision.test_explicit_level", level=logging.DEBUG)
assert logger.level == logging.DEBUG
def test_log_level_env_var(self):
"""Logger respects the LOG_LEVEL environment variable."""
with patch.dict("os.environ", {"LOG_LEVEL": "DEBUG"}):
logger = _get_logger("supervision.test_env_level")
assert logger.level == logging.DEBUG
def test_log_level_env_var_warning(self):
"""Logger respects the LOG_LEVEL=WARNING environment variable."""
with patch.dict("os.environ", {"LOG_LEVEL": "WARNING"}):
logger = _get_logger("supervision.test_env_warning")
assert logger.level == logging.WARNING
def test_two_handlers_configured(self):
"""Logger has exactly two handlers: one for stdout, one for stderr."""
logger = _get_logger("supervision.test_handlers")
assert len(logger.handlers) == 2
def test_stdout_handler_present(self):
"""Logger has a StreamHandler pointing to stdout."""
logger = _get_logger("supervision.test_stdout")
stdout_handlers = [
h
for h in logger.handlers
if isinstance(h, logging.StreamHandler) and h.stream is sys.stdout
]
assert len(stdout_handlers) == 1
def test_stderr_handler_present(self):
"""Logger has a StreamHandler pointing to stderr for warnings."""
logger = _get_logger("supervision.test_stderr")
stderr_handlers = [
h
for h in logger.handlers
if isinstance(h, logging.StreamHandler) and h.stream is sys.stderr
]
assert len(stderr_handlers) == 1
def test_no_propagation(self):
"""Logger does not propagate to the root logger."""
logger = _get_logger("supervision.test_propagation")
assert not logger.propagate
def test_idempotent_no_duplicate_handlers(self):
"""Calling _get_logger twice with the same name does not add extra handlers."""
name = "supervision.test_idempotent"
logger1 = _get_logger(name)
handler_count = len(logger1.handlers)
logger2 = _get_logger(name)
assert len(logger2.handlers) == handler_count
assert logger1 is logger2