Compare commits

..

2 Commits

Author SHA1 Message Date
49c3734526 feat: added hooks for collection on append
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
2024-06-04 10:11:46 +02:00
bc19a94e7f feat: added parent artwork options 2024-06-04 10:09:17 +02:00
2 changed files with 72 additions and 21 deletions

View File

@ -1,18 +1,14 @@
from __future__ import annotations from __future__ import annotations
from typing import List, Optional, Dict, Tuple, Type, Union, TypedDict from typing import Dict, List, Optional, Set, Tuple, Type, TypedDict, Union
from .collection import Collection
from .metadata import (
Mapping as id3Mapping,
ID3Timestamp,
Metadata
)
from ..utils.string_processing import unify, hash_url
from .parents import OuterProxy as Base
from ..utils.config import main_settings from ..utils.config import main_settings
from ..utils.string_processing import hash_url, unify
from .collection import Collection
from .metadata import ID3Timestamp
from .metadata import Mapping as id3Mapping
from .metadata import Metadata
from .parents import OuterProxy as Base
class ArtworkVariant(TypedDict): class ArtworkVariant(TypedDict):
@ -23,7 +19,9 @@ class ArtworkVariant(TypedDict):
class Artwork: class Artwork:
def __init__(self, *variants: List[ArtworkVariant]) -> None: def __init__(self, *variants: List[ArtworkVariant], parent_artworks: Set[Artwork] = None) -> None:
self.parent_artworks: Set[Artwork] = parent_artworks or set()
self._variant_mapping: Dict[str, ArtworkVariant] = {} self._variant_mapping: Dict[str, ArtworkVariant] = {}
for variant in variants: for variant in variants:
@ -36,7 +34,7 @@ class Artwork:
def append(self, url: str, width: int = main_settings["preferred_artwork_resolution"], height: int = main_settings["preferred_artwork_resolution"], **kwargs) -> None: def append(self, url: str, width: int = main_settings["preferred_artwork_resolution"], height: int = main_settings["preferred_artwork_resolution"], **kwargs) -> None:
if url is None: if url is None:
return return
self._variant_mapping[hash_url(url=url)] = { self._variant_mapping[hash_url(url=url)] = {
"url": url, "url": url,
"width": width, "width": width,
@ -44,21 +42,36 @@ class Artwork:
"deviation": self._calculate_deviation(width, height), "deviation": self._calculate_deviation(width, height),
} }
@property
def flat_empty(self) -> bool:
return len(self._variant_mapping.keys()) <= 0
def _get_best_from_list(self, artwork_variants: List[ArtworkVariant]) -> Optional[ArtworkVariant]:
return min(artwork_variants, key=lambda x: x["deviation"])
@property @property
def best_variant(self) -> ArtworkVariant: def best_variant(self) -> ArtworkVariant:
if len(self._variant_mapping.keys()) <= 0: if self.flat_empty:
return None return self._get_best_from_list([parent.best_variant for parent in self.parent_artworks])
return min(self._variant_mapping.values(), key=lambda x: x["deviation"]) return self._get_best_from_list(self._variant_mapping.values())
def get_variant_name(self, variant: ArtworkVariant) -> str: def get_variant_name(self, variant: ArtworkVariant) -> str:
return f"artwork_{variant['width']}x{variant['height']}_{hash_url(variant['url']).replace('/', '_')}" return f"artwork_{variant['width']}x{variant['height']}_{hash_url(variant['url']).replace('/', '_')}"
def __merge__(self, other: Artwork, **kwargs) -> None: def __merge__(self, other: Artwork, **kwargs) -> None:
self.parent_artworks.update(other.parent_artworks)
for key, value in other._variant_mapping.items(): for key, value in other._variant_mapping.items():
if key not in self._variant_mapping: if key not in self._variant_mapping:
self._variant_mapping[key] = value self._variant_mapping[key] = value
def __hash__(self) -> int:
return id(self)
def __eq__(self, other: Artwork) -> bool: def __eq__(self, other: Artwork) -> bool:
if hash(self) == hash(other):
return True
if not isinstance(other, Artwork): if not isinstance(other, Artwork):
return False return False
return any(a == b for a, b in zip(self._variant_mapping.keys(), other._variant_mapping.keys())) return any(a == b for a, b in zip(self._variant_mapping.keys(), other._variant_mapping.keys()))

