blob: 7d4fc4dbeadd38a3f6738b874c9e4801aa989cac [file] [log] [blame]
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Authors:
# Tristan Maat <tristan.maat@codethink.co.uk>
import os
from . import utils
from . import _yaml
from ._exceptions import LoadError
from .exceptions import LoadErrorReason
BST_WORKSPACE_FORMAT_VERSION = 4
BST_WORKSPACE_PROJECT_FORMAT_VERSION = 1
WORKSPACE_PROJECT_FILE = ".bstproject.yaml"
# WorkspaceProject()
#
# An object to contain various helper functions and data required for
# referring from a workspace back to buildstream.
#
# Args:
# directory (str): The directory that the workspace exists in.
#
class WorkspaceProject:
def __init__(self, directory):
self._projects = []
self._directory = directory
# get_default_project_path()
#
# Retrieves the default path to a project.
#
# Returns:
# (str): The path to a project
#
def get_default_project_path(self):
return self._projects[0]["project-path"]
# get_default_element()
#
# Retrieves the name of the element that owns this workspace.
#
# Returns:
# (str): The name of an element
#
def get_default_element(self):
return self._projects[0]["element-name"]
# to_dict()
#
# Turn the members data into a dict for serialization purposes
#
# Returns:
# (dict): A dict representation of the WorkspaceProject
#
def to_dict(self):
ret = {
"projects": self._projects,
"format-version": BST_WORKSPACE_PROJECT_FORMAT_VERSION,
}
return ret
# from_dict()
#
# Loads a new WorkspaceProject from a simple dictionary
#
# Args:
# directory (str): The directory that the workspace exists in
# dictionary (dict): The dict to generate a WorkspaceProject from
#
# Returns:
# (WorkspaceProject): A newly instantiated WorkspaceProject
#
@classmethod
def from_dict(cls, directory, dictionary):
# Only know how to handle one format-version at the moment.
format_version = int(dictionary["format-version"])
assert format_version == BST_WORKSPACE_PROJECT_FORMAT_VERSION, "Format version {} not found in {}".format(
BST_WORKSPACE_PROJECT_FORMAT_VERSION, dictionary
)
workspace_project = cls(directory)
for item in dictionary["projects"]:
workspace_project.add_project(item["project-path"], item["element-name"])
return workspace_project
# load()
#
# Loads the WorkspaceProject for a given directory.
#
# Args:
# directory (str): The directory
# Returns:
# (WorkspaceProject): The created WorkspaceProject, if in a workspace, or
# (NoneType): None, if the directory is not inside a workspace.
#
@classmethod
def load(cls, directory):
workspace_file = os.path.join(directory, WORKSPACE_PROJECT_FILE)
if os.path.exists(workspace_file):
data_dict = _yaml.roundtrip_load(workspace_file)
return cls.from_dict(directory, data_dict)
else:
return None
# write()
#
# Writes the WorkspaceProject to disk
#
def write(self):
os.makedirs(self._directory, exist_ok=True)
_yaml.roundtrip_dump(self.to_dict(), self.get_filename())
# get_filename()
#
# Returns the full path to the workspace local project file
#
def get_filename(self):
return os.path.join(self._directory, WORKSPACE_PROJECT_FILE)
# add_project()
#
# Adds an entry containing the project's path and element's name.
#
# Args:
# project_path (str): The path to the project that opened the workspace.
# element_name (str): The name of the element that the workspace belongs to.
#
def add_project(self, project_path, element_name):
assert project_path and element_name
self._projects.append({"project-path": project_path, "element-name": element_name})
# WorkspaceProjectCache()
#
# A class to manage workspace project data for multiple workspaces.
#
class WorkspaceProjectCache:
def __init__(self):
self._projects = {} # Mapping of a workspace directory to its WorkspaceProject
# get()
#
# Returns a WorkspaceProject for a given directory, retrieving from the cache if
# present.
#
# Args:
# directory (str): The directory to search for a WorkspaceProject.
#
# Returns:
# (WorkspaceProject): The WorkspaceProject that was found for that directory.
# or (NoneType): None, if no WorkspaceProject can be found.
#
def get(self, directory):
try:
workspace_project = self._projects[directory]
except KeyError:
workspace_project = WorkspaceProject.load(directory)
if workspace_project:
self._projects[directory] = workspace_project
return workspace_project
# add()
#
# Adds the project path and element name to the WorkspaceProject that exists
# for that directory
#
# Args:
# directory (str): The directory to search for a WorkspaceProject.
# project_path (str): The path to the project that refers to this workspace
# element_name (str): The element in the project that was refers to this workspace
#
# Returns:
# (WorkspaceProject): The WorkspaceProject that was found for that directory.
#
def add(self, directory, project_path, element_name):
workspace_project = self.get(directory)
if not workspace_project:
workspace_project = WorkspaceProject(directory)
self._projects[directory] = workspace_project
workspace_project.add_project(project_path, element_name)
return workspace_project
# remove()
#
# Removes the project path and element name from the WorkspaceProject that exists
# for that directory.
#
# NOTE: This currently just deletes the file, but with support for multiple
# projects opening the same workspace, this will involve decreasing the count
# and deleting the file if there are no more projects.
#
# Args:
# directory (str): The directory to search for a WorkspaceProject.
#
def remove(self, directory):
workspace_project = self.get(directory)
if not workspace_project:
raise LoadError(
"Failed to find a {} file to remove".format(WORKSPACE_PROJECT_FILE), LoadErrorReason.MISSING_FILE
)
path = workspace_project.get_filename()
try:
os.unlink(path)
except FileNotFoundError:
pass
# Workspace()
#
# An object to contain various helper functions and data required for
# workspaces.
#
# last_build and path are intended to be public
# properties, but may be best accessed using this classes' helper
# methods.
#
# Args:
# toplevel_project (Project): Top project. Will be used for resolving relative workspace paths.
# path (str): The path that should host this workspace
# last_build (str): The key of the last attempted build of this workspace
#
class Workspace:
def __init__(self, toplevel_project, *, last_build=None, path=None):
self.last_build = last_build
self._path = path
self._toplevel_project = toplevel_project
self._key = None
# to_dict()
#
# Convert a list of members which get serialized to a dict for serialization purposes
#
# Returns:
# (dict) A dict representation of the workspace
#
def to_dict(self):
ret = {"path": self._path}
if self.last_build is not None:
ret["last_build"] = self.last_build
return ret
# from_dict():
#
# Loads a new workspace from a simple dictionary, the dictionary
# is expected to be generated from Workspace.to_dict(), or manually
# when loading from a YAML file.
#
# Args:
# toplevel_project (Project): Top project. Will be used for resolving relative workspace paths.
# dictionary: A simple dictionary object
#
# Returns:
# (Workspace): A newly instantiated Workspace
#
@classmethod
def from_dict(cls, toplevel_project, dictionary):
# Just pass the dictionary as kwargs
return cls(toplevel_project, **dictionary)
# differs()
#
# Checks if two workspaces are different in any way.
#
# Args:
# other (Workspace): Another workspace instance
#
# Returns:
# True if the workspace differs from 'other', otherwise False
#
def differs(self, other):
return self.to_dict() != other.to_dict()
# get_absolute_path():
#
# Returns: The absolute path of the element's workspace.
#
def get_absolute_path(self):
return os.path.join(self._toplevel_project.directory, self._path)
# Workspaces()
#
# A class to manage Workspaces for multiple elements.
#
# Args:
# toplevel_project (Project): Top project used to resolve paths.
# workspace_project_cache (WorkspaceProjectCache): The cache of WorkspaceProjects
#
class Workspaces:
def __init__(self, toplevel_project, workspace_project_cache):
self._toplevel_project = toplevel_project
self._workspace_project_cache = workspace_project_cache
# A project without a directory can happen
if toplevel_project.directory:
self._bst_directory = os.path.join(toplevel_project.directory, ".bst2")
self._workspaces = self._load_config()
else:
self._bst_directory = None
self._workspaces = {}
# list()
#
# Generator function to enumerate workspaces.
#
# Yields:
# A tuple in the following format: (str, Workspace), where the
# first element is the name of the workspaced element.
def list(self):
for element in self._workspaces.keys():
yield (element, self._workspaces[element])
# create_workspace()
#
# Create a workspace in the given path for the given element, and potentially
# checks-out the target into it.
#
# Args:
# target (Element) - The element to create a workspace for
# path (str) - The path in which the workspace should be kept
# checkout (bool): Whether to check-out the element's sources into the directory
#
def create_workspace(self, target, path, *, checkout):
element_name = target._get_full_name()
project_dir = self._toplevel_project.directory
if path.startswith(project_dir):
workspace_path = os.path.relpath(path, project_dir)
else:
workspace_path = path
self._workspaces[element_name] = Workspace(self._toplevel_project, path=workspace_path)
if checkout:
with target.timed_activity("Staging sources to {}".format(path)):
target._open_workspace()
workspace_project = self._workspace_project_cache.add(path, project_dir, element_name)
project_file_path = workspace_project.get_filename()
if os.path.exists(project_file_path):
target.warn("{} was staged from this element's sources".format(WORKSPACE_PROJECT_FILE))
workspace_project.write()
self.save_config()
# get_workspace()
#
# Get the path of the workspace source associated with the given
# element's source at the given index
#
# Args:
# element_name (str) - The element name whose workspace to return
#
# Returns:
# (None|Workspace)
#
def get_workspace(self, element_name):
if element_name not in self._workspaces:
return None
return self._workspaces[element_name]
# update_workspace()
#
# Update the datamodel with a new Workspace instance
#
# Args:
# element_name (str): The name of the element to update a workspace for
# workspace_dict (Workspace): A serialized workspace dictionary
#
# Returns:
# (bool): Whether the workspace has changed as a result
#
def update_workspace(self, element_name, workspace_dict):
assert element_name in self._workspaces
workspace = Workspace.from_dict(self._toplevel_project, workspace_dict)
if self._workspaces[element_name].differs(workspace):
self._workspaces[element_name] = workspace
return True
return False
# delete_workspace()
#
# Remove the workspace from the workspace element. Note that this
# does *not* remove the workspace from the stored yaml
# configuration, call save_config() afterwards.
#
# Args:
# element_name (str) - The element name whose workspace to delete
#
def delete_workspace(self, element_name):
workspace = self.get_workspace(element_name)
del self._workspaces[element_name]
# Remove from the cache if it exists
try:
self._workspace_project_cache.remove(workspace.get_absolute_path())
except LoadError as e:
# We might be closing a workspace with a deleted directory
if e.reason == LoadErrorReason.MISSING_FILE:
pass
else:
raise
# save_config()
#
# Dump the current workspace element to the project configuration
# file. This makes any changes performed with delete_workspace or
# create_workspace permanent
#
def save_config(self):
assert utils._is_in_main_thread()
config = {
"format-version": BST_WORKSPACE_FORMAT_VERSION,
"workspaces": {element: workspace.to_dict() for element, workspace in self._workspaces.items()},
}
os.makedirs(self._bst_directory, exist_ok=True)
_yaml.roundtrip_dump(config, self._get_filename())
# _load_config()
#
# Loads and parses the workspace configuration
#
# Returns:
# (dict) The extracted workspaces
#
# Raises: LoadError if there was a problem with the workspace config
#
def _load_config(self):
workspace_file = self._get_filename()
try:
node = _yaml.load(workspace_file, shortname="workspaces.yml")
except LoadError as e:
if e.reason == LoadErrorReason.MISSING_FILE:
# Return an empty dict if there was no workspace file
return {}
raise
return self._parse_workspace_config(node)
# _parse_workspace_config_format()
#
# If workspace config is in old-style format, i.e. it is using
# source-specific workspaces, try to convert it to element-specific
# workspaces.
#
# Args:
# workspaces (dict): current workspace config, usually output of _load_workspace_config()
#
# Returns:
# (dict) The extracted workspaces
#
# Raises: LoadError if there was a problem with the workspace config
#
def _parse_workspace_config(self, workspaces):
try:
version = workspaces.get_int("format-version", default=0)
except ValueError:
raise LoadError(
"Format version is not an integer in workspace configuration", LoadErrorReason.INVALID_DATA
)
if version < 4:
# bst 1.x workspaces do not separate source and build files.
raise LoadError(
"Workspace configuration format version {} not supported."
"Please recreate this workspace.".format(version),
LoadErrorReason.INVALID_DATA,
)
if 4 <= version <= BST_WORKSPACE_FORMAT_VERSION:
workspaces = workspaces.get_mapping("workspaces", default={})
res = {element: self._load_workspace(node) for element, node in workspaces.items()}
else:
raise LoadError(
"Workspace configuration format version {} not supported."
"Your version of buildstream may be too old. Max supported version: {}".format(
version, BST_WORKSPACE_FORMAT_VERSION
),
LoadErrorReason.INVALID_DATA,
)
return res
# _load_workspace():
#
# Loads a new workspace from a YAML node
#
# Args:
# node: A YAML dict
#
# Returns:
# (Workspace): A newly instantiated Workspace
#
def _load_workspace(self, node):
dictionary = {
"path": node.get_str("path"),
"last_build": node.get_str("last_build", default=None),
}
return Workspace.from_dict(self._toplevel_project, dictionary)
# _get_filename():
#
# Get the workspaces.yml file path.
#
# Returns:
# (str): The path to workspaces.yml file.
def _get_filename(self):
return os.path.join(self._bst_directory, "workspaces.yml")