Coverage for src/nendo/schema/plugin.py: 83%
283 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-11-23 10:03 +0100
1# -*- encoding: utf-8 -*-
2"""Plugin classes of Nendo Core."""
3from __future__ import annotations
5import functools
6import os
7from abc import abstractmethod
8from typing import (
9 TYPE_CHECKING,
10 Any,
11 Callable,
12 Dict,
13 Iterator,
14 List,
15 Optional,
16 Tuple,
17 Union,
18)
20from pydantic import ConfigDict, DirectoryPath, FilePath
22from nendo.schema.core import (
23 NendoBlob,
24 NendoCollection,
25 NendoCollectionSlim,
26 NendoPlugin,
27 NendoPluginData,
28 NendoStorage,
29 NendoTrack,
30)
31from nendo.schema.exception import NendoError, NendoPluginRuntimeError
32from nendo.utils import ensure_uuid
34if TYPE_CHECKING:
35 import uuid
37 import numpy as np
40class NendoAnalysisPlugin(NendoPlugin):
41 """Basic class for nendo analysis plugins.
43 Analysis plugins are plugins that analyze a track or a collection
44 and add metadata and other properties to the track or collection.
45 Decorate your methods with `@NendoAnalysisPlugin.plugin_data` to add the
46 return values of your methods as plugin data to the track or collection.
48 Decorate your methods with `@NendoAnalysisPlugin.run_track` to run your method
49 on a track and use `@NendoAnalysisPlugin.run_collection` to run your method on
50 a collection.
52 Examples:
53 ```python
54 from nendo import Nendo, NendoConfig
56 class MyPlugin(NendoAnalysisPlugin):
57 ...
59 @NendoAnalysisPlugin.plugin_data
60 def my_plugin_data_function_one(self, track):
61 # do something analysis on the track
62 return {"key": "value"}
64 @NendoAnalysisPlugin.plugin_data
65 def my_plugin_data_function_two(self, track):
66 # do some more analysis on the track
67 return {"key": "value"}
69 @NendoAnalysisPlugin.run_track
70 def my_run_track_function(self, track):
71 my_plugin_data_function_one(track)
72 my_plugin_data_function_two(track)
73 ```
74 """
76 # decorators
77 # ----------
78 @staticmethod
79 def plugin_data(
80 func: Callable[[NendoPlugin, NendoTrack], Dict[str, Any]],
81 ) -> Callable[[NendoPlugin, NendoTrack], Dict[str, Any]]:
82 """Decorator to enrich a NendoTrack with data from a plugin.
84 Args:
85 func: Callable[[NendoPlugin, NendoTrack], Dict[str, Any]]: The function to register.
87 Returns:
88 Callable[[NendoPlugin, NendoTrack], Dict[str, Any]]: The wrapped function.
89 """
91 def wrapper(self, track: NendoTrack):
92 try:
93 f_result = func(self, track)
94 except NendoError as e:
95 raise NendoPluginRuntimeError(
96 f"Error running plugin function: {e}",
97 ) from None
98 for k, v in f_result.items():
99 track.add_plugin_data(
100 plugin_name=self.plugin_name,
101 plugin_version=self.plugin_version,
102 key=str(k),
103 value=v,
104 # replace=False,
105 )
106 return f_result
108 return wrapper
110 # ----------
112 @property
113 def plugin_type(self) -> str:
114 """Return type of plugin."""
115 return "AnalysisPlugin"
117 @staticmethod
118 def run_collection(
119 func: Callable[[NendoPlugin, NendoCollection, Any], None],
120 ) -> Callable[[NendoPlugin, Any], NendoCollection]:
121 """Decorator to register a function as a collection running function for a `NendoAnalysisPlugin`.
123 This decorator wraps the function and allows a plugin user to call the plugin with either a collection or a track.
125 Args:
126 func: Callable[[NendoPlugin, NendoCollection, Any], None]: The function to register.
128 Returns:
129 Callable[[NendoPlugin, Any], NendoCollection]: The wrapped function.
130 """
132 @functools.wraps(func)
133 def wrapper(self, **kwargs: Any) -> NendoCollection:
134 track_or_collection, kwargs = self._get_track_or_collection_from_args(
135 **kwargs,
136 )
137 if isinstance(track_or_collection, NendoCollection):
138 func(self, track_or_collection, **kwargs)
139 return self.nendo_instance.library.get_collection(
140 track_or_collection.id,
141 )
143 tmp_collection = self.nendo_instance.library.add_collection(
144 name="tmp",
145 track_ids=[track_or_collection.id],
146 collection_type="temp",
147 )
148 func(self, tmp_collection, **kwargs)
149 return self.nendo_instance.library.get_track(track_or_collection.id)
151 return wrapper
153 @staticmethod
154 def run_track(
155 func: Callable[[NendoPlugin, NendoTrack, Any], None],
156 ) -> Callable[[NendoPlugin, Any], Union[NendoTrack, NendoCollection]]:
157 """Decorator to register a function as a track running function for a `NendoAnalysisPlugin`.
159 This decorator wraps the function and allows a plugin user to call the plugin with either a collection or a track.
161 Args:
162 func: Callable[[NendoPlugin, NendoTrack, Any], None]: The function to register.
164 Returns:
165 Callable[[NendoPlugin, Any], NendoCollection]: The wrapped function.
166 """
168 @functools.wraps(func)
169 def wrapper(self, **kwargs: Any) -> Union[NendoTrack, NendoCollection]:
170 track_or_collection, kwargs = self._get_track_or_collection_from_args(
171 **kwargs,
172 )
173 if isinstance(track_or_collection, NendoTrack):
174 func(self, track_or_collection, **kwargs)
175 return self.nendo_instance.library.get_track(track_or_collection.id)
176 [func(self, track, **kwargs) for track in track_or_collection.tracks()]
177 return self.nendo_instance.library.get_collection(track_or_collection.id)
179 return wrapper
182class NendoGeneratePlugin(NendoPlugin):
183 """Basic class for nendo generate plugins.
185 Generate plugins are plugins that generate new tracks or collections, either from scratch or
186 based on existing tracks or collections.
187 Decorate your methods with `@NendoGeneratePlugin.run_track` to run your method
188 on a track, use `@NendoGeneratePlugin.run_collection` to run your method on
189 a collection and use `@NendoGeneratePlugin.run_signal` to run your method on
190 a signal.
192 Examples:
193 ```python
194 from nendo import Nendo, NendoConfig
196 class MyPlugin(NendoGeneratePlugin):
197 ...
199 @NendoAnalysisPlugin.run_track
200 def my_generate_track_function(self, track, arg_one="foo"):
201 # generate some new audio
203 # add audio to the nendo library
204 return self.nendo_instance.library.add_related_track_from_signal(
205 signal,
206 sr,
207 related_track_id=track.id,
208 )
209 ```
210 """
212 @property
213 def plugin_type(self) -> str:
214 """Return type of plugin."""
215 return "GeneratePlugin"
217 @staticmethod
218 def run_collection(
219 func: Callable[[NendoPlugin, Optional[NendoCollection], Any], NendoCollection],
220 ) -> Callable[[NendoPlugin, Any], NendoCollection]:
221 """Decorator to register a function as a collection running function for a `NendoGeneratePlugin`.
223 This decorator wraps the function and allows a plugin user to call the plugin with either a collection or a track.
225 Args:
226 func: Callable[[NendoPlugin, NendoCollection, Any], NendoCollection]: The function to register.
228 Returns:
229 Callable[[NendoPlugin, Any], NendoCollection]: The wrapped function.
230 """
232 @functools.wraps(func)
233 def wrapper(self, **kwargs: Any) -> NendoCollection:
234 track_or_collection, kwargs = self._get_track_or_collection_from_args(
235 **kwargs,
236 )
237 if track_or_collection is None:
238 return func(self, **kwargs)
240 if isinstance(track_or_collection, NendoCollection):
241 return func(self, track_or_collection, **kwargs)
243 tmp_collection = self.nendo_instance.library.add_collection(
244 name="tmp",
245 track_ids=[track_or_collection.id],
246 collection_type="temp",
247 )
248 return func(self, tmp_collection, **kwargs)
250 return wrapper
252 @staticmethod
253 def run_signal(
254 func: Callable[
255 [NendoPlugin, Optional[np.ndarray], Optional[int], Any],
256 Tuple[np.ndarray, int],
257 ],
258 ) -> Callable[[NendoPlugin, Any], Union[NendoTrack, NendoCollection]]:
259 """Decorator to register a function as a signal running function for a `NendoGeneratePlugin`.
261 This decorator wraps the function and allows a plugin user to call the plugin with either a collection or a track.
263 Args:
264 func: Callable[[NendoPlugin, np.ndarray, int, Any], Tuple[np.ndarray, int]]: The function to register.
266 Returns:
267 Callable[[NendoPlugin, Any], NendoCollection]: The wrapped function.
268 """
270 @functools.wraps(func)
271 def wrapper(self, **kwargs: Any) -> Union[NendoTrack, NendoCollection]:
272 track_or_collection, kwargs = self._get_track_or_collection_from_args(
273 **kwargs,
274 )
275 if track_or_collection is None:
276 signal, sr = func(self, **kwargs)
277 return self.nendo_instance.library.add_track_from_signal(
278 signal,
279 sr,
280 )
281 if isinstance(track_or_collection, NendoTrack):
282 signal, sr = track_or_collection.signal, track_or_collection.sr
283 new_signal, new_sr = func(self, signal, sr, **kwargs)
284 return self.nendo_instance.library.add_related_track_from_signal(
285 new_signal,
286 sr,
287 related_track_id=track_or_collection.id,
288 )
289 processed_tracks = []
290 for track in track_or_collection.tracks():
291 new_signal, new_sr = func(
292 self,
293 track.signal,
294 track.sr,
295 **kwargs,
296 )
297 processed_tracks.append(
298 self.nendo_instance.library.add_related_track_from_signal(
299 new_signal,
300 track.sr,
301 related_track_id=track.id,
302 ),
303 )
304 return self.nendo_instance.library.add_collection(
305 name="tmp",
306 track_ids=[track.id for track in processed_tracks],
307 collection_type="temp",
308 )
310 return wrapper
312 @staticmethod
313 def run_track(
314 func: Callable[
315 [NendoPlugin, Optional[NendoTrack], Any],
316 Union[NendoTrack, List[NendoTrack]],
317 ],
318 ) -> Callable[[NendoPlugin, Any], Union[NendoTrack, NendoCollection]]:
319 """Decorator to register a function as a track running function for a `NendoGeneratePlugin`.
321 This decorator wraps the function and allows a plugin user to call the plugin with either a collection or a track.
323 Args:
324 func: Callable[[NendoPlugin, NendoTrack, Any], Union[NendoTrack, List[NendoTrack]]]: The function to register.
326 Returns:
327 Callable[[NendoPlugin, Any], NendoCollection]: The wrapped function.
328 """
330 @functools.wraps(func)
331 def wrapper(self, **kwargs: Any) -> Union[NendoTrack, NendoCollection]:
332 track_or_collection, kwargs = self._get_track_or_collection_from_args(
333 **kwargs,
334 )
335 processed_tracks = []
336 if track_or_collection is None:
337 track = func(self, **kwargs)
339 # may be multiple tracks as a result
340 if not isinstance(track, list):
341 return track
342 processed_tracks.extend(track)
343 elif isinstance(track_or_collection, NendoTrack):
344 track = func(self, track_or_collection, **kwargs)
346 # may be multiple tracks as a result
347 if not isinstance(track, list):
348 return track
349 processed_tracks.extend(track)
350 else:
351 for track in track_or_collection.tracks():
352 processed_track = func(self, track, **kwargs)
354 # may be multiple tracks as a result
355 if isinstance(processed_track, list):
356 processed_tracks.extend(processed_track)
357 else:
358 processed_tracks.append(processed_track)
360 return self.nendo_instance.library.add_collection(
361 name="tmp",
362 track_ids=[track.id for track in processed_tracks],
363 collection_type="temp",
364 )
366 return wrapper
369class NendoEffectPlugin(NendoPlugin):
370 """Basic class for nendo effects plugins.
372 Effects plugins are plugins that add effects to tracks or collections.
373 Decorate your methods with `@NendoGeneratePlugin.run_track` to run your method
374 on a track, use `@NendoGeneratePlugin.run_collection` to run your method on
375 a collection and use `@NendoGeneratePlugin.run_signal` to run your method on
376 a signal.
378 Examples:
379 ```python
380 from nendo import Nendo, NendoConfig
382 class MyPlugin(NendoGeneratePlugin):
383 ...
385 @NendoAnalysisPlugin.run_signal
386 def my_effect_function(self, signal, sr, arg_one="foo"):
387 # add some effect to the signal
388 new_signal = apply_effect(signal, sr, arg_one)
390 return new_signal, sr
391 ```
392 """
394 @property
395 def plugin_type(self) -> str:
396 """Return type of plugin."""
397 return "EffectPlugin"
399 @staticmethod
400 def run_collection(
401 func: Callable[[NendoPlugin, NendoCollection, Any], NendoCollection],
402 ) -> Callable[[NendoPlugin, Any], NendoCollection]:
403 """Decorator to register a function as a collection running function for a `NendoEffectPlugin`.
405 This decorator wraps the function and allows a plugin user to call the plugin with either a collection or a track.
407 Args:
408 func: Callable[[NendoPlugin, NendoCollection, Any], NendoCollection]: The function to register.
410 Returns:
411 Callable[[NendoPlugin, Any], NendoCollection]: The wrapped function.
412 """
414 @functools.wraps(func)
415 def wrapper(self, **kwargs: Any) -> NendoCollection:
416 track_or_collection, kwargs = self._get_track_or_collection_from_args(
417 **kwargs,
418 )
419 if isinstance(track_or_collection, NendoCollection):
420 return func(self, track_or_collection, **kwargs)
422 tmp_collection = self.nendo_instance.library.add_collection(
423 name="tmp",
424 track_ids=[track_or_collection.id],
425 collection_type="temp",
426 )
427 return func(self, tmp_collection, **kwargs)
429 return wrapper
431 @staticmethod
432 def run_signal(
433 func: Callable[[NendoPlugin, np.ndarray, int, Any], Tuple[np.ndarray, int]],
434 ) -> Callable[[NendoPlugin, Any], Union[NendoTrack, NendoCollection]]:
435 """Decorator to register a function as a signal running function for a `NendoEffectPlugin`.
437 This decorator wraps the function and allows a plugin user to call the plugin with either a collection or a track.
439 Args:
440 func: Callable[[NendoPlugin, np.ndarray, int, Any], Tuple[np.ndarray, int]]: The function to register.
442 Returns:
443 Callable[[NendoPlugin, Any], NendoTrack]: The wrapped function.
444 """
446 @functools.wraps(func)
447 def wrapper(self, **kwargs: Any) -> Union[NendoTrack, NendoCollection]:
448 track_or_collection, kwargs = self._get_track_or_collection_from_args(
449 **kwargs,
450 )
451 processed_tracks = []
452 if isinstance(track_or_collection, NendoTrack):
453 signal, sr = track_or_collection.signal, track_or_collection.sr
454 new_signal, new_sr = func(self, signal, sr, **kwargs)
456 # TODO update track instead of adding a new one to the library
457 return self.nendo_instance.library.add_related_track_from_signal(
458 new_signal,
459 sr,
460 related_track_id=track_or_collection.id,
461 )
463 for track in track_or_collection.tracks():
464 new_signal, new_sr = func(
465 self,
466 track.signal,
467 track.sr,
468 **kwargs,
469 )
471 # TODO update track instead of adding a new one to the library
472 processed_tracks.append(
473 self.nendo_instance.library.add_related_track_from_signal(
474 new_signal,
475 track.sr,
476 related_track_id=track.id,
477 ),
478 )
480 return self.nendo_instance.library.add_collection(
481 name="tmp",
482 track_ids=[track.id for track in processed_tracks],
483 collection_type="temp",
484 )
486 return wrapper
488 @staticmethod
489 def run_track(
490 func: Callable[
491 [NendoPlugin, NendoTrack, Any],
492 Union[NendoTrack, List[NendoTrack]],
493 ],
494 ) -> Callable[[NendoPlugin, Any], NendoTrack]:
495 """Decorator to register a function as a track running function for a `NendoEffectPlugin`.
497 This decorator wraps the function and allows a plugin user to call the plugin with either a collection or a track.
499 Args:
500 func: Callable[[NendoPlugin, NendoTrack, Any], Union[NendoTrack, List[NendoTrack]]]: The function to register.
502 Returns:
503 Callable[[NendoPlugin, Any], NendoTrack]: The wrapped function.
504 """
506 @functools.wraps(func)
507 def wrapper(self, **kwargs: Any) -> Union[NendoTrack, NendoCollection]:
508 track_or_collection, kwargs = self._get_track_or_collection_from_args(
509 **kwargs,
510 )
511 if isinstance(track_or_collection, NendoTrack):
512 return func(self, track_or_collection, **kwargs)
514 processed_tracks = [
515 func(self, track, **kwargs) for track in track_or_collection.tracks()
516 ]
517 return self.nendo_instance.library.add_collection(
518 name="tmp",
519 track_ids=[track.id for track in processed_tracks],
520 collection_type="temp",
521 )
523 return wrapper
526class NendoLibraryPlugin(NendoPlugin):
527 """Basic class for nendo library plugins."""
529 model_config = ConfigDict(
530 arbitrary_types_allowed=True,
531 )
533 storage_driver: Optional[NendoStorage] = None
535 # ==========================
536 #
537 # TRACK MANAGEMENT FUNCTIONS
538 #
539 # ==========================
541 @abstractmethod
542 def add_track(
543 self,
544 file_path: Union[FilePath, str],
545 track_type: str = "track",
546 copy_to_library: Optional[bool] = None,
547 skip_duplicate: Optional[bool] = None,
548 user_id: Optional[uuid.UUID] = None,
549 meta: Optional[dict] = None,
550 ) -> NendoTrack:
551 """Add the track given by path to the library.
553 Args:
554 file_path (Union[FilePath, str]): Path to the file to add as track.
555 track_type (str): Type of the track. Defaults to "track".
556 copy_to_library (bool, optional): Flag that specifies whether
557 the file should be copied into the library directory.
558 Defaults to None.
559 skip_duplicate (bool, optional): Flag that specifies whether a
560 file should be added that already exists in the library, based on its
561 file checksum. Defaults to None.
562 user_id (UUID, optional): ID of user adding the track.
563 meta (dict, optional): Metadata to attach to the track upon adding.
565 Returns:
566 NendoTrack: The track that was added to the library.
567 """
568 raise NotImplementedError
570 @abstractmethod
571 def add_track_from_signal(
572 self,
573 signal: np.ndarray,
574 sr: int,
575 track_type: str = "track",
576 user_id: Optional[uuid.UUID] = None,
577 meta: Optional[Dict[str, Any]] = None,
578 ) -> NendoTrack:
579 """Add a track to the library that is described by the given signal.
581 Args:
582 signal (np.ndarray): The numpy array containing the audio signal.
583 sr (int): Sample rate
584 track_type (str): Track type. Defaults to "track".
585 user_id (UUID, optional): The ID of the user adding the track.
586 meta (Dict[str, Any], optional): Track metadata. Defaults to {}.
588 Returns:
589 schema.NendoTrack: The added NendoTrack
590 """
591 raise NotImplementedError
593 @abstractmethod
594 def add_related_track(
595 self,
596 file_path: Union[FilePath, str],
597 related_track_id: Union[str, uuid.UUID],
598 track_type: str = "str",
599 user_id: Optional[Union[str, uuid.UUID]] = None,
600 track_meta: Optional[Dict[str, Any]] = None,
601 relationship_type: str = "relationship",
602 meta: Optional[Dict[str, Any]] = None,
603 ) -> NendoTrack:
604 """Add track that is related to another `NendoTrack`.
606 Add the track found in the given path to the library and create a relationship
607 in the new track that points to the track identified by related_to.
609 Args:
610 file_path (Union[FilePath, str]): Path to the file to add as track.
611 related_track_id (Union[str, uuid.UUID]): ID of the related track.
612 track_type (str): Track type. Defaults to "track".
613 user_id (Union[str, UUID], optional): ID of the user adding the track.
614 track_meta (dict, optional): Dictionary containing the track metadata.
615 relationship_type (str): Type of the relationship.
616 Defaults to "relationship".
617 meta (dict): Dictionary containing metadata about
618 the relationship. Defaults to {}.
620 Returns:
621 NendoTrack: The track that was added to the Library
622 """
623 raise NotImplementedError
625 @abstractmethod
626 def add_related_track_from_signal(
627 self,
628 signal: np.ndarray,
629 sr: int,
630 related_track_id: Union[str, uuid.UUID],
631 track_type: str = "track",
632 user_id: Optional[uuid.UUID] = None,
633 track_meta: Optional[Dict[str, Any]] = None,
634 relationship_type: str = "relationship",
635 meta: Optional[Dict[str, Any]] = None,
636 ) -> NendoTrack:
637 """Add signal as track that is related to another `NendoTrack`.
639 Add the track represented by the provided signal to the library and create a
640 relationship in the new track that points to the track passed as related_to.
642 Args:
643 signal (np.ndarray): Waveform of the track in numpy array form.
644 sr (int): Sampling rate of the waveform.
645 related_track_id (str | uuid.UUID): ID to which the relationship
646 should point to.
647 track_type (str): Track type. Defaults to "track".
648 user_id (UUID, optional): ID of the user adding the track.
649 track_meta (dict, optional): Dictionary containing the track metadata.
650 relationship_type (str): Type of the relationship.
651 Defaults to "relationship".
652 meta (dict): Dictionary containing metadata about
653 the relationship. Defaults to {}.
655 Returns:
656 NendoTrack: The added track with the relationship.
657 """
658 raise NotImplementedError
660 @abstractmethod
661 def add_tracks(
662 self,
663 path: Union[DirectoryPath, str],
664 track_type: str = "track",
665 user_id: Optional[Union[str, uuid.UUID]] = None,
666 copy_to_library: Optional[bool] = None,
667 skip_duplicate: bool = True,
668 ) -> NendoCollection:
669 """Scan the provided path and upsert the information into the library.
671 Args:
672 path (Union[DirectoryPath, str]): Path to the directory to scan.
673 track_type (str): Track type. Defaults to "track".
674 user_id (UUID, optional): The ID of the user adding the tracks.
675 copy_to_library (Optional[bool], optional): Flag that specifies whether
676 the file should be copied into the library directory.
677 Defaults to None.
678 skip_duplicate (Optional[bool], optional): Flag that specifies whether a
679 file should be added that already exists in the library, based on its
680 file checksum. Defaults to None.
682 Returns:
683 collection (NendoCollection): The collection of tracks that were added to the Library
684 """
685 raise NotImplementedError
687 @abstractmethod
688 def update_track(
689 self,
690 track: NendoTrack,
691 ) -> NendoTrack:
692 """Updates the given collection by storing it to the database.
694 Args:
695 track (NendoTrack): The track to be stored to the database.
697 Raises:
698 NendoTrackNotFoundError: If the track passed to the function
699 does not exist in the database.
701 Returns:
702 NendoTrack: The updated track.
703 """
704 raise NotImplementedError
706 @abstractmethod
707 def add_plugin_data(
708 self,
709 track_id: Union[str, uuid.UUID],
710 plugin_name: str,
711 plugin_version: str,
712 key: str,
713 value: str,
714 user_id: Optional[Union[str, uuid.UUID]] = None,
715 replace: bool = False,
716 ) -> NendoPluginData:
717 """Add plugin data to a NendoTrack and persist changes into the DB.
719 Args:
720 track_id (Union[str, uuid.UUID]): ID of the track to which
721 the plugin data should be added.
722 plugin_name (str): Name of the plugin.
723 plugin_version (str): Version of the plugin.
724 key (str): Key under which to save the data.
725 value (str): Data to save.
726 user_id (uuid4, optional): ID of user adding the plugin data.
727 replace (bool, optional): Flag that determines whether
728 the last existing data point for the given plugin name and -version
729 is overwritten or not. Defaults to False.
731 Returns:
732 NendoPluginData: The saved plugin data as a NendoPluginData object.
733 """
735 @abstractmethod
736 def get_track(self, track_id: Any) -> NendoTrack:
737 """Get a single track from the library by ID.
739 If no track with the given ID was found, return None.
741 Args:
742 track_id (Any): The ID of the track to get
744 Returns:
745 track (NendoTrack): The track with the given ID
746 """
747 raise NotImplementedError
749 @abstractmethod
750 @NendoPlugin.stream_output
751 def get_tracks(
752 self,
753 user_id: Optional[Union[str, uuid.UUID]] = None,
754 order_by: Optional[str] = None,
755 order: str = "asc",
756 limit: Optional[int] = None,
757 offset: Optional[int] = None,
758 ) -> Union[List, Iterator]:
759 """Get tracks based on the given query parameters.
761 Args:
762 user_id (Union[str, UUID], optional): ID of user getting the tracks.
763 order_by (Optional[str]): Key used for ordering the results.
764 order (Optional[str]): Order in which to retrieve results ("asc" or "desc").
765 limit (Optional[int]): Limit the number of returned results.
766 offset (Optional[int]): Offset into the paginated results (requires limit).
768 Returns:
769 Union[List, Iterator]: List or generator of tracks, depending on the
770 configuration variable stream_mode
771 """
772 raise NotImplementedError
774 @abstractmethod
775 def get_related_tracks(
776 self,
777 track_id: Union[str, uuid.UUID],
778 user_id: Optional[Union[str, uuid.UUID]] = None,
779 order_by: Optional[str] = None,
780 order: Optional[str] = "asc",
781 limit: Optional[int] = None,
782 offset: Optional[int] = None,
783 ) -> Union[List, Iterator]:
784 """Get tracks with a relationship to the track with track_id.
786 Args:
787 track_id (str): ID of the track to be searched for.
788 user_id (Union[str, UUID], optional): The user ID to filter for.
789 order_by (Optional[str]): Key used for ordering the results.
790 order (Optional[str]): Order in which to retrieve results ("asc" or "desc").
791 limit (Optional[int]): Limit the number of returned results.
792 offset (Optional[int]): Offset into the paginated results (requires limit).
794 Returns:
795 Union[List, Iterator]: List or generator of tracks, depending on the
796 configuration variable stream_mode
797 """
798 raise NotImplementedError
800 @abstractmethod
801 def find_tracks(
802 self,
803 value: str,
804 user_id: Optional[Union[str, uuid.UUID]] = None,
805 order_by: Optional[str] = None,
806 limit: Optional[int] = None,
807 offset: Optional[int] = None,
808 ) -> Union[List, Iterator]:
809 """Find tracks by searching for a string through the resource metadata.
811 Args:
812 value (str): The search value to filter by.
813 user_id (Union[str, UUID], optional): The user ID to filter for.
814 order_by (str, optional): Ordering.
815 limit (str, optional): Pagination limit.
816 offset (str, optional): Pagination offset.
818 Returns:
819 Union[List, Iterator]: List or generator of tracks, depending on the
820 configuration variable stream_mode
821 """
822 raise NotImplementedError
824 @abstractmethod
825 def filter_tracks(
826 self,
827 filters: Optional[dict] = None,
828 resource_filters: Optional[Dict[str, Any]] = None,
829 track_type: Optional[Union[str, List[str]]] = None,
830 user_id: Optional[Union[str, uuid.UUID]] = None,
831 collection_id: Optional[Union[str, uuid.UUID]] = None,
832 plugin_names: Optional[List[str]] = None,
833 order_by: Optional[str] = None,
834 order: str = "asc",
835 limit: Optional[int] = None,
836 offset: Optional[int] = None,
837 ) -> Union[List, Iterator]:
838 """Obtain tracks from the db by filtering over plugin data.
840 Args:
841 filters (Optional[dict]): Dictionary containing the filters to apply.
842 Defaults to None.
843 resource_filters (dict): Dictionary containing the keywords to search for
844 over the track.resource.meta field. The dictionary's values
845 should contain singular search tokens and the keys currently have no
846 effect but might in the future. Defaults to {}.
847 track_type (Union[str, List[str]], optional): Track type to filter for.
848 Can be a singular type or a list of types. Defaults to None.
849 user_id (Union[str, UUID], optional): The user ID to filter for.
850 collection_id (Union[str, uuid.UUID], optional): Collection id to
851 which the filtered tracks must have a relationship. Defaults to None.
852 plugin_names (list, optional): List used for applying the filter only to
853 data of certain plugins. If None, all plugin data related to the track
854 is used for filtering.
855 order_by (str, optional): Key used for ordering the results.
856 order (str, optional): Ordering ("asc" vs "desc"). Defaults to "asc".
857 limit (int, optional): Limit the number of returned results.
858 offset (int, optional): Offset into the paginated results (requires limit).
860 Returns:
861 Union[List, Iterator]: List or generator of tracks, depending on the
862 configuration variable stream_mode
863 """
864 raise NotImplementedError
866 @abstractmethod
867 def remove_track(
868 self,
869 track_id: Union[str, uuid.UUID],
870 remove_relationships: bool = False,
871 remove_plugin_data: bool = False,
872 remove_resources: bool = True,
873 user_id: Optional[Union[str, uuid.UUID]] = None,
874 ) -> bool:
875 """Delete track from library by ID.
877 Args:
878 track_id (Union[str, uuid.UUID]): The ID of the track to remove.
879 remove_relationships (bool):
880 If False prevent deletion if related tracks exist,
881 if True delete relationships together with the object.
882 remove_plugin_data (bool):
883 If False prevent deletion if related plugin data exist,
884 if True delete plugin data together with the object.
885 remove_resources (bool):
886 If False, keep the related resources, e.g. files,
887 if True, delete the related resources.
888 user_id (Union[str, UUID], optional): The ID of the user
889 owning the track.
891 Returns:
892 success (bool): True if removal was successful, False otherwise
893 """
894 raise NotImplementedError
896 def export_track(
897 self,
898 track_id: Union[str, uuid.UUID],
899 file_path: str,
900 file_format: str = "wav",
901 ) -> str:
902 """Export the track to a file.
904 Args:
905 track_id (Union[str, uuid.UUID]): The ID of the target track to export.
906 file_path (str): Path to the exported file. Can be either a full
907 file path or a directory path. If a directory path is given,
908 a filename will be automatically generated and the file will be
909 exported to the format specified as file_format. If a full file
910 path is given, the format will be deduced from the path and the
911 file_format parameter will be ignored.
912 file_format (str, optional): Format of the exported track. Ignored if
913 file_path is a full file path. Defaults to "wav".
915 Returns:
916 str: The path to the exported file.
917 """
918 raise NotImplementedError
920 # ===============================
921 #
922 # COLLECTION MANAGEMENT FUNCTIONS
923 #
924 # ===============================
926 @abstractmethod
927 def add_collection(
928 self,
929 name: str,
930 user_id: Optional[Union[str, uuid.UUID]] = None,
931 track_ids: Optional[List[Union[str, uuid.UUID]]] = None,
932 description: str = "",
933 collection_type: str = "collection",
934 meta: Optional[Dict[str, Any]] = None,
935 ) -> NendoCollection:
936 """Creates a new collection and saves it into the DB.
938 Args:
939 track_ids (List[Union[str, uuid.UUID]]): List of track ids
940 to be added to the collection.
941 name (str): Name of the collection.
942 user_id (UUID, optional): The ID of the user adding the collection.
943 description (str): Description of the collection.
944 collection_type (str): Type of the collection. Defaults to "collection".
945 meta (Dict[str, Any]): Metadata of the collection.
947 Returns:
948 schema.NendoCollection: The newly created NendoCollection object.
949 """
950 raise NotImplementedError
952 @abstractmethod
953 def add_related_collection(
954 self,
955 track_ids: List[Union[str, uuid.UUID]],
956 collection_id: Union[str, uuid.UUID],
957 name: str,
958 description: str = "",
959 user_id: Optional[Union[str, uuid.UUID]] = None,
960 relationship_type: str = "relationship",
961 meta: Optional[Dict[str, Any]] = None,
962 ) -> NendoCollection:
963 """Add a collection that is related to another `NendoCollection`.
965 Add a new collection with a relationship to and from the collection
966 with the given collection_id.
968 Args:
969 track_ids (List[Union[str, uuid.UUID]]): List of track ids.
970 collection_id (Union[str, uuid.UUID]): Existing collection id.
971 name (str): Name of the new related collection.
972 description (str): Description of the new related collection.
973 user_id (UUID, optional): The ID of the user adding the collection.
974 relationship_type (str): Type of the relationship.
975 meta (Dict[str, Any]): Meta of the new related collection.
977 Returns:
978 schema.NendoCollection: The newly added NendoCollection object.
979 """
980 raise NotImplementedError
982 @abstractmethod
983 def add_track_to_collection(
984 self,
985 track_id: Union[str, uuid.UUID],
986 collection_id: Union[str, uuid.UUID],
987 position: Optional[int] = None,
988 meta: Optional[Dict[str, Any]] = None,
989 ) -> NendoCollection:
990 """Creates a relationship from the track to the collection.
992 Args:
993 collection_id (Union[str, uuid.UUID]): Collection id.
994 track_id (Union[str, uuid.UUID]): Track id.
996 Returns:
997 schema.NendoCollection: The updated NendoCollection object.
998 """
999 raise NotImplementedError
1001 @abstractmethod
1002 def get_collection_tracks(
1003 self,
1004 collection_id: Union[str, uuid.UUID],
1005 ) -> List[NendoTrack]:
1006 """Get all tracks of a collection.
1008 Args:
1009 collection_id (Union[str, uuid.UUID]): Collection id.
1011 Returns:
1012 List[schema.NendoTrack]: List of tracks in the collection.
1013 """
1014 raise NotImplementedError
1016 @abstractmethod
1017 def get_collection(
1018 self,
1019 collection_id: uuid.uuid4,
1020 details: bool = True,
1021 ) -> Union[NendoCollection, NendoCollectionSlim]:
1022 """Get a collection by its ID.
1024 Args:
1025 collection_id (uuid.uuid4): ID of the target collection.
1026 details (bool, optional): Flag that defines whether the result should
1027 contain all fields or only a Defaults to True.
1029 Returns:
1030 Union[NendoCollection, NendoCollectionSlim]: Collection object, compact
1031 version if the `details` flag has been set to False.
1032 """
1033 raise NotImplementedError
1035 @abstractmethod
1036 @NendoPlugin.stream_output
1037 def get_collections(
1038 self,
1039 user_id: Optional[Union[str, uuid.UUID]] = None,
1040 order_by: Optional[str] = None,
1041 order: Optional[str] = "asc",
1042 limit: Optional[int] = None,
1043 offset: Optional[int] = None,
1044 ) -> Union[List, Iterator]:
1045 """Get a list of collections.
1047 Args:
1048 user_id (Union[str, UUID], optional): The user ID to filter for.
1049 order_by (Optional[str]): Key used for ordering the results.
1050 order (Optional[str]): Order in which to retrieve results ("asc" or "desc").
1051 limit (Optional[int]): Limit the number of returned results.
1052 offset (Optional[int]): Offset into the paginated results (requires limit).
1054 Returns:
1055 Union[List, Iterator]: List or generator of collections, depending on the
1056 configuration variable stream_mode
1057 """
1058 raise NotImplementedError
1060 @abstractmethod
1061 def find_collections(
1062 self,
1063 value: str = "",
1064 user_id: Optional[Union[str, uuid.UUID]] = None,
1065 order_by: Optional[str] = None,
1066 order: Optional[str] = "asc",
1067 limit: Optional[int] = None,
1068 offset: Optional[int] = None,
1069 ) -> Union[List, Iterator]:
1070 """Find collections with a search term in the description or meta field.
1072 Args:
1073 value (str): Term to be searched for in the description and meta field.
1074 user_id (Union[str, UUID], optional): The user ID to filter for.
1075 order_by (Optional[str]): Key used for ordering the results.
1076 order (Optional[str]): Order in which to retrieve results ("asc" or "desc").
1077 limit (Optional[int]): Limit the number of returned results.
1078 offset (Optional[int]): Offset into the paginated results (requires limit).
1080 Returns:
1081 Union[List, Iterator]: List or generator of collections, depending on the
1082 configuration variable stream_mode
1083 """
1084 raise NotImplementedError
1086 @abstractmethod
1087 def get_related_collections(
1088 self,
1089 collection_id: Union[str, uuid.UUID],
1090 user_id: Optional[Union[str, uuid.UUID]] = None,
1091 order_by: Optional[str] = None,
1092 order: Optional[str] = "asc",
1093 limit: Optional[int] = None,
1094 offset: Optional[int] = None,
1095 ) -> Union[List, Iterator]:
1096 """Get collections with a relationship to the collection with collection_id.
1098 Args:
1099 collection_id (str): ID of the collection to be searched for.
1100 user_id (Union[str, UUID], optional): The user ID to filter for.
1101 order_by (Optional[str]): Key used for ordering the results.
1102 order (Optional[str]): Order in which to retrieve results ("asc" or "desc").
1103 limit (Optional[int]): Limit the number of returned results.
1104 offset (Optional[int]): Offset into the paginated results (requires limit).
1106 Returns:
1107 Union[List, Iterator]: List or generator of collections, depending on the
1108 configuration variable stream_mode
1109 """
1110 raise NotImplementedError
1112 @abstractmethod
1113 def update_collection(
1114 self,
1115 collection: NendoCollection,
1116 ) -> NendoCollection:
1117 """Updates the given collection by storing it to the database.
1119 Args:
1120 collection (NendoCollection): The collection to store.
1122 Raises:
1123 NendoCollectionNotFoundError: If the collection with
1124 the given ID was not found.
1126 Returns:
1127 NendoCollection: The updated collection.
1128 """
1129 raise NotImplementedError
1131 @abstractmethod
1132 def remove_track_from_collection(
1133 self,
1134 track_id: Union[str, uuid.UUID],
1135 collection_id: Union[str, uuid.UUID],
1136 ) -> bool:
1137 """Deletes a relationship from the track to the collection.
1139 Args:
1140 collection_id (Union[str, uuid.UUID]): Collection id.
1141 track_id (Union[str, uuid.UUID]): Track id.
1143 Returns:
1144 success (bool): True if removal was successful, False otherwise.
1145 """
1146 raise NotImplementedError
1148 @abstractmethod
1149 def remove_collection(
1150 self,
1151 collection_id: uuid.UUID,
1152 remove_relationships: bool = False,
1153 ) -> bool:
1154 """Deletes the collection identified by `collection_id`.
1156 Args:
1157 collection_id (uuid.UUID): ID of the collection to remove.
1158 remove_relationships (bool, optional):
1159 If False prevent deletion if related tracks exist,
1160 if True delete relationships together with the object.
1161 Defaults to False.
1163 Returns:
1164 bool: True if deletion was successful, False otherwise.
1165 """
1166 raise NotImplementedError
1168 def export_collection(
1169 self,
1170 collection_id: Union[str, uuid.UUID],
1171 export_path: str,
1172 filename_suffix: str = "_nendo",
1173 file_format: str = "wav",
1174 ) -> List[str]:
1175 """Export the track to a file.
1177 Args:
1178 collection_id (Union[str, uuid.UUID]): The ID of the target
1179 collection to export.
1180 export_path (str): Path to a directory into which the collection's tracks
1181 should be exported.
1182 filename_suffix (str): The suffix which should be appended to each
1183 exported track's filename.
1184 file_format (str, optional): Format of the exported track. Ignored if
1185 file_path is a full file path. Defaults to "wav".
1187 Returns:
1188 List[str]: A list with all full paths to the exported files.
1189 """
1190 raise NotImplementedError
1192 # =========================
1193 #
1194 # BLOB MANAGEMENT FUNCTIONS
1195 #
1196 # =========================
1198 @abstractmethod
1199 def store_blob(
1200 self,
1201 file_path: Union[FilePath, str],
1202 user_id: Optional[Union[str, uuid.UUID]] = None,
1203 ) -> NendoBlob:
1204 """Stores a blob of data.
1206 Args:
1207 file_path (Union[FilePath, str]): Path to the file to store as blob.
1208 user_id (Optional[Union[str, uuid.UUID]], optional): ID of the user
1209 who's storing the file to blob.
1211 Returns:
1212 schema.NendoBlob: The stored blob.
1213 """
1214 raise NotImplementedError
1216 @abstractmethod
1217 def store_blob_from_bytes(
1218 self,
1219 data: bytes,
1220 user_id: Optional[Union[str, uuid.UUID]] = None,
1221 ) -> NendoBlob:
1222 """Stores a data of type `bytes` to a blob.
1224 Args:
1225 data (bytes): The blob to store.
1226 user_id (Optional[Union[str, uuid.UUID]], optional): ID of the user
1227 who's storing the bytes to blob.
1229 Returns:
1230 schema.NendoBlob: The stored blob.
1231 """
1232 raise NotImplementedError
1234 @abstractmethod
1235 def load_blob(
1236 self,
1237 blob_id: uuid.UUID,
1238 user_id: Optional[Union[str, uuid.UUID]] = None,
1239 ) -> NendoBlob:
1240 """Loads a blob of data into memory.
1242 Args:
1243 blob_id (uuid.UUID): The UUID of the blob.
1244 user_id (Optional[Union[str, uuid.UUID]], optional): ID of the user
1245 who's loading the blob.
1247 Returns:
1248 schema.NendoBlob: The loaded blob.
1249 """
1250 raise NotImplementedError
1252 @abstractmethod
1253 def remove_blob(
1254 self,
1255 blob_id: uuid.UUID,
1256 remove_resources: bool = True,
1257 user_id: Optional[uuid.UUID] = None,
1258 ) -> bool:
1259 """Deletes a blob of data.
1261 Args:
1262 blob_id (uuid.UUID): The UUID of the blob.
1263 remove_resources (bool): If True, remove associated resources.
1264 user_id (Optional[Union[str, uuid.UUID]], optional): ID of the user
1265 who's removing the blob.
1267 Returns:
1268 success (bool): True if removal was successful, False otherwise
1269 """
1270 raise NotImplementedError
1272 # ==================================
1273 #
1274 # MISCELLANEOUS MANAGEMENT FUNCTIONS
1275 #
1276 # ==================================
1278 def get_track_or_collection(
1279 self,
1280 target_id: Union[str, uuid.UUID],
1281 ) -> Union[NendoTrack, NendoCollection]:
1282 """Return a track or a collection based on the given target_id.
1284 Args:
1285 target_id (Union[str, uuid.UUID]): The target ID to obtain.
1287 Returns:
1288 Union[NendoTrack, NendoCollection]: The track or the collection.
1289 """
1290 target_id = ensure_uuid(target_id)
1291 collection = self.get_collection(target_id)
1292 if collection is not None:
1293 return collection
1295 # assume the id is a track id
1296 return self.get_track(target_id)
1298 def verify(self, action: Optional[str] = None, user_id: str = "") -> None:
1299 """Verify the library's integrity.
1301 Args:
1302 action (Optional[str], optional): Default action to choose when an
1303 inconsistency is detected. Choose between (i)gnore and (r)emove.
1304 """
1305 original_config = {}
1306 try:
1307 original_config["stream_mode"] = self.config.stream_mode
1308 original_config["stream_chunk_size"] = self.config.stream_chunk_size
1309 self.config.stream_mode = False
1310 self.config.stream_chunk_size = 16
1311 for track in self.get_tracks():
1312 if not self.storage_driver.file_exists(
1313 file_name=track.resource.file_name,
1314 user_id=user_id,
1315 ):
1316 action = (
1317 action
1318 or input(
1319 f"Inconsistency detected: {track.resource.src} "
1320 "does not exist. Please choose an action:\n"
1321 "(i) ignore - (r) remove",
1322 ).lower()
1323 )
1324 if action == "i":
1325 self.logger.warning(
1326 "Detected missing file "
1327 f"{track.resource.src} but instructed "
1328 "to ignore.",
1329 )
1330 continue
1331 if action == "r":
1332 self.logger.info(
1333 f"Removing track with ID {track.id} "
1334 f"due to missing file {track.resource.src}",
1335 )
1336 self.remove_track(
1337 track_id=track.id,
1338 remove_plugin_data=True,
1339 remove_relationships=True,
1340 remove_resources=False,
1341 )
1342 for library_file in self.storage_driver.list_files(user_id=user_id):
1343 file_without_ext = os.path.splitext(library_file)[0]
1344 if len(self.find_tracks(value=file_without_ext)) == 0:
1345 action = (
1346 action
1347 or input(
1348 f"Inconsistency detected: File {library_file} "
1349 "cannot be fonud in database. Please choose an action:\n"
1350 "(i) ignore - (r) remove",
1351 ).lower()
1352 )
1353 if action == "i":
1354 self.logger.warning(
1355 f"Detected orphaned file {library_file} "
1356 f"but instructed to ignore.",
1357 )
1358 continue
1359 if action == "r":
1360 self.logger.info(f"Removing orphaned file {library_file}")
1361 self.storage_driver.remove_file(
1362 file_name=library_file,
1363 user_id=user_id,
1364 )
1366 finally:
1367 self.config.stream_mode = original_config["stream_mode"]
1368 self.config.stream_chunk_size = original_config["stream_chunk_size"]
1370 @abstractmethod
1371 def reset(
1372 self,
1373 force: bool = False,
1374 user_id: Optional[Union[str, uuid.UUID]] = None,
1375 ) -> None:
1376 """Reset the nendo library.
1378 Erase all tracks, collections and relationships.
1379 Ask before erasing.
1381 Args:
1382 force (bool, optional): Flag that specifies whether to ask the user for
1383 confirmation of the operation. Default is to ask the user.
1384 user_id (Optional[Union[str, uuid.UUID]], optional): ID of the user
1385 who's resetting the library. If none is given, the configured
1386 nendo default user will be used.
1387 """
1388 raise NotImplementedError
1390 def __str__(self):
1391 output = f"{self.plugin_name}, version {self.plugin_version}:\n"
1392 output += f"{len(self)} tracks"
1393 return output
1395 @property
1396 def plugin_type(self) -> str:
1397 """Return type of plugin."""
1398 return "LibraryPlugin"