ext.py

# extension testbed

from datetime import datetime, timedelta
from itertools import islice
from operator import itemgetter
from typing import Dict, List

from sphinx.domains import Domain
from sphinx.util import logging

logger = logging.getLogger(__name__)


class ReadingDurationDomain(Domain):
    name = 'duration'

    @property
    def durations(self) -> Dict[str, timedelta]:
        return self.data.setdefault('durations', {})

    def note_duration(self, duration: timedelta):
        self.durations[self.env.docname] = duration

    def clear(self):
        self.durations.clear()

    def clear_doc(self, docname: str) -> None:
        self.durations.pop(docname, None)

    def merge_domaindata(self, docnames: List[str], otherdata: Dict[str, timedelta]) -> None:
        for docname, duration in otherdata.items():
            if docname in docnames:
                self.durations[docname] = duration


def on_builder_inited(app):
    domain = app.env.get_domain('duration')
    domain.clear()


def on_source_read(app, docname: str, content: List[str]) -> None:
    app.env.temp_data['started_at'] = datetime.now()


def on_doctree_read(app, doctree):
    started_at = app.env.temp_data.get('started_at')
    duration = datetime.now() - started_at
    domain = app.env.get_domain('duration')
    domain.note_duration(duration)


def on_build_finished(app, error):
    domain = app.env.get_domain('duration')
    durations = sorted(domain.durations.items(), key=itemgetter(1), reverse=True)
    if not durations:
        return

    logger.info('')
    logger.info('=========================== slowest reading durations ============================')
    for docname, d in islice(durations, 5):
        logger.info('%d.%03d %s ', d.seconds, d.microseconds / 1000, docname)


def setup(app):
    app.add_domain(ReadingDurationDomain)
    app.connect('builder-inited', on_builder_inited)
    app.connect('source-read', on_source_read)
    app.connect('doctree-read', on_doctree_read)
    app.connect('build-finished', on_build_finished)

    return {
        'parallel_read_safe': True,
    }