Coverage for src/nendo/schema/core.py: 73%
639 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-11-30 09:47 +0100
1# -*- encoding: utf-8 -*-
2# ruff: noqa: A003, TCH002, TCH001
3"""Core schema definitions.
5Contains class definitions for all common nendo objects.
6"""
8from __future__ import annotations
10import functools
11import inspect
12import logging
13import os
14import pickle
15import re
16import shutil
17import time
18import uuid
19from abc import ABC, abstractmethod
20from concurrent.futures import ThreadPoolExecutor, as_completed
21from datetime import datetime
22from enum import Enum
23from typing import Any, ClassVar, Dict, List, Optional, Tuple, Union
25import librosa
26import numpy as np
27import soundfile as sf
28from pydantic import BaseModel, ConfigDict, Field, FilePath
30from nendo.config import NendoConfig
31from nendo.main import Nendo
32from nendo.schema.exception import NendoError, NendoPluginRuntimeError
33from nendo.utils import (
34 ensure_uuid,
35 get_wrapped_methods,
36 md5sum,
37 play_signal,
38 pretty_print,
39)
41logger = logging.getLogger("nendo")
44class ResourceType(str, Enum):
45 """Enum representing different types of resources used in Nendo."""
47 audio: str = "audio"
48 image: str = "image"
49 model: str = "model"
50 blob: str = "blob"
53class ResourceLocation(str, Enum):
54 """Enum representing differnt types of resource locations.
56 (e.g. original filepath, local FS library path, S3 bucket, etc.)
57 """
59 original: str = "original"
60 local: str = "local"
61 gcs: str = "gcs"
62 s3: str = "s3"
65class Visibility(str, Enum):
66 """Enum representing different visibilities of information in the nendo Library.
68 Mostly relevant when sharing a nendo library between different users.
69 """
71 public: str = "public"
72 private: str = "private"
73 deleted: str = "deleted"
76class NendoUserBase(BaseModel): # noqa: D101
77 model_config = ConfigDict(
78 arbitrary_types_allowed=True,
79 from_attributes=True,
80 use_enum_values=True,
81 )
83 name: str
84 password: str
85 email: str
86 avatar: str = "/assets/images/default_avatar.png"
87 verified: bool = False
88 last_login: datetime = Field(default_factory=datetime.now)
91class NendoUser(NendoUserBase):
92 """Basic class representing a Nendo user."""
94 id: uuid.UUID = Field(default_factory=uuid.uuid4)
95 created_at: datetime = Field(default_factory=datetime.now)
98class NendoUserCreate(NendoUserBase): # noqa: D101
99 pass
102class NendoTrackSlim(BaseModel): # noqa: D101
103 model_config = ConfigDict(
104 from_attributes=True,
105 use_enum_values=True,
106 )
108 id: uuid.UUID
109 user_id: uuid.UUID
110 track_type: str = "track"
111 meta: Dict[str, Any] = Field(default_factory=dict)
114class NendoCollectionSlim(BaseModel): # noqa: D101
115 model_config = ConfigDict(
116 from_attributes=True,
117 use_enum_values=True,
118 )
120 id: uuid.UUID
121 name: str
122 description: str = ""
123 collection_type: str = "collection"
124 user_id: uuid.UUID
125 meta: Dict[str, Any] = Field(default_factory=dict)
128class NendoResource(BaseModel, ABC):
129 """Basic class representing a resource in Nendo.
131 For example, every `NendoTrack` has at least one associated `NendoResource`,
132 namely the file containing its waveform. But it can also have other associated
133 resources in the form of images, etc.
134 """
136 model_config = ConfigDict(
137 arbitrary_types_allowed=True,
138 from_attributes=True,
139 use_enum_values=True,
140 )
142 id: uuid.UUID = Field(default_factory=uuid.uuid4)
143 created_at: datetime = Field(default_factory=datetime.now)
144 updated_at: datetime = Field(default_factory=datetime.now)
145 file_path: str
146 file_name: str
147 resource_type: ResourceType = ResourceType.audio
148 location: ResourceLocation = ResourceLocation.local
149 meta: Dict[str, Any] = Field(default_factory=dict)
151 @property
152 def src(self): # noqa: D102
153 return os.path.join(self.file_path, self.file_name)
156class NendoRelationshipBase(BaseModel, ABC):
157 """Base class representing a relationship between two Nendo Core objects."""
159 model_config = ConfigDict(use_enum_values=True)
161 source_id: uuid.UUID
162 target_id: uuid.UUID
163 created_at: datetime = Field(default_factory=datetime.now)
164 updated_at: datetime = Field(default_factory=datetime.now)
165 relationship_type: str
166 meta: Dict[str, Any] = Field(default_factory=dict)
169class NendoRelationship(NendoRelationshipBase):
170 """Base class for Nendo Core relationships.
172 All relationship classes representing relationships between specific
173 types of Nendo Core objects inherit from this class.
174 """
176 model_config = ConfigDict(
177 from_attributes=True,
178 use_enum_values=True,
179 )
181 id: uuid.UUID
184class NendoTrackTrackRelationship(NendoRelationship):
185 """Class representing a relationship between two `NendoTrack`s."""
187 source: NendoTrackSlim
188 target: NendoTrackSlim
191class NendoTrackCollectionRelationship(NendoRelationship):
192 """Class representing a relationship between a `NendoTrack` and a `NendoCollection`."""
194 source: NendoTrackSlim
195 target: NendoCollectionSlim
196 relationship_position: int
199class NendoCollectionCollectionRelationship(NendoRelationship):
200 """Class representing a relationship between two `NendoCollection`s."""
202 source: NendoCollectionSlim
203 target: NendoCollectionSlim
206class NendoRelationshipCreate(NendoRelationshipBase): # noqa: D101
207 pass
210class NendoPluginDataBase(BaseModel, ABC): # noqa: D101
211 model_config = ConfigDict(
212 from_attributes=True,
213 )
215 track_id: uuid.UUID
216 user_id: Optional[uuid.UUID] = None
217 plugin_name: str
218 plugin_version: str
219 key: str
220 value: str
223class NendoPluginData(NendoPluginDataBase):
224 """Class representing basic plugin data attached to a track."""
226 id: uuid.UUID = Field(default_factory=uuid.uuid4)
227 created_at: datetime = Field(default_factory=datetime.now)
228 updated_at: datetime = Field(default_factory=datetime.now)
230 def __str__(self):
231 # output = f"id: {self.id}"
232 output = "----------------"
233 output += f"\nplugin name: {self.plugin_name}"
234 output += f"\nplugin version: {self.plugin_version}"
235 # output += f"\nuser id: {self.user_id}"
236 output += f"\nkey: {self.key}"
237 output += f"\nvalue: {self.value}"
238 return output
241class NendoPluginDataCreate(NendoPluginDataBase): # noqa: D101
242 pass
245class NendoTrackBase(BaseModel):
246 """Base class for tracks in Nendo."""
248 model_config = ConfigDict(
249 arbitrary_types_allowed=True,
250 from_attributes=True,
251 use_enum_values=True,
252 )
254 nendo_instance: Optional[Nendo] = None
255 user_id: uuid.UUID
256 track_type: str = "track"
257 visibility: Visibility = Visibility.private
258 images: List[NendoResource] = Field(default_factory=list)
259 resource: NendoResource
260 related_tracks: List[NendoTrackTrackRelationship] = Field(default_factory=list)
261 related_collections: List[NendoTrackCollectionRelationship] = Field(
262 default_factory=list,
263 )
264 meta: Dict[str, Any] = Field(default_factory=dict)
265 plugin_data: List[NendoPluginData] = Field(default_factory=list)
267 def __init__(self, **kwargs: Any) -> None: # noqa: D107
268 super().__init__(**kwargs)
269 self.nendo_instance = Nendo()
271 @property
272 def signal(self) -> np.ndarray:
273 """Lazy-load the signal from the track using librosa.
275 Returns:
276 np.ndarray: The signal of the track.
277 """
278 signal = self.__dict__.get("signal")
279 if signal is None:
280 track_local = self.nendo_instance.library.storage_driver.as_local(
281 file_path=self.resource.src,
282 location=self.resource.location,
283 user_id=self.nendo_instance.config.user_id,
284 )
285 signal, sr = librosa.load(track_local, sr=self.sr, mono=False)
286 self.__dict__["sr"] = sr
287 self.__dict__["signal"] = signal
288 return signal
290 @property
291 def sr(self) -> int:
292 """Lazy-load the sample rate of the track.
294 Returns:
295 int: The sample rate of the track.
296 """
297 sr = self.__dict__.get("sr") or self.get_meta("sr")
298 if sr is None:
299 track_local = self.nendo_instance.library.storage_driver.as_local(
300 file_path=self.resource.src,
301 location=self.resource.location,
302 user_id=self.nendo_instance.config.user_id,
303 )
304 sr = librosa.get_samplerate(track_local)
305 self.set_meta({"sr": sr})
306 self.__dict__["sr"] = sr
307 return sr
309 def model_dump(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: # noqa: D102
310 result = super().model_dump(*args, **kwargs)
311 # remove properties
312 for name, value in self.__class__.__dict__.items():
313 if isinstance(value, property):
314 result.pop(name, None)
315 return result
317 def __getitem__(self, key: str) -> Any:
318 return self.meta[key]
321class NendoTrack(NendoTrackBase):
322 """Basic class representing a nendo track."""
324 model_config = ConfigDict(
325 from_attributes=True,
326 )
328 id: uuid.UUID
329 created_at: datetime = Field(default_factory=datetime.now)
330 updated_at: datetime = Field(default_factory=datetime.now)
332 def __str__(self):
333 output = f"\nid: {self.id}"
334 output += f"\nsample rate: {self.sr}"
335 output += f"{pretty_print(self.meta)}"
336 return output
338 def __len__(self):
339 """Return the length of the track in seconds."""
340 return self.signal.shape[1] / self.sr
342 @classmethod
343 def model_validate(cls, *args, **kwargs):
344 """Inject the nendo instance upon conversion from ORM."""
345 instance = super().model_validate(*args, **kwargs)
346 instance.nendo_instance = Nendo()
347 return instance
349 def resample(self, rsr: int = 44100) -> np.ndarray:
350 """Resample track."""
351 new_signal = librosa.resample(self.signal, orig_sr=self.sr, target_sr=rsr)
352 self.__dict__["signal"] = new_signal
353 self.__dict__["sr"] = rsr
354 return new_signal
356 def local(self) -> str:
357 """Get a path to a local file handle on the track."""
358 return self.nendo_instance.library.storage_driver.as_local(
359 file_path=self.resource.src,
360 location=self.resource.location,
361 user_id=self.nendo_instance.config.user_id,
362 )
364 def overlay(self, track: NendoTrack, gain_db: Optional[float] = 0) -> NendoTrack:
365 """Overlay two tracks using gain control in decibels.
367 The gain gets applied to the second track.
368 This function creates a new related track in the library.
370 Args:
371 track (NendoTrack): The track to overlay with.
372 gain_db (Optional[float], optional): The gain to apply to the second track.
373 Defaults to 0.
375 Returns:
376 NendoTrack: The resulting mixed track.
377 """
378 if self.sr > track.sr:
379 self.resample(track.sr)
380 elif self.sr < track.sr:
381 track.resample(self.sr)
383 if self.signal.shape[1] > track.signal.shape[1]:
384 signal_one = self.signal[:, : track.signal.shape[1]]
385 signal_two = track.signal
386 else:
387 signal_one = self.signal
388 signal_two = track.signal[:, : self.signal.shape[1]]
390 # Convert dB gain to linear gain factor
391 gain_factor_during_overlay = 10 ** (gain_db / 20)
393 new_signal = signal_one + (signal_two * gain_factor_during_overlay)
394 return self.nendo_instance.library.add_related_track_from_signal(
395 signal=new_signal,
396 sr=self.sr,
397 track_type="track",
398 related_track_id=self.id,
399 track_meta={"overlay_parameters": {"gain_db": gain_db}},
400 )
402 def slice(self, end: float, start: Optional[float] = 0) -> np.ndarray:
403 """Slice a track.
405 Args:
406 end (float): End of the slice in seconds.
407 start (Optional[float], optional): Start of the slice in seconds.
408 Defaults to 0.
410 Returns:
411 np.ndarray: The sliced track.
412 """
413 start_frame = int(start * self.sr)
414 end_frame = int(end * self.sr)
415 return self.signal[:, start_frame:end_frame]
417 def save(self) -> NendoTrack:
418 """Save the track to the library.
420 Returns:
421 NendoTrack: The track itself.
422 """
423 self.nendo_instance.library.update_track(track=self)
424 return self
426 def delete(
427 self,
428 remove_relationships: bool = False,
429 remove_plugin_data: bool = True,
430 remove_resources: bool = True,
431 user_id: Optional[Union[str, uuid.UUID]] = None,
432 ) -> NendoTrack:
433 """Delete the track from the library.
435 Args:
436 remove_relationships (bool):
437 If False prevent deletion if related tracks exist,
438 if True delete relationships together with the object
439 remove_plugin_data (bool):
440 If False prevent deletion if related plugin data exist
441 if True delete plugin data together with the object
442 remove_resources (bool):
443 If False, keep the related resources, e.g. files
444 if True, delete the related resources
445 user_id (Union[str, UUID], optional): The ID of the user owning the track.
447 Returns:
448 NendoTrack: The track itself.
449 """
450 self.nendo_instance.library.remove_track(
451 track_id=self.id,
452 remove_relationships=remove_relationships,
453 remove_plugin_data=remove_plugin_data,
454 remove_resources=remove_resources,
455 user_id=user_id,
456 )
457 return self
459 def set_meta(self, meta: Dict[str, Any]) -> NendoTrack:
460 """Set metadata of track.
462 Args:
463 meta (Dict[str, Any]): Dictionary containing the metadata to be set.
465 Returns:
466 NendoTrack: The track itself.
467 """
468 try:
469 self.meta.update(meta)
470 self.nendo_instance.library.update_track(track=self)
471 except NendoError as e:
472 logger.exception("Error updating meta: %s", e)
473 return self
475 def has_meta(self, key: str) -> bool:
476 """Check if a given track has the given key in its meta dict.
478 Args:
479 key (str): The key to check for.
481 Returns:
482 bool: True if the key exists, False otherwise.
483 """
484 return any(k == key for k in self.meta)
486 def get_meta(self, key: str) -> Dict[str, Any]:
487 """Get the meta entry for the given key.
489 Args:
490 key (str): The key to get metadata for.
492 Returns:
493 Dict[str, Any]: Meta entry for given key.
494 """
495 if not self.has_meta(key):
496 logger.error("Key not found in meta: %s", key)
497 return None
498 return self.meta[key]
500 def remove_meta(self, key: str) -> NendoTrack:
501 """Remove the meta entry for the given key.
503 Args:
504 key (str): The key to remove metadata for.
506 Returns:
507 NendoTrack: The track itself.
508 """
509 if not self.has_meta(key):
510 logger.error("Key not found in meta: %s", key)
511 return None
512 _ = self.meta.pop(key, None)
513 self.nendo_instance.library.update_track(track=self)
514 return self
516 def add_plugin_data(
517 self,
518 plugin_name: str,
519 plugin_version: str,
520 key: str,
521 value: str,
522 user_id: Optional[Union[str, uuid.UUID]] = None,
523 replace: bool = True,
524 ) -> NendoTrack:
525 """Add plugin data to a NendoTrack and persist changes into the DB.
527 Args:
528 plugin_name (str): Name of the plugin.
529 plugin_version (str): Version of the plugin.
530 key (str): Key under which to save the data.
531 value (Any): Data to save.
532 user_id (Union[str, UUID], optional): ID of user adding the plugin data.
533 replace (bool, optional): Flag that determines whether
534 the last existing data point for the given plugin name and -version
535 is overwritten or not. Defaults to True.
537 Returns:
538 NendoTrack: The track itself.
539 """
540 pd = self.nendo_instance.library.add_plugin_data(
541 track_id=self.id,
542 plugin_name=plugin_name,
543 plugin_version=plugin_version,
544 key=key,
545 value=value,
546 user_id=user_id,
547 replace=replace,
548 )
549 self.plugin_data.append(pd)
550 return self
552 def get_plugin_data(
553 self,
554 plugin_name: str = "",
555 key: str = "",
556 ) -> List[NendoPluginData]:
557 """Get all plugin data related to the given plugin name and the given key.
559 Note: Function behavior
560 - If no plugin_name is specified, all plugin data found with the given
561 key is returned.
562 - If no key is specified, all plugin data found with the given
563 plugin_name is returned.
564 - If neither key, nor plugin_name is specified, all plugin data
565 is returned.
566 - If the return value is a single item, it's `value` will be returned
567 directly, otherwise a list of `NendoPluginData` will be returned.
568 - Certain kinds of plugin data are actually stored as blobs
569 and the corresponding blob id is stored in the plugin data's value
570 field. Those will be automatically loaded from the blob into memory
571 and a `NendoBlob` object will be returned inside `plugin_data.value`.
573 Args:
574 plugin_name (str): The name of the plugin to get the data for.
575 Defaults to "".
576 key (str): The key to filter plugin data for.
577 Defaults to "".
579 Returns:
580 List[NendoPluginData]: List of nendo plugin data entries.
581 """
582 uuid_pattern = re.compile(
583 r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-"
584 r"[89ab][0-9a-f]{3}-[0-9a-f]{12}\Z",
585 re.I,
586 )
587 plugin_data = []
588 for pd in self.plugin_data:
589 if (pd.plugin_name == plugin_name or len(plugin_name) == 0) and (
590 pd.key == key or len(key) == 0
591 ):
592 # if we have a UUID, load the corresponding blob
593 if uuid_pattern.match(pd.value):
594 loaded_blob = self.nendo_instance.library.load_blob(
595 blob_id=uuid.UUID(pd.value),
596 )
597 pd.value = loaded_blob
598 plugin_data.append(pd)
599 return plugin_data
601 def add_related_track(
602 self,
603 file_path: FilePath,
604 track_type: str = "str",
605 user_id: Optional[uuid.UUID] = None,
606 track_meta: Optional[Dict[str, Any]] = None,
607 relationship_type: str = "relationship",
608 meta: Optional[Dict[str, Any]] = None,
609 ) -> NendoTrack:
610 """Adds a new track with a relationship to the current one.
612 Args:
613 file_path (FilePath): Path to the file to add as track.
614 track_type (str): Track type. Defaults to "track".
615 user_id (Union[str, UUID], optional): ID of the user adding the track.
616 track_meta (dict, optional): Dictionary containing the track metadata.
617 relationship_type (str): Type of the relationship.
618 Defaults to "relationship".
619 meta (dict): Dictionary containing metadata about
620 the relationship. Defaults to {}.
622 Returns:
623 NendoTrack: The track itself.
624 """
625 related_track = self.nendo_instance.library.add_related_track(
626 file_path=file_path,
627 related_track_id=self.id,
628 track_type=track_type,
629 user_id=user_id or self.user_id,
630 track_meta=track_meta,
631 relationship_type=relationship_type,
632 meta=meta,
633 )
634 self.related_tracks.append(related_track.related_tracks[0])
635 return self
637 def add_related_track_from_signal(
638 self,
639 signal: np.ndarray,
640 sr: int,
641 track_type: str = "track",
642 user_id: Optional[uuid.UUID] = None,
643 track_meta: Optional[Dict[str, Any]] = None,
644 relationship_type: str = "relationship",
645 meta: Optional[Dict[str, Any]] = None,
646 ) -> NendoTrack:
647 """Adds a new track with a relationship to the current one.
649 Args:
650 signal (np.ndarray): Waveform of the track in numpy array form.
651 sr (int): Sampling rate of the waveform.
652 track_type (str): Track type. Defaults to "track".
653 user_id (UUID, optional): ID of the user adding the track.
654 track_meta (dict, optional): Dictionary containing the track metadata.
655 relationship_type (str): Type of the relationship.
656 Defaults to "relationship".
657 meta (dict): Dictionary containing metadata about
658 the relationship. Defaults to {}.
660 Returns:
661 NendoTrack: The track itself.
662 """
663 related_track = self.nendo_instance.library.add_related_track_from_signal(
664 signal=signal,
665 sr=sr,
666 track_type=track_type,
667 related_track_id=self.id,
668 user_id=user_id,
669 track_meta=track_meta,
670 relationship_type=relationship_type,
671 meta=meta,
672 )
673 self.related_tracks.append(related_track.related_tracks[0])
674 return self
676 def has_relationship(self, relationship_type: str = "relationship") -> bool:
677 """Check whether the track has any relationships of the specified type.
679 Args:
680 relationship_type (str): Type of the relationship to check for.
681 Defaults to "relationship".
683 Returns:
684 bool: True if a relationship of the given type exists, False otherwise.
685 """
686 all_relationships = self.related_tracks + self.related_collections
687 if len(all_relationships) == 0:
688 return False
689 return any(r.relationship_type == relationship_type for r in all_relationships)
691 def has_relationship_to(self, track_id: Union[str, uuid.UUID]) -> bool:
692 """Check if the track has a relationship to the track with the given track_id.
694 Args:
695 track_id (Union[str, uuid.UUID]): ID of the track to which to check
696 for relationships.
698 Returns:
699 bool: True if a relationship to the track with the given track_id exists.
700 False otherwise.
701 """
702 track_id = ensure_uuid(track_id)
703 if self.related_tracks is None:
704 return False
705 return any(r.target_id == track_id for r in self.related_tracks)
707 def get_related_tracks(
708 self,
709 user_id: Optional[Union[str, uuid.UUID]] = None,
710 order_by: Optional[str] = None,
711 order: Optional[str] = "asc",
712 limit: Optional[int] = None,
713 offset: Optional[int] = None,
714 ) -> List[NendoTrack]:
715 """Get all tracks to which the current track has a relationship.
717 Args:
718 user_id (Union[str, UUID], optional): The user ID to filter for.
719 order_by (Optional[str]): Key used for ordering the results.
720 order (Optional[str]): Order in which to retrieve results ("asc" or "desc").
721 limit (Optional[int]): Limit the number of returned results.
722 offset (Optional[int]): Offset into the paginated results (requires limit).
724 Returns:
725 List[NendoTrack]: List containting all related NendoTracks
726 """
727 return self.nendo_instance.library.get_related_tracks(
728 track_id=self.id,
729 user_id=user_id,
730 order_by=order_by,
731 order=order,
732 limit=limit,
733 offset=offset,
734 )
736 def add_to_collection(
737 self,
738 collection_id: Union[str, uuid.UUID],
739 position: Optional[int] = None,
740 meta: Optional[Dict[str, Any]] = None,
741 ) -> NendoTrack:
742 """Adds the track to the collection given as collection_id.
744 Args:
745 collection_id (Union[str, uuid.UUID]): ID of the collection to
746 which to add the track.
747 position (int, optional): Target position of the track inside
748 the collection.
749 meta (Dict[str, Any]): Metadata of the relationship.
751 Returns:
752 NendoTrack: The track itself.
753 """
754 self.nendo_instance.library.add_track_to_collection(
755 track_id=self.id,
756 collection_id=collection_id,
757 position=position,
758 meta=meta,
759 )
760 return self
762 def remove_from_collection(
763 self,
764 collection_id: Union[str, uuid.UUID],
765 ) -> NendoTrack:
766 """Remove the track from the collection specified by collection_id.
768 Args:
769 collection_id (Union[str, uuid.UUID]): ID of the collection from which
770 to remove the track.
772 Returns:
773 NendoTrack: The track itself.
774 """
775 self.nendo_instance.library.remove_track_from_collection(
776 track_id=self.id,
777 collection_id=collection_id,
778 )
779 return self
781 def process(self, plugin: str, **kwargs: Any) -> Union[NendoTrack, NendoCollection]:
782 """Process the track with the specified plugin.
784 Args:
785 plugin (str): Name of the plugin to run on the track.
787 Returns:
788 Union[NendoTrack, NendoCollection]: The resulting track or collection,
789 depending on what the plugin returns.
790 """
791 registered_plugin: RegisteredNendoPlugin = getattr(
792 self.nendo_instance.plugins,
793 plugin,
794 )
795 wrapped_method = get_wrapped_methods(registered_plugin.plugin_instance)
797 if len(wrapped_method) > 1:
798 raise NendoError(
799 "Plugin has more than one wrapped method. Please use `nd.plugins.<plugin_name>.<method_name>` instead.",
800 )
801 return getattr(self.nendo_instance.plugins, plugin)(track=self, **kwargs)
803 def export(self, file_path: str, file_format: str = "wav") -> NendoTrack:
804 """Export the track to a file.
806 Args:
807 file_path (str): Path to the exported file. Can be either a full
808 file path or a directory path. If a directory path is given,
809 a filename will be automatically generated and the file will be
810 exported to the format specified as file_format. If a full file
811 path is given, the format will be deduced from the path and the
812 file_format parameter will be ignored.
813 file_format (str, optional): Format of the exported track. Ignored if
814 file_path is a full file path. Defaults to "wav".
816 Returns:
817 NendoTrack: The track itself.
818 """
819 self.nendo_instance.library.export_track(
820 track_id=self.id,
821 file_path=file_path,
822 file_format=file_format,
823 )
824 return self
826 def play(self):
827 """Play the track."""
828 play_signal(self.signal, self.sr)
830 def loop(self):
831 """Loop the track."""
832 play_signal(self.signal, self.sr, loop=True)
835class NendoTrackCreate(NendoTrackBase): # noqa: D101
836 pass
839class NendoCollectionBase(BaseModel): # noqa: D101
840 model_config = ConfigDict(
841 arbitrary_types_allowed=True,
842 from_attributes=True,
843 use_enum_values=True,
844 )
846 nendo_instance: Optional[Nendo] = None
847 name: str
848 description: str = ""
849 collection_type: str = "collection"
850 user_id: uuid.UUID
851 visibility: Visibility = Visibility.private
852 meta: Dict[str, Any] = Field(default_factory=dict)
853 related_tracks: List[NendoTrackCollectionRelationship] = Field(default_factory=list)
854 related_collections: List[NendoCollectionCollectionRelationship] = Field(
855 default_factory=list,
856 )
858 def __init__(self, **kwargs: Any) -> None: # noqa: D107
859 super().__init__(**kwargs)
860 self.nendo_instance = Nendo()
863class NendoCollection(NendoCollectionBase):
864 """Basic class representing a nendo collection."""
866 model_config = ConfigDict(
867 from_attributes=True,
868 )
870 id: uuid.UUID
871 created_at: datetime = Field(default_factory=datetime.now)
872 updated_at: datetime = Field(default_factory=datetime.now)
874 def __str__(self):
875 output = f"id: {self.id}"
876 output += f"\ntype: {self.collection_type}"
877 output += f"\ndescription: {self.description}"
878 output += f"\nuser id: {self.user_id}"
879 output += f"\nvisibility: {self.visibility}"
880 output += f"{pretty_print(self.meta)}"
881 return output
883 @classmethod
884 def model_validate(cls, *args, **kwargs): # noqa: D102
885 instance = super().model_validate(*args, **kwargs)
886 instance.nendo_instance = Nendo()
887 return instance
889 def __getitem__(self, index: int) -> NendoTrack:
890 """Return the track at the specified index."""
891 return self.tracks()[index]
893 def __len__(self):
894 """Return the number of tracks in the collection."""
895 return len(self.tracks())
897 def tracks(self) -> List[NendoTrack]:
898 """Return all tracks listed in the collection.
900 Collection will be loaded from the DB if not already loaded.
902 Returns:
903 List[NendoTrack]: List of tracks.
904 """
905 ts = self.__dict__.get("loaded_tracks")
906 if ts is None:
907 ts = self.nendo_instance.library.get_collection_tracks(self.id)
908 self.__dict__["loaded_tracks"] = ts
909 return ts
911 def save(self) -> NendoCollection:
912 """Save the collection to the nendo library.
914 Returns:
915 NendoCollection: The collection itself.
916 """
917 self.nendo_instance.library.update_collection(collection=self)
918 return self
920 def delete(
921 self,
922 remove_relationships: bool = False,
923 ) -> NendoCollection:
924 """Deletes the collection from the nendo library.
926 Args:
927 remove_relationships (bool, optional):
928 If False prevent deletion if related tracks exist,
929 if True delete relationships together with the object.
930 Defaults to False.
932 Returns:
933 NendoCollection: The collection itself.
934 """
935 self.nendo_instance.library.remove_collection(
936 collection_id=self.id,
937 remove_relationships=remove_relationships,
938 )
939 return self
941 def process(self, plugin: str, **kwargs: Any) -> NendoCollection:
942 """Process the collection with the specified plugin.
944 Args:
945 plugin (str): Name of the plugin to run on the collection.
947 Returns:
948 NendoCollection: The collection that was created by the plugin.
949 """
950 registered_plugin: RegisteredNendoPlugin = getattr(
951 self.nendo_instance.plugins,
952 plugin,
953 )
954 wrapped_method = get_wrapped_methods(registered_plugin.plugin_instance)
956 if len(wrapped_method) > 1:
957 raise NendoError(
958 "Plugin has more than one wrapped method. Please use `nd.plugins.<plugin_name>.<method_name>` instead.",
959 )
960 return getattr(self.nendo_instance.plugins, plugin)(collection=self, **kwargs)
962 def has_relationship(self, relationship_type: str = "relationship") -> bool:
963 """Check if the collection has the specified relationship type.
965 Args:
966 relationship_type (str): Type of the relationship to check for.
967 Defaults to "relationship".
969 Returns:
970 bool: True if a relationship of the given type exists, False otherwise.
971 """
972 if self.related_collections is None:
973 return False
975 return any(
976 r.relationship_type == relationship_type for r in self.related_collections
977 )
979 def has_relationship_to(self, collection_id: Union[str, uuid.UUID]) -> bool:
980 """Check if the collection has a relationship to the specified collection ID."""
981 collection_id = ensure_uuid(collection_id)
982 if self.related_collections is None:
983 return False
985 return any(r.target_id == collection_id for r in self.related_collections)
987 def add_track(
988 self,
989 track_id: Union[str, uuid.UUID],
990 position: Optional[int] = None,
991 meta: Optional[Dict[str, Any]] = None,
992 ) -> NendoCollection:
993 """Creates a relationship from the track to the collection.
995 Args:
996 track_id (Union[str, uuid.UUID]): ID of the track to add.
997 position (int, optional): Target position of the track inside
998 the collection.
999 meta (Dict[str, Any]): Metadata of the relationship.
1001 Returns:
1002 NendoCollection: The collection itself.
1003 """
1004 updated_collection = self.nendo_instance.library.add_track_to_collection(
1005 track_id=track_id,
1006 collection_id=self.id,
1007 position=position,
1008 meta=meta,
1009 )
1010 self.related_tracks = updated_collection.related_tracks
1011 return self
1013 def remove_track(
1014 self,
1015 track_id: Union[str, uuid.UUID],
1016 ) -> NendoCollection:
1017 """Removes the track from the collection.
1019 Args:
1020 track_id (Union[str, uuid.UUID]): ID of the track to remove
1021 from the collection.
1023 Returns:
1024 NendoCollection: The collection itself.
1025 """
1026 self.nendo_instance.library.remove_track_from_collection(
1027 track_id=track_id,
1028 collection_id=self.id,
1029 )
1030 # NOTE the following removes _all_ relationships between the track and
1031 # the collection. In the future, this could be refined to account for cases
1032 # where multiple relationships of different types exist between a track
1033 # and a collection
1034 self.related_tracks = [t for t in self.related_tracks if t.id != track_id]
1035 return self
1037 def add_related_collection(
1038 self,
1039 track_ids: List[Union[str, uuid.UUID]],
1040 name: str,
1041 description: str = "",
1042 user_id: Optional[Union[str, uuid.UUID]] = None,
1043 relationship_type: str = "relationship",
1044 meta: Optional[Dict[str, Any]] = None,
1045 ) -> NendoCollection:
1046 """Create a new collection with a relationship to the current collection.
1048 Args:
1049 track_ids (List[Union[str, uuid.UUID]]): List of track ids.
1050 name (str): Name of the new related collection.
1051 description (str): Description of the new related collection.
1052 user_id (UUID, optional): The ID of the user adding the collection.
1053 relationship_type (str): Type of the relationship.
1054 meta (Dict[str, Any]): Meta of the new related collection.
1056 Returns:
1057 schema.NendoCollection: The newly added NendoCollection object.
1058 """
1059 self.nendo_instance.library.add_related_collection(
1060 track_ids=track_ids,
1061 collection_id=self.id,
1062 name=name,
1063 description=description,
1064 user_id=user_id,
1065 relationship_type=relationship_type,
1066 meta=meta,
1067 )
1068 return self
1070 def set_meta(self, meta: Dict[str, Any]) -> NendoCollection:
1071 """Set metadata of collection.
1073 Args:
1074 meta (Dict[str, Any]): Dictionary containing the metadata to be set.
1076 Returns:
1077 NendoCollection: The collection itself.
1078 """
1079 try:
1080 self.meta.update(meta)
1081 self.nendo_instance.library.update_collection(collection=self)
1082 except NendoError as e:
1083 logger.exception("Error updating meta: %s", e)
1084 return self
1086 def get_related_collections(
1087 self,
1088 user_id: Optional[Union[str, uuid.UUID]] = None,
1089 order_by: Optional[str] = None,
1090 order: Optional[str] = "asc",
1091 limit: Optional[int] = None,
1092 offset: Optional[int] = None,
1093 ) -> List[NendoCollection]:
1094 """Get all collections to which the current collection has a relationship.
1096 Args:
1097 user_id (Union[str, UUID], optional): The user ID to filter for.
1098 order_by (Optional[str]): Key used for ordering the results.
1099 order (Optional[str]): Order in which to retrieve results ("asc" or "desc").
1100 limit (Optional[int]): Limit the number of returned results.
1101 offset (Optional[int]): Offset into the paginated results (requires limit).
1103 Returns:
1104 List[NendoCollection]: List containting all related NendoCollections
1105 """
1106 return self.nendo_instance.library.get_related_collections(
1107 track_id=self.id,
1108 user_id=user_id,
1109 order_by=order_by,
1110 order=order,
1111 limit=limit,
1112 offset=offset,
1113 )
1115 def has_meta(self, key: str) -> bool:
1116 """Check if a given collection has the given key in its meta dict.
1118 Args:
1119 key (str): The key to check for.
1121 Returns:
1122 bool: True if the key exists, False otherwise.
1123 """
1124 return any(k == key for k in self.meta)
1126 def get_meta(self, key: str) -> Dict[str, Any]:
1127 """Get the meta entry for the given key.
1129 Args:
1130 key (str): The key to get metadata for.
1132 Returns:
1133 Dict[str, Any]: Meta entry for given key.
1134 """
1135 if not self.has_meta(key):
1136 logger.error("Key not found in meta: %s", key)
1137 return None
1138 return self.meta[key]
1140 def remove_meta(self, key: str) -> NendoCollection:
1141 """Remove the meta entry for the given key.
1143 Args:
1144 key (str): The key to remove metadata for.
1146 Returns:
1147 NendoCollection: The track itself.
1148 """
1149 if not self.has_meta(key):
1150 logger.error("Key not found in meta: %s", key)
1151 return None
1152 _ = self.meta.pop(key, None)
1153 self.nendo_instance.library.update_collection(collection=self)
1154 return self
1156 def export(
1157 self,
1158 export_path: str,
1159 filename_suffix: str = "nendo",
1160 file_format: str = "wav",
1161 ) -> NendoCollection:
1162 """Export the collection to a directory.
1164 Args:
1165 export_path (str): Path to a directory into which the collection's tracks
1166 should be exported.
1167 filename_suffix (str): The suffix which should be appended to each
1168 exported track's filename.
1169 file_format (str, optional): Format of the exported track. Ignored if
1170 file_path is a full file path. Defaults to "wav".
1172 Returns:
1173 NendoTrack: The track itself.
1174 """
1175 self.nendo_instance.library.export_collection(
1176 collection_id=self.id,
1177 export_path=export_path,
1178 filename_suffix=filename_suffix,
1179 file_format=file_format,
1180 )
1181 return self
1184class NendoCollectionCreate(NendoCollectionBase): # noqa: D101
1185 pass
1188class NendoPlugin(BaseModel, ABC):
1189 """Base class for all nendo plugins."""
1191 model_config = ConfigDict(
1192 arbitrary_types_allowed=True,
1193 )
1195 nendo_instance: Nendo
1196 config: NendoConfig
1197 logger: logging.Logger
1198 plugin_name: str
1199 plugin_version: str
1201 # --------------
1202 # Decorators
1204 @staticmethod
1205 def stream_output(func):
1206 """Decorator to turn on streaming mode for functions.
1208 The requirement for this decorator to work on a function is that it would
1209 normally return a list.
1210 """
1212 @functools.wraps(func)
1213 def wrapper(self, *args, **kwargs):
1214 result = func(self, *args, **kwargs)
1215 if self.config.stream_mode:
1216 return result
1217 # function is yielding single tracks if stream_chunk_size == 1
1218 elif self.config.stream_chunk_size > 1: # noqa: RET505
1219 return [track for chunk in result for track in chunk]
1220 else:
1221 return list(result)
1223 return wrapper
1225 @staticmethod
1226 def batch_process(func):
1227 """Decorator to run functions multithreaded in batches.
1229 This decorator function transforms the given function to run
1230 in multiple threads. It expects that the first argument to the function
1231 is a list of items, which will be processed in parallel,
1232 in batches of a given size.
1233 """
1235 @functools.wraps(func)
1236 def wrapper(self, track=None, file_paths=None, *args, **kwargs):
1237 target = track or file_paths
1238 if isinstance(target, NendoTrack):
1239 return func(self, track=target, **kwargs)
1240 elif isinstance(target, list): # noqa: RET505
1241 max_threads = self.config.max_threads
1242 batch_size = self.config.batch_size
1243 total = len(target)
1244 batches = [
1245 target[i : i + batch_size] for i in range(0, total, batch_size)
1246 ]
1247 start_time = time.time()
1248 futures = []
1250 def run_batch(batch_index, batch):
1251 try:
1252 batch_start_time = time.time()
1253 results = []
1254 if track:
1255 for _, item in enumerate(batch):
1256 result = func(
1257 self,
1258 track=item,
1259 *args, # noqa: B026
1260 **kwargs,
1261 )
1262 results.extend(result)
1263 elif file_paths:
1264 result = func(
1265 self,
1266 file_paths=batch,
1267 *args, # noqa: B026
1268 **kwargs,
1269 )
1270 results.extend(result)
1271 batch_end_time = time.time()
1272 batch_time = time.strftime(
1273 "%H:%M:%S",
1274 time.gmtime(batch_end_time - batch_start_time),
1275 )
1276 total_elapsed_time = batch_end_time - start_time
1277 average_time_per_batch = total_elapsed_time / (batch_index + 1)
1278 estimated_total_time = average_time_per_batch * len(batches)
1279 estimated_total_time_print = time.strftime(
1280 "%H:%M:%S",
1281 time.gmtime(estimated_total_time),
1282 )
1283 remaining_time = time.strftime(
1284 "%H:%M:%S",
1285 time.gmtime(estimated_total_time - total_elapsed_time),
1286 )
1287 logger.info(
1288 f"Finished batch {batch_index + 1}/{len(batches)}.\n"
1289 f"Time taken for this batch: {batch_time} - "
1290 f"Estimated total time: {estimated_total_time_print} - "
1291 f"Estimated remaining time: {remaining_time}\n",
1292 )
1293 return results
1294 except NendoError as e:
1295 logger.exception(
1296 "Error processing batch %d: %s",
1297 batch_index,
1298 e,
1299 )
1301 with ThreadPoolExecutor(max_workers=max_threads) as executor:
1302 for batch_index, batch in enumerate(batches):
1303 futures.append(executor.submit(run_batch, batch_index, batch))
1305 all_results = []
1306 for future in as_completed(futures):
1307 result = future.result()
1308 if result:
1309 all_results.extend(future.result())
1310 return all_results
1311 else:
1312 raise TypeError("Expected NendoTrack or list of NendoTracks")
1314 return wrapper
1316 # ------------------
1318 def _get_track_or_collection_from_args(
1319 self,
1320 **kwargs,
1321 ) -> Tuple[Union[NendoTrack, NendoCollection], Dict]:
1322 """Get the track or collection from the kwargs."""
1323 track = kwargs.pop("track", None)
1324 if track is not None:
1325 return track, kwargs
1327 collection = kwargs.pop("collection", None)
1328 if collection is not None:
1329 return collection, kwargs
1331 track_or_collection_id = kwargs.pop("id", None)
1332 return (
1333 self.nendo_instance.library.get_track_or_collection(track_or_collection_id),
1334 kwargs,
1335 )
1337 def _run_default_wrapped_method(
1338 self,
1339 **kwargs: Any,
1340 ) -> Optional[Union[NendoTrack, NendoCollection]]:
1341 """Check if the plugin has a wrapped method and run it if it exists.
1343 If the plugin has more than one wrapped method, a warning is logged and
1344 None is returned.
1346 Returns:
1347 Optional[Union[NendoTrack, NendoCollection]]: The track or collection
1348 returned by the wrapped method.
1349 """
1350 # get the wrapped functions and ignore any methods of pydantic.BaseModel
1351 wrapped_methods = get_wrapped_methods(self)
1353 if len(wrapped_methods) > 1:
1354 warning_module = (
1355 inspect.getmodule(type(self))
1356 .__name__.split(".")[0]
1357 .replace("nendo_plugin_", "")
1358 )
1359 warning_methods = [
1360 f"nendo.plugins.{warning_module}.{m.__name__}()"
1361 for m in wrapped_methods
1362 ]
1363 self.logger.warning(
1364 f" Warning: Multiple wrapped methods found in plugin. Please call the plugin via one of the following methods: {', '.join(warning_methods)}.",
1365 )
1366 return None
1368 run_func = wrapped_methods[0]
1369 return run_func(self, **kwargs)
1371 @property
1372 def plugin_type(self) -> str:
1373 """Return type of plugin."""
1374 return "NendoPlugin"
1376 def __str__(self):
1377 return f"{self.plugin_type} | name: {self.name} | version: {self.version}"
1379 # batching is deactivated for now
1380 # @NendoPlugin.batch_process
1381 def __call__(self, **kwargs: Any) -> Optional[Union[NendoTrack, NendoCollection]]:
1382 """Call the plugin.
1384 Runs a registered run function of a plugin on a track, a collection, or a signal.
1385 If the plugin has more than one run function, a warning is raised and all the possible options are listed.
1387 Args:
1388 **kwargs: Arbitrary keyword arguments.
1390 Returns:
1391 Optional[Union[NendoTrack, NendoCollection]]: The track or collection.
1392 """
1393 return self._run_default_wrapped_method(**kwargs)
1396class NendoBlobBase(BaseModel): # noqa: D101
1397 model_config = ConfigDict(
1398 arbitrary_types_allowed=True,
1399 from_attributes=True,
1400 use_enum_values=True,
1401 )
1403 user_id: uuid.UUID
1404 resource: NendoResource
1405 visibility: Visibility = Visibility.private
1408class NendoBlobCreate(NendoBlobBase): # noqa: D101
1409 pass
1412class NendoBlob(NendoBlobBase):
1413 """Class representing a blob object in Nendo.
1415 Used to store binary data, e.g. large numpy matrices in
1416 the nendo library. Not used for storing waveforms,
1417 as they are considered native `NendoResource`s.
1418 """
1420 id: uuid.UUID = Field(default_factory=uuid.uuid4)
1421 created_at: datetime = Field(default_factory=datetime.now)
1422 updated_at: datetime = Field(default_factory=datetime.now)
1423 data: Optional[bytes] = None
1425 class Config: # noqa: D106
1426 from_attributes = True
1429class NendoStorage(BaseModel, ABC):
1430 """Basic class representing a Nendo storage driver."""
1432 @abstractmethod
1433 def init_storage_for_user(self, user_id: str) -> Any:
1434 """Initializes the storage location for the given user.
1436 Args:
1437 user_id (str): ID of the user for whom the storage is to be initialized.
1439 Returns:
1440 Any: Storage object
1441 """
1442 raise NotImplementedError
1444 @abstractmethod
1445 def generate_filename(self, filetype: str, user_id: str) -> str:
1446 """Generate a collision-free new filename.
1448 Args:
1449 filetype (str): The filetype to append (without the dot).
1450 user_id (str): The ID of the user requesting the file.
1452 Returns:
1453 str: The generated filename.
1454 """
1455 raise NotImplementedError
1457 @abstractmethod
1458 def file_exists(self, file_name: str, user_id: str) -> bool:
1459 """Check if the given file_name exists in the storage.
1461 Args:
1462 file_name (str): Name of the file for which to check existence
1463 user_id (str): User ID
1465 Returns:
1466 bool: True if it exists, false otherwise
1467 """
1468 raise NotImplementedError
1470 @abstractmethod
1471 def as_local(self, file_path: str, location: ResourceLocation, user_id: str) -> str:
1472 """Get a local handle to the file.
1474 Return a file given as `file_path` as a locally accessible file,
1475 even if it originally resides in a remote storage location
1477 Args:
1478 file_path (str): Name of the file to obtain a local copy of.
1479 location (ResourceLocation): Location of the file, as provided by the
1480 corresponding NendoResource.
1481 user_id (str): User ID
1483 Returns:
1484 str: Local path to the original file (if it was local) or to a
1485 temporary copy (if it was remote)
1486 """
1487 raise NotImplementedError
1489 @abstractmethod
1490 def save_file(self, file_name: str, file_path: str, user_id: str) -> str:
1491 """Saves the file with the given file_name to the path specified by file_path.
1493 Args:
1494 file_name (str): The name of the target file.
1495 file_path (str): The path to the target file.
1496 user_id (str): The ID of the user requesting the file.
1498 Returns:
1499 str: The full path to the saved file resource.
1500 """
1501 raise NotImplementedError
1503 @abstractmethod
1504 def save_signal(
1505 self,
1506 file_name: str,
1507 signal: np.ndarray,
1508 sr: int,
1509 user_id: str,
1510 ) -> str:
1511 """Save a signal given as a numpy array to storage.
1513 Args:
1514 file_name (str): Name of the target file to save the signal to.
1515 signal (np.ndarray): The signal to write as a numpy array.
1516 sr (int): The sample rate of the signal.
1517 user_id (str): The ID of the user writing the file.
1519 Returns:
1520 str: The full path to the saved file resource.
1521 """
1522 raise NotImplementedError
1524 @abstractmethod
1525 def save_bytes(self, file_name: str, data: bytes, user_id: str) -> str:
1526 """Save a data given as bytes to the FS by pickling them first.
1528 Args:
1529 file_name (str): Name of the target file to save the data to.
1530 data (np.ndarray): The data to write to file in pickled form.
1531 user_id (str): The ID of the user writing the file.
1533 Returns:
1534 str: The full path to the saved file resource.
1535 """
1536 raise NotImplementedError
1538 @abstractmethod
1539 def remove_file(self, file_name: str, user_id: str) -> bool:
1540 """Remove the file given by file_name and user_id from the storage.
1542 Args:
1543 file_name (str): Name of the file to remove.
1544 user_id (str): ID of the user requesting the removal.
1546 Returns:
1547 bool: True if deletion was successful, False otherwise.
1548 """
1549 raise NotImplementedError
1551 @abstractmethod
1552 def get_file_path(self, src: str, user_id: str) -> str:
1553 """Returns the path of a resource.
1555 Args:
1556 src (str): The full path to the resource.
1557 user_id (str): The ID of the user requesting the path.
1559 Returns:
1560 str: The resource path, minus the file name.
1561 """
1562 raise NotImplementedError
1564 @abstractmethod
1565 def get_file_name(self, src: str, user_id: str) -> str:
1566 """Returns the filename of a resource.
1568 Args:
1569 src (str): The full path to the resource.
1570 user_id (str): The ID of the user requesting the path.
1572 Returns:
1573 str: The resource file name, minus the path.
1574 """
1575 raise NotImplementedError
1577 @abstractmethod
1578 def get_file(self, file_name: str, user_id: str) -> str:
1579 """Obtains the full path to the file by the name of file_name from storage.
1581 Args:
1582 file_name (str): The name of the target file.
1583 user_id (str): The ID of the user requesting the file.
1585 Returns:
1586 str: The full path to the file.
1587 """
1588 raise NotImplementedError
1590 @abstractmethod
1591 def list_files(self, user_id: str) -> List[str]:
1592 """Lists all files found in the user's library.
1594 Args:
1595 user_id (str): The ID of the user.
1597 Returns:
1598 List[str]: List of paths to files.
1599 """
1600 raise NotImplementedError
1602 @abstractmethod
1603 def get_bytes(self, file_name: str, user_id: str) -> Any:
1604 """Load the data bytes from the storage.
1606 Loading includes unpickling the file given as `file_name`.
1608 Args:
1609 file_name (str): Name of the target file to load.
1610 user_id (str): The ID of the user writing the file.
1612 Returns:
1613 The deserialized data bytes
1614 """
1615 raise NotImplementedError
1617 @abstractmethod
1618 def get_checksum(self, file_name: str, user_id: str) -> str:
1619 """Compute the checksum for the given file and user_id.
1621 Args:
1622 file_name (str): The name of the file in the library for which
1623 to compute the checksum.
1624 user_id (str): The ID of the user requesting the checksum.
1626 Returns:
1627 str: The checksum of the target file.
1628 """
1629 raise NotImplementedError
1631 @abstractmethod
1632 def get_driver_location(self) -> str:
1633 """Get the default resource location of the storage driver.
1635 e.g. "original", "local", "gcs", "s3", etc.
1637 Returns:
1638 str: Location type.
1639 """
1642class NendoStorageLocalFS(NendoStorage):
1643 """Implementation of the base storage driver for local filesystem access."""
1645 library_path: str = None
1647 def __init__( # noqa: D107
1648 self,
1649 library_path: str,
1650 user_id: str,
1651 **kwargs: Any,
1652 ):
1653 super().__init__(**kwargs)
1654 self.library_path = os.path.join(library_path, user_id)
1655 self.init_storage_for_user(user_id=user_id)
1657 def init_storage_for_user(self, user_id: str) -> str: # noqa: ARG002
1658 """Initialize local storage for user."""
1659 if not os.path.isdir(self.library_path):
1660 logger.info(
1661 f"Library path {self.library_path} does not exist, creating now.",
1662 )
1663 os.makedirs(self.library_path)
1664 return self.library_path
1666 def generate_filename(self, filetype: str, user_id: str) -> str: # noqa: ARG002
1667 """Generate a unique filename."""
1668 return f"{uuid.uuid4()!s}.{filetype}"
1670 def file_exists(self, file_name: str, user_id: str) -> bool:
1671 """Check if the given file exists."""
1672 return os.path.isfile(self.get_file(file_name, user_id))
1674 def as_local(self, file_path: str, location: ResourceLocation, user_id: str) -> str:
1675 """Get a local handle to the file."""
1676 if location == self.get_driver_location():
1677 return self.get_file(os.path.basename(file_path), user_id)
1678 return file_path
1680 def save_file(self, file_name: str, file_path: str, user_id: str) -> str:
1681 """Copy the source file given by file_path to the library."""
1682 target_file = self.get_file(file_name=file_name, user_id=user_id)
1683 shutil.copy2(file_path, target_file)
1684 return target_file
1686 def save_signal(
1687 self,
1688 file_name: str,
1689 signal: np.ndarray,
1690 sr: int,
1691 user_id: str, # noqa: ARG002
1692 ) -> str:
1693 """Save the given signal to storage."""
1694 target_file_path = self.get_file(file_name=file_name, user_id="")
1695 sf.write(target_file_path, signal, sr, subtype="PCM_16")
1696 return target_file_path
1698 def save_bytes(
1699 self,
1700 file_name: str,
1701 data: bytes,
1702 user_id: str, # noqa: ARG002
1703 ) -> str:
1704 """Save the given bytes to storage."""
1705 target_file_path = self.get_file(file_name=file_name, user_id="")
1706 with open(target_file_path, "wb") as target_file:
1707 pickle.dump(data, target_file)
1708 return target_file_path
1710 def remove_file(self, file_name: str, user_id: str) -> bool:
1711 """Remove the given file."""
1712 target_file = self.get_file(file_name=file_name, user_id=user_id)
1713 try:
1714 os.remove(target_file)
1715 return True
1716 except OSError as e:
1717 logger.error("Removing file %s failed: %s", target_file, e)
1718 return False
1720 def get_file_path(self, src: str, user_id: str) -> str: # noqa: ARG002
1721 """Get the path to the file (without the file name)."""
1722 return os.path.dirname(src)
1724 def get_file_name(self, src: str, user_id: str) -> str: # noqa: ARG002
1725 """Get the file name (without the path)."""
1726 return os.path.basename(src)
1728 def get_file(self, file_name: str, user_id: str) -> str: # noqa: ARG002
1729 """Get the full path to the file."""
1730 return os.path.join(self.library_path, file_name)
1732 def list_files(self, user_id: str) -> List[str]: # noqa: ARG002
1733 """List all files contained in the storage."""
1734 with os.scandir(self.library_path) as entries:
1735 return [entry.name for entry in entries if entry.is_file()]
1737 def get_bytes(self, file_name: str, user_id: str) -> Any:
1738 """Get bytes from a stored file by unpickling it."""
1739 target_file_path = self.get_file(file_name=file_name, user_id=user_id)
1740 with open(target_file_path, "rb") as target_file:
1741 return pickle.loads(target_file.read()) # noqa: S301
1743 def get_checksum(self, file_name: str, user_id: str) -> str:
1744 """Compute the MD5 checksum of the given file."""
1745 return md5sum(self.get_file(file_name=file_name, user_id=user_id))
1747 def get_driver_location(self) -> ResourceLocation:
1748 """Get the default resource location of the storage driver."""
1749 return ResourceLocation.local
1752class RegisteredNendoPlugin(BaseModel):
1753 """A registered `NendoPlugin`.
1755 Used by the `NendoPluginRegistry` to manage `NendoPlugins`.
1756 """
1758 name: str
1759 version: str = "n/a"
1760 plugin_instance: NendoPlugin
1762 def __getattr__(self, func_name: str):
1763 try:
1764 attr = getattr(self.plugin_instance, func_name)
1765 except AttributeError:
1766 raise NendoPluginRuntimeError(
1767 f"Plugin {self.name} has no function {func_name}",
1768 ) from None
1769 if not callable(attr):
1770 return attr
1772 def method(*args, **kwargs):
1773 return attr(*args, **kwargs)
1775 return method
1777 def __call__(self, **kwargs: Any) -> Any: # noqa: D102
1778 return self.plugin_instance(**kwargs)
1780 def __str__(self):
1781 return f"Nendo Library Plugin | name: {self.name} | version: {self.version}"
1784RegisteredNendoPlugin.model_rebuild()
1787class NendoPluginRegistry:
1788 """Class for registering and managing of nendo plugins."""
1790 _plugins: ClassVar[Dict[str, RegisteredNendoPlugin]] = {}
1792 def __getattr__(self, plugin_name: str):
1793 if plugin_name in self._plugins:
1794 return self._plugins[plugin_name]
1795 if f"nendo_plugin_{plugin_name}" in self._plugins:
1796 return self._plugins[f"nendo_plugin_{plugin_name}"]
1797 raise AttributeError(f"Plugin '{plugin_name}' not found")
1799 def __str__(self):
1800 output = ""
1801 if not self._plugins:
1802 return "No plugins registered"
1803 output = f"{len(self._plugins)} registered plugins:"
1804 for k, v in self._plugins.items():
1805 output += "\n"
1806 output += f"{k} - {v.name} ({v.version})"
1807 return output
1809 def __call__(self):
1810 """Return string representation upon direct access to the registered plugin."""
1811 return self.__str__()
1813 def all_names(self) -> List[str]:
1814 """Return all plugins that are registered as a list of their names.
1816 Returns:
1817 List[str]: List containing all plugin names.
1818 """
1819 return [k for k, v in self._plugins.items()]
1821 def add(
1822 self,
1823 plugin_name: str,
1824 version: str,
1825 plugin_instance: NendoPlugin,
1826 ) -> RegisteredNendoPlugin:
1827 """Add a Registered plugin to the plugin registry.
1829 Args:
1830 plugin_name (str): Name of the plugin.
1831 version (str): Version of the plugin.
1832 plugin_instance (schema.NendoPlugin): Instantiated plugin class.
1834 Returns:
1835 RegisteredNendoPlugin: The registered nendo plugin.
1836 """
1837 self._plugins[plugin_name] = RegisteredNendoPlugin(
1838 name=plugin_name,
1839 version=version,
1840 plugin_instance=plugin_instance,
1841 )
1843 def remove(self, plugin_name: str) -> None:
1844 """Remove a plugin from the plugin registry.
1846 Args:
1847 plugin_name (str): Name of the plugin to remove.
1848 """
1849 del self._plugins[plugin_name]