Source code for pclean
"""pclean — Parallel CLEAN imaging with Dask and CASA tools.
Provides a ``pclean()`` function with a tclean-compatible interface and
transparent Dask-based parallelism for cube (channel) and continuum
(visibility-row) imaging.
"""
import logging
import time
from typing import Literal
from pclean.pclean import pclean
try:
from pclean._version import version as __version__
except ModuleNotFoundError: # editable install without build
__version__ = '0.0.0.dev0'
__all__ = ['pclean', 'CustomFormatter']
[docs]
class CustomFormatter(logging.Formatter):
"""Custom logging formatter that defaults to CASA-style output and UTC time.
Format: YYYY-MM-DD HH:MM:SS LEVEL LoggerName Message
"""
# Enforce UTC time conversion at the class level
converter = time.gmtime
def __init__(self, fmt: str | None = None, datefmt: str | None = None, style: Literal['%', '{', '$'] = '%') -> None:
"""Initialize with optional format/datefmt, falling back to defaults.
Both arguments default to the strings used in :mod:`pclean` if not
supplied. The caller can override either or both selections; the
formatter still enforces UTC timing via the class-level ``converter``
attribute.
"""
# Set the default format strings if none are explicitly provided
if fmt is None:
fmt = '%(asctime)s %(levelname)-7s %(name)s %(message)s'
if datefmt is None:
datefmt = '%Y-%m-%d %H:%M:%S'
super().__init__(fmt=fmt, datefmt=datefmt, style=style)
# Configure a default stream handler so that log messages from pclean
# (and its submodules) are visible without extra user setup.
def _configure_default_logging() -> None:
_log = logging.getLogger(__name__)
if _log.handlers:
return # already configured by the caller; don't override
_handler = logging.StreamHandler()
_handler.setFormatter(CustomFormatter())
_log.addHandler(_handler)
_log.setLevel(logging.INFO) # ensure INFO is visible by default
_configure_default_logging()