127 lines
3.8 KiB
Python
127 lines
3.8 KiB
Python
"""
|
|
General projector code.
|
|
|
|
Functions that handel the registration of projector elements and the rendering
|
|
of the data to present it on the projector.
|
|
"""
|
|
|
|
from typing import Any, Awaitable, Callable, Dict, List
|
|
|
|
from .cache import element_cache
|
|
|
|
|
|
AllData = Dict[str, Dict[int, Dict[str, Any]]]
|
|
ProjectorSlide = Callable[[AllData, Dict[str, Any], int], Awaitable[Dict[str, Any]]]
|
|
|
|
|
|
projector_slides: Dict[str, ProjectorSlide] = {}
|
|
|
|
|
|
class ProjectorElementException(Exception):
|
|
"""
|
|
Exception for errors in one element on the projector.
|
|
"""
|
|
|
|
|
|
def register_projector_slide(name: str, slide: ProjectorSlide) -> None:
|
|
"""
|
|
Registers a projector slide.
|
|
|
|
Has to be called in the app.ready method.
|
|
"""
|
|
projector_slides[name] = slide
|
|
|
|
|
|
async def get_projector_data(
|
|
projector_ids: List[int] = None,
|
|
) -> Dict[int, List[Dict[str, Any]]]:
|
|
"""
|
|
Calculates and returns the data for one or all projectors.
|
|
|
|
The keys of the returned data are the projector ids as int. When converted
|
|
to json, the numbers will changed to strings like "1".
|
|
|
|
The data for each projector is a list of elements.
|
|
|
|
Each element is a dict where the keys are "elements", "data". "elements"
|
|
contains the projector elements. It is the same as the projector elements in
|
|
the database. "data" contains all necessary data to render the projector
|
|
element. The key can also be "error" if there is a generall error for the
|
|
slide. In this case the values "elements" and "data" are optional.
|
|
|
|
The returned value looks like this:
|
|
|
|
projector_data = {
|
|
1: [
|
|
{
|
|
"element": {
|
|
"name": "agenda/item-list",
|
|
},
|
|
"data": {
|
|
"items": []
|
|
},
|
|
},
|
|
],
|
|
}
|
|
"""
|
|
if projector_ids is None:
|
|
projector_ids = []
|
|
|
|
all_data = await element_cache.get_all_data_dict()
|
|
projector_data: Dict[int, List[Dict[str, Any]]] = {}
|
|
|
|
for projector_id, projector in all_data.get("core/projector", {}).items():
|
|
if projector_ids and projector_id not in projector_ids:
|
|
# only render the projector in question.
|
|
continue
|
|
|
|
if not projector["elements"]:
|
|
# Skip empty elements.
|
|
continue
|
|
|
|
projector_data[projector_id] = []
|
|
for element in projector["elements"]:
|
|
projector_slide = projector_slides[element["name"]]
|
|
try:
|
|
data = await projector_slide(all_data, element, projector_id)
|
|
except ProjectorElementException as err:
|
|
data = {"error": str(err)}
|
|
projector_data[projector_id].append({"data": data, "element": element})
|
|
|
|
return projector_data
|
|
|
|
|
|
async def get_config(all_data: AllData, key: str) -> Any:
|
|
"""
|
|
Returns a config value from all_data.
|
|
"""
|
|
from ..core.config import config
|
|
|
|
config_id = (await config.async_get_key_to_id())[key]
|
|
|
|
return all_data[config.get_collection_string()][config_id]["value"]
|
|
|
|
|
|
def get_model(all_data: AllData, collection: str, id: Any) -> Dict[str, Any]:
|
|
"""
|
|
Tries to get the model identified by the collection and id.
|
|
If the id is invalid or the model not found, ProjectorElementExceptions will be raised.
|
|
"""
|
|
if id is None:
|
|
raise ProjectorElementException(f"id is required for {collection} slide")
|
|
|
|
try:
|
|
model = all_data[collection][id]
|
|
except KeyError:
|
|
raise ProjectorElementException(f"{collection} with id {id} does not exist")
|
|
return model
|
|
|
|
|
|
def get_models(
|
|
all_data: AllData, collection: str, ids: List[Any]
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Tries to fetch all given models. Models are required to be all of the collection `collection`.
|
|
"""
|
|
return [get_model(all_data, collection, id) for id in ids]
|