Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
kiara / models / context.py
Size: Mime:
# -*- coding: utf-8 -*-
import uuid
from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Union

import humanfriendly
from humanfriendly import format_size
from pydantic import Field, PrivateAttr, RootModel
from rich import box
from rich.console import Group, RenderableType
from rich.markdown import Markdown
from rich.table import Table

from kiara.models import KiaraModel
from kiara.models.archives import ArchiveGroupInfo

if TYPE_CHECKING:
    from kiara.context import Kiara, KiaraConfig, KiaraContextConfig, KiaraRuntimeConfig


class ContextInfo(KiaraModel):
    @classmethod
    def create_from_context_config(
        cls,
        config: "KiaraContextConfig",
        context_name: Union[str, None] = None,
        runtime_config: Union["KiaraRuntimeConfig", None] = None,
    ) -> "ContextInfo":

        from kiara.context import Kiara

        kiara = Kiara(config=config, runtime_config=runtime_config)
        return cls.create_from_context(kiara=kiara, context_name=context_name)

    @classmethod
    def create_from_context(
        cls, kiara: "Kiara", context_name: Union[str, None] = None
    ) -> "ContextInfo":

        errors = {}
        try:
            value_ids = list(kiara.data_registry.retrieve_all_available_value_ids())
        except Exception as e:
            errors["values"] = e
            value_ids = []

        try:
            aliases = {
                a.full_alias: a.value_id for a in kiara.alias_registry.aliases.values()
            }
        except Exception as e:
            errors["aliases"] = e
            aliases = {}

        try:
            archives_info = ArchiveGroupInfo.create_from_context(kiara=kiara)
        except Exception as e:
            errors["archives"] = e
            archives_info = ArchiveGroupInfo(item_infos={})

        comment = None
        invalid = False
        if errors:
            invalid = True
            comment = "Errors creating this context, this is most likely a bug:\n\n"
            for k, err in errors.items():
                comment = f"{comment}  - {k}: {err}"

        result = ContextInfo(
            kiara_id=kiara.id,
            value_ids=value_ids,
            aliases=aliases,
            context_name=context_name,
            archives=archives_info,
            invalid=invalid,
            comment=comment,
        )
        result._kiara = kiara
        return result

    kiara_id: uuid.UUID = Field(
        description="The (globally unique) id of the kiara context."
    )
    context_name: Union[str, None] = Field(
        description="The local alias for this context."
    )
    value_ids: List[uuid.UUID] = Field(
        description="The ids of all stored values in this context."
    )
    aliases: Dict[str, uuid.UUID] = Field(
        description="All available aliases within this context (and the value ids they refer to)."
    )
    archives: ArchiveGroupInfo = Field(
        description="The archives registered in this context."
    )
    invalid: bool = Field(description="Whether this context has errors.", default=False)
    comment: Union[str, None] = Field(
        default=None, description="(Optional) comment about this context."
    )

    _kiara: Union["Kiara", None] = PrivateAttr()

    @property
    def kiara_context(self) -> "Kiara":
        if self._kiara is None:
            raise Exception("Kiara context object not set.")
        return self._kiara

    def value_summary(self) -> Dict[str, Any]:

        sum_size = 0
        types: Dict[str, int] = {}
        internal_types: Dict[str, int] = {}
        no_of_values = len(self.value_ids)

        for value_id in self.value_ids:
            value = self.kiara_context.data_registry.get_value(value=value_id)
            sum_size = sum_size + value.value_size
            if self.kiara_context.type_registry.is_internal_type(value.data_type_name):
                if value.data_type_name not in internal_types.keys():
                    internal_types[value.data_type_name] = 1
                else:
                    internal_types[value.data_type_name] += 1
            else:
                if value.data_type_name not in types.keys():
                    types[value.data_type_name] = 1
                else:
                    types[value.data_type_name] += 1

            types.setdefault(value.data_type_name, 0)

        return {
            "size": sum_size,
            "no_values": no_of_values,
            "types": types,
            "internal_types": internal_types,
        }

    def alias_summary(self) -> Dict[str, Any]:

        sum_size = 0
        types: Dict[str, int] = {}
        internal_types: Dict[str, int] = {}
        no_of_values = len(self.value_ids)

        for alias, value_id in self.aliases.items():
            value = self.kiara_context.data_registry.get_value(value=value_id)
            sum_size = sum_size + value.value_size
            if self.kiara_context.type_registry.is_internal_type(value.data_type_name):
                if value.data_type_name not in internal_types.keys():
                    internal_types[value.data_type_name] = 1
                else:
                    internal_types[value.data_type_name] += 1
            else:
                if value.data_type_name not in types.keys():
                    types[value.data_type_name] = 1
                else:
                    types[value.data_type_name] += 1

            types.setdefault(value.data_type_name, 0)

        return {
            "size": sum_size,
            "no_values": no_of_values,
            "types": types,
            "internal_types": internal_types,
        }

    def create_renderable(self, **config: Any) -> RenderableType:

        full_details = config.get("full_details", False)
        show_value_ids = config.get("show_value_ids", False)
        show_archive_info = config.get("show_archive_info", True)

        table = Table(box=box.SIMPLE, show_header=False)

        table.add_column("Property", style="i")
        table.add_column("Value")

        if self.invalid:
            msg = "[red]This context is invalid![/red]\n"
            cmt = self.comment if self.comment else "-- no details --"
            md = Markdown(cmt)
            table.add_row(
                "",
                Group(msg, md, ""),
            )

        if self.context_name:
            table.add_row("context name", self.context_name)
        table.add_row("kiara_id", str(self.kiara_id))

        size_on_disk = humanfriendly.format_size(self.archives.combined_size)
        table.add_row("size on disk", size_on_disk)

        value_sum = self.value_summary()
        v_table = Table(box=box.SIMPLE, show_header=False)
        v_table.add_column("Property")
        v_table.add_column("Value")
        v_table.add_row("no. values", str(value_sum["no_values"]))
        v_table.add_row("combined size", format_size(value_sum["size"]))
        if full_details and show_value_ids:
            if self.value_ids:
                value_ids = sorted((str(v) for v in self.value_ids))
                v_table.add_row("value_ids", value_ids[0])
                for v_id in value_ids[1:]:
                    v_table.add_row("", v_id)
            else:
                v_table.add_row("value_ids", "")
        table.add_row("values", v_table)

        alias_sum = self.alias_summary()
        a_table = Table(box=box.SIMPLE, show_header=False)
        a_table.add_column("Property")
        a_table.add_column("Value")
        a_table.add_row("no. aliases", str(len(self.aliases)))
        a_table.add_row("combined size", format_size(alias_sum["size"]))
        if full_details:
            if self.aliases:
                aliases = sorted(self.aliases.keys())
                a_table.add_row(
                    "aliases", f"{aliases[0]} -> {self.aliases[aliases[0]]}"
                )
                for alias in aliases[1:]:
                    a_table.add_row("", f"{alias} -> {self.aliases[alias]}")
            else:
                a_table.add_row("aliases", "")
        table.add_row("aliases", a_table)

        if show_archive_info:
            table.add_row("archives", self.archives)

        return table


