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 typing import List, Optional, Dict, Tuple, Type, Union, TypedDict
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 typing import Dict, List, Optional, Set, Tuple, Type, TypedDict, Union
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):
@ -23,7 +19,9 @@ class ArtworkVariant(TypedDict):
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] = {}
for variant in variants:
@ -44,21 +42,36 @@ class Artwork:
"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
def best_variant(self) -> ArtworkVariant:
if len(self._variant_mapping.keys()) <= 0:
return None
return min(self._variant_mapping.values(), key=lambda x: x["deviation"])
if self.flat_empty:
return self._get_best_from_list([parent.best_variant for parent in self.parent_artworks])
return self._get_best_from_list(self._variant_mapping.values())
def get_variant_name(self, variant: ArtworkVariant) -> str:
return f"artwork_{variant['width']}x{variant['height']}_{hash_url(variant['url']).replace('/', '_')}"
def __merge__(self, other: Artwork, **kwargs) -> None:
self.parent_artworks.update(other.parent_artworks)
for key, value in other._variant_mapping.items():
if key not in self._variant_mapping:
self._variant_mapping[key] = value
def __hash__(self) -> int:
return id(self)
def __eq__(self, other: Artwork) -> bool:
if hash(self) == hash(other):
return True
if not isinstance(other, Artwork):
return False
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 collections import defaultdict
from typing import TypeVar, Generic, Dict, Optional, Iterable, List, Iterator, Tuple, Generator, Union, Any, Set
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 object_trace
from ..utils import output, BColors
from ..utils import BColors, object_trace, output
from .parents import InnerData, 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]):
__is_collection__ = True
@ -27,6 +55,7 @@ class Collection(Generic[T]):
sync_on_append: Dict[str, Collection] = None,
append_object_to_attribute: Dict[str, T] = None,
extend_object_to_attribute: Dict[str, Collection] = None,
append_callbacks: List[Callable[[AppendHookArguments], None]] = None,
) -> None:
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.pull_from: 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
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():
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):
"""
If an object, that represents the same entity exists in a relevant collection,