adamops.utils package

Submodules

adamops.utils.config module

AdamOps Configuration Module

Provides centralized configuration management for the entire library. Supports YAML, JSON, and environment variable configurations.

class adamops.utils.config.AdamOpsConfig(data: DataConfig = <factory>, model: ModelConfig = <factory>, automl: AutoMLConfig = <factory>, deployment: DeploymentConfig = <factory>, monitoring: MonitoringConfig = <factory>, logging: LoggingConfig = <factory>, registry_backend: str = 'json', registry_path: str = '.adamops_registry', cache_enabled: bool = True, cache_path: str = '.adamops_cache')[source]

Bases: object

Main configuration class for AdamOps.

automl: AutoMLConfig

cache_enabled: bool = True

cache_path: str = '.adamops_cache'

data: DataConfig

deployment: DeploymentConfig

logging: LoggingConfig

model: ModelConfig

monitoring: MonitoringConfig

registry_backend: str = 'json'

registry_path: str = '.adamops_registry'

class adamops.utils.config.AutoMLConfig(time_limit: int = 3600, max_trials: int = 100, tuning_method: str = 'bayesian', optimization_metric: str = 'auto', early_stopping: bool = True)[source]

Bases: object

Configuration for AutoML module.

early_stopping: bool = True

max_trials: int = 100

optimization_metric: str = 'auto'

time_limit: int = 3600

tuning_method: str = 'bayesian'

class adamops.utils.config.DataConfig(default_encoding: str = 'utf-8', missing_threshold: float = 0.5, outlier_method: str = 'iqr', outlier_threshold: float = 1.5, validation_sample_size: int = 10000, auto_detect_types: bool = True)[source]

Bases: object

Configuration for data module.

auto_detect_types: bool = True

default_encoding: str = 'utf-8'

missing_threshold: float = 0.5

outlier_method: str = 'iqr'

outlier_threshold: float = 1.5

validation_sample_size: int = 10000

class adamops.utils.config.DeploymentConfig(default_port: int = 8000, default_host: str = '0.0.0.0', api_framework: str = 'fastapi', enable_cors: bool = True, log_requests: bool = True)[source]

Bases: object

Configuration for deployment module.

api_framework: str = 'fastapi'

default_host: str = '0.0.0.0'

default_port: int = 8000

enable_cors: bool = True

log_requests: bool = True

class adamops.utils.config.LoggingConfig(level: str = 'INFO', format: str = '%(asctime)s - %(name)s - %(levelname)s - %(message)s', file: str | None = None, console: bool = True, max_bytes: int = 10485760, backup_count: int = 5)[source]

Bases: object

Configuration for logging.

backup_count: int = 5

console: bool = True

file: str | None = None

format: str = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'

level: str = 'INFO'

max_bytes: int = 10485760

class adamops.utils.config.ModelConfig(default_random_state: int = 42, cv_folds: int = 5, early_stopping_rounds: int = 50, n_jobs: int = -1, verbose: int = 0)[source]

Bases: object

Configuration for model module.

cv_folds: int = 5

default_random_state: int = 42

early_stopping_rounds: int = 50

n_jobs: int = -1

verbose: int = 0

class adamops.utils.config.MonitoringConfig(drift_threshold: float = 0.05, alert_email: str | None = None, check_interval: int = 3600, log_predictions: bool = True)[source]

Bases: object

Configuration for monitoring module.

alert_email: str | None = None

check_interval: int = 3600

drift_threshold: float = 0.05

log_predictions: bool = True

adamops.utils.config.get_config() → AdamOpsConfig[source]

Get the global configuration instance.

Returns:: The global configuration object.
Return type:: AdamOpsConfig

Example

>>> config = get_config()
>>> print(config.model.cv_folds)
5

adamops.utils.config.load_config_from_env(prefix: str = 'ADAMOPS') → AdamOpsConfig[source]

Load configuration from environment variables.

Environment variables should be named as {prefix}_{SECTION}_{KEY}. For example: ADAMOPS_MODEL_CV_FOLDS=10

Parameters:: prefix – Prefix for environment variables.
Returns:: Configuration with values from environment.
Return type:: AdamOpsConfig

Example

>>> # Set env: ADAMOPS_MODEL_CV_FOLDS=10
>>> config = load_config_from_env()
>>> print(config.model.cv_folds)
10

adamops.utils.config.load_config_from_file(filepath: str | Path) → AdamOpsConfig[source]