class ContextInfos(RootModel):
    root: Dict[str, ContextInfo]

    @classmethod
    def create_context_infos(
        cls, contexts: Union[Mapping[str, "KiaraContextConfig"], None] = None
    ) -> "ContextInfos":

        if not contexts:
            kc = KiaraConfig()
            contexts = kc.context_configs

        return ContextInfos(
            root={
                a: ContextInfo.create_from_context_config(c, context_name=a)
                for a, c in contexts.items()
            }
        )

    def create_renderable(self, **config: Any) -> RenderableType:

        full_details = config.get("full_details", False)

        if not full_details:
            table = Table(box=box.SIMPLE, show_header=True, show_lines=False)
            table.add_column("context name", style="i")
            table.add_column("context id", style="i")
            table.add_column("size on disk")
            table.add_column("size of all values")
            table.add_column("no. values")
            table.add_column("no. aliaes")
            for context_name, context_summary in self.root.items():
                size_on_disk = context_summary.archives.combined_size
                value_summary = context_summary.value_summary()
                size = humanfriendly.format_size(value_summary["size"])
                no_values = str(value_summary["no_values"])
                no_aliases = str(len(context_summary.aliases))
                table.add_row(
                    context_name,
                    str(context_summary.kiara_id),
                    humanfriendly.format_size(size_on_disk),
                    size,
                    no_values,
                    no_aliases,
                )
        else:

            table = Table(box=box.MINIMAL, show_header=True, show_lines=True)
            table.add_column("context_name", style="i")
            table.add_column("details")

            for context_name, context_summary in self.root.items():

                table.add_row(context_name, context_summary.create_renderable(**config))

        return table