1"""
2    sphinx.ext.duration
3    ~~~~~~~~~~~~~~~~~~~
4
5    Measure durations of Sphinx processing.
6
7    :copyright: Copyright 2007-2021 by the Sphinx team, see AUTHORS.
8    :license: BSD, see LICENSE for details.
9"""
10
11from datetime import datetime, timedelta
12from itertools import islice
13from operator import itemgetter
14from typing import Any, Dict, List, cast
15
16from docutils import nodes
17
18from sphinx.application import Sphinx
19from sphinx.domains import Domain
20from sphinx.locale import __
21from sphinx.util import logging
22
23logger = logging.getLogger(__name__)
24
25
26class DurationDomain(Domain):
27    """A domain for durations of Sphinx processing."""
28    name = 'duration'
29
30    @property
31    def reading_durations(self) -> Dict[str, timedelta]:
32        return self.data.setdefault('reading_durations', {})
33
34    def note_reading_duration(self, duration: timedelta) -> None:
35        self.reading_durations[self.env.docname] = duration
36
37    def clear(self) -> None:
38        self.reading_durations.clear()
39
40    def clear_doc(self, docname: str) -> None:
41        self.reading_durations.pop(docname, None)
42
43    def merge_domaindata(self, docnames: List[str], otherdata: Dict[str, timedelta]) -> None:
44        for docname, duration in otherdata.items():
45            if docname in docnames:
46                self.reading_durations[docname] = duration
47
48
49def on_builder_inited(app: Sphinx) -> None:
50    """Initialize DurationDomain on bootstrap.
51
52    This clears results of last build.
53    """
54    domain = cast(DurationDomain, app.env.get_domain('duration'))
55    domain.clear()
56
57
58def on_source_read(app: Sphinx, docname: str, content: List[str]) -> None:
59    """Start to measure reading duration."""
60    app.env.temp_data['started_at'] = datetime.now()
61
62
63def on_doctree_read(app: Sphinx, doctree: nodes.document) -> None:
64    """Record a reading duration."""
65    started_at = app.env.temp_data.get('started_at')
66    duration = datetime.now() - started_at
67    domain = cast(DurationDomain, app.env.get_domain('duration'))
68    domain.note_reading_duration(duration)
69
70
71def on_build_finished(app: Sphinx, error: Exception) -> None:
72    """Display duration ranking on current build."""
73    domain = cast(DurationDomain, app.env.get_domain('duration'))
74    durations = sorted(domain.reading_durations.items(), key=itemgetter(1), reverse=True)
75    if not durations:
76        return
77
78    logger.info('')
79    logger.info(__('====================== slowest reading durations ======================='))
80    for docname, d in islice(durations, 5):
81        logger.info('%d.%03d %s', d.seconds, d.microseconds / 1000, docname)
82
83
84def setup(app: Sphinx) -> Dict[str, Any]:
85    app.add_domain(DurationDomain)
86    app.connect('builder-inited', on_builder_inited)
87    app.connect('source-read', on_source_read)
88    app.connect('doctree-read', on_doctree_read)
89    app.connect('build-finished', on_build_finished)
90
91    return {
92        'version': 'builtin',
93        'parallel_read_safe': True,
94        'parallel_write_safe': True,
95    }
96