Load configuration from a YAML or JSON file.

Parameters:

filepath – Path to the configuration file.

Returns:

Loaded configuration object.

Return type:

AdamOpsConfig

Raises:

FileNotFoundError – If the file doesn’t exist.
ValueError – If the file format is not supported.

Example

>>> config = load_config_from_file("config.yaml")

adamops.utils.config.reset_config() → None[source]

Reset the global configuration to defaults.

Example

>>> reset_config()
>>> config = get_config()
>>> print(config.model.cv_folds)
5

adamops.utils.config.save_config_to_file(config: AdamOpsConfig, filepath: str | Path) → None[source]

Save configuration to a YAML or JSON file.

Parameters:

config – Configuration object to save.
filepath – Path to save the configuration to.

Example

>>> config = get_config()
>>> save_config_to_file(config, "config.yaml")

adamops.utils.config.set_config(config: AdamOpsConfig) → None[source]

Set the global configuration instance.

Parameters:: config – The configuration object to set as global.

Example

>>> custom_config = AdamOpsConfig()
>>> custom_config.model.cv_folds = 10
>>> set_config(custom_config)

adamops.utils.config.update_config(**kwargs) → AdamOpsConfig[source]

Update specific configuration values.

Parameters:: **kwargs – Configuration values in format section__key=value.
Returns:: Updated configuration object.
Return type:: AdamOpsConfig

Example

>>> config = update_config(model__cv_folds=10, automl__time_limit=7200)
>>> print(config.model.cv_folds)
10

adamops.utils.helpers module

AdamOps Helpers Module

Provides common utility functions used across the library.

class adamops.utils.helpers.ProgressTracker(total: int, description: str = 'Progress')[source]

Bases: object

Simple progress tracker for long-running operations.

Example

>>> tracker = ProgressTracker(total=100)
>>> for i in range(100):
...     tracker.update()
>>> tracker.finish()

finish() → None[source]: Mark progress as complete.

update(n: int = 1) → None[source]: Update progress by n steps.

adamops.utils.helpers.deprecated(message: str = '', version: str = '') → Callable[source]

Decorator to mark a function as deprecated.

Parameters:

message – Deprecation message.
version – Version when the function will be removed.

Returns:

Decorator function.

adamops.utils.helpers.ensure_dir(path: str | Path) → Path[source]

Ensure a directory exists, creating it if necessary.

Parameters:: path – Path to directory.
Returns:: Path object.
Return type:: Path

adamops.utils.helpers.get_file_hash(filepath: str | Path, algorithm: str = 'md5') → str[source]

Calculate hash of a file.

Parameters:

filepath – Path to file.
algorithm – Hash algorithm (md5, sha1, sha256).

Returns:

Hex digest of file hash.

Return type:

str

adamops.utils.helpers.get_memory_usage(obj: Any) → int[source]

Get memory usage of an object in bytes.

Parameters:: obj – Object to measure.
Returns:: Memory usage in bytes.
Return type:: int

adamops.utils.helpers.infer_task_type(y: ndarray | Series) → str[source]

Infer the task type from the target variable.

Parameters:: y – Target variable.
Returns:: Task type (“classification”, “regression”, or “multiclass”).
Return type:: str

Example

>>> y = np.array([0, 1, 0, 1])
>>> infer_task_type(y)
'classification'

adamops.utils.helpers.is_categorical(series: Series, threshold: float = 0.05) → bool[source]

Check if a pandas Series is likely categorical.

Parameters:

series – Pandas Series to check.
threshold – Ratio of unique values to total values.

Returns:

True if series is likely categorical.

Return type:

bool

Example

>>> df = pd.DataFrame({"cat": ["a", "b", "a", "b"]})
>>> is_categorical(df["cat"])
True

adamops.utils.helpers.is_numeric(value: Any) → bool[source]

Check if a value is numeric.

Parameters:: value – Value to check.
Returns:: True if value is numeric.
Return type:: bool

Example

>>> is_numeric(42)
True
>>> is_numeric("hello")
False

adamops.utils.helpers.load_object(filepath: str | Path, format: str = 'pickle') → Any[source]

Load an object from file.

Parameters:

filepath – Path to load from.
format – Load format (pickle, json, joblib).

Returns:

Loaded object.

adamops.utils.helpers.now_str(format: str = '%Y%m%d_%H%M%S') → str[source]

Get current timestamp as formatted string.

