from __future__ import annotations
import datetime
import functools
import logging
import os
import re
import shutil
import sys
import traceback
import uuid
from pathlib import Path
from typing import TYPE_CHECKING, Any, Union
import attr
import attrs
import omegaconf
import omegaconf.errors
import pharaoh
import pharaoh.log
import pharaoh.util.oc_resolvers
from pharaoh.assetlib import finder, resource
from pharaoh.assetlib.context import context_stack
from pharaoh.errors import AssetGenerationError, ProjectInconsistentError
from pharaoh.plugins.plugin_manager import PM
from pharaoh.util.contextlib_chdir import chdir
if TYPE_CHECKING:
from collections.abc import Iterable, Iterator
import jinja2
try:
PHARAOH_CLI_PATH = Path(shutil.which("pharaoh.exe")).as_posix() # type: ignore[arg-type]
except TypeError:
PHARAOH_CLI_PATH = "pharaoh.exe"
PathLike = Union[str, Path]
LRU_PHARAOH_PROJECT: PharaohProject | None = None
DEFAULT_MISSING = object()
log = pharaoh.log.log
[docs]
def get_project(lookup_path: str | Path | None = None) -> PharaohProject:
"""
Returns an instance of PharaohProject.
:param lookup_path: If None, the function tries to return the Pharaoh singleton instance that is already loaded in
memory and raises and exception if there is none.
If given a string or Path instance, searches the given path and all its parent folders
for a ``pharaoh.yaml`` project file and returns a PharaohProject instance.
"""
if lookup_path is None:
# Enforce having a singleton per process
if LRU_PHARAOH_PROJECT is None:
msg = "No PharaohProject instance created yet!"
raise RuntimeError(msg)
return LRU_PHARAOH_PROJECT
if isinstance(lookup_path, (str, Path)):
lookup_path = Path(lookup_path)
if lookup_path.is_file():
lookup_path = lookup_path.parent
else:
msg = "Unsupported value for argument 'lookup_path'."
raise ValueError(msg)
path = Path(lookup_path).absolute()
if path.is_file():
path = path.parent
while True:
if (path / PharaohProject.PROJECT_SETTINGS_YAML).exists():
return PharaohProject(project_root=path)
if path == path.parent:
# We reached the path's anchor level
msg = (
f"Cannot find a Pharaoh project file {PharaohProject.PROJECT_SETTINGS_YAML} in "
f"{lookup_path} or one of its parent directories!"
)
raise LookupError(msg)
path = path.parent
[docs]
class PharaohProject:
PROJECT_SETTINGS_YAML = "pharaoh.yaml"
def __new__(cls, *args, **kwargs):
# Cache the last recently used project instance in a global variable,
# so it can be accessed from anywhere after creation
global LRU_PHARAOH_PROJECT # noqa: PLW0603
LRU_PHARAOH_PROJECT = object.__new__(cls)
return LRU_PHARAOH_PROJECT
def __del__(self):
try: # noqa: SIM105
pharaoh.log.remove_filehandlers(self._project_root)
except Exception:
pass
[docs]
def __init__(
self,
project_root: PathLike,
overwrite: bool = False,
templates: str | list[str] | tuple[str] = ("pharaoh.default_project",),
template_context: dict[str, Any] | None = None,
**kwargs,
):
"""
Instantiates a Pharaoh project instance using either an existing project root
(directory containing pharaoh.yaml) or creates a new project.
:param project_root: A directory containing a pharaoh.yaml file
:param overwrite: If the project directory already contains files, overwrite them.
:param templates: The project template(s) to use for creating the Sphinx project files.
Maybe be any number of template names or paths to directories.
:param template_context: The Jinja rendering context for the selected project templates.
:keyword custom_settings: A path to a YAML file containing settings to overwrite the default project settings.
"""
self._settings_map: dict[str, omegaconf.DictConfig] = {}
self._merged_settings: omegaconf.DictConfig = omegaconf.DictConfig({})
self._project_root: Path = Path(project_root).absolute().resolve()
self._asset_finder: finder.AssetFinder | None = None
logging_add_filehandler = kwargs.pop("logging_add_filehandler", True)
custom_settings = kwargs.pop("custom_settings", None)
if kwargs:
msg = f"Unknown keyword arguments {tuple(kwargs.keys())}!"
raise ValueError(msg)
pharaoh.log.add_streamhandler()
if logging_add_filehandler:
pharaoh.log.add_filehandler(self._project_root)
pharaoh.log.add_warning_filehandler(self._project_root)
if isinstance(templates, str):
templates = [templates]
self._ensure_project(
templates=list(templates),
template_context=template_context,
recreate=overwrite,
custom_settings=custom_settings,
)
self.load_settings(namespace="all")
log.setLevel(self.get_setting("logging.level").upper())
[docs]
def load_settings(self, namespace: str = "all"):
"""
Loads settings from various namespaces.
default
The default setting in the Pharaoh library
project
The project settings that may be modified by the user
env
The settings defined by environment variables starting with PHARAO
all
Loads all of the above
:param namespace: The namespace to load settings from. all, default, project or env.
"""
if namespace in ("all", "default"):
self._settings_map["default"] = self._load_default_settings()
if namespace in ("all", "project"):
project_settings = self._project_root / self.PROJECT_SETTINGS_YAML
with open(project_settings) as fp:
self._settings_map["project"] = omegaconf.OmegaConf.load(fp) # type: ignore[assignment]
if namespace in ("all", "env"):
dotlist = []
env_pat = re.compile(r"^PHARAOH?((\.)|(__)).+$", re.IGNORECASE)
for k, v in os.environ.items():
if env_pat.fullmatch(k):
keys = re.split(r"\.|(?<!_)__(?!_)", k)
key = ".".join(keys[1:]).lower()
dotlist.append(f"{key}={v}")
if len(dotlist):
self._settings_map["env"] = omegaconf.OmegaConf.from_dotlist(dotlist)
else:
self._settings_map["env"] = omegaconf.OmegaConf.create({})
self._update_merged_settings()
def _update_merged_settings(self):
"""
Merges settings from all namespaces.
"""
self._merged_settings = omegaconf.OmegaConf.create(
omegaconf.OmegaConf.merge(
self._settings_map["default"],
self._settings_map["project"],
self._settings_map["env"],
)
) # type: ignore[assignment]
[docs]
def get_settings(self) -> omegaconf.DictConfig:
"""
Returns merged settings from all namespaces.
"""
if self._merged_settings is None:
self._update_merged_settings()
return self._merged_settings
[docs]
def put_setting(self, key: str, value: Any):
"""
Sets a setting by its dot-separated name, e.g. "core.debug".
This has no effect, if there is an environment variable defining the same setting!
"""
# todo: Let put_setting be able to overwrite settings defined in env vars.
# Currently env vars have priority.
paths = key.lower().split(".")
node = self._project_settings
for path in paths[:-1]:
node = getattr(node, path)
setattr(node, paths[-1], value)
self._update_merged_settings()
[docs]
def get_setting(self, key: str, default=DEFAULT_MISSING, to_container: bool = False, resolve: bool = True):
"""
Gets a setting by its dot-separated name, e.g. "core.debug".
The settings are preferably taken from environment variables (e.g. PHARAOH.CORE.DEBUG), whereas the values
must be TOML compatible values (e.g. "true" for boolean).
If no environment variable is found, the user defined settings in the pharaoh.yaml file in the
project root will be used. If the setting is not specified in there or the file is not existing,
the Pharaoh default settings are queried.
If the setting is not found, a LookupError is returned unless "default" is set.
Examples::
proj.get_setting("report.title")
proj.get_setting("cloud.metadata", to_container=True, resolve=False) == {
"author": "${report.author}",
"title": "${report.title}",
}
proj.get_setting("cloud.metadata", to_container=True, resolve=True) == {
"author": "loibljoh",
"title": "Pharaoh Report",
}
proj.get_setting("cloud.metadata.title") == "Pharaoh Report"
proj.get_setting("cloud.metadata") == omegaconf.dictconfig.DictConfig(...)
:param key: Dot-separated name of the setting value to return
:param default: The returned default value if the key does not exist.
:param to_container: If set, recursively converts an OmegaConf config to a primitive Python container type
(dict or list).
:param resolve: True to resolve all values if "to_container" is set to True
"""
if not key:
msg = "Must be a string with a minimum length of 1!"
raise ValueError(msg)
merged_settings = self.get_settings()
key = key.lower()
ret = omegaconf.OmegaConf.to_container(merged_settings, resolve=resolve) if to_container else merged_settings
try:
paths = key.split(".")
for path in paths:
ret = ret[path]
return ret
except KeyError:
if default is DEFAULT_MISSING:
msg = f"No setting key called {key!r}!"
raise LookupError(msg) from None
return default
[docs]
def save_settings(self, include_env: bool = False):
"""
Saves back the project settings to the project YAML file.
:param include_env: If True, Pharaoh settings that are set via environment variables will be persisted
to the project settings YAML file.
"""
project_settings = self._project_root / self.PROJECT_SETTINGS_YAML
if include_env:
settings = omegaconf.OmegaConf.merge(
self._settings_map["project"],
self._settings_map["env"],
)
else:
settings = self._project_settings
with open(project_settings, "w") as fp:
omegaconf.OmegaConf.save(config=settings, f=fp)
@property
def _project_settings(self):
return self._settings_map["project"]
@property
def settings_file(self):
return self._project_root / self.PROJECT_SETTINGS_YAML
@property
def project_root(self):
return self._project_root
@property
def sphinx_report_project(self):
return self.project_root / "report-project"
@property
def sphinx_report_project_components(self):
return self.sphinx_report_project / "components"
@property
def sphinx_report_build(self):
return self.project_root / "report-build"
@property
def asset_build_dir(self):
return self.sphinx_report_project / ".asset_build"
@property
def asset_finder(self) -> finder.AssetFinder:
if self._asset_finder is None:
self._asset_finder = finder.AssetFinder(self.asset_build_dir)
return self._asset_finder
[docs]
def add_component(
self,
component_name: str,
templates: str | (Path | (Iterable[str] | Iterable[Path])) = ("pharaoh.empty",),
render_context: dict | None = None,
resources: list[resource.Resource] | None = None,
metadata: dict[str, Any] | None = None,
index: int = -1,
overwrite: bool = False,
):
"""
Adds a new component to the Pharaoh project
Example::
from pharaoh.api import FileResource, PharaohProject
proj = PharaohProject(".")
proj.add_component(
component_name="component_ABC",
templates=[
"plugin_abc.template_xyz", # plugin template
"path/to/template/directory", # template directory
"path/to/template/file/tmpl.pharaoh.py" # template file
],
render_context={"foo": "bar"},
resources=[FileResource(alias="dlh5_result", pattern="C:/temp/**/*.dlh5")],
metadata={"some tag": "some value"},
index=-1, # append
overwrite=False
)
:param component_name: The name of the component. Must be a valid Python identifier.
:param templates: A list of component templates to use for creating the components project files.
Those may be the template identifier (e.g. ``plugin_abc.template_xyz``) of a registered plugin template,
a path to a template directory or a path to a template file (single-file template).
Since multiple template may be specified, their order matters in cases where different templates
create equally-named files, thus templates might overwrite files of the previous templates.
This enables *template-composition*, where template designers can chunk their bigger templates into
smaller reusable building blocks.
If omitted, an empty default template is used.
:param render_context: The Jinja rendering context for the selected template.
The actual template rendering context will have ``component_name``, ``resources`` and
``metadata`` available under the respective keys.
:param resources: A list of Resource instances, defining the component's resources used in asset scripts
:param metadata: A dictionary of metadata that may be used to find the component via
:func:`PharaohProject.find_component()
<pharaoh.project.PharaohProject.find_component>` method.
:param overwrite: If True, an already existing component will be overwritten
"""
from pharaoh.templating.first_level.find import find_template
from pharaoh.templating.first_level.render import render_template
# Check if component with this name already added
components = []
for component in self.iter_components():
if component["name"] == component_name:
if overwrite:
pass
else:
msg = f"Component {component_name!r} already exists!"
raise KeyError(msg)
else:
components.append(component)
render_context = render_context or {}
render_context["pharaoh_cli_path"] = PHARAOH_CLI_PATH
if isinstance(templates, (str, Path)):
templates = [str(templates)]
templates = [str(t) for t in templates]
if not templates:
msg = "No templates specified!"
raise ValueError(msg)
new_component = Component(
name=component_name,
templates=templates,
render_context=render_context,
resources=resources or [],
metadata=metadata or {},
)
# Support negative indexing where -1 is the last item. Normal insert with -1 will insert at second last index.
if index < 0:
index = max(0, len(components) + 1 + index)
components.insert(
index,
attrs.asdict(new_component), # type: ignore[arg-type]
)
for templ in templates:
template_path = find_template(templ)
if (template_path / "[[ component_name ]]").exists(): # pragma: no cover
msg = (
"The usage of [[ component_name ]] as top level directory name of component "
"templates is not required anymore. "
)
raise RuntimeError(msg)
render_template(
template_path=template_path,
outputdir=self.sphinx_report_project_components / component_name,
context=new_component.get_render_context(),
)
self._project_settings.components = components
self.save_settings()
log.info(f"Added component {component_name!r}. Saved project.")
[docs]
def add_template_to_component(
self, component_name: str, templates: str | Path | Iterable[str], render_context: dict | None = None
):
"""
Adds additional templates to an existing component, that may overwrite existing files during rendering.
:param component_name: The name of the component. Must be a valid Python identifier.
:param templates: The component template(s) to use for creating the components project files.
Maybe be the a single or multiple template names or paths to a directory.
:param render_context: The Jinja rendering context for the selected template.
The actual template rendering context will have component_name, resources and metadata
available under the respective keys.
"""
from pharaoh.templating.first_level.find import find_template
from pharaoh.templating.first_level.render import render_template
# Check if component with this name already added
for component_index, comp in enumerate(self.iter_components()): # noqa: B007
if comp["name"] == component_name:
component = Component.from_dictconfig(comp)
break
else:
msg = f"Component {component_name!r} does not exist!"
raise KeyError(msg)
component.render_context.update(render_context or {})
if isinstance(templates, (str, Path)):
templates = [str(templates)]
templates = [str(t) for t in templates]
if not templates:
msg = "No templates specified!"
raise ValueError(msg)
component.templates.extend(templates)
for templ in templates:
template_path = find_template(templ)
render_template(
template_path=template_path,
outputdir=self.sphinx_report_project_components / component_name,
context=component.get_render_context(),
)
self._project_settings.components[component_index] = attrs.asdict(component)
self._update_merged_settings()
self.save_settings()
log.info(f"Updated component {component_name!r}. Saved project.")
[docs]
def remove_component(
self,
filter: str,
regex: bool = False,
) -> list[str]:
"""
Removes one or multiple existing components.
:param filter: A case-insensitive component filter. Either a full-match or regular expression, depending on
regex argument.
:param regex: If True, the filter argument will be treated as regular expression. Components
that partially match the regular expression are removed.
:returns: A list of component names that got removed
"""
# Check if component with this name already added
if regex:
rex = re.compile(filter, flags=re.IGNORECASE)
else:
filter = filter.lower()
removed = []
components = []
for comp in self.iter_components():
if (regex and rex.match(comp["name"]) is not None) or comp["name"].lower() == filter:
comp_files = self.sphinx_report_project_components / comp["name"]
if comp_files.exists() and comp_files.is_dir(): # pragma: no cover
shutil.rmtree(comp_files)
removed.append(comp["name"])
else:
components.append(comp)
if len(removed):
self._project_settings.components = components
self.save_settings()
log.info(f"Removed components {','.join(removed)}. Saved project.")
return removed
[docs]
def iter_components(self) -> Iterator[omegaconf.DictConfig]:
"""
Returns an iterator over all components from a project.
"""
yield from self._project_settings.get("components", []) or []
[docs]
def find_components(self, expression: str = "") -> list[omegaconf.DictConfig]:
"""
Find components by their metadata using an evaluated expression.
The expression must be a valid Python expression and following local variables may be used:
- name: The name of the component
- templates: A list of templates used to render the component
- metadata: A dict of metadata specified
- render_context: The Jinja rendering context specified
- resources: A list of resource definitions
:param expression: A Python expression that will be evaluated. If it evaluates to a truthy result,
the component name is included in the returned list.
Example: ``name == "dummy" and metadata.foo in (1,2,3)``.
A failing evaluation will be treated as False. An empty expression will always match.
"""
found = []
for comp in self.iter_components():
if not expression:
found.append(comp)
continue
try:
result = eval(expression, {}, comp)
except Exception:
result = False
if result:
found.append(comp)
return found
[docs]
def get_resource(self, alias: str, component: str) -> resource.Resource:
"""
Finds a Resource from a project component by its alias.
:param component: The component's name
:param alias: The resource's alias
"""
for r in self.iter_resources(component):
if r["alias"] == alias:
obj = resource.Resource.from_dict(r)
obj._cachedir = str(self.sphinx_report_project / ".resource_cache" / component)
return obj
msg = f"Cannot find resource {alias!r} in component {component!r}!"
raise LookupError(msg)
[docs]
def iter_resources(self, component: str) -> Iterator[omegaconf.DictConfig]:
"""
Returns an iterator over all resources from a project component.
:param component: The component's name
"""
for comp in self.iter_components():
if comp["name"] == component:
yield from comp["resources"]
return
msg = f"Component {component} does not exist!"
raise LookupError(msg)
[docs]
def update_resource(self, alias: str, component: str, resource: resource.Resource):
"""
Updates a Resource from a project component by its alias.
:param component: The component's name
:param alias: The resource's alias
:param resource: The resource's alias
"""
for comp in self.iter_components():
if comp["name"] == component:
for res in comp["resources"]:
if res["alias"] == alias:
res.update(resource.to_dict())
self.save_settings()
return
comp["resources"].append(resource.to_dict())
self.save_settings()
return
msg = f"Component {component} does not exist!"
raise LookupError(msg)
[docs]
def get_default_sphinx_configuration(self, confdir: PathLike):
"""
Provides Sphinx project configurations, that will be dynamically included by the generated conf.py.
"""
confdir = Path(confdir)
config = {
"pharaoh_jinja_templates": [],
"pharaoh_jinja_context": {},
"extensions": [
"sphinx_design", # https://github.com/executablebooks/sphinx-design
"pharaoh.templating.second_level.sphinx_ext.jinja_ext",
"pharaoh.templating.second_level.sphinx_ext.asset_ext",
],
"project": self.get_setting("report.title"),
"author": self.get_setting("report.author"),
"copyright": f"{datetime.datetime.now(tz=datetime.timezone.utc).date()}",
# The suffix(es) of source filenames.
# You can specify multiple suffix as a list of string:
"source_suffix": [".rst", ".rst.jinja2", ".txt"],
"master_doc": "index",
"templates_path": ["_templates"],
# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
# This pattern also affects html_static_path and html_extra_path.
"exclude_patterns": [".asset_build", "**/asset_scripts"],
# Make sure the target is unique
"autosectionlabel_prefix_document": True,
# Latex Builder (PDF)
}
# HTML Builder
static_path = confdir / "_static"
assert static_path.exists()
css_files = list((static_path / "css").glob("*.css"))
js_files = list((static_path / "js").glob("*.js"))
config.update(
{
"html_title": self.get_setting("report.title"),
"html_use_smartypants": True,
"html_last_updated_fmt": "%b %d, %Y",
"html_split_index": False,
"html_sidebars": {
"**": ["searchbox.html", "globaltoc.html", "sourcelink.html"],
},
# Custom index page
# https://ofosos.org/2018/12/28/landing-page-template/
# html_additional_pages={"index": "index.html"},
# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "pharaoh.css" will overwrite the builtin "pharaoh.css".
"html_static_path": ["_static"],
"html_css_files": [f"css/{cssfile.name}" for cssfile in css_files],
"html_js_files": [f"js/{jsfile.name}" for jsfile in js_files],
"html_show_sphinx": False,
"html_show_copyright": True,
"html_context": {
"pharaoh_version": pharaoh.__version__,
"pharaoh_logo": "_static/html_logo.png",
"pharaoh_show_logo": self.get_setting("report.html.show_pharaoh_logo"),
"pharaoh_homepage": "", # todo: change to RTD
},
"html_style": "sphinx_rtd_theme_overrides.css",
"html_theme": "sphinx_rtd_theme",
"html_theme_options": {
"display_version": False,
"prev_next_buttons_location": "both",
"style_external_links": False,
"collapse_navigation": False,
"sticky_navigation": True,
"navigation_depth": 5,
"includehidden": False,
},
}
)
PM.pharaoh_configure_sphinx(self, config, confdir)
return config
[docs]
def generate_assets(self, component_filters: Iterable[str] = (".*",)) -> list[Path]:
"""
Generate all assets by executing the asset scripts of a selected or all components.
All asset scripts are executed in separate parallel child processes
(number of workers determined by ``asset_gen.worker_processes`` setting;
setting 0 executes all asset scripts sequentially in the current process).
Setting ``asset_gen.script_ignore_pattern`` determines if a script is ignored.
Putting the comment ``# pharaoh: ignore`` at the start of a script will also ignore the file.
:param component_filters: A list of regular expressions that are matched against each component name.
If a component name matches any of the regular expressions, the component's
assets are regenerated (containing directory will be cleared)
"""
from pharaoh.assetlib.generation import generate_assets, generate_assets_parallel
pharaoh.log.log_version_info()
log.info("Generating assets...")
PM.pharaoh_asset_gen_started(self)
sources = []
for comp in self.iter_components():
comp_name = comp["name"]
comp_asset_build_dir = self.asset_build_dir / comp_name
for cfilter in component_filters:
if re.match(cfilter, comp_name, re.IGNORECASE) is not None:
if comp_asset_build_dir.exists():
shutil.rmtree(comp_asset_build_dir, ignore_errors=False)
cachedir = self.sphinx_report_project / ".resource_cache" / comp_name
log.info(f"Preparing resources of component {comp_name!r} for asset generation...")
cachedir.mkdir(parents=True, exist_ok=True)
resources: dict[str, resource.Resource] = {
resource_dict["alias"]: resource.Resource.from_dict(resource_dict)
for resource_dict in comp["resources"]
}
for r in resources.values():
r._cachedir = str(cachedir)
PM.pharaoh_asset_gen_prepare_resources(project=self, resources=resources)
# TransformedResource may depend on others, so process them first.
for r in resources.values():
r._cachedir = str(cachedir)
if isinstance(r, resource.TransformedResource):
r.transform(resources=resources)
for script in (self.sphinx_report_project_components / comp_name / "asset_scripts").rglob("*"):
sources.append((comp_name, script))
break
workers = self.get_setting("asset_gen.worker_processes", 0)
if workers == 0: # Run in same process - used for easier debugging
results: list[tuple[Path, str | None]] = []
for component_name, asset_source in sources:
try:
generate_assets(self.project_root, asset_src=asset_source, component_name=component_name)
results.append((asset_source, None))
except Exception:
results.append((asset_source, traceback.format_exc()))
else:
results = generate_assets_parallel(self.project_root, asset_sources=sources, workers=workers)
msg = "At least one error occurred while asset script execution:\n"
i = 1
processed_asset_scripts = []
for script, ex in results:
if ex:
msg += f"\n\nError #{i}: {ex}\n"
i += 1
else:
processed_asset_scripts.append(script)
if i > 1:
pharaoh.log.log_debug_info()
raise AssetGenerationError(msg)
return processed_asset_scripts
[docs]
def build_report(self, catch_errors=True) -> int:
"""
Builds the Sphinx project and returns the status code.
:param catch_errors: If True, Sphinx build errors will not raise an exception but return a -1 instead.
"""
import multiprocessing
from sphinx.util.docutils import docutils_namespace, patch_docutils
from pharaoh.sphinx_app import PharaohSphinx
builder = self.get_setting("report.builder")
PM.pharaoh_build_started(self, builder)
self._check_template_dependencies()
with chdir(self.sphinx_report_project):
pharaoh.log.log_version_info()
log.info(f"Building Sphinx project using {builder.upper()} builder...")
resolved_settings = omegaconf.OmegaConf.create(
omegaconf.OmegaConf.merge(
self._settings_map["default"],
self._settings_map["project"],
self._settings_map["env"],
)
)
omegaconf.OmegaConf.resolve(resolved_settings)
self.sphinx_report_build.mkdir(parents=True, exist_ok=True)
with open(self.sphinx_report_build / "pharaoh.resolved.yaml", "w") as fp:
omegaconf.OmegaConf.save(config=resolved_settings, f=fp)
# Sphinx is used via its API instead of the CLI because we want to modify the info/warning stream
try:
sourcedir = "."
confdir = sourcedir
outputdir = str(self.sphinx_report_build)
doctreedir = f"{outputdir}\\.doctrees"
confoverrides: dict | None = {}
status = pharaoh.log.SphinxLogRedirector(logging.DEBUG)
warning = pharaoh.log.SphinxLogRedirector(logging.WARNING)
freshenv = True
warningiserror = True
tags = None
verbosity = int(self.get_setting("report.verbosity", 0))
jobs = multiprocessing.cpu_count()
keep_going = True
pdb = False
with patch_docutils(confdir), docutils_namespace():
app = PharaohSphinx(
sourcedir,
confdir,
outputdir,
doctreedir,
builder,
confoverrides,
status,
warning,
freshenv,
warningiserror,
tags,
verbosity,
jobs,
keep_going,
pdb,
)
PM.pharaoh_sphinx_app_inited(app)
app.build(force_all=False, filenames=None)
if app.statuscode:
log.error(
f"Sphinx build finished with non-zero exit code {app.statuscode}. "
f"See warnings/errors in log above!"
)
else:
log.info(f"Sphinx build finished successfully! Report at:\n{self.sphinx_report_build}")
PM.pharaoh_build_finished(self, None)
return app.statuscode
except Exception:
log.error("Errors occurred during Sphinx build:", exc_info=True)
pharaoh.log.log_debug_info()
PM.pharaoh_build_finished(self, sys.exc_info())
if not catch_errors:
raise
return -1
[docs]
def open_report(self): # pragma: no cover
"""
Opens the generated report (if possible, e.g. for local HTML reports).
"""
builder = self.get_setting("report.builder").lower()
if builder == "html" and os.name == "nt":
report = self.sphinx_report_build / "index.html"
if not report.is_file():
log.warning(f"Could not find HTML report {report}!")
return
import webbrowser
webbrowser.WindowsDefault().open_new_tab(f"file://{report}")
else:
log.warning(f"The open_report method is not implemented for builder {builder}!")
[docs]
def archive_report(self, dest: PathLike | None = None) -> Path:
"""
Create an archive from the build folder.
:param dest: A destination path to create the archive. Relative paths are relative to the project root.
If omitted, the filename will be taken from the ``report.archive_name`` setting.
:return: The path to the archive
"""
if not self.sphinx_report_build.exists():
msg = "Pharaoh report not built yet!"
raise Exception(msg)
dest_path = Path(self.get_setting("report.archive_name") if dest is None else dest)
if not dest_path.is_absolute():
dest_path = (self.project_root / dest_path).absolute()
if not dest_path.suffix:
dest_path /= self.get_setting("report.archive_name")
if dest_path.exists():
os.remove(dest_path)
dest_path.parent.mkdir(parents=True, exist_ok=True)
base_name = dest_path.parent / dest_path.stem
fmt = dest_path.suffix.replace(".", "")
shutil.make_archive(str(base_name), fmt, self.sphinx_report_build)
log.info(f"Created archive at {dest_path}")
return dest_path
def _check_template_dependencies(self):
"""
Checks if the used templates of all components are have dependencies to templates that are not used in any
component.
E.g. "plugin_abc.template_A" requires an additional component that renders the "plugin_abc.template_B" template,
in order for all glossary references to work properly.
"""
l1_templates = PM.pharaoh_collect_l1_templates()
used_templates = set()
dependent_templates = set()
for comp in self._project_settings.get("components", []):
for template in comp.templates:
if template in l1_templates:
used_templates.add(template)
for need in l1_templates[template].needs:
dependent_templates.add(need)
if not dependent_templates.issubset(used_templates):
msg = (
f"The used project templates requires the existence of components that render "
f"following templates: {','.join(dependent_templates - used_templates)}"
)
raise ProjectInconsistentError(msg)
def _ensure_project(
self,
templates: list[str],
template_context: dict[str, Any] | None = None,
recreate=False,
custom_settings: str | Path | None = None,
):
"""
Creates a new project by rendering the Sphinx default project at the desired project location.
"""
self.project_root.mkdir(exist_ok=True, parents=True)
existing_files = tuple(self.project_root.glob("*"))
if len(existing_files):
if recreate:
shutil.rmtree(
self.project_root, onerror=lambda func, path, exc_info: log.warning(f"Could not delete file {path}")
)
else:
# There are existing files, check if those are the required pharaoh project files/folders
for path in (
self.settings_file,
self.project_root,
self.sphinx_report_project,
):
if not path.exists():
msg = f"Project structure inconsistent. Missing {path}!"
raise ProjectInconsistentError(msg)
self.asset_build_dir.mkdir(exist_ok=True, parents=True)
return
# Create a new project
self.project_root.mkdir(exist_ok=True, parents=True)
self.asset_build_dir.mkdir(exist_ok=True, parents=True)
default_settings = self._load_default_settings()
if isinstance(custom_settings, (str, Path)):
custom_settings = omegaconf.OmegaConf.load(custom_settings) # type: ignore[assignment]
default_settings = omegaconf.OmegaConf.unsafe_merge(default_settings, custom_settings) # type: ignore[assignment]
self.settings_file.write_text(omegaconf.OmegaConf.to_yaml(default_settings, resolve=False))
gitignore = self.project_root / ".gitignore"
gitignore_lines = [
"# Auto-generated by Pharaoh",
"/log*.txt",
"/report-build",
"/report-project/.asset_build",
"/report-project/.resource_cache",
"/*.zip",
"/*.idea",
]
PM.pharaoh_modify_gitignore(gitignore_lines)
gitignore.write_text("\n".join(gitignore_lines), encoding="utf-8")
from .templating.first_level.render import render_sphinx_base_project
render_sphinx_base_project(
outputdir=self.sphinx_report_project,
templates=templates,
context={**(template_context or {}), **{"pharaoh_cli_path": PHARAOH_CLI_PATH}}, # noqa: PIE800
)
PM.pharaoh_project_created(self)
return
def _load_default_settings(self) -> omegaconf.DictConfig:
default_settings_collection = PM.pharaoh_collect_default_settings()
loaded = []
for default_settings in default_settings_collection:
if isinstance(default_settings, (str, Path)):
if not Path(default_settings).exists():
msg = f"Custom setting YAMl file {default_settings} does not exist!"
raise FileNotFoundError(msg)
loaded.append(omegaconf.OmegaConf.load(default_settings))
elif isinstance(default_settings, dict):
loaded.append(omegaconf.OmegaConf.create(default_settings))
else:
msg = (
f"Unsupported settings type {type(default_settings)} collected from plugin hook "
f"pharaoh_collect_default_settings!"
)
raise TypeError(msg)
return functools.reduce(omegaconf.OmegaConf.unsafe_merge, loaded) # type: ignore[return-value]
def _update_template_env(self, env: jinja2.Environment):
"""
Called by PharaohTemplateEnv.sphinx_config_inited_hook to update the template env by
project specific Jinja globals or filters
:param env: the Jinja env to update
"""
def search_error_assets_global():
"""
Find all error traceback assets in the project grouped by component name.
"""
error_assets = {}
for comp in self.iter_components():
assets = self.asset_finder.search_assets("asset_type == 'error_traceback'", comp.name)
if assets:
error_assets[comp.name] = assets
return error_assets
env.globals["get_setting"] = self.get_setting
env.globals["search_assets_global"] = self.asset_finder.search_assets
env.globals["search_error_assets_global"] = search_error_assets_global
def _build_asset_filepath(self, file: PathLike, component_name: str | None = None) -> Path:
"""
Returns a new file name inside the asset build directory with the same filename as the input file,
except its file stem is suffixed with a unique uuid4 hash (first 8 chars).
E.g. `foo/bar/iris_scatter_plot.html` --> `<asset-build-dir>/<component_name>/iris_scatter_plot_ab8b4081.html`
"""
file = Path(file)
try:
component = component_name or context_stack.get_parent_context("generate_assets")["asset"]["component_name"]
except Exception:
component = "unknown_component"
(self.asset_build_dir / component).mkdir(parents=True, exist_ok=True)
return self.asset_build_dir / component / f"{file.stem}_{str(uuid.uuid4())[:8]}{file.suffix}"
@attrs.define(frozen=True, slots=False)
class Component:
name: str = attrs.field(validator=(attr.validators.instance_of(str), attr.validators.matches_re(r"\w+")))
templates: list[str] = attrs.field(
validator=attrs.validators.deep_iterable(
member_validator=attrs.validators.instance_of(str),
iterable_validator=attrs.validators.instance_of(list),
)
)
render_context: dict = attrs.field(factory=dict)
resources: list[dict] = attrs.field(factory=list, converter=lambda resources: [r.to_dict() for r in resources])
metadata: dict = attrs.field(factory=dict)
def get_render_context(self) -> dict:
for reserved_key in ("component_name", "resources", "metadata"):
if reserved_key in self.render_context:
msg = f"{reserved_key!r} is a reserved context key!"
raise ValueError(msg)
return dict(**self.render_context, component_name=self.name, resources=self.resources, metadata=self.metadata)
@staticmethod
def from_dictconfig(cfg: omegaconf.DictConfig):
return Component(**omegaconf.OmegaConf.to_container(cfg, resolve=False)) # type: ignore[arg-type]
if __name__ == "__main__":
pass