"""Core objects for interfacing e3 modules."""
import logging
from typing import Dict, List, Optional
from git import Repo
from gitlab.v4.objects import Project
from e3_build_tools import utils
from e3_build_tools.exceptions import (
FetchDataException,
NotInitialisedError,
FileNotFoundException,
MissingReferenceException,
)
from e3_build_tools.git.registry import WrapperRegistry
[docs]
logger = logging.getLogger(__name__)
[docs]
class ModuleSource:
"""Class for e3 modules."""
[docs]
install_config_file = "configure/RELEASE"
[docs]
install_config_var = "EPICS_BASE"
[docs]
version_config_file = "configure/CONFIG_MODULE"
[docs]
version_config_var = "E3_MODULE_VERSION"
[docs]
validate_build_number = True
def __init__(
self,
name: str,
*,
versions: List[str],
dependencies: Optional[List[str]] = None,
) -> None:
"""Initialise the ModuleSource object.
Initialises versions and dependencies attributes if they are not
provided.
"""
self._project: Optional[Project] = None
self._repo: Optional[Repo] = None
[docs]
self.versions: Dict[str, Dict] = {v: {} for v in versions}
[docs]
self.dependencies = set(
dependencies + ["require"] if dependencies else ["require"]
)
[docs]
self.targets = ["init", "clean", "patch", "build", "install"]
@property
[docs]
def project(self) -> Project:
"""Return the module's remote repository."""
if self._project is None:
raise NotInitialisedError
return self._project
@property
[docs]
def repo(self) -> Repo:
"""Return the module's local repository."""
if self._repo is None:
raise NotInitialisedError
return self._repo
@repo.setter
def repo(self, value: Repo) -> None:
self._repo = value
[docs]
def __str__(self) -> str:
"""Return the module's name."""
return self.name
[docs]
def __repr__(self) -> str:
"""Return serialized object."""
strng = f"{self.__class__.__name__}('{self.name}'"
if self.versions:
strng += f", versions={[name for name in self.versions]}"
if self.dependencies:
strng += f", dependencies={[name for name in self.dependencies]}"
return strng + ")"
[docs]
def remove_version(self, version: str) -> Dict[str, Dict]:
"""Remove version."""
return self.versions.pop(version)
[docs]
def fetch_remote_data(self, registry: WrapperRegistry) -> None:
"""Fetch all of the module's config data.
Raises:
FetchDataException: If data was not fetched.
"""
logger.debug(f"Fetching data for {self.name}")
self._fetch_project(registry)
for version in self.versions:
if version == "latest":
ref = registry.get_project(self.name).default_branch
logger.debug(f"Found reference 'latest' - instead reading {ref!r}")
else:
ref = version
self._fetch_config(version, ref, registry)
self._read_version_string(version)
[docs]
def set_config_data(self, config_content: Dict[str, str], version: str) -> None:
"""Write config data for the specified version."""
logger.debug(f"Config data: {config_content}, version: {version}")
self.versions[version]["config"] = config_content
[docs]
def update_deps(self, ref: str) -> None:
"""Extract dependencies from parsed config data.
Updates the set of dependencies from the configuration data.
"""
# TODO: Check whether this function should be removing any values from
# the dependencies set if they no longer exist in the configuration
# file. Alternatively, clear the dependencies set first, and add all
# values from the new configuration file.
config_data = self.versions[ref]["config"]
dep_suffix = "_DEP_VERSION"
self.dependencies.update(
dep_version[: -len(dep_suffix)].lower()
for dep_version in config_data
if dep_version.endswith(dep_suffix)
)
def _fetch_project(self, registry: WrapperRegistry) -> None:
"""Fetch the project data from the registry.
Raises:
FetchDataException: if the project does not exist in the registry.
"""
try:
self._project = registry.get_project(self.name)
except KeyError as e:
raise FetchDataException(
f"Could not fetch project data for {self.name} from registry."
) from e
def _fetch_config(self, version: str, ref: str, registry: WrapperRegistry) -> None:
"""Fetch the configuration data from the registry for the given reference.
Raises:
FetchDataException: if the version configuration file does not exist for the specified reference.
"""
try:
raw_config_content = registry.get_config_from(
self.name, ref, file=self.version_config_file
)
except (FileNotFoundException, MissingReferenceException) as e:
logger.error(
f"Failed to fetch config data for '{self.name}' version '{version}'. File not found or reference is missing."
)
raise FetchDataException from e
self.set_config_data(
utils.read_makefile_definitions(raw_config_content), version
)
def _read_version_string(self, version: str) -> None:
"""Fetch the version string from the registry for the given reference.
Raises:
FetchDataException: if the version string is not defined for the specified reference.
"""
try:
version_string = self.versions[version]["config"][self.version_config_var]
except KeyError as e:
raise FetchDataException(
f"Configuration not laoded correctly for '{self.name}' version '{version}'."
) from e
if self.validate_build_number:
version_string = utils.ensure_build_number(version_string)
self.versions[version]["version_string"] = version_string
[docs]
class EPICSBaseSource(ModuleSource):
"""Class for e3 base."""
[docs]
install_config_file = "configure/CONFIG_BASE"
[docs]
install_config_var = "E3_EPICS_PATH"
[docs]
version_config_file = "configure/CONFIG_BASE"
[docs]
version_config_var = "E3_BASE_VERSION"
[docs]
validate_build_number = False
def __init__(self, *, version: str) -> None:
"""Initialize e3-base object."""
super().__init__("base", versions=[version] if version else [], dependencies=[])
self.dependencies.remove("require")
self.targets.remove("clean") # Not a valid target for e3-base
self.targets.remove("install") # e3-base is installed during `build`
@property
[docs]
def version(self) -> Optional[str]:
"""Return version."""
try:
value = next(iter(self.versions))
except StopIteration:
value = None
return value
[docs]
def __repr__(self) -> str:
"""Return serialised object."""
return f"EPICSBaseSource(version={self.version!r})"
[docs]
class RequireSource(ModuleSource):
"""Class for e3 require."""
[docs]
version_config_file = "configure/RELEASE"
[docs]
version_config_var = "E3_REQUIRE_VERSION"
[docs]
validate_build_number = False
def __init__(self, *, version: str) -> None:
"""Initialize e3-require object."""
super().__init__(
"require", versions=[version] if version else [], dependencies=["base"]
)
self.dependencies.remove("require")
@property
[docs]
def version(self) -> Optional[str]:
"""Return version."""
try:
value = next(iter(self.versions))
except StopIteration:
value = None
return value
[docs]
def __repr__(self) -> str:
"""Return serialised object."""
return f"RequireSource(version={self.version!r})"