Создать Kedro PartitionedDataset из PartitionedDatasets. ⇐ Python

[Anonymous]

Я работаю над проектом kedro, где я хочу автоматически маркировать тысячи аудиофайлов, применять к ним преобразования, а затем сохранять их в папке из папок, каждая подпапка соответствует одной метке. Я хочу, чтобы эта папка папок была записью каталога в моем файле yml.
Я следовал этому руководству Kedro и создал свой собственный набор данных для сохранения/загрузки файлов .wav в каталоге kedro. Я также могу создавать записи каталога PartitionedDataset в файлеcatalog.yml, например
audio_folder:
type: partitions.PartitionedDataset
dataset: my_kedro_project.datasets.audio_dataset.SoundDataset
path: data/output/audios/
filename_suffix: ".WAV"

для сохранения/загрузки папок с файлами .WAV в каталоге Kedro.
Следующий уровень абстракции, который мне потребуется, — это возможность создать запись каталога, соответствующая папке, содержащей папки, такие как audio_folder выше. Я хочу делать это не путем динамического создания записей каталога, а путем расширения класса PartitionedDataset. Это потому, что я хочу, чтобы папка папок была частью моего каталога.yml. Мои вопросы:

• Возможно ли это? Кто-нибудь из вас когда-нибудь пробовал что-то подобное?
• Если это возможно, мой собственный класс должен включать только _load, _save и _describe, как при настройке AbstractDataset?

EDIT
Наконец-то я решил создать еще один собственный класс, расширяющий AbstractDataset. Вот некоторые подробности о методах _load и _save:
def _load(self):
subfolder_names=[ subfolder_name
for subfolder_name in os.listdir(self._mainfolderpath)
if os.path.isdir(os.path.join(self._mainfolderpath, subfolder_name))
]

wav_paths_dict={}
for subfolder_name in subfolder_names:
subfolder_path=os.path.join(self._mainfolderpath, subfolder_name)
wav_files=[]
for root, dirs, files in os.walk(subfolder_path):
for file in files:
if file.lower().endswith('.wav'):
wav_file_path=os.path.join(root, file)
wav_file_name=os.path.split(wav_file_path)[-1].replace('.wav','').replace('.WAV','')
wav_files.append((wav_file_name,wav_file_path))
wav_paths_dict[subfolder_name]=dict(wav_files)

partitioned_dataset_dict={}
for subfolder_name, sub_dict in wav_paths_dict.items():
partitioned_dataset=[(wav_file_name,SoundDataset(wav_file_path).load()) for wav_file_name,wav_file_path in sub_dict.items()]
partitioned_dataset_dict[subfolder_name]=dict(partitioned_dataset)

return partitioned_dataset_dict

И
def _save(self, subfolders_dictionary):
if os.path.isdir(self._mainfolderpath):
for root, dirs, files in os.walk(self._mainfolderpath,topdown=False):
for name in files:
os.remove(os.path.join(root, name))
for name in dirs:
os.rmdir(os.path.join(root, name))
os.rmdir(self._mainfolderpath)
os.mkdir(self._mainfolderpath)
for subfolder_name in subfolders_dictionary.keys():
subfolder_path=os.path.join(self._mainfolderpath, subfolder_name)
os.mkdir(os.path.normpath(subfolder_path))

#print(subfolder_name, subfolder_path)
partitioned_dataset = PartitionedDataset(
path=subfolder_path,
dataset=SoundDataset,
filename_suffix=".WAV",
)

partitioned_dataset.save(subfolders_dictionary[subfolder_name])


Подробнее здесь: https://stackoverflow.com/questions/790 ... eddatasets