source: rtems-central/rtemsqual/items.py @ c0ac12a

Last change on this file since c0ac12a was c0ac12a, checked in by Sebastian Huber <sebastian.huber@…>, on Feb 25, 2020 at 12:54:17 PM

Initial import

  • Property mode set to 100644
File size: 5.4 KB
Line 
1# SPDX-License-Identifier: BSD-2-Clause
2""" This module provides specification items and an item cache. """
3
4# Copyright (C) 2019, 2020 embedded brains GmbH (http://www.embedded-brains.de)
5#
6# Redistribution and use in source and binary forms, with or without
7# modification, are permitted provided that the following conditions
8# are met:
9# 1. Redistributions of source code must retain the above copyright
10#    notice, this list of conditions and the following disclaimer.
11# 2. Redistributions in binary form must reproduce the above copyright
12#    notice, this list of conditions and the following disclaimer in the
13#    documentation and/or other materials provided with the distribution.
14#
15# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
16# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
17# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
18# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
19# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
20# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
21# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
22# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
23# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
24# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
25# POSSIBILITY OF SUCH DAMAGE.
26
27# pylint: disable=useless-object-inheritance
28
29import os
30import pickle
31import stat
32from typing import Any, List, Dict
33import yaml
34
35ItemList = List["Item"]
36ItemMap = Dict[str, "Item"]
37
38
39class Item(object):
40    """ Objects of this class represent a specification item. """
41    def __init__(self, uid: str, data: Any):
42        self._uid = uid
43        self._data = data
44        self._links = []  # type: ItemList
45        self._children = []  # type: ItemList
46
47    def __getitem__(self, name: str) -> Any:
48        return self._data[name]
49
50    @property
51    def uid(self) -> str:
52        """ Returns the UID of the item. """
53        return self._uid
54
55    @property
56    def parents(self) -> ItemList:
57        """ Returns the list of parents of this items. """
58        return self._links
59
60    @property
61    def children(self) -> ItemList:
62        """ Returns the list of children of this items. """
63        return self._children
64
65    def init_parents(self, item_cache: "ItemCache"):
66        """ Initializes the list of parents of this items. """
67        for link in self._data["links"]:
68            self._links.append(item_cache[list(link.keys())[0]])
69
70    def add_child(self, child: "Item"):
71        """ Adds a child to this item. """
72        self._children.append(child)
73
74
75def _is_one_item_newer(path: str, mtime: float) -> bool:
76    for name in os.listdir(path):
77        path2 = os.path.join(path, name)
78        if name.endswith(".yml") and not name.startswith("."):
79            mtime2 = os.path.getmtime(path2)
80            if mtime <= mtime2:
81                return True
82        else:
83            mode = os.lstat(path2).st_mode
84            if stat.S_ISDIR(mode) and _is_one_item_newer(path2, mtime):
85                return True
86    return False
87
88
89def _must_update_item_cache(path: str, cache_file: str) -> bool:
90    try:
91        mtime = os.path.getmtime(cache_file)
92        return _is_one_item_newer(path, mtime)
93    except FileNotFoundError:
94        return True
95
96
97def _load_from_yaml(data_by_uid: Dict[str, Any], path: str) -> None:
98    for name in os.listdir(path):
99        path2 = os.path.join(path, name)
100        if name.endswith(".yml") and not name.startswith("."):
101            uid = os.path.basename(name).replace(".yml", "")
102            with open(path2, "r") as src:
103                data_by_uid[uid] = yaml.safe_load(src.read())
104        else:
105            mode = os.lstat(path2).st_mode
106            if stat.S_ISDIR(mode):
107                _load_from_yaml(data_by_uid, path2)
108
109
110class ItemCache(object):
111    """ This class provides a cache of specification items. """
112    def __init__(self, config: Any):
113        self._items = {}  # type: ItemMap
114        self._top_level = {}  # type: ItemMap
115        self._load_items(config)
116
117    def __getitem__(self, uid: str) -> Item:
118        return self._items[uid]
119
120    @property
121    def top_level(self) -> ItemMap:
122        """ Returns the list of top-level specification items. """
123        return self._top_level
124
125    def _load_items_in_path(self, path: str, cache_file: str) -> None:
126        data_by_uid = {}  # type: Dict[str, Any]
127        if _must_update_item_cache(path, cache_file):
128            _load_from_yaml(data_by_uid, path)
129            with open(cache_file, "wb") as out:
130                pickle.dump(data_by_uid, out)
131        else:
132            with open(cache_file, "rb") as out:
133                data_by_uid = pickle.load(out)
134        for uid, data in data_by_uid.items():
135            item = Item(uid, data)
136            self._items[uid] = item
137            if not item["links"]:
138                self._top_level[uid] = item
139
140    def _init_parents(self) -> None:
141        for item in self._items.values():
142            item.init_parents(self)
143
144    def _init_children(self) -> None:
145        for item in self._items.values():
146            for parent in item.parents:
147                parent.add_child(item)
148
149    def _load_items(self, config: Any) -> None:
150        cache_file = config["cache-file"]
151        for path in config["paths"]:
152            self._load_items_in_path(path, cache_file)
153        self._init_parents()
154        self._init_children()
Note: See TracBrowser for help on using the repository browser.