Cat Feeder  1.0.0
The Cat feeder project
Loading...
Searching...
No Matches
ff_downloader.py
Go to the documentation of this file.
1r"""
2# +==== BEGIN CatFeeder =================+
3# LOGO:
4# ..............(..../\
5# ...............)..(.')
6# ..............(../..)
7# ...............\‍(__)|
8# Inspired by Joan Stark
9# source https://www.asciiart.eu/
10# animals/cats
11# /STOP
12# PROJECT: CatFeeder
13# FILE: downloader.py
14# CREATION DATE: 11-10-2025
15# LAST Modified: 1:59:55 17-01-2026
16# DESCRIPTION:
17# This is the backend server in charge of making the actual website work.
18#
19# This file raises:
20# ArchitectureNotSupported: This means that the architecure you are running this file on is unknown to it.
21# PackageNotSupported: This class informs the user that the system architecture they are using is not supported by this script
22# PackageNotInstalled: This class informs the user that the ff dependency(ies) they required does not exist.
23# NotImplementedError: This class inform the user that a required operation is not implemented and thus not completable.
24# ValueError: This is used for any error that does not fit into any specific class.
25# /STOP
26# COPYRIGHT: (c) Cat Feeder
27# PURPOSE: File in charge of downloading and extracting the FFmpeg binaries for the current system.
28# // AR
29# +==== END CatFeeder =================+
30"""
31import os
32import sys
33import math
34import struct
35import shutil
36import zipfile
37import tarfile
38import platform
39from typing import Union, Callable, Optional
40from pathlib import Path as PPath
41import importlib.util
42from pathlib import Path
43
44
45# Ensure a missing `audioop` C-extension can be satisfied by a repository
46# shim `pyaudioop.py` (keeps containers/venvs untouched). This tries the
47# stdlib first, then looks for `pyaudioop.py` up the filesystem from this
48# module and injects it into `sys.modules` as both `pyaudioop` and
49# `audioop` so downstream imports (e.g. `pydub`) succeed.
50try:
51 import audioop # type: ignore
52except Exception:
53 import sys
54 shim_loaded = False
55 here = Path(__file__).resolve()
56 for parent in here.parents:
57 candidate = parent / "pyaudioop.py"
58 if candidate.exists():
59 spec = importlib.util.spec_from_file_location(
60 "pyaudioop", str(candidate))
61 if spec and spec.loader:
62 mod = importlib.util.module_from_spec(spec)
63 spec.loader.exec_module(mod) # type: ignore[attr-defined]
64 sys.modules.setdefault("pyaudioop", mod)
65 sys.modules.setdefault("audioop", mod)
66 shim_loaded = True
67 break
68 if not shim_loaded:
69 # fall back to attempting to import a top-level `pyaudioop` package
70 try:
71 import backend.src.libs.fffamily.pyaudioop as _pa # type: ignore
72 sys.modules.setdefault("audioop", _pa)
73 except Exception:
74 # leave the original ImportError to be raised later when pydub is imported
75 pass
76
77from pydub import AudioSegment, playback
78import requests
79from .ff_exceptions import ArchitectureNotSupported, PackageNotInstalled, PackageNotSupported
80from . import ff_constants as CONST
81from ..core import FinalClass
82
83
84class FFMPEGDownloader(metaclass=FinalClass):
85 """
86 The class in charge of downloading and extracting the FFmpeg binaries for the current system.
87 This class is there so that ffmpeg (and it's familly binaries) ca be run without requiring the user to install it.
88
89 Raises:
90 ArchitectureNotSupported: This means that the architecure you are running this file on is unknown to it.
91 PackageNotSupported: This class informs the user that the system architecture they are using is not supported by this script
92 PackageNotInstalled: This class informs the user that the ff dependency(ies) they required does not exist.
93 NotImplementedError: This class inform the user that a required operation is not implemented and thus not completable.
94 ValueError: This is used for any error that does not fit into any specific class.
95 RuntimeError: Any error that was unexpected and caused the program to stop.
96 """
97
98 available_binaries: list = [CONST.FFMPEG_KEY,
99 CONST.FFPROBE_KEY, CONST.FFPLAY_KEY]
100
101 _instance: Optional["FFMPEGDownloader"] = None
102
103 def __new__(cls, *args, **kwargs) -> "FFMPEGDownloader":
104 if cls._instance is None:
105 cls._instance = super(FFMPEGDownloader, cls).__new__(cls)
106 return cls._instance
107
108 def __init__(self, cwd: str = os.getcwd(), query_timeout: int = 10, success: int = 0, error: int = 84, debug: bool = False):
109 self.cwd: str = cwd
110 self.error: int = error
111 self.debug: bool = debug
112 self.success: int = success
113 self.file_url: Union[str, None] = None
114 self.file_path: Union[str, None] = None
115 self.fold_path: Union[str, None] = None
116 self.query_timeout: int = query_timeout
117 self.new_folder_path: Union[str, None] = None
118 self.extracted_folder: Union[str, None] = None
119 self.system: str = self.get_system_name()
120 self.architecture: str = self.get_platform()
121 self.available_binaries: list = FFMPEGDownloader.available_binaries
122
123 @staticmethod
124 def process_file_path(*args: Union[str, PPath], cwd: Union[str, PPath, None] = None) -> str:
125 """_summary_
126 This is a rebind function of the process_file_path defined globaly
127 """
128 return CONST.process_file_path(
129 *args, cwd=cwd
130 )
131
132 @staticmethod
133 def generate_audio_sample(tone: int) -> AudioSegment:
134 """
135 Generates a pure tone audio sample using the pydub library.
136
137 Args:
138 tone (int): Frequency of the tone in Hz.
139
140 Returns:
141 AudioSegment: The generated audio segment.
142 """
143 print(f"Generating audio sample for tone {tone} Hz")
144 sample_rate = 44100 # Hz
145 frequency = tone # Hz
146 duration = 3.0 # seconds
147
148 # Generate samples manually
149 num_samples = int(sample_rate * duration)
150 samples = []
151 for i in range(num_samples):
152 t = i / sample_rate
153 # Generate a sine wave
154 sample_value = 0.5 * math.sin(2 * math.pi * frequency * t)
155 # Convert to 16-bit integer
156 waveform_integer = int(sample_value * 32767)
157 samples.append(waveform_integer)
158
159 # Pack the waveform integers into a byte array
160 raw_audio_data = struct.pack("<" + "h" * len(samples), *samples)
161
162 # Create the AudioSegment
163 audio_segment = AudioSegment(
164 raw_audio_data,
165 frame_rate=sample_rate,
166 sample_width=2, # 16-bit audio is 2 bytes
167 channels=1 # Mono audio
168 )
169 print("Audio sample generated")
170 return audio_segment
171
172 @staticmethod
173 def play_audio_sample(audio_segment: AudioSegment) -> None:
174 """
175 Plays an audio sample using the pydub library.
176
177 Args:
178 audio_segment (AudioSegment): The audio segment to play.
179 """
180 print("Playing audio sample")
181 playback.play(audio_segment)
182 print("Audio sample played")
183
184 @staticmethod
185 def save_audio_sample(audio_segment: AudioSegment, file_path: str) -> None:
186 """
187 Saves an audio sample to a file using the pydub library.
188
189 Args:
190 audio_segment (AudioSegment): _description_
191 file_path (str): _description_
192
193 Returns:
194 _type_: _description_
195 """
196 print(f"Saving audio sample to {file_path}")
197 audio_segment.export(file_path, format="wav")
198 print(f"Audio sample saved to {file_path}")
199
200 @staticmethod
201 def _extract_package(file_path: str, destination: str) -> None:
202 """_summary_
203
204 Args:
205 file_path (_type_): _description_
206 destination (_type_): _description_
207
208 Raises:
209 NotImplementedError: _description_
210 ValueError: _description_
211 """
212 print(f"Extracting {file_path} to {destination}")
213 if file_path.endswith(".zip"):
214 with zipfile.ZipFile(file_path, 'r') as zip_ref:
215 zip_ref.extractall(destination)
216 elif file_path.endswith(".tar.xz"):
217 with tarfile.open(file_path, 'r:xz') as tar_ref:
218 tar_ref.extractall(destination)
219 elif file_path.endswith(".7z"):
220 # Handling 7z files
221 raise NotImplementedError("7z extraction not implemented")
222 else:
223 raise ValueError("Unsupported file format")
224 print("Extraction complete")
225
226 @staticmethod
227 def get_system_name() -> str:
228 """_summary_
229
230 Raises:
231 ArchitectureNotSupported: _description_
232 PackageNotSupported: _description_
233 PackageNotInstalled: _description_
234 PackageNotSupported: _description_
235 PackageNotInstalled: _description_
236
237 Returns:
238 str: _description_
239 """
240 return platform.system().lower()
241
242 @staticmethod
243 def get_platform() -> str:
244 """_summary_
245
246 Raises:
247 NotImplementedError: _description_
248 ValueError: _description_
249
250 Returns:
251 str: _description_
252 """
253 return platform.architecture()[0]
254
255 @staticmethod
256 def _create_path_if_not_exists(path: str) -> None:
257 """_summary_
258
259 Args:
260 path (_type_): _description_
261
262 Raises:
263 NotImplementedError: _description_
264 ValueError: _description_
265 """
266 if not os.path.exists(path):
267 print(f"Creating directory: {path}")
268 os.makedirs(path, exist_ok=True)
269
270 @staticmethod
271 def _download_file(file_url: str, file_path: str, query_timeout: int = 10) -> None:
272 """_summary_
273
274 Args:
275 file_url (_type_): _description_
276 file_path (_type_): _description_
277
278 Raises:
279 PackageNotInstalled: _description_
280 """
281 print(f"Downloading FFmpeg from {file_url}")
282 response_data = requests.get(file_url, timeout=query_timeout)
283 if response_data.status_code != 200:
284 raise PackageNotInstalled("Could not download the package")
285 print(f"Saving to {file_path}")
286 with open(file_path, "wb") as file_descriptor:
287 file_descriptor.write(response_data.content)
288 print("Download complete")
289
290 @staticmethod
291 def _grant_executable_rights(file_path: Union[str, None] = None) -> None:
292 """_summary_
293
294 Args:
295 file_path (str, optional): _description_. Defaults to None.
296 """
297 if file_path is None:
298 return
299 print(f"Giving executable rights to {file_path}")
300 if os.path.exists(file_path):
301 os.chmod(file_path, 0o755)
302 print(f"Executable rights granted to {file_path}")
303 else:
304 print(f"{file_path} does not exist, could not grant executable rights")
305
306 @staticmethod
307 def _rename_extracted_folder(old_name: str, new_name: str) -> None:
308 """_summary_
309
310 Args:
311 old_name (str): _description_
312 new_name (str): _description_
313 """
314
315 if os.path.exists(old_name):
316 print(f"Renaming {old_name} to {new_name}")
317 shutil.move(old_name, new_name)
318
319 @staticmethod
320 def get_ff_family_path(download_if_not_present: bool = True, cwd: str = os.getcwd(), query_timeout: int = 10, success: int = 0, error: int = 1, debug: bool = False) -> str:
321 """_summary_
322 The general path for ff related libraries
323
324 Args:
325 download_if_not_present (bool, optional): _description_. Defaults to True.
326 cwd (str, optional): _description_. Defaults to os.getcwd().
327 query_timeout (int, optional): _description_. Defaults to 10.
328 success (int, optional): _description_. Defaults to 0.
329 error (int, optional): _description_. Defaults to 1.
330 debug (bool, optional): _description_. Defaults to False.
331
332 Raises:
333 PackageNotSupported: _description_
334 PackageNotSupported: _description_
335 PackageNotInstalled: _description_
336 ArchitectureNotSupported: _description_
337 PackageNotSupported: _description_
338 RuntimeError: _description_
339
340 Returns:
341 str: _description_
342 """
343 system = FFMPEGDownloader.get_system_name()
344 if system not in ("windows", "linux", "darwin"):
345 raise PackageNotSupported("Unsupported system")
346 precompiled_ffmpeg = os.path.join(cwd, "ffmpeg", system)
347 precompiled_ffprobe = os.path.join(cwd, "ffprobe", system)
348 precompiled_ffplay = os.path.join(cwd, "ffplay", system)
349 if not os.path.isdir(precompiled_ffmpeg) or not os.path.isdir(precompiled_ffprobe) or not os.path.isdir(precompiled_ffplay):
350 if not download_if_not_present:
351 raise PackageNotInstalled("FF_family not found")
352 print("FF_family not found in precompiled paths, setting up")
353 fdi = FFMPEGDownloader(
354 cwd=cwd,
355 query_timeout=query_timeout,
356 success=success,
357 error=error,
358 debug=debug
359 )
360 status = fdi.main()
361 if status != 0:
362 raise RuntimeError("FF_family could not be installed")
363 if os.path.exists(cwd) and os.path.isdir(cwd):
364 return cwd
365 raise PackageNotInstalled("FF_family not found")
366
367 @staticmethod
368 def get_ffmpeg_binary_path(download_if_not_present: bool = True, cwd: str = os.getcwd(), query_timeout: int = 10, success: int = 0, error: int = 1, debug: bool = False) -> str:
369 """_summary_
370
371 Args:
372 download_if_not_present (bool, optional): _description_. Defaults to True.
373 cwd (str, optional): _description_. Defaults to os.getcwd().
374 query_timeout (int, optional): _description_. Defaults to 10.
375 success (int, optional): _description_. Defaults to 0.
376 error (int, optional): _description_. Defaults to 1.
377 debug (bool, optional): _description_. Defaults to False.
378
379 Raises:
380 PackageNotSupported: _description_
381 PackageNotSupported: _description_
382 PackageNotInstalled: _description_
383
384 Returns:
385 str: _description_
386 """
387 system = FFMPEGDownloader.get_system_name()
388 ffmpeg_system_path = FFMPEGDownloader.get_ff_family_path(
389 download_if_not_present=download_if_not_present,
390 cwd=cwd,
391 query_timeout=query_timeout,
392 success=success,
393 error=error,
394 debug=debug
395 )
396 ffmpeg_precompiled_path = os.path.join(
397 ffmpeg_system_path, "ffmpeg", system
398 )
399 path = None
400 if system == "windows":
401 path = os.path.join(
402 ffmpeg_precompiled_path,
403 "ffmpeg.exe"
404 )
405 elif system == "linux":
406 path = os.path.join(
407 ffmpeg_precompiled_path,
408 "ffmpeg"
409 )
410 FFMPEGDownloader._grant_executable_rights(path)
411 elif system == "darwin":
412 path = os.path.join(
413 ffmpeg_precompiled_path,
414 "ffmpeg"
415 )
416 FFMPEGDownloader._grant_executable_rights(path)
417 else:
418 raise PackageNotSupported("Unsupported OS")
419 print(f"FFmpeg path = '{path}'")
420 if os.path.exists(path):
421 if os.path.isfile(path):
422 return path
423 raise PackageNotSupported("Path is not a file")
424 raise PackageNotInstalled("ffmpeg is not properly installed")
425
426 @staticmethod
427 def get_ffplay_binary_path(download_if_not_present: bool = True, cwd: str = os.getcwd(), query_timeout: int = 10, success: int = 0, error: int = 1, debug: bool = False) -> str:
428 """_summary_
429
430 Args:
431 download_if_not_present (bool, optional): _description_. Defaults to True.
432 cwd (str, optional): _description_. Defaults to os.getcwd().
433 query_timeout (int, optional): _description_. Defaults to 10.
434 success (int, optional): _description_. Defaults to 0.
435 error (int, optional): _description_. Defaults to 1.
436 debug (bool, optional): _description_. Defaults to False.
437
438 Raises:
439 PackageNotSupported: _description_
440 PackageNotSupported: _description_
441 PackageNotInstalled: _description_
442
443 Returns:
444 str: _description_
445 """
446 system = FFMPEGDownloader.get_system_name()
447 ffmpeg_system_path = FFMPEGDownloader.get_ff_family_path(
448 download_if_not_present=download_if_not_present,
449 cwd=cwd,
450 query_timeout=query_timeout,
451 success=success,
452 error=error,
453 debug=debug
454 )
455 ffmpeg_precompiled_path = os.path.join(
456 ffmpeg_system_path, "ffplay", system
457 )
458 path = None
459 if system == "windows":
460 path = os.path.join(
461 ffmpeg_precompiled_path,
462 "ffplay.exe"
463 )
464 elif system == "linux":
465 path = os.path.join(
466 ffmpeg_precompiled_path,
467 "ffplay"
468 )
469 FFMPEGDownloader._grant_executable_rights(path)
470 elif system == "darwin":
471 path = os.path.join(
472 ffmpeg_precompiled_path,
473 "ffplay"
474 )
475 FFMPEGDownloader._grant_executable_rights(path)
476 else:
477 raise PackageNotSupported("Unsupported OS")
478 print(f"FFplay path = '{path}'")
479 if os.path.exists(path):
480 if os.path.isfile(path):
481 return path
482 raise PackageNotSupported("Path is not a file")
483 raise PackageNotInstalled("ffplay is not properly installed")
484
485 @staticmethod
486 def get_ffprobe_binary_path(download_if_not_present: bool = True, cwd: str = os.getcwd(), query_timeout: int = 10, success: int = 0, error: int = 1, debug: bool = False) -> str:
487 """_summary_
488
489 Args:
490 download_if_not_present (bool, optional): _description_. Defaults to True.
491 cwd (str, optional): _description_. Defaults to os.getcwd().
492 query_timeout (int, optional): _description_. Defaults to 10.
493 success (int, optional): _description_. Defaults to 0.
494 error (int, optional): _description_. Defaults to 1.
495 debug (bool, optional): _description_. Defaults to False.
496
497 Raises:
498 PackageNotSupported: _description_
499 PackageNotSupported: _description_
500 PackageNotInstalled: _description_
501
502 Returns:
503 str: _description_
504 """
505 system = FFMPEGDownloader.get_system_name()
506 ffmpeg_system_path = FFMPEGDownloader.get_ff_family_path(
507 download_if_not_present=download_if_not_present,
508 cwd=cwd,
509 query_timeout=query_timeout,
510 success=success,
511 error=error,
512 debug=debug
513 )
514 ffmpeg_precompiled_path = os.path.join(
515 ffmpeg_system_path, "ffprobe", system
516 )
517 path = None
518 if system == "windows":
519 path = os.path.join(
520 ffmpeg_precompiled_path,
521 "ffprobe.exe"
522 )
523 elif system == "linux":
524 path = os.path.join(
525 ffmpeg_precompiled_path,
526 "ffprobe"
527 )
528 FFMPEGDownloader._grant_executable_rights(path)
529 elif system == "darwin":
530 path = os.path.join(
531 ffmpeg_precompiled_path,
532 "ffprobe"
533 )
534 FFMPEGDownloader._grant_executable_rights(path)
535 else:
536 raise PackageNotSupported("Unsupported OS")
537 print(f"FFprobe path = '{path}'")
538 if os.path.exists(path):
539 if os.path.isfile(path):
540 return path
541 raise PackageNotSupported("Path is not a file")
542 raise PackageNotInstalled("ffprobe is not properly installed")
543
544 @staticmethod
545 def add_ff_family_to_path(ffmpeg_path: Union[str, None] = None, ffplay_path: Union[str, None] = None, ffprobe_path: Union[str, None] = None, download_if_not_present: bool = True, cwd: str = os.getcwd(), query_timeout: int = 10, success: int = 0, error: int = 1, debug: bool = False) -> None:
546 """_summary_
547 Add the FF family to the system path.
548
549 Args:
550 ffmpeg_path (str, optional): _description_. Defaults to None.
551 ffplay_path (str, optional): _description_. Defaults to None.
552 ffprobe_path (str, optional): _description_. Defaults to None.
553 download_if_not_present (bool, optional): _description_. Defaults to True.
554 cwd (str, optional): _description_. Defaults to os.getcwd().
555 query_timeout (int, optional): _description_. Defaults to 10.
556 success (int, optional): _description_. Defaults to 0.
557 error (int, optional): _description_. Defaults to 1.
558 debug (bool, optional): _description_. Defaults to False.
559 """
560 if ffmpeg_path is None:
561 try:
562 print("Getting ffmpeg path")
563 ffmpeg_path = FFMPEGDownloader.get_ffmpeg_binary_path(
564 download_if_not_present=download_if_not_present,
565 cwd=cwd,
566 query_timeout=query_timeout,
567 success=success,
568 error=error,
569 debug=debug
570 )
571 except (PackageNotSupported, PackageNotSupported, PackageNotInstalled) as e:
572 print(f"Failed to expose ffmpeg, error: {e}")
573 if ffplay_path is None:
574 try:
575 print("Getting ffplay path")
576 ffplay_path = FFMPEGDownloader.get_ffplay_binary_path(
577 download_if_not_present=download_if_not_present,
578 cwd=cwd,
579 query_timeout=query_timeout,
580 success=success,
581 error=error,
582 debug=debug
583 )
584 except (PackageNotSupported, PackageNotSupported, PackageNotInstalled) as e:
585 print(f"Failed to expose ffplay, error: {e}")
586 if ffprobe_path is None:
587 try:
588 print("Getting ffprobe path")
589 ffprobe_path = FFMPEGDownloader.get_ffprobe_binary_path(
590 download_if_not_present=download_if_not_present,
591 cwd=cwd,
592 query_timeout=query_timeout,
593 success=success,
594 error=error,
595 debug=debug
596 )
597 except (PackageNotSupported, PackageNotSupported, PackageNotInstalled) as e:
598 print(f"Failed to expose ffprobe, error: {e}")
599 msg = "Converting the direct paths (which include the binaries at the end) into system paths (which only include the directories containing the binaries)"
600 print(msg)
601 if ffmpeg_path:
602 ffmpeg_path = os.path.abspath(os.path.dirname(ffmpeg_path))
603 if ffplay_path:
604 ffplay_path = os.path.abspath(os.path.dirname(ffplay_path))
605 if ffprobe_path:
606 ffprobe_path = os.path.abspath(os.path.dirname(ffprobe_path))
607 print("Adding FF family to PATH")
608 for ff_path in [ffmpeg_path, ffplay_path, ffprobe_path]:
609 if not ff_path:
610 print("Path not defined, skipping")
611 continue
612 if ff_path not in os.environ["PATH"]:
613 print(f"Adding {ff_path} to PATH")
614 os.environ["PATH"] = ff_path + os.pathsep + os.environ["PATH"]
615 print(f"Added {ff_path} to PATH")
616 else:
617 print(f"{ff_path} is already in PATH")
618 if ff_path not in sys.path:
619 print(f"Adding {ff_path} to sys.path")
620 sys.path.append(ff_path)
621 print(f"Added {ff_path} to sys.path")
622 else:
623 print(f"{ff_path} is already in sys.path")
624
625 def _clean_platform_name(self) -> None:
626 """_summary_
627
628 Raises:
629 NotImplementedError: _description_
630 ValueError: _description_
631 """
632 msg = "Current system (before filtering): "
633 msg += f"{self.system} {self.architecture}"
634 print(msg)
635
636 if "bit" in self.architecture:
637 self.architecture = self.architecture.replace("bit", "")
638 elif "x" in self.architecture:
639 self.architecture = self.architecture.replace("x", "")
640
641 msg = "Current system (after filtering): "
642 msg += f"{self.system} {self.architecture}"
643 print(msg)
644
645 def _get_correct_download_and_file_path(self, binary_name: str = CONST.FFMPEG_KEY) -> None:
646 """_summary_
647
648 Args:
649 binary_name (str, optional): _description_. Defaults to FFMPEG_KEY.
650
651 Raises:
652 ArchitectureNotSupported: _description_
653 PackageNotSupported: _description_
654 PackageNotInstalled: _description_
655 """
656 if binary_name not in CONST.BUNDLE_DOWNLOAD:
657 raise PackageNotSupported("Unknown binary choice" + binary_name)
658 if self.system in CONST.BUNDLE_DOWNLOAD[binary_name]:
659 if self.architecture in CONST.BUNDLE_DOWNLOAD[binary_name][self.system]:
660 tmp_path: Union[str, Callable] = CONST.BUNDLE_DOWNLOAD[binary_name][
661 self.system][self.architecture][CONST.FILE_PATH_TOKEN]
662 if callable(tmp_path):
663 self.file_path = str(tmp_path(cwd=self.cwd))
664 else:
665 self.file_path = tmp_path
666 self.file_url = CONST.BUNDLE_DOWNLOAD[binary_name][
667 self.system
668 ][self.architecture][CONST.FILE_URL_TOKEN]
669 else:
670 raise ArchitectureNotSupported("Unknown architecture")
671 else:
672 raise PackageNotSupported("Unknown system")
673
674 def _get_all_binaries(self) -> None:
675 """_summary_
676
677 Raises:
678 PackageNotSupported: _description_
679 PackageNotSupported: _description_
680 PackageNotInstalled: _description_
681 """
682 for binary in self.available_binaries:
683 print(f"Downloading {binary}")
685 if self.file_path:
686 self.fold_path = os.path.dirname(self.file_path)
687 else:
688 self.fold_path = os.path.dirname(__file__)
690 if self.file_path and os.path.exists(self.file_path):
691 print(f"{binary} already downloaded")
692 continue
693 if not self.file_url or not self.file_path:
694 print(f"{binary} contains corrupted data, skipping")
695 continue
696 self._download_file(
697 self.file_url,
698 self.file_path,
699 self.query_timeout
700 )
701
702 def _install_all_binaries(self) -> None:
703 """_summary_
704 This is the function that will install all the FF_family binaries if they are already downloaded.
705 """
706 for binary in self.available_binaries:
707 print(f"Installing {binary}")
709 if not self.file_path:
710 print(
711 f"{binary} filepath is empty, the binary meta data appears to be corrupted, skipping")
712 continue
713 self.fold_path = os.path.dirname(self.file_path)
714 extract_to = os.path.join(
715 self.cwd,
716 binary
717 )
718 final_name = os.path.join(
719 extract_to,
720 self.system
721 )
722 if os.path.isdir(final_name):
723 print(f"{binary} already installed")
724 continue
725 if binary == CONST.FFPLAY_KEY:
726 extract_to = os.path.join(
727 extract_to,
728 self.system
729 )
730 self._create_path_if_not_exists(extract_to)
731 self._extract_package(self.file_path, extract_to)
732 self.extracted_folder = os.listdir(extract_to)[0]
733 new_folder_path = os.path.join(
734 self.cwd,
735 binary,
736 self.system
737 )
738 old_folder_path = os.path.join(
739 extract_to,
741 )
742 if binary != CONST.FFPLAY_KEY:
744 old_folder_path,
745 new_folder_path
746 )
747 print(f"{binary} installed")
748
749 def main(self, audio_segment_node: Union[AudioSegment, None] = None) -> int:
750 """_summary_
751 The function in charge of downloading and extracting the FF_family binaries for the current system.
752
753 Raises:
754 PackageNotInstalled: _description_
755 PackageNotSupported: _description_
756 PackageNotInstalled: _description_
757
758 Returns:
759 int: _description_
760 """
761 try:
762 found_path = self.get_ff_family_path(download_if_not_present=False)
763 print(f"FF_family already installed at {found_path}")
764 ffmpeg_path = self.get_ffmpeg_binary_path(
765 download_if_not_present=False,
766 cwd=self.cwd,
767 query_timeout=self.query_timeout,
768 success=self.success,
769 error=self.error,
770 debug=self.debug
771 )
772 ffplay_path = self.get_ffplay_binary_path(
773 download_if_not_present=False,
774 cwd=self.cwd,
775 query_timeout=self.query_timeout,
776 success=self.success,
777 error=self.error,
778 debug=self.debug
779 )
780 ffprobe_path = self.get_ffprobe_binary_path(
781 download_if_not_present=False,
782 cwd=self.cwd,
783 query_timeout=self.query_timeout,
784 success=self.success,
785 error=self.error,
786 debug=self.debug
787 )
788 print("Updating pydub ffmpeg path")
789 if audio_segment_node is not None:
790 audio_segment_node.ffmpeg = ffmpeg_path
791 AudioSegment.ffmpeg = ffmpeg_path
793 ffmpeg_path, ffplay_path, ffprobe_path, download_if_not_present=False
794 )
795 print("FF_family already installed and ready to use!")
796 return self.success
797 except PackageNotInstalled:
798 print("FFmpeg not found. Installing...")
799 except PackageNotSupported as e:
800 raise RuntimeError(
801 "FFmpeg cannot be installed on this device because the system is unknown to this script."
802 ) from e
804 self._get_all_binaries()
806 ffmpeg_path = self.get_ffmpeg_binary_path(
807 download_if_not_present=False,
808 cwd=self.cwd,
809 query_timeout=self.query_timeout,
810 success=self.success,
811 error=self.error,
812 debug=self.debug
813 )
814 ffplay_path = self.get_ffplay_binary_path(
815 download_if_not_present=False,
816 cwd=self.cwd,
817 query_timeout=self.query_timeout,
818 success=self.success,
819 error=self.error,
820 debug=self.debug
821 )
822 ffprobe_path = self.get_ffprobe_binary_path(
823 download_if_not_present=False,
824 cwd=self.cwd,
825 query_timeout=self.query_timeout,
826 success=self.success,
827 error=self.error,
828 debug=self.debug
829 )
830 print(f"FFmpeg installed at {ffmpeg_path}")
831 print(f"FFplay installed at {ffplay_path}")
832 print(f"FFprobe installed at {ffprobe_path}")
833 if audio_segment_node is not None:
834 audio_segment_node.ffmpeg = ffmpeg_path
835 AudioSegment.ffmpeg = ffmpeg_path
837 ffmpeg_path, ffplay_path, ffprobe_path, download_if_not_present=False
838 )
839 print("FF_family installed and ready to use!")
840 return self.success
841
842
843if __name__ == "__main__":
845 FDI.main()
846 AUDIO_WAVE = 440
847 AUDIO_SAMPLE_PATH = f"./{AUDIO_WAVE}.wav"
848 AUDIO_SAMPLE = FDI.generate_audio_sample(AUDIO_WAVE)
849 FDI.save_audio_sample(AUDIO_SAMPLE, AUDIO_SAMPLE_PATH)
850 FDI.play_audio_sample(AUDIO_SAMPLE)
None _extract_package(str file_path, str destination)
None add_ff_family_to_path(Union[str, None] ffmpeg_path=None, Union[str, None] ffplay_path=None, Union[str, None] ffprobe_path=None, bool download_if_not_present=True, str cwd=os.getcwd(), int query_timeout=10, int success=0, int error=1, bool debug=False)
str get_ffplay_binary_path(bool download_if_not_present=True, str cwd=os.getcwd(), int query_timeout=10, int success=0, int error=1, bool debug=False)
"FFMPEGDownloader" __new__(cls, *args, **kwargs)
None _download_file(str file_url, str file_path, int query_timeout=10)
str get_ffprobe_binary_path(bool download_if_not_present=True, str cwd=os.getcwd(), int query_timeout=10, int success=0, int error=1, bool debug=False)
__init__(self, str cwd=os.getcwd(), int query_timeout=10, int success=0, int error=84, bool debug=False)
None _grant_executable_rights(Union[str, None] file_path=None)
None save_audio_sample(AudioSegment audio_segment, str file_path)
int main(self, Union[AudioSegment, None] audio_segment_node=None)
None _rename_extracted_folder(str old_name, str new_name)
str get_ffmpeg_binary_path(bool download_if_not_present=True, str cwd=os.getcwd(), int query_timeout=10, int success=0, int error=1, bool debug=False)
str process_file_path(*Union[str, PPath] args, Union[str, PPath, None] cwd=None)
str get_ff_family_path(bool download_if_not_present=True, str cwd=os.getcwd(), int query_timeout=10, int success=0, int error=1, bool debug=False)
None _get_correct_download_and_file_path(self, str binary_name=CONST.FFMPEG_KEY)