OpenSlides/openslides/utils/projector.py

"""
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]