# SPDX-License-Identifier: BSD-2-Clause
""" This module provides specification items and an item cache. """
# Copyright (C) 2019, 2022 embedded brains GmbH & Co. KG
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
from contextlib import contextmanager
import base64
import hashlib
import os
import pickle
import re
import string
import stat
from typing import Any, Callable, Dict, Iterable, Iterator, List, Match, \
NamedTuple, Optional, Set, TextIO, Tuple, Union
import json
import yaml
try:
from yaml import CSafeLoader as SafeLoader
except ImportError: # pragma: no cover
from yaml import SafeLoader # type: ignore
class ItemGetValueContext(NamedTuple):
""" Context used to get an item value. """
item: "Item"
path: str
value: Any
key: str
index: Any # should be int, but this triggers a mypy error
args: Optional[str]
def arg(self, name: str, value: Optional[str] = None) -> str:
""" Get argument value by name. """
args = dict(
kv.split("=") # type: ignore
for kv in self.args.split(",")) # type: ignore
if value:
return args.get(name, value)
return args[name]
ItemMap = Dict[str, "Item"]
ItemGetValue = Callable[[ItemGetValueContext], Any]
ItemGetValueMap = Dict[str, Tuple[ItemGetValue, Any]]
def _is_enabled_op_and(enabled: List[str], enabled_by: Any) -> bool:
for next_enabled_by in enabled_by:
if not is_enabled(enabled, next_enabled_by):
return False
return True
def _is_enabled_op_not(enabled: List[str], enabled_by: Any) -> bool:
return not is_enabled(enabled, enabled_by)
def _is_enabled_op_or(enabled: List[str], enabled_by: Any) -> bool:
for next_enabled_by in enabled_by:
if is_enabled(enabled, next_enabled_by):
return True
return False
_IS_ENABLED_OP = {
"and": _is_enabled_op_and,
"not": _is_enabled_op_not,
"or": _is_enabled_op_or
}
def is_enabled(enabled: List[str], enabled_by: Any) -> bool:
""" Verifies if the given parameter is enabled by specific enables. """
if isinstance(enabled_by, bool):
return enabled_by
if isinstance(enabled_by, list):
return _is_enabled_op_or(enabled, enabled_by)
if isinstance(enabled_by, dict):
key, value = next(iter(enabled_by.items()))
return _IS_ENABLED_OP[key](enabled, value)
return enabled_by in enabled
def _str_representer(dumper, data):
return dumper.represent_scalar("tag:yaml.org,2002:str",
data,
style="|" if "\n" in data else "")
yaml.add_representer(str, _str_representer)
class Link:
""" A link to an item. """
def __init__(self, item: "Item", data: Any):
self._item = item
self._data = data
@classmethod
def create(cls, link: "Link", item: "Item") -> "Link":
""" Creates a link using an existing link with a new target item. """
return cls(item, link._data) # pylint: disable=protected-access
def __getitem__(self, name: str) -> Any:
return self._data[name]
def __setitem__(self, key: str, value: Any) -> None:
self._data[key] = value
@property
def item(self) -> "Item":
""" The item referenced by this link. """
return self._item
@property
def role(self) -> str:
""" The link role. """
return self._data["role"]
def _get_value(ctx: ItemGetValueContext) -> Any:
value = ctx.value[ctx.key]
if ctx.index >= 0:
return value[ctx.index]
return value
def normalize_key_path(key_path: str, prefix: str = "") -> str:
""" Normalizes the key path with an optional prefix path. """
if not os.path.isabs(key_path):
key_path = os.path.join(prefix, key_path)
return os.path.normpath(key_path)
_TYPES = {
type(True): "B".encode("utf-8"),
type(1.0): "F".encode("utf-8"),
type(1): "I".encode("utf-8"),
type(None): "N".encode("utf-8"),
type(""): "S".encode("utf-8"),
}
def _hash_data(data, state) -> None:
if isinstance(data, list):
for value in data:
_hash_data(value, state)
elif isinstance(data, dict):
for key, value in sorted(data.items()):
if not key.startswith("_"):
state.update(key.encode("utf-8"))
_hash_data(value, state)
else:
state.update(_TYPES[type(data)])
state.update(str(data).encode("utf-8"))
def data_digest(data: Any) -> str:
""" Returns a digest of the data. """
state = hashlib.sha256()
_hash_data(data, state)
return base64.urlsafe_b64encode(state.digest()).decode("ascii")
_UID_TO_UPPER = re.compile(r"[/_-]+(.)")
def _match_to_upper(match: Match) -> str:
return match.group(1).upper()
def _is_link_enabled(link: Link) -> bool:
return link.item._data["_enabled"] # pylint: disable=protected-access
class Item:
""" Objects of this class represent a specification item. """
# pylint: disable=too-many-public-methods
def __init__(self, item_cache: "ItemCache", uid: str, data: Any):
self._cache = item_cache
self._ident = _UID_TO_UPPER.sub(_match_to_upper, uid)
self._uid = uid
self._data = data
self._links_to_parents: List[Link] = []
self._links_to_children: List[Link] = []
self._resolved_proxy = False
def __eq__(self, other: Any) -> bool:
if not isinstance(other, Item):
return NotImplemented
return self._uid == other._uid # pylint: disable=protected-access
def __lt__(self, other: Any) -> bool:
if not isinstance(other, Item):
return NotImplemented
return self._uid < other._uid # pylint: disable=protected-access
def __hash__(self) -> int:
return hash(self._uid)
def __contains__(self, key: str) -> bool:
return key in self._data
def __getitem__(self, key: str) -> Any:
return self._data[key]
def __setitem__(self, key: str, value: Any) -> None:
self._data[key] = value
@property
def cache(self) -> "ItemCache":
""" Returns the cache of the item. """
return self._cache
@property
def digest(self) -> str:
""" Returns the digest of the item data. """
return data_digest(self._data)
def get(self, key: str, default: Any) -> Any:
"""
Gets the attribute value if the attribute exists, otherwise the
specified default value is returned.
"""
return self._data.get(key, default)
@property
def uid(self) -> str:
""" Returns the UID of the item. """
return self._uid
@property
def ident(self) -> str:
""" Returns the identifier of the item. """
return self._ident
@property
def spec(self) -> str:
""" Returns the UID of the item with an URL-like format. """
return f"spec:{self._uid}"
@property
def spec_2(self) -> str:
"""
Returns the UID of the item with an URL-like format with invisible
white space to allow line breaks.
"""
uid = self._uid.replace("/", "/\u200b")
return f"spec:{uid}"
def to_abs_uid(self, abs_or_rel_uid: str) -> str:
"""
Returns the absolute UID of an absolute UID or an UID relative to this
item.
"""
if abs_or_rel_uid == ".":
return self._uid
if os.path.isabs(abs_or_rel_uid):
return abs_or_rel_uid
return os.path.normpath(
os.path.join(os.path.dirname(self.uid), abs_or_rel_uid))
def map(self, abs_or_rel_uid: str) -> "Item":
"""
Maps the absolute UID or the UID relative to this item to the
corresponding item.
"""
return self._cache[self.to_abs_uid(abs_or_rel_uid)]
def links_to_parents(
self,
role: Optional[Union[str, Iterable[str]]] = None,
is_link_enabled: Callable[[Link], bool] = _is_link_enabled
) -> Iterator[Link]:
""" Yields the links to the parents of this items. """
if role is None:
for link in self._links_to_parents:
if is_link_enabled(link):
yield link
elif isinstance(role, str):
for link in self._links_to_parents:
if link.role == role and is_link_enabled(link):
yield link
else:
for link in self._links_to_parents:
if link.role in role and is_link_enabled(link):
yield link
def parents(
self,
role: Optional[Union[str, Iterable[str]]] = None,
is_link_enabled: Callable[[Link], bool] = _is_link_enabled
) -> Iterator["Item"]:
""" Yields the parents of this items. """
for link in self.links_to_parents(role, is_link_enabled):
yield link.item
def parent(
self,
role: Optional[Union[str, Iterable[str]]] = None,
index: Optional[int] = 0,
is_link_enabled: Callable[[Link],
bool] = _is_link_enabled) -> "Item":
""" Returns the parent with the specified role and index. """
for item_index, item in enumerate(self.parents(role, is_link_enabled)):
if item_index == index:
return item
raise IndexError
def parent_link(
self,
role: Optional[Union[str, Iterable[str]]] = None,
index: Optional[int] = 0,
is_link_enabled: Callable[[Link],
bool] = _is_link_enabled) -> Link:
""" Returns the parent link with the specified role and index. """
for link_index, link in enumerate(
self.links_to_parents(role, is_link_enabled)):
if link_index == index:
return link
raise IndexError
def links_to_children(
self,
role: Optional[Union[str, Iterable[str]]] = None,
is_link_enabled: Callable[[Link], bool] = _is_link_enabled
) -> Iterator[Link]:
""" Yields the links to the children of this items. """
if role is None:
for link in self._links_to_children:
if is_link_enabled(link):
yield link
elif isinstance(role, str):
for link in self._links_to_children:
if link.role == role and is_link_enabled(link):
yield link
else:
for link in self._links_to_children:
if link.role in role and is_link_enabled(link):
yield link
def children(
self,
role: Optional[Union[str, Iterable[str]]] = None,
is_link_enabled: Callable[[Link], bool] = _is_link_enabled
) -> Iterator["Item"]:
""" Yields the children of this items. """
for link in self.links_to_children(role, is_link_enabled):
yield link.item
def child(
self,
role: Optional[Union[str, Iterable[str]]] = None,
index: Optional[int] = 0,
is_link_enabled: Callable[[Link],
bool] = _is_link_enabled) -> "Item":
""" Returns the child with the specified role and index. """
for item_index, item in enumerate(self.children(role,
is_link_enabled)):
if item_index == index:
return item
raise IndexError
def child_link(
self,
role: Optional[Union[str, Iterable[str]]] = None,
index: Optional[int] = 0,
is_link_enabled: Callable[[Link],
bool] = _is_link_enabled) -> Link:
""" Returns the child link with the specified role and index. """
for link_index, link in enumerate(
self.links_to_children(role, is_link_enabled)):
if link_index == index:
return link
raise IndexError
def init_parents(self, item_cache: "ItemCache") -> None:
""" Initializes the list of links to parents of this items. """
for data in self._data["links"]:
try:
link = Link(item_cache[self.to_abs_uid(data["uid"])], data)
self._links_to_parents.append(link)
except KeyError as err:
msg = (f"item '{self.uid}' links "
f"to non-existing item '{data['uid']}'")
raise KeyError(msg) from err
def init_children(self) -> None:
""" Initializes the list of links to children of this items. """
for link in self._links_to_parents:
link.item.add_link_to_child(Link.create(link, self))
def add_link_to_parent(self, link: Link):
""" Adds the link as a parent item link to this item. """
self._links_to_parents.append(link)
def add_link_to_child(self, link: Link):
""" Adds the link as a child item link to this item. """
self._links_to_children.append(link)
def is_enabled(self, enabled: List[str]):
"""
Returns true if the item is enabled by the enabled set, otherwise
returns false.
"""
return is_enabled(enabled, self._data["enabled-by"])
@property
def enabled(self) -> bool:
""" Returns true if the item is enabled, otherwise returns false. """
return self._data["_enabled"]
@property
def resolved_proxy(self) -> bool:
""" Is true if the item is a resolved proxy, otherwise false. """
return self._resolved_proxy
@property
def data(self) -> Any:
""" The item data. """
return self._data
@property
def file(self) -> str:
""" Returns the file of the item. """
return self._data["_file"]
@file.setter
def file(self, value: str):
""" Sets the file of the item. """
self._data["_file"] = value
@property
def type(self) -> str:
""" Returns the type of the item. """
return self._data["_type"]
def save(self):
""" Saves the item to the corresponding file. """
self._cache.save_data(self.file, self._data)
def load(self):
""" Loads the item from the corresponding file. """
self._data = self._cache.load_data(self.file, self._uid)
def create_unique_link(child: Item, parent: Item, data: Any) -> None:
"""
Creates a unique link from the child to the parent item and vice versa
using the data for the link.
"""
for item in parent.children(data["role"]):
if item.uid == child.uid:
break
else:
parent.add_link_to_child(Link(child, data))
child.add_link_to_parent(Link(parent, data))
class ItemTemplate(string.Template):
""" String template for item mapper identifiers. """
idpattern = "[a-zA-Z0-9._/-]+:[\\[\\]a-zA-Z0-9._/-]+(:[^${}]*)?"
class _ItemMapperContext(dict):
""" Context to map identifiers to items and attribute values. """
def __init__(self, mapper: "ItemMapper", item: Optional[Item],
prefix: Optional[str], recursive: bool):
super().__init__()
self._mapper = mapper
self._item = item
self._prefix = prefix
self._recursive = recursive
def __getitem__(self, identifier):
item, key_path, value = self._mapper.map(identifier, self._item,
self._prefix)
if self._recursive:
return self._mapper.substitute(value, item,
os.path.dirname(key_path))
return value
class _GetValueDictionary(dict):
def __init__(self, get_value: ItemGetValue):
super().__init__()
self._get_value = get_value
def get(self, _key, _default):
return (self._get_value, {})
class ItemMapper:
""" Maps identifiers to items and attribute values. """
def __init__(self, item: Item, recursive: bool = False):
self._item = item
self._recursive = recursive
self._prefix = [""]
self._get_value_map: Dict[str, ItemGetValueMap] = {}
@property
def item(self) -> Item:
""" The item of the mapper. """
return self._item
@item.setter
def item(self, item: Item) -> None:
""" Sets the item of the mapper. """
self._item = item
def _add_get_value_map(
self, type_path_key: str, new_get_value_map: Tuple[ItemGetValue,
Dict]) -> None:
type_name, path_key = type_path_key.split(":")
keys = path_key.strip("/").split("/")
get_value_map = self._get_value_map.setdefault(type_name, {})
for key in keys[:-1]:
_, get_value_map = get_value_map.setdefault(key, (_get_value, {}))
get_value_map[keys[-1]] = new_get_value_map
def add_get_value(self, type_path_key: str,
get_value: ItemGetValue) -> None:
"""
Adds a get value for the specified type and key path.
"""
self._add_get_value_map(type_path_key, (get_value, {}))
def add_get_value_dictionary(self, type_path_key: str,
get_value: ItemGetValue) -> None:
"""
Adds a get value dictionary for the specified type and key path.
"""
self._add_get_value_map(type_path_key,
(_get_value, _GetValueDictionary(get_value)))
def push_prefix(self, prefix: str) -> None:
""" Pushes a key path prefix. """
self._prefix.append(prefix)
def pop_prefix(self) -> None:
""" Pops a key path prefix. """
self._prefix.pop()
@contextmanager
def prefix(self, prefix: str) -> Iterator[None]:
""" Opens a key path prefix context. """
self.push_prefix(prefix)
yield
self.pop_prefix()
def get_value_map(self, item: Item) -> ItemGetValueMap:
""" Returns the get value map for the item. """
return self._get_value_map.get(item.type, {})
def _get_by_normalized_key_path(self, item: Item, normalized_key_path: str,
args: Optional[str]) -> Any:
"""
Gets the attribute value associated with the normalized key path.
"""
get_value_map = self.get_value_map(item)
path = "/"
value = item.data
for key in normalized_key_path.strip("/").split("/"):
parts = key.split("[")
try:
index = int(parts[1].split("]")[0])
except IndexError:
index = -1
ctx = ItemGetValueContext(item, path, value, parts[0], index, args)
get_value, get_value_map = get_value_map.get(
parts[0], (_get_value, {}))
value = get_value(ctx)
path = os.path.join(path, key)
return value
def map(self,
identifier: str,
item: Optional[Item] = None,
prefix: Optional[str] = None) -> Tuple[Item, str, Any]:
"""
Maps the identifier with item and prefix to the associated item, key
path, and attribute value.
"""
colon = identifier.find(":")
uid = identifier[:colon]
more = identifier[colon + 1:]
colon = more.find(":")
if colon < 0:
key_path = more
args = None
else:
key_path = more[:colon]
args = more[colon + 1:]
if item is None:
item = self._item
if uid == ".":
if prefix is None:
prefix = "/".join(self._prefix)
else:
prefix = ""
try:
item = item.map(uid)
except KeyError as err:
msg = (f"item '{uid}' relative to {item.spec} "
f"specified by '{identifier}' does not exist")
raise ValueError(msg) from err
key_path = normalize_key_path(key_path, prefix)
try:
value = self._get_by_normalized_key_path(item, key_path, args)
except Exception as err:
msg = (f"cannot get value for '{key_path}' of {item.spec} "
f"specified by '{identifier}'")
raise ValueError(msg) from err
return item, key_path, value
def __getitem__(self, identifier):
item, key_path, value = self.map(identifier)
if self._recursive:
return self.substitute(value, item, os.path.dirname(key_path))
return value
def substitute(self,
text: Optional[str],
item: Optional[Item] = None,
prefix: Optional[str] = None) -> str:
"""
Performs a variable substitution using the item mapper with the item
and prefix.
"""
if not text:
return ""
try:
context = _ItemMapperContext(self, item, prefix, self._recursive)
return ItemTemplate(text).substitute(context)
except Exception as err:
spec = self._item.spec if item is None else item.spec
if prefix is None:
prefix = "/".join(self._prefix)
msg = (f"substitution for {spec} using prefix '{prefix}' "
f"failed for text: {text}")
raise ValueError(msg) from err
class _SpecType(NamedTuple):
key: str
refinements: Dict[str, Any]
def _gather_spec_refinements(item: Item) -> Optional[_SpecType]:
new_type: Optional[_SpecType] = None
for link in item._links_to_children: # pylint: disable=protected-access
if link.role == "spec-refinement":
key = link["spec-key"]
if new_type is None:
new_type = _SpecType(key, {})
assert new_type.key == key
new_type.refinements[
link["spec-value"]] = _gather_spec_refinements(link.item)
return new_type
def _load_yaml_data(path: str, uid: str) -> Any:
with open(path, "r", encoding="utf-8") as src:
try:
data = yaml.load(src.read(), Loader=SafeLoader)
except yaml.YAMLError as err:
msg = ("YAML error while loading specification item file "
f"'{path}': {str(err)}")
raise IOError(msg) from err
data["_file"] = os.path.abspath(path)
data["_uid"] = uid
return data
def _load_json_data(path: str, uid: str) -> Any:
with open(path, "r", encoding="utf-8") as src:
try:
data = json.load(src)
except json.JSONDecodeError as err:
msg = ("JSON error while loading specification item file "
f"'{path}': {str(err)}")
raise IOError(msg) from err
data["_file"] = os.path.abspath(path)
data["_uid"] = uid
return data
def _is_item_enabled(enabled: List[str], item: Item) -> bool:
return is_enabled(enabled, item["enabled-by"])
def item_is_enabled(_enabled: List[str], _item: Item) -> bool:
""" Returns true. """
return True
def _resolve_proxy(proxy: Item, is_link_enabled: Callable[[Link],
bool]) -> None:
# pylint: disable=protected-access
try:
member = proxy.child("proxy-member", is_link_enabled=is_link_enabled)
except IndexError:
pass
else:
member._links_to_parents.extend(proxy._links_to_parents)
member._links_to_children.extend(proxy._links_to_children)
proxy._data = member._data
proxy._ident = member._ident
proxy._resolved_proxy = True
proxy._uid = member._uid
for link in proxy._links_to_parents:
for link_2 in link.item._links_to_children:
if link_2.item == proxy:
link_2._item = member
for link in proxy._links_to_children:
for link_2 in link.item._links_to_parents:
if link_2.item == proxy:
link_2._item = member
proxy._links_to_children = member._links_to_children
proxy._links_to_parents = member._links_to_parents
class ItemCache:
""" This class provides a cache of specification items. """
def __init__(self,
config: Any,
post_process_load: Optional[Callable[[ItemMap], None]] = None,
is_item_enabled: Callable[[List[str], Item],
bool] = _is_item_enabled):
self._items: ItemMap = {}
self._types: Set[str] = set()
self.items_by_type: Dict[str, List[Item]] = {}
self._updates = 0
self._load_items(config)
if post_process_load:
post_process_load(self._items)
if config.get("initialize-links", True):
self._init_parents()
self._init_children()
spec_root = config["spec-type-root-uid"]
if spec_root:
self._root_type = _gather_spec_refinements(self[spec_root])
else:
self._root_type = None
self._enabled = config.get("enabled", [])
self._is_enabled = is_item_enabled
for item in self._items.values():
self._set_type(item)
item["_enabled"] = is_item_enabled(self._enabled, item)
if config.get("resolve-proxies", False):
self.resolve_proxies()
def __getitem__(self, uid: str) -> Item:
return self._items[uid]
@property
def updates(self) -> bool:
"""
Returns true if the item cache updates occurred due to new, modified,
or removed files.
"""
return self._updates > 0
@property
def all(self) -> ItemMap:
""" Returns the map of all specification items. """
return self._items
@property
def types(self) -> Set[str]:
""" Returns the types of the items. """
return self._types
@property
def enabled(self) -> List[str]:
""" Returns the enabled set. """
return self._enabled
def set_enabled(self,
enabled: List[str],
is_item_enabled: Callable[[List[str], Item],
bool] = _is_item_enabled):
"""
Sets the enabled status of all items according to the enabled set using
the is item enabled function.
"""
self._enabled = enabled
self._is_enabled = is_item_enabled
for item in self._items.values():
item["_enabled"] = is_item_enabled(enabled, item)
def resolve_proxies(
self,
is_link_enabled: Callable[[Link],
bool] = _is_link_enabled) -> None:
""" Resolves each proxy item to the its first enabled member. """
for item in self.items_by_type.get("proxy", []):
_resolve_proxy(item, is_link_enabled)
def add_volatile_item(self, uid: str, data: Any) -> Item:
"""
Adds an item with the specified data to the cache and returns it.
The item is not added to the persistent cache storage.
"""
item = self._add_item(uid, data)
item.init_parents(self)
item.init_children()
self._set_type(item)
item["_enabled"] = self._is_enabled(self._enabled, item)
return item
def add_volatile_item_from_file(self, uid: str, path: str) -> Item:
"""
Adds an item stored in the specified file to the cache and returns it.
The item is not added to the persistent cache storage.
"""
return self.add_volatile_item(uid, self.load_data(path, uid))
def _add_item(self, uid: str, data: Any) -> Item:
item = Item(self, uid, data)
self._items[uid] = item
return item
def _load_items_in_dir(self, base: str, path: str, cache_file: str,
update_cache: bool) -> None:
data_by_uid: Dict[str, Any] = {}
if update_cache:
self._updates += 1
for name in os.listdir(path):
path2 = os.path.join(path, name)
if name.endswith(".yml") and not name.startswith("."):
uid = "/" + os.path.relpath(path2, base).replace(
".yml", "")
data_by_uid[uid] = _load_yaml_data(path2, uid)
os.makedirs(os.path.dirname(cache_file), exist_ok=True)
with open(cache_file, "wb") as out:
pickle.dump(data_by_uid, out)
else:
with open(cache_file, "rb") as pickle_src:
data_by_uid = pickle.load(pickle_src)
for uid, data in iter(data_by_uid.items()):
self._add_item(uid, data)
def _load_items_recursive(self, index: str, base: str, path: str,
cache_dir: str) -> None:
mid = os.path.abspath(path)
mid = mid.replace(os.path.commonpath([cache_dir, mid]), "").strip("/")
cache_file = os.path.join(cache_dir, index, mid, "spec.pickle")
try:
mtime = os.path.getmtime(cache_file)
update_cache = False
except FileNotFoundError:
update_cache = True
else:
update_cache = mtime <= os.path.getmtime(path)
for name in os.listdir(path):
path2 = os.path.join(path, name)
if name.endswith(".yml") and not name.startswith("."):
if not update_cache:
update_cache = mtime <= os.path.getmtime(path2)
elif stat.S_ISDIR(os.lstat(path2).st_mode):
self._load_items_recursive(index, base, path2, cache_dir)
self._load_items_in_dir(base, path, cache_file, update_cache)
def _load_items(self, config: Any):
cache_dir = os.path.abspath(config["cache-directory"])
for index, path in enumerate(config["paths"]):
self._load_items_recursive(str(index), path, path, cache_dir)
def load_data(self, path: str, uid: str) -> Any:
""" Loads the item data from the file specified by path. """
return _load_yaml_data(path, uid)
def _save_data(self, file: TextIO, data: Any) -> None:
file.write(
yaml.dump(data, default_flow_style=False, allow_unicode=True))
def save_data(self, path: str, data: Any) -> None:
""" Saves the item data to the file specified by path. """
with open(path, "w", encoding="utf-8") as file:
data2 = {}
for key, value in data.items():
if not key.startswith("_"):
data2[key] = value
self._save_data(file, data2)
def _init_parents(self) -> None:
for item in self._items.values():
item.init_parents(self)
def _init_children(self) -> None:
for uid in sorted(self._items):
self._items[uid].init_children()
def _set_type(self, item: Item) -> None:
spec_type = self._root_type
value = item.data
path: List[str] = []
while spec_type is not None:
type_name = value[spec_type.key]
path.append(type_name)
spec_type = spec_type.refinements[type_name]
the_type = "/".join(path)
item["_type"] = the_type
self._types.add(the_type)
self.items_by_type.setdefault(the_type, []).append(item)
class EmptyItemCache(ItemCache):
""" This class provides a empty cache of specification items. """
def __init__(self):
super().__init__({
"cache-directory": ".",
"paths": [],
"spec-type-root-uid": None
})
class JSONItemCache(ItemCache):
""" This class provides a cache of specification items using JSON. """
def _load_json_items(self, base: str, path: str) -> None:
for name in os.listdir(path):
path2 = os.path.join(path, name)
if name.endswith(".json") and not name.startswith("."):
uid = "/" + os.path.relpath(path2, base).replace(".json", "")
self._add_item(uid, _load_json_data(path2, uid))
elif stat.S_ISDIR(os.lstat(path2).st_mode):
self._load_json_items(base, path2)
def _load_items(self, config: Any):
for path in config["paths"]:
self._load_json_items(path, path)
def load_data(self, path: str, uid: str) -> Any:
return _load_json_data(path, uid)
def _save_data(self, file: TextIO, data: Any) -> None:
json.dump(data, file, sort_keys=True, indent=2)
class EmptyItem(Item):
""" Objects of this class represent empty items. """
def __init__(self):
super().__init__(EmptyItemCache(), "", {})
