feat: add duration-related methods to SourceMedia

pipeline/utils.py

@@ -45,12 +45,35 @@ class Source():
         if not provider:
             raise ValueError("Provider must be provided") # TODO: #API -- decide if this is necessary
         self.provider = provider
+
     def __str__(self):
         """See: 'accessing the object should return the path to the media file'"""
         return self.path
 
     def __repr__(self):
         return f"Source({self.source}, {self.path}, {self.provider})"
+
+    def duration(self):
+        """Return the duration of the media file at self.path (result is cached)"""
+        return self._duration or self._get_duration(self.path)
+
+    def _get_duration(self, file):
+        """Use ffprobe to get the duration of the media file at self.path and cache result (_duration)
+
+        usage: ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 <file>
+        """
+        # test if file exists
+        try:
+            with open(file) as _:
+                pass
+        except FileNotFoundError:
+            raise FileNotFoundError(f"File not found: {file}")
+        # cache the result
+        self._duration = 0.0 or float(subprocess.check_output(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", file]))
+        return self._duration
+
+
+
 class Interval():
     """An interval of time in a media file