Parameters:: format – Datetime format string.
Returns:: Formatted timestamp.
Return type:: str

adamops.utils.helpers.parse_timestamp(timestamp: str | int | float | datetime) → datetime[source]

Parse various timestamp formats to datetime.

Parameters:: timestamp – Timestamp to parse.
Returns:: Parsed datetime object.
Return type:: datetime

adamops.utils.helpers.reduce_memory_usage(df: DataFrame, verbose: bool = False) → DataFrame[source]

Reduce memory usage of a DataFrame by downcasting numeric types.

Parameters:

df – DataFrame to optimize.
verbose – Whether to print memory savings.

Returns:

Optimized DataFrame.

Return type:

pd.DataFrame

adamops.utils.helpers.retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0, exceptions: ~typing.Tuple[~typing.Type[Exception], ...] = (<class 'Exception'>,)) → Callable[source]

Decorator to retry a function on failure.

Parameters:

max_attempts – Maximum retry attempts.
delay – Initial delay between retries.
backoff – Multiplier for delay after each retry.
exceptions – Exceptions to catch and retry.

Returns:

Decorator function.

Example

>>> @retry(max_attempts=3, delay=1.0)
... def unstable_function():
...     # May fail sometimes
...     pass

adamops.utils.helpers.safe_cast(value: Any, target_type: Type[T], default: T | None = None) → T | None[source]

Safely cast a value to a target type.

Parameters:

value – Value to cast.
target_type – Target type.
default – Default value if casting fails.

Returns:

Cast value or default.

Example

>>> safe_cast("42", int)
42
>>> safe_cast("hello", int, default=0)
0

adamops.utils.helpers.save_object(obj: Any, filepath: str | Path, format: str = 'pickle') → None[source]

Save an object to file.

Parameters:

obj – Object to save.
filepath – Path to save to.
format – Save format (pickle, json, joblib).

adamops.utils.helpers.set_random_seed(seed: int = 42) → None[source]

Set random seed for reproducibility.

Parameters:: seed – Random seed value.

adamops.utils.helpers.slugify(text: str) → str[source]

Convert text to URL-friendly slug.

Parameters:: text – Text to convert.
Returns:: Slugified text.
Return type:: str

Example

>>> slugify("Hello World!")
'hello-world'

adamops.utils.helpers.timeit(func: Callable) → Callable[source]

Decorator to time function execution.

Parameters:: func – Function to time.
Returns:: Wrapped function.

Example

>>> @timeit
... def slow_function():
...     time.sleep(1)

Convert data to pandas DataFrame.

Parameters:

data – Data to convert.
columns – Optional column names.

Returns:

Pandas DataFrame.

Return type:

pd.DataFrame

adamops.utils.helpers.to_numpy(data: ndarray | DataFrame | Series | list) → ndarray[source]

Convert data to numpy array.

Parameters:: data – Data to convert.
Returns:: Numpy array.
Return type:: np.ndarray

adamops.utils.helpers.truncate(text: str, max_length: int = 100, suffix: str = '...') → str[source]

Truncate text to maximum length.

Parameters:

text – Text to truncate.
max_length – Maximum length.
suffix – Suffix to add if truncated.

Returns:

Truncated text.

Return type:

str

adamops.utils.helpers.validate_args(**validators: Callable[[Any], bool]) → Callable[source]

Decorator to validate function arguments.

Parameters:: **validators – Mapping of argument names to validation functions.
Returns:: Decorator function.

Example

>>> @validate_args(x=lambda x: x > 0)
... def process(x):
...     return x * 2

adamops.utils.helpers.validate_dataframe(df: DataFrame, required_columns: List[str] | None = None) → bool[source]

Validate a pandas DataFrame.

Parameters:

df – DataFrame to validate.
required_columns – List of required column names.

Returns:

True if valid.

Return type:

bool

Raises:

ValueError – If validation fails.

adamops.utils.logging module

AdamOps Logging Module

Provides centralized logging functionality for the entire library. Supports console and file logging with configurable levels and formats.

class adamops.utils.logging.AdamOpsLogger(name: str, level: str = 'INFO', log_file: str | None = None, console: bool = True, use_colors: bool = True, format_string: str | None = None, max_bytes: int = 10485760, backup_count: int = 5)[source]

Bases: object

Custom logger class for AdamOps.

Provides a unified logging interface with features like: - Console and file logging - Colored output - Automatic log rotation - Context managers for temporary log level changes

