umpyre#

Umpyre - Code analysis and quality metrics tracking.

Functions

`modules_info_gen`(root[, filepath_filt, on_error])	Yields statistics (as dicts) of modules under the root module or directory of a python package.
`modules_info_df`(root[, filepath_filt, ...])	A pandas DataFrame of stats of the root (package or directory thereof).
`modules_info_df_stats`(root[, filepath_filt, ...])	A pandas Series of statistics over all modules of some root (package or directory thereof).
`stats_of`(modules[, filepath_filt, ...])	A dataframe of stats of the input modules.
`get_objs`(root, k)
`serialize_metrics`(data, output_path[, format])	Serialize metrics to file.
`deserialize_metrics`(file_path[, format])	Deserialize metrics from file.

Classes

`Config`([config_path, config_dict])	Configuration manager for umpyre metrics collection.
`MetricSchema`([version])	Versioned schema for metrics storage with migration support.
`MetricCollector`()	Abstract base for metric collectors using Mapping interface.
`CollectorRegistry`()	Registry for managing available collectors.
`UmpyreCollector`([root_path, exclude_dirs])	Collect code statistics using AST parsing (no code execution).
`WorkflowStatusCollector`(repo[, token, ...])	Collect GitHub workflow status using GitHub API.
`WilyCollector`([repo_path, max_revisions, ...])	Collect complexity metrics using wily.
`CoverageCollector`([repo_path, source, ...])	Collect test coverage metrics from coverage reports.
`GitBranchStorage`(repo_path[, remote, token])	Store metrics in a separate git branch.

class umpyre.CollectorRegistry[source]#

Bases: object

Registry for managing available collectors.

get(name)[source]#

Get collector class by name.

Return type:: type[MetricCollector]

list_collectors()[source]#

List all registered collector names.

Return type:: list[str]

register(name, collector_class)[source]#

Parameters:

name (str) – Collector identifier
collector_class (type[MetricCollector]) – MetricCollector subclass

class umpyre.Config(config_path=None, config_dict=None)[source]#

Bases: object

Configuration manager for umpyre metrics collection.

collector_config(collector_name)[source]#

Get configuration for a specific collector.

Return type:: dict

classmethod from_dict(config)[source]#

Create config from dictionary.

Return type:: Config

classmethod from_file(path)[source]#

Load config from YAML file.

Return type:: Config

get(*keys, default=None)[source]#

Get nested config value.

Parameters:

*keys (str) – Nested keys (e.g., ‘collectors’, ‘wily’, ‘enabled’)
default (Any) – Default value if key not found

Return type:

Any

Returns:

Config value or default

>>> config = Config(config_dict={'collectors': {'wily': {'enabled': True}}})
>>> config.get('collectors', 'wily', 'enabled')
True
>>> config.get('collectors', 'missing', default='default_value')
'default_value'

is_collector_enabled(collector_name)[source]#

Check if a collector is enabled.

Return type:: bool

property retention_strategy: str#: Get data retention strategy.

property storage_branch: str#: Get the storage branch name.

property storage_formats: list[str]#: Get enabled storage formats.

to_dict()[source]#

Export config as dictionary.

Return type:: dict

class umpyre.CoverageCollector(repo_path=None, source='pytest-cov', coverage_file=None)[source]#

Bases: MetricCollector

Collect test coverage metrics from coverage reports.

Supports:

coverage.py JSON reports
coverage.py XML reports (Cobertura format)

Tracks:

Line coverage percentage
Branch coverage percentage (if available)
Statements covered/total

Example

>>> # This requires coverage report files
>>> # Skip in doctest as it requires actual test runs
>>> True

collect()[source]#

Collect coverage metrics from report files.

Returns:

line_coverage: Line coverage percentage
branch_coverage: Branch coverage percentage (may be 0 if not tracked)
statements_covered: Number of statements covered
statements_total: Total number of statements

Return type:

dict

class umpyre.GitBranchStorage(repo_path, remote='origin', *, token=None)[source]#

Bases: object

Store metrics in a separate git branch.

This provides versioned, branch-based storage without polluting the main branch or requiring external artifacts.

Example

>>> storage = GitBranchStorage(repo_path="/path/to/repo")
>>> metrics = {"lines": 100, "functions": 10}
>>> storage.store_metrics(metrics, "abc123", branch="code-metrics")

store_metrics(metrics, commit_sha, branch='code-metrics', formats=None)[source]#

Store metrics to git branch.

Parameters:

metrics (dict) – Metric dictionary to store
commit_sha (str) – Git commit SHA
branch (str) – Target branch name
formats (Optional[list[str]]) – List of formats (‘json’, ‘csv’)

class umpyre.MetricCollector[source]#

Bases: Mapping, ABC

Abstract base for metric collectors using Mapping interface.

Collectors implement the Mapping protocol to provide dict-like access to collected metrics, enabling flexible composition and iteration.

Example