View File

@ -1,16 +1,44 @@
from __future__ import annotations from __future__ import annotations
from collections import defaultdict
from typing import TypeVar, Generic, Dict, Optional, Iterable, List, Iterator, Tuple, Generator, Union, Any, Set
import copy import copy
from collections import defaultdict
from dataclasses import dataclass
from typing import (Any, Callable, Dict, Generator, Generic, Iterable,
Iterator, List, Optional, Set, Tuple, TypeVar, Union)
from .parents import OuterProxy from ..utils import BColors, object_trace, output
from ..utils import object_trace from .parents import InnerData, OuterProxy
from ..utils import output, BColors
T = TypeVar('T', bound=OuterProxy) T = TypeVar('T', bound=OuterProxy)
@dataclass
class AppendHookArguments:
"""
This class is used to store the arguments for the append hook.
The best explanation is with an examples:
```
# this is the action that triggers the append hook
album = Album()
song = Song()
album.song_collection.append(song)
```
In this case, the append hook is triggered with the following arguments:
```
AppendHookArguments(
collection=album.song_collection,
new_object=song,
collection_root_objects=[album]
)
```
"""
collection: Collection
new_object: T
collection_root_objects: Set[InnerData]
class Collection(Generic[T]): class Collection(Generic[T]):
__is_collection__ = True __is_collection__ = True
@ -27,6 +55,7 @@ class Collection(Generic[T]):
sync_on_append: Dict[str, Collection] = None, sync_on_append: Dict[str, Collection] = None,
append_object_to_attribute: Dict[str, T] = None, append_object_to_attribute: Dict[str, T] = None,
extend_object_to_attribute: Dict[str, Collection] = None, extend_object_to_attribute: Dict[str, Collection] = None,
append_callbacks: List[Callable[[AppendHookArguments], None]] = None,
) -> None: ) -> None:
self._collection_for: dict = dict() self._collection_for: dict = dict()
@ -41,6 +70,7 @@ class Collection(Generic[T]):
self.sync_on_append: Dict[str, Collection] = sync_on_append or {} self.sync_on_append: Dict[str, Collection] = sync_on_append or {}
self.pull_from: List[Collection] = [] self.pull_from: List[Collection] = []
self.push_to: List[Collection] = [] self.push_to: List[Collection] = []
self.append_callbacks: List[Callable[[AppendHookArguments], None]] = append_callbacks or []
# This is to cleanly unmap previously mapped items by their id # This is to cleanly unmap previously mapped items by their id
self._indexed_from_id: Dict[int, Dict[str, Any]] = defaultdict(dict) self._indexed_from_id: Dict[int, Dict[str, Any]] = defaultdict(dict)
@ -141,6 +171,14 @@ class Collection(Generic[T]):
for attribute, new_object in self.append_object_to_attribute.items(): for attribute, new_object in self.append_object_to_attribute.items():
other.__getattribute__(attribute).append(new_object, **kwargs) other.__getattribute__(attribute).append(new_object, **kwargs)
append_hook_args = AppendHookArguments(
collection=self,
new_object=other,
collection_root_objects=self._collection_for.keys(),
)
for callback in self.append_callbacks:
callback(append_hook_args)
def append(self, other: Optional[T], **kwargs): def append(self, other: Optional[T], **kwargs):
""" """
If an object, that represents the same entity exists in a relevant collection, If an object, that represents the same entity exists in a relevant collection,