Example

>>> logger = AdamOpsLogger("my_module")
>>> logger.info("This is an info message")
>>> logger.debug("This is a debug message")

Initialize the logger.

Parameters:

name – Logger name (usually module name).
level – Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL).
log_file – Optional path to log file.
console – Whether to log to console.
use_colors – Whether to use colored output.
format_string – Custom format string.
max_bytes – Maximum log file size before rotation.
backup_count – Number of backup files to keep.

critical(msg: str, *args, **kwargs) → None[source]: Log a CRITICAL level message.

debug(msg: str, *args, **kwargs) → None[source]: Log a DEBUG level message.

error(msg: str, *args, **kwargs) → None[source]: Log an ERROR level message.

exception(msg: str, *args, **kwargs) → None[source]: Log an exception with traceback.

info(msg: str, *args, **kwargs) → None[source]: Log an INFO level message.

log(level: int, msg: str, *args, **kwargs) → None[source]: Log a message at the specified level.

set_level(level: str) → None[source]: Set the log level.

trace(msg: str, *args, **kwargs) → None[source]: Log a TRACE level message.

warn(msg: str, *args, **kwargs) → None[source]: Alias for warning.

warning(msg: str, *args, **kwargs) → None[source]: Log a WARNING level message.

class adamops.utils.logging.ColoredFormatter(fmt: str | None = None, datefmt: str | None = None, use_colors: bool = True)[source]

Bases: Formatter

Colored formatter for console output.

Adds color codes to log messages based on log level.

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{’ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

BOLD = '\x1b[1m'

COLORS = {'CRITICAL': '\x1b[35m', 'DEBUG': '\x1b[36m', 'ERROR': '\x1b[31m', 'INFO': '\x1b[32m', 'TRACE': '\x1b[37m', 'WARNING': '\x1b[33m'}

RESET = '\x1b[0m'

format(record: LogRecord) → str[source]: Format log record with optional colors.

class adamops.utils.logging.LogContext(logger: AdamOpsLogger | Logger, level: str)[source]

Bases: object

Context manager for temporary log level changes.

Example

>>> logger = get_logger("my_module")
>>> with LogContext(logger, "DEBUG"):
...     logger.debug("This will be logged")
>>> logger.debug("This might not be logged")

Initialize context manager.

Parameters:

logger – Logger to modify.
level – Temporary log level.

class adamops.utils.logging.Timer(operation: str, logger: AdamOpsLogger | None = None, level: str = 'INFO')[source]

Bases: object

Context manager for timing operations with optional logging.

Example

>>> logger = get_logger("my_module")
>>> with Timer("Data loading", logger):
...     load_data()
[INFO] Data loading completed in 2.34s

Initialize timer.

Parameters:

operation – Name of the operation being timed.
logger – Optional logger for timing output.
level – Log level for timing message.

property elapsed: float: Get elapsed time in seconds.

end_time: datetime | None

start_time: datetime | None

adamops.utils.logging.get_logger(name: str = 'adamops', level: str | None = None, log_file: str | None = None, console: bool = True, use_colors: bool = True) → AdamOpsLogger[source]

Get or create a logger with the specified name.

Parameters:

name – Logger name.
level – Log level (default: INFO).
log_file – Optional path to log file.
console – Whether to log to console.
use_colors – Whether to use colored output.

Returns:

Logger instance.

Return type:

AdamOpsLogger

Example

>>> logger = get_logger("my_module")
>>> logger.info("Processing data...")

adamops.utils.logging.log_function_call(logger: AdamOpsLogger | None = None, level: str = 'DEBUG')[source]

Decorator to log function calls and their results.

Parameters:

logger – Logger to use (creates one if not provided).
level – Log level for function call messages.

Example

>>> @log_function_call()
... def process_data(df):
...     return df.dropna()

adamops.utils.logging.setup_logging(level: str = 'INFO', log_file: str | None = None, console: bool = True, use_colors: bool = True, format_string: str | None = None) → None[source]

Set up global logging configuration for AdamOps.

Parameters:

level – Global log level.
log_file – Path to log file.
console – Whether to log to console.
use_colors – Whether to use colored output.
format_string – Custom format string.

Example

>>> setup_logging(level="DEBUG", log_file="adamops.log")

Module contents

AdamOps Utils Module

Provides utility functions: - config: Configuration management - logging: Centralized logging - helpers: Common helper functions