>>> class SimpleCollector(MetricCollector):
...     def collect(self) -> dict:
...         return {'lines': 100, 'functions': 10}
>>> collector = SimpleCollector()
>>> collector['lines']
100
>>> list(collector)
['lines', 'functions']

abstractmethod collect()[source]#

Collect metrics and return as dictionary.

Return type:: dict
Returns:: Dictionary of metric names to values

refresh()[source]#: Force re-collection of metrics.

to_dict()[source]#

Export all metrics as dictionary.

Return type:: dict

class umpyre.MetricSchema(version='1.0')[source]#

Bases: object

Versioned schema for metrics storage with migration support.

classmethod create_metric_data(commit_sha, metrics, commit_message=None, python_version=None, workflow_status=None, collection_duration=None)[source]#

Create a standardized metric data structure.

Parameters:

commit_sha (str) – Git commit SHA
metrics (dict) – Dictionary of collected metrics
commit_message (Optional[str]) – Optional commit message
python_version (Optional[str]) – Optional Python version string
workflow_status (Optional[dict]) – Optional workflow status dict
collection_duration (Optional[float]) – Optional collection time in seconds

Return type:

dict

Returns:

Standardized metric dictionary

classmethod current_version()[source]#

Return the current schema version.

Return type:: str

classmethod migrate(data, from_version)[source]#

Migrate data from old schema version to current version.

Parameters:

data (dict) – Metric data in old format
from_version (str) – Version of the input data

Return type:

dict

Returns:

Migrated data in current schema format

classmethod validate(data)[source]#

Validate that data conforms to schema.

Parameters:: data (dict) – Metric data to validate
Return type:: bool
Returns:: True if valid, raises ValueError otherwise

class umpyre.UmpyreCollector(root_path=None, exclude_dirs=None)[source]#

Bases: MetricCollector

Collect code statistics using AST parsing (no code execution).

Provides metrics like:

Number of functions and classes
Lines of code (total, empty, comments, docs)
Code ratios (comment/empty/function lines)
Mean lines per function

Uses AST parsing instead of dynamic imports for safety.

Example

>>> import tempfile
>>> from pathlib import Path
>>> # Create a simple Python file for testing
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     test_file = Path(tmpdir) / "test.py"
...     test_file.write_text('''def hello():
...     """Say hello."""
...     return "hello"
...
... class Greeter:
...     """A greeter."""
...     pass
... ''')
...     collector = UmpyreCollector(tmpdir)
...     metrics = collector.collect()
...     metrics['num_functions'] >= 1 and metrics['num_classes'] >= 1
True

collect()[source]#

Collect code statistics from root directory using AST parsing.

Returns:

num_functions: Total number of functions
num_classes: Total number of classes
total_lines: Total lines of code
empty_lines: Number of empty lines
comment_lines: Number of comment lines
docs_lines: Number of documentation lines
function_lines: Lines in functions
empty_lines_ratio: Ratio of empty lines
comment_lines_ratio: Ratio of comment lines
function_lines_ratio: Ratio of function lines
mean_lines_per_function: Average lines per function
files_analyzed: Number of files successfully analyzed

Return type:

dict

class umpyre.WilyCollector(repo_path=None, max_revisions=5, operators=None)[source]#

Bases: MetricCollector

Collect complexity metrics using wily.

Tracks:

Cyclomatic complexity (average)
Maintainability index
Total lines of code
Files analyzed

Note

Limited to recent commits (max_revisions) for performance.

Example

>>> # This requires wily to be installed and a git repo
>>> # Skip in doctest as it requires git setup
>>> True

collect()[source]#

Collect complexity metrics using wily.

Returns:

cyclomatic_avg: Average cyclomatic complexity
maintainability_index: Maintainability index score
total_loc: Total lines of code
files_analyzed: Number of files analyzed

Return type:

dict

class umpyre.WorkflowStatusCollector(repo, token=None, lookback_runs=10, workflow_name=None)[source]#

Bases: MetricCollector

Collect GitHub workflow status using GitHub API.

Tracks:

Last workflow run status (success/failure)
Last successful run timestamp
Recent failure count
Workflow run URL

Example

>>> # Mock test (requires actual GitHub API)
>>> import os
>>> if os.getenv('GITHUB_TOKEN'):
...     collector = WorkflowStatusCollector(
...         repo="thorwhalen/astate",
...         token=os.getenv('GITHUB_TOKEN')
...     )
...     metrics = collector.collect()
...     'last_run_status' in metrics
... else:
...     True  # Skip if no token
True

collect()[source]#

Collect workflow status from GitHub API.

Returns:

last_run_status: ‘success’, ‘failure’, or ‘other’
last_success_timestamp: ISO timestamp or None
recent_failure_count: Number of failures in lookback window
run_url: URL of last run
total_runs_analyzed: Number of runs examined

Return type:

dict

umpyre.deserialize_metrics(file_path, format='json')[source]#

