Repository URL to install this package:
|
Version:
0.5.10.dev2+gb5d94239a ▾
|
# -*- coding: utf-8 -*-
import uuid
from typing import (
TYPE_CHECKING,
Any,
ClassVar,
Iterable,
List,
Literal,
Mapping,
Type,
Union,
)
import orjson
from pydantic import Field
from rich import box
from rich.console import RenderableType
from rich.panel import Panel
from rich.syntax import Syntax
from rich.table import Table
from kiara.interfaces.python_api.models.info import (
InfoItemGroup,
ItemInfo,
TypeInfo,
TypeInfoItemGroup,
)
from kiara.models.documentation import (
AuthorsMetadataModel,
ContextMetadataModel,
DocumentationMetadataModel,
)
from kiara.models.python_class import PythonClass
from kiara.registries import ArchiveDetails, KiaraArchive
from kiara.utils.json import orjson_dumps
if TYPE_CHECKING:
from kiara.context import Kiara
class ArchiveTypeInfo(TypeInfo):
_kiara_model_id: ClassVar = "info.archive_type"
@classmethod
def create_from_type_class(
self, type_cls: Type[KiaraArchive], kiara: "Kiara"
) -> "ArchiveTypeInfo":
authors_md = AuthorsMetadataModel.from_class(type_cls)
doc = DocumentationMetadataModel.from_class_doc(type_cls)
python_class = PythonClass.from_class(type_cls)
properties_md = ContextMetadataModel.from_class(type_cls)
type_name = type_cls._archive_type_name # type: ignore
return ArchiveTypeInfo(
type_name=type_name,
documentation=doc,
authors=authors_md,
context=properties_md,
python_class=python_class,
supported_item_types=list(type_cls.supported_item_types()),
)
@classmethod
def base_class(self) -> Type[KiaraArchive]:
return KiaraArchive
@classmethod
def category_name(cls) -> str:
return "archive_type"
is_writable: bool = Field(
description="Whether this archive is writeable.", default=False
)
supported_item_types: List[str] = Field(
description="The item types this archive suports."
)
def create_renderable(self, **config: Any) -> RenderableType:
include_doc = config.get("include_doc", True)
table = Table(box=box.SIMPLE, show_header=False, padding=(0, 0, 0, 0))
table.add_column("property", style="i")
table.add_column("value")
if include_doc:
table.add_row(
"Documentation",
Panel(self.documentation.create_renderable(), box=box.SIMPLE),
)
table.add_row("Author(s)", self.authors.create_renderable())
table.add_row("Context", self.context.create_renderable())
table.add_row("Python class", self.python_class.create_renderable())
table.add_row("is_writeable", "yes" if self.is_writable else "no")
table.add_row(
"supported_item_types", ", ".join(sorted(self.supported_item_types))
)
return table
class ArchiveTypeClassesInfo(TypeInfoItemGroup):
_kiara_model_id: ClassVar = "info.archive_types"
@classmethod
def base_info_class(cls) -> Type[ArchiveTypeInfo]:
return ArchiveTypeInfo
type_name: Literal["archive_type"] = "archive_type"
item_infos: Mapping[str, ArchiveTypeInfo] = Field( # type: ignore
description="The archive info instances for each type."
)
class ArchiveInfo(ItemInfo):
@classmethod
def base_instance_class(cls) -> Type[KiaraArchive]:
return KiaraArchive
@classmethod
def create_from_instance(cls, kiara: "Kiara", instance: KiaraArchive, **kwargs):
return cls.create_from_archive(kiara=kiara, archive=instance, **kwargs)
@classmethod
def create_from_archive(
cls,
kiara: "Kiara",
archive: KiaraArchive,
archive_aliases: Union[Iterable[str], None] = None,
):
archive_type_info = ArchiveTypeInfo.create_from_type_class(
archive.__class__, kiara=kiara
)
if archive_aliases is None:
archive_aliases = []
else:
archive_aliases = list(archive_aliases)
return ArchiveInfo(
archive_type_info=archive_type_info,
type_name=str(archive.archive_id),
documentation=archive_type_info.documentation,
authors=archive_type_info.authors,
context=archive_type_info.context,
archive_id=archive.archive_id,
details=archive.get_archive_details(),
config=archive.config.model_dump(),
aliases=archive_aliases,
)
@classmethod
def category_name(cls) -> str:
return "info.archive"
archive_id: uuid.UUID = Field(description="The (globally unique) archive id.")
archive_type_info: ArchiveTypeInfo = Field(
description="Information about this archives' type."
)
config: Mapping[str, Any] = Field(description="The configuration of this archive.")
details: ArchiveDetails = Field(
description="Type dependent (runtime) details for this archive."
)
aliases: List[str] = Field(
description="Aliases for this archive.", default_factory=list
)
class ArchiveGroupInfo(InfoItemGroup):
_kiara_model_id: ClassVar = "info.archives"
@classmethod
def base_info_class(cls) -> Type[ItemInfo]:
return ArchiveInfo
@classmethod
def create_from_context(
cls, kiara: "Kiara", group_title: Union[str, None] = None
) -> "ArchiveGroupInfo":
archives = {}
for archive, aliases in kiara.get_all_archives().items():
title = str(archive.archive_id) + ", ".join(aliases)
archives[title] = ArchiveInfo.create_from_archive(
kiara=kiara, archive=archive, archive_aliases=aliases
)
# archives[str(archive.archive_id)] = ArchiveInfo.create_from_archive(
# kiara=kiara, archive=archive, archive_aliases=aliases
# )
info = cls(group_title=group_title, item_infos=archives)
return info
item_infos: Mapping[str, ArchiveInfo] = Field(
description="The info for each archive."
)
@property
def combined_size(self) -> int:
combined = 0
archive_ids = set()
for archive_info in self.item_infos.values():
if archive_info.archive_id in archive_ids:
continue
archive_ids.add(archive_info.archive_id)
size = archive_info.details.size
if size and size > 0:
combined = combined + size
return combined
def create_renderable(self, **config: Any) -> RenderableType:
show_archive_id = config.get("show_archive_id", False)
show_config = config.get("show_config", True)
show_details = config.get("show_details", False)
# by_type: Dict[str, Dict[str, ArchiveInfo]] = {}
# for archive_id, archive in sorted(self.item_infos.items()):
# for item_type in archive.archive_type_info.supported_item_types:
# by_type.setdefault(item_type, {})[archive.type_name] = archive
table = Table(show_header=True, box=box.SIMPLE)
if show_archive_id:
table.add_column("archive id")
table.add_column("alias(es)", style="i")
table.add_column("item type(s)", style="i")
if show_config:
table.add_column("config")
if show_details:
table.add_column("details")
for archive in self.item_infos.values():
row: List[RenderableType] = []
if show_archive_id:
row.append(str(archive.archive_id))
row.append("\n".join(archive.aliases))
row.append("\n".join(archive.archive_type_info.supported_item_types))
if show_config:
config_json = Syntax(
orjson_dumps(archive.config, option=orjson.OPT_INDENT_2),
"json",
background_color="default",
)
row.append(config_json)
if show_details:
details_json = Syntax(
orjson_dumps(archive.details, option=orjson.OPT_INDENT_2),
"json",
background_color="default",
)
row.append(details_json)
table.add_row(*row)
return table