Deserialize metrics from file.

Parameters:

file_path (Path) – Path to input file
format (str) – Format (‘json’ or ‘csv’)

Return type:

dict

Returns:

Metric dictionary

umpyre.modules_info_df(root, filepath_filt=<function <lambda>>, index_field=None, on_error='ignore')[source]#

A pandas DataFrame of stats of the root (package or directory thereof).

Parameters:

root – module or directory path
filepath_filt – filepath filter function or regular expression
index_field – function or field string that should be used for the indexing of modules
on_error – What to do when an error occurs when extracting information from a module. Values are ‘ignore’, ‘print’, or ‘raise’

Returns:

A DataFrame whose rows contain information for each module

>>> import urllib
>>> df = modules_info_df(urllib)
>>> df
                    lines  empty_lines  ...  num_of_functions  num_of_classes
urllib.error           78           18  ...                 0               3
urllib.request       2774          410  ...                23              28
urllib.__init__         1            1  ...                 0               0
urllib.response        81           24  ...                 0               4
urllib.robotparser    274           36  ...                 0               4
urllib.parse         1166          199  ...                46              16

[6 rows x 7 columns]

umpyre.modules_info_df_stats(root, filepath_filt=<function <lambda>>, index_field=None, on_error='ignore')[source]#

A pandas Series of statistics over all modules of some root (package or directory thereof).

Parameters:

root – module or directory path
filepath_filt – filepath filter function or regular expression
index_field – function or field string that should be used for the indexing of modules
on_error – What to do when an error occurs when extracting information from a module. Values are ‘ignore’, ‘print’, or ‘raise’

Returns:

A Series whose rows containing statistics

>>> import json
>>> df = modules_info_df_stats(json)
>>> df
lines                      1301.000000
empty_lines                 178.000000
comment_lines                50.000000
docs_lines                  218.000000
function_lines              585.000000
num_of_functions             14.000000
num_of_classes                3.000000
empty_lines_ratio             0.136818
comment_lines_ratio           0.038432
function_lines_ratio          0.449654
mean_lines_per_function      41.785714
dtype: float64
>>> modules_info_df_stats('collections.abc')
lines                      1276.000000
empty_lines                 190.000000
comment_lines                73.000000
docs_lines                  133.000000
function_lines              138.000000
num_of_functions              1.000000
num_of_classes               34.000000
empty_lines_ratio             0.148903
comment_lines_ratio           0.057210
function_lines_ratio          0.108150
mean_lines_per_function     138.000000
dtype: float64

umpyre.modules_info_gen(root, filepath_filt=<function <lambda>>, on_error='ignore')[source]#

Yields statistics (as dicts) of modules under the root module or directory of a python package.

Parameters:

root – module instance, dot path, or directory path
filepath_filt – filepath filter function or regular expression
on_error – What to do when an error occurs when extracting information from a module. Values are ‘ignore’, ‘print’, ‘raise’, or ‘yield’

Returns:

A generator of dicts

umpyre.serialize_metrics(data, output_path, format='json')[source]#

Serialize metrics to file.

Parameters:

data (dict) – Metric dictionary
output_path (Path) – Path to output file
format (str) – Format (‘json’ or ‘csv’)

umpyre.stats_of(modules, filepath_filt=<function <lambda>>, index_field=None, on_error='ignore')[source]#

A dataframe of stats of the input modules.

Parameters:

modules – list of importable names
root – module or directory path
filepath_filt – filepath filter function or regular expression
index_field – function or field string that should be used for the indexing of modules
on_error – What to do when an error occurs when extracting information from a module. Values are ‘ignore’, ‘print’, or ‘raise’

Returns:

>>> df = stats_of(['urllib', 'json', 'collections'])
>>> assert set(df.index.values).issuperset(
...     {'empty_lines_ratio', 'comment_lines_ratio', 'function_lines_ratio', 'mean_lines_per_function'})
>>> stats_of(['urllib', 'json', 'collections'])
                              urllib         json  collections
empty_lines_ratio           0.157293     0.136503     0.148090
comment_lines_ratio         0.075217     0.038344     0.061574
function_lines_ratio        0.212391     0.448620     0.107560
mean_lines_per_function    13.463768    41.785714   138.000000
lines                    4374.000000  1304.000000  1283.000000
empty_lines               688.000000   178.000000   190.000000
comment_lines             329.000000    50.000000    79.000000
docs_lines                425.000000   218.000000   133.000000
function_lines            929.000000   585.000000   138.000000
num_of_functions           69.000000    14.000000     1.000000
num_of_classes             55.000000     3.000000    34.000000

Modules

`cli`	Command-line interface for umpyre.
`collectors`	Collectors package - metric collection implementations.
`config`	Configuration loading and validation for umpyre.
`python_code_stats`	Get stats about packages.
`schema`	Versioned schema for metrics storage and migration support.
`storage`	Storage operations for metrics.