diff --git a/deffcode/ffhelper.py b/deffcode/ffhelper.py index 3f35602..0b0b9fe 100644 --- a/deffcode/ffhelper.py +++ b/deffcode/ffhelper.py @@ -42,8 +42,16 @@ logger.addHandler(logger_handler()) logger.setLevel(logging.DEBUG) +# set default timeout for subprocesses +MAX_TIMEOUT_SUBPROCESS: float = float(os.getenv("MAX_TIMEOUT_SUBPROCESS", 10.0)) +# set grace period (seconds) between SIGTERM and SIGKILL when terminating +# a timed-out subprocess. Allows FFmpeg to flush buffers and release hardware +# resources before being force-killed. +TERMINATE_TIMEOUT_SUBPROCESS: float = float( + os.getenv("TERMINATE_TIMEOUT_SUBPROCESS", 2.0) +) # set default timer for download requests -DEFAULT_TIMEOUT = 3 +DEFAULT_TIMEOUT_REQUESTS: float = float(os.getenv("DEFAULT_TIMEOUT_REQUESTS", 3.0)) class TimeoutHTTPAdapter(HTTPAdapter): @@ -52,13 +60,15 @@ class TimeoutHTTPAdapter(HTTPAdapter): """ def __init__(self, *args: Any, **kwargs: Any) -> None: - self.timeout: float = DEFAULT_TIMEOUT + self.timeout: float = DEFAULT_TIMEOUT_REQUESTS if "timeout" in kwargs: self.timeout = kwargs["timeout"] del kwargs["timeout"] super().__init__(*args, **kwargs) - def send(self, request: requests.PreparedRequest, **kwargs: Any) -> requests.Response: + def send( + self, request: requests.PreparedRequest, **kwargs: Any + ) -> requests.Response: timeout = kwargs.get("timeout") if timeout is None: kwargs["timeout"] = self.timeout @@ -131,7 +141,9 @@ def get_valid_ffmpeg_path( final_path = os.path.join(final_path, "ffmpeg.exe") else: # else return False - verbose and logger.debug("No valid FFmpeg executables found at Custom FFmpeg path!") + verbose and logger.debug( + "No valid FFmpeg executables found at Custom FFmpeg path!" + ) return False else: # otherwise perform test for Unix @@ -145,7 +157,9 @@ def get_valid_ffmpeg_path( final_path = os.path.join(custom_ffmpeg, "ffmpeg") else: # else return False - verbose and logger.debug("No valid FFmpeg executables found at Custom FFmpeg path!") + verbose and logger.debug( + "No valid FFmpeg executables found at Custom FFmpeg path!" + ) return False else: # otherwise assign ffmpeg binaries from system @@ -157,7 +171,9 @@ def get_valid_ffmpeg_path( return final_path if validate_ffmpeg(final_path, verbose=verbose) else False -def download_ffmpeg_binaries(path: str, os_windows: bool = False, os_bit: str = "") -> str: +def download_ffmpeg_binaries( + path: str, os_windows: bool = False, os_bit: str = "" +) -> str: """ ## download_ffmpeg_binaries @@ -177,7 +193,9 @@ def download_ffmpeg_binaries(path: str, os_windows: bool = False, os_bit: str = os_bit ) - file_name = os.path.join(os.path.abspath(path), "ffmpeg-static-{}-gpl.zip".format(os_bit)) + file_name = os.path.join( + os.path.abspath(path), "ffmpeg-static-{}-gpl.zip".format(os_bit) + ) file_path = os.path.join( os.path.abspath(path), "ffmpeg-static-{}-gpl/bin/ffmpeg.exe".format(os_bit), @@ -192,7 +210,8 @@ def download_ffmpeg_binaries(path: str, os_windows: bool = False, os_bit: str = # check if given path has write access assert os.access(path, os.W_OK), ( - "[Helper:ERROR] :: Permission Denied, Cannot write binaries to directory = " + path + "[Helper:ERROR] :: Permission Denied, Cannot write binaries to directory = " + + path ) # remove leftovers if exists os.path.isfile(file_name) and delete_file_safe(file_name) @@ -210,7 +229,9 @@ def download_ffmpeg_binaries(path: str, os_windows: bool = False, os_bit: str = status_forcelist=[429, 500, 502, 503, 504], ) # Mount it for https usage - adapter = TimeoutHTTPAdapter(timeout=2.0, max_retries=retries) + adapter = TimeoutHTTPAdapter( + timeout=MAX_TIMEOUT_SUBPROCESS, max_retries=retries + ) http.mount("https://", adapter) response = http.get(file_url, stream=True) response.raise_for_status() @@ -219,9 +240,9 @@ def download_ffmpeg_binaries(path: str, os_windows: bool = False, os_bit: str = if "content-length" in response.headers else len(response.content) ) - assert total_length is not None, ( - "[Helper:ERROR] :: Failed to retrieve files, check your Internet connectivity!" - ) + assert ( + total_length is not None + ), "[Helper:ERROR] :: Failed to retrieve files, check your Internet connectivity!" bar = tqdm(total=int(total_length), unit="B", unit_scale=True) for data in response.iter_content(chunk_size=4096): f.write(data) @@ -259,7 +280,9 @@ def validate_ffmpeg(path: str, verbose: bool = False) -> bool: if verbose: # log if test are passed logger.debug("FFmpeg validity Test Passed!") logger.debug( - "Found valid FFmpeg Version: `{}` installed on this system".format(version) + "Found valid FFmpeg Version: `{}` installed on this system".format( + version + ) ) except Exception as e: # log if test are failed @@ -286,14 +309,20 @@ def get_supported_pixfmts(path: str) -> list[tuple[str, str, str]]: srtindex = [i for i, s in enumerate(splitted) if b"-----" in s] # extract video encoders supported_pxfmts = [ - x.decode("utf-8").strip() for x in splitted[srtindex[0] + 1 :] if x.decode("utf-8").strip() + x.decode("utf-8").strip() + for x in splitted[srtindex[0] + 1 :] + if x.decode("utf-8").strip() ] # compile regex finder = re.compile(r"([A-Z]*[\.]+[A-Z]*\s[a-z0-9_-]*)(\s+[0-4])(\s+[0-9]+)") # find all outputs outputs = finder.findall("\n".join(supported_pxfmts)) # return output findings - return [(list(o[0].split(" "))[-1], o[1].strip(), o[2].strip()) for o in outputs if len(o) == 3] + return [ + (list(o[0].split(" "))[-1], o[1].strip(), o[2].strip()) + for o in outputs + if len(o) == 3 + ] def get_supported_vdecoders(path: str) -> list[str]: @@ -369,9 +398,9 @@ def extract_device_n_demuxer( **Returns:** Tuple of list of supported device(s) path/name/index and OS specific demuxer used. """ # validate `machine_OS` parameter value - assert machine_OS is not None and isinstance(machine_OS, str), ( - "`machine_OS` parameter value is empty or invalid type. Aborting!" - ) + assert machine_OS is not None and isinstance( + machine_OS, str + ), "`machine_OS` parameter value is empty or invalid type. Aborting!" # initialize params devices: list[Any] = [] # handles devices discovered @@ -395,14 +424,16 @@ def extract_device_n_demuxer( verbose and logger.debug("Auto-Searching for valid devices...") # assert if demuxer is supported by provided ffmpeg. - assert req_demuxer in get_supported_demuxers(path), ( - "Required `{}` demuxer isn't supported by provided FFmpeg binaries. Kindly compile FFmpeg with \ + assert req_demuxer in get_supported_demuxers( + path + ), "Required `{}` demuxer isn't supported by provided FFmpeg binaries. Kindly compile FFmpeg with \ suitable flags or manually assign `source` and `source_demuxer` parameter values. Aborting!".format( - valid_demuxers[machine_OS] - ) + valid_demuxers[machine_OS] ) # create default ffmpeg command (for Windows and MacOS) - default_ffcommand = "-hide_banner -list_devices true -f {} -i dummy".format(req_demuxer) + default_ffcommand = "-hide_banner -list_devices true -f {} -i dummy".format( + req_demuxer + ) # find all OS specific FFmpeg devices path and demuxer if machine_OS == "Windows": @@ -410,6 +441,7 @@ def extract_device_n_demuxer( metadata = check_sp_output( [path, *default_ffcommand.split(" ")], force_retrieve_stderr=True, + timeout=MAX_TIMEOUT_SUBPROCESS, ) # clean and split metadata splitted = [x.decode("utf-8").strip() for x in metadata.split(b"\n")] @@ -434,7 +466,10 @@ def extract_device_n_demuxer( if ( not decoded or {"command", "not", "found"}.issubset(decoded.split(" ")) - or ({"Cannot", "open", "device"}.issubset(decoded.split(" ")) and "):" not in decoded) + or ( + {"Cannot", "open", "device"}.issubset(decoded.split(" ")) + and "):" not in decoded + ) ): logger.error( "Cannot execute `v4l2-ctl` command. " @@ -447,7 +482,9 @@ def extract_device_n_demuxer( else: # clean metadata clean_n_splitted = [ - x.strip() for x in decoded.split("\n\n") if "/dev/video" in x and "):" in x + x.strip() + for x in decoded.split("\n\n") + if "/dev/video" in x and "):" in x ] # compile regex finder = re.compile(r"^[a-zA-Z0-9_.\- ]*") @@ -469,10 +506,14 @@ def extract_device_n_demuxer( # search in path properties metadata_path = check_sp_output( ["v4l2-ctl", "--device={}".format(path), "--all"], + timeout=MAX_TIMEOUT_SUBPROCESS, ) # decode path metadata decoded_path = metadata_path.decode("utf-8").strip() - if "Width/Height" in decoded_path and "Pixel Format" in decoded_path: + if ( + "Width/Height" in decoded_path + and "Pixel Format" in decoded_path + ): # append once required Width/Height and Pixel Format detected devices.append({path: device_name}) else: @@ -488,6 +529,7 @@ def extract_device_n_demuxer( metadata = check_sp_output( [path, *default_ffcommand.split(" ")], force_retrieve_stderr=True, + timeout=MAX_TIMEOUT_SUBPROCESS, ) # clean and split metadata splitted = [x.decode("utf-8").strip() for x in metadata.split(b"\n")] @@ -532,7 +574,9 @@ def extract_device_n_demuxer( ) -def validate_imgseqdir(source: str, extension: str = "jpg", verbose: bool = False) -> bool: +def validate_imgseqdir( + source: str, extension: str = "jpg", verbose: bool = False +) -> bool: """ ## validate_imgseqdir @@ -558,7 +602,9 @@ def validate_imgseqdir(source: str, extension: str = "jpg", verbose: bool = Fals return False -def is_valid_image_seq(path: str, source: str | None = None, verbose: bool = False) -> bool: +def is_valid_image_seq( + path: str, source: str | None = None, verbose: bool = False +) -> bool: """ ## is_valid_image_seq @@ -578,7 +624,9 @@ def is_valid_image_seq(path: str, source: str | None = None, verbose: bool = Fal # extract all FFmpeg supported protocols formats = check_sp_output([path, "-hide_banner", "-formats"]) extract_formats = re.findall(r"\w+_pipe", formats.decode("utf-8").strip()) - supported_image_formats = [x.split("_")[0] for x in extract_formats if x.endswith("_pipe")] + supported_image_formats = [ + x.split("_")[0] for x in extract_formats if x.endswith("_pipe") + ] _filename, extension = os.path.splitext(source) # Test and return result whether scheme is supported if extension and source.endswith(tuple(supported_image_formats)): @@ -623,7 +671,9 @@ def is_valid_url(path: str, url: str | None = None, verbose: bool = False) -> bo supported_protocols = splitted[splitted.index("Output:") + 1 : len(splitted) - 1] # RTSP is a demuxer somehow # support both RTSP and RTSPS(over SSL) - supported_protocols += ["rtsp", "rtsps"] if "rtsp" in get_supported_demuxers(path) else [] + supported_protocols += ( + ["rtsp", "rtsps"] if "rtsp" in get_supported_demuxers(path) else [] + ) # Test and return result whether scheme is supported if extracted_scheme_url and extracted_scheme_url in supported_protocols: verbose and logger.debug( @@ -641,44 +691,88 @@ def check_sp_output(*args: Any, **kwargs: Any) -> bytes: """ ## check_sp_output - Returns FFmpeg `stdout` output from subprocess module. + Executes a subprocess command and returns its `stdout` (or `stderr` when + requested). On timeout, performs a two-step graceful shutdown — sends + `SIGTERM` first to allow FFmpeg to flush buffers and release hardware + resources (decoders, capture devices), then escalates to `SIGKILL` if the + process fails to exit within a short grace period. Parameters: args (based on input): Non Keyword Arguments kwargs (based on input): Keyword Arguments + force_retrieve_stderr (bool): If True, returns stderr. Also + suppresses `CalledProcessError` on non-zero exit, since some + FFmpeg diagnostic commands (e.g. `-list_devices`) emit useful + output on stderr while exiting non-zero by design. + timeout (float): Seconds to wait before terminating the process. - **Returns:** A string value. + **Returns:** A bytes value. """ # workaround for python bug: https://bugs.python.org/issue37380 if platform.system() == "Windows": # see comment https://bugs.python.org/msg370334 sp._cleanup = lambda: None + # handle additional params retrieve_stderr = kwargs.pop("force_retrieve_stderr", False) + timeout = kwargs.pop("timeout", None) + # execute command in subprocess process = sp.Popen( *args, stdout=sp.PIPE, - stderr=sp.DEVNULL if not (retrieve_stderr) else sp.PIPE, + stderr=sp.PIPE if retrieve_stderr else sp.DEVNULL, **kwargs, ) - # communicate and poll process - output, stderr = process.communicate() + + # communicate and poll process with two-step timeout handling + timeout_occurred = False + try: + output, stderr = process.communicate(timeout=timeout) + except sp.TimeoutExpired: + timeout_occurred = True + logger.warning( + f"[Pipeline-Warning] :: Process exceeded timeout of {timeout}s. " + "Attempting graceful termination..." + ) + # Step 1: polite SIGTERM, give process a chance to clean up + process.terminate() + try: + output, stderr = process.communicate(timeout=TERMINATE_TIMEOUT_SUBPROCESS) + except sp.TimeoutExpired: + # Step 2: process ignored SIGTERM, force kill + logger.error( + "[Pipeline-Error] :: Process unresponsive to SIGTERM. " + "Hard killing..." + ) + process.kill() + output, stderr = process.communicate() + retcode = process.poll() + # handle return code - if retcode and not (retrieve_stderr): - logger.error("[Pipeline-Error] :: {}".format(output.decode("utf-8"))) + # Bypass CalledProcessError if caller wants stderr (some FFmpeg commands + # exit non-zero by design) or if we killed the process via our timeout. + if retcode and not retrieve_stderr and not timeout_occurred: + logger.error( + "[Pipeline-Error] :: {}".format( + output.decode("utf-8") if output else "No output" + ) + ) cmd = kwargs.get("args") if cmd is None: cmd = args[0] error = sp.CalledProcessError(retcode, cmd) error.output = output raise error - # raise error if no output - bool(output) or bool(stderr) or logger.error( - "[Pipeline-Error] :: Pipeline failed to exact any data from command: {}!".format( - args[0] if args else [] + + # warn if process emitted nothing on either stream + if not (bool(output) or bool(stderr)): + logger.error( + "[Pipeline-Error] :: Pipeline failed to extract any data from command: {}!".format( + args[0] if args else [] + ) ) - ) - # return output otherwise + + # return stderr when explicitly requested (and present), else stdout return stderr if retrieve_stderr and stderr else output diff --git a/deffcode/sourcer.py b/deffcode/sourcer.py index 5045d9a..4420350 100644 --- a/deffcode/sourcer.py +++ b/deffcode/sourcer.py @@ -33,6 +33,7 @@ import numpy as np from .ffhelper import ( + MAX_TIMEOUT_SUBPROCESS, check_sp_output, extract_device_n_demuxer, get_supported_demuxers, @@ -107,13 +108,17 @@ def __init__( # sanitize sourcer_params self.__sourcer_params = { str(k).strip(): ( - str(v).strip() if not isinstance(v, (dict, list, int, float, tuple)) else v + str(v).strip() + if not isinstance(v, (dict, list, int, float, tuple)) + else v ) for k, v in sourcer_params.items() } # handle whether to force validate source - self.__forcevalidatesource = self.__sourcer_params.pop("-force_validate_source", False) + self.__forcevalidatesource = self.__sourcer_params.pop( + "-force_validate_source", False + ) if not isinstance(self.__forcevalidatesource, bool): # reset improper values self.__forcevalidatesource = False @@ -231,10 +236,10 @@ def __init__( # assign if valid demuxer value self.__source_demuxer = source_demuxer.strip().lower() # assign if valid demuxer value - assert self.__source_demuxer != "auto" or validate_device_index(source), ( - "Invalid `source_demuxer='auto'` value detected with source: `{}`. Aborting!".format( - source - ) + assert self.__source_demuxer != "auto" or validate_device_index( + source + ), "Invalid `source_demuxer='auto'` value detected with source: `{}`. Aborting!".format( + source ) else: # otherwise find valid default source demuxer value @@ -294,7 +299,9 @@ def __init__( # check whether metadata probed or not? self.__metadata_probed = False - def probe_stream(self, default_stream_indexes: list[int] | tuple[int, int] = (0, 0)) -> Sourcer: + def probe_stream( + self, default_stream_indexes: list[int] | tuple[int, int] = (0, 0) + ) -> Sourcer: """ This method Parses/Probes FFmpeg `subprocess` pipe's Standard Output for given input source and Populates the information in private class variables. @@ -312,7 +319,9 @@ def probe_stream(self, default_stream_indexes: list[int] | tuple[int, int] = (0, self.__ffsp_output = self.__validate_source( self.__source, source_demuxer=self.__source_demuxer, - forced_validate=(self.__forcevalidatesource if self.__source_demuxer is None else True), + forced_validate=( + self.__forcevalidatesource if self.__source_demuxer is None else True + ), ) # parse resolution and framerate video_rfparams = self.__extract_resolution_framerate( @@ -393,7 +402,9 @@ def probe_stream(self, default_stream_indexes: list[int] | tuple[int, int] = (0, # source's flat metadata is captured first via retrieve_metadata; # the guard inside retrieve_metadata (checks for non-empty # __multi_source_metadata) prevents `sources: []` self-pollution. - self.__multi_source_metadata.append(self.retrieve_metadata(force_retrieve_missing=True)) + self.__multi_source_metadata.append( + self.retrieve_metadata(force_retrieve_missing=True) + ) for idx in range(1, len(self.__source_list)): _src = self.__source_list[idx] _demux = self.__source_demuxer_list[idx] @@ -439,9 +450,9 @@ def retrieve_metadata( **Returns:** `metadata` or `(metadata, metadata_missing)`, formatted as JSON string or python dictionary. """ # check if metadata has been probed or not - assert self.__metadata_probed, ( - "Source Metadata not been probed yet! Check if you called `probe_stream()` method." - ) + assert ( + self.__metadata_probed + ), "Source Metadata not been probed yet! Check if you called `probe_stream()` method." # log it self.__verbose_logs and logger.debug("Extracting Metadata...") # create metadata dictionary from information populated in private class variables @@ -524,7 +535,9 @@ def retrieve_metadata( ) # log it - self.__verbose_logs and logger.debug("Metadata Extraction completed successfully!") + self.__verbose_logs and logger.debug( + "Metadata Extraction completed successfully!" + ) # parse as JSON string(`json.dumps`), if defined metadata = json.dumps(metadata, indent=2) if pretty_json else metadata metadata_missing = ( @@ -542,9 +555,9 @@ def enumerate_devices(self) -> dict[int, Any]: **Returns:** Probed Camera Devices as python dictionary. """ # check if metadata has been probed or not - assert self.__metadata_probed, ( - "Source Metadata not been probed yet! Check if you called `probe_stream()` method." - ) + assert ( + self.__metadata_probed + ), "Source Metadata not been probed yet! Check if you called `probe_stream()` method." # log if specified self.__verbose_logs and logger.debug("Enumerating all probed Camera Devices.") @@ -616,9 +629,15 @@ def __validate_source( ( self.__extracted_devices_list[index] if self.__machine_OS != "Linux" - else next(iter(self.__extracted_devices_list[index].values()))[0] + else next( + iter(self.__extracted_devices_list[index].values()) + )[0] + ), + ( + index + if index >= 0 + else len(self.__extracted_devices_list) + index ), - (index if index >= 0 else len(self.__extracted_devices_list) + index), self.__source_demuxer, ) ) @@ -642,7 +661,9 @@ def __validate_source( pass # assert if valid source - assert source and isinstance(source, str), "Input `source` parameter is of invalid type!" + assert source and isinstance( + source, str + ), "Input `source` parameter is of invalid type!" # Differentiate input if forced_validate: @@ -652,7 +673,9 @@ def __validate_source( self.__source = source elif os.path.isfile(source): self.__source = os.path.abspath(source) - elif is_valid_image_seq(self.__ffmpeg, source=source, verbose=self.__verbose_logs): + elif is_valid_image_seq( + self.__ffmpeg, source=source, verbose=self.__verbose_logs + ): self.__source = source self.__contains_images = True elif is_valid_url(self.__ffmpeg, url=source, verbose=self.__verbose_logs): @@ -687,13 +710,14 @@ def __validate_source( check_sp_output( meta_cmd, force_retrieve_stderr=True, + timeout=MAX_TIMEOUT_SUBPROCESS, ) .decode("utf-8") .strip() ) # separate input and output metadata (if available) if "Output #" in metadata: - (metadata, self.__metadata_output) = metadata.split("Output #") + metadata, self.__metadata_output = metadata.split("Output #") # return metadata based on params return metadata @@ -720,7 +744,9 @@ def __extract_video_bitrate(self, default_stream: int = 0) -> str: else 0 ) ] - filtered_bitrate = re.findall(r",\s[0-9]+\s\w\w[\/]s", selected_stream.strip()) + filtered_bitrate = re.findall( + r",\s[0-9]+\s\w\w[\/]s", selected_stream.strip() + ) if len(filtered_bitrate): default_video_bitrate = filtered_bitrate[0].split(" ")[1:3] final_bitrate = "{}{}".format( @@ -748,14 +774,22 @@ def __extract_video_decoder(self, default_stream: int = 0) -> str: ] if meta_text: selected_stream = meta_text[ - (default_stream if default_stream > 0 and default_stream < len(meta_text) else 0) + ( + default_stream + if default_stream > 0 and default_stream < len(meta_text) + else 0 + ) ] - filtered_pixfmt = re.findall(r"Video:\s[a-z0-9_-]*", selected_stream.strip()) + filtered_pixfmt = re.findall( + r"Video:\s[a-z0-9_-]*", selected_stream.strip() + ) if filtered_pixfmt: return filtered_pixfmt[0].split(" ")[-1] return "" - def __extract_video_pixfmt(self, default_stream: int = 0, extract_output: bool = False) -> str: + def __extract_video_pixfmt( + self, default_stream: int = 0, extract_output: bool = False + ) -> str: """ This Internal method parses default video-stream pixel-format from metadata. @@ -780,14 +814,22 @@ def __extract_video_pixfmt(self, default_stream: int = 0, extract_output: bool = ) if meta_text: selected_stream = meta_text[ - (default_stream if default_stream > 0 and default_stream < len(meta_text) else 0) + ( + default_stream + if default_stream > 0 and default_stream < len(meta_text) + else 0 + ) ] - filtered_pixfmt = re.findall(r",\s[a-z][a-z0-9_-]*", selected_stream.strip()) + filtered_pixfmt = re.findall( + r",\s[a-z][a-z0-9_-]*", selected_stream.strip() + ) if filtered_pixfmt: return filtered_pixfmt[0].split(" ")[-1] return "" - def __extract_audio_bitrate_nd_samplerate(self, default_stream: int = 0) -> dict[str, str]: + def __extract_audio_bitrate_nd_samplerate( + self, default_stream: int = 0 + ) -> dict[str, str]: """ This Internal method parses default audio-stream bitrate and sample-rate from metadata. @@ -805,13 +847,19 @@ def __extract_audio_bitrate_nd_samplerate(self, default_stream: int = 0) -> dict result = {} if meta_text: selected_stream = meta_text[ - (default_stream if default_stream > 0 and default_stream < len(meta_text) else 0) + ( + default_stream + if default_stream > 0 and default_stream < len(meta_text) + else 0 + ) ] # filter data filtered_audio_bitrate = re.findall( r"fltp,\s[0-9]+\s\w\w[\/]s", selected_stream.strip() ) - filtered_audio_samplerate = re.findall(r",\s[0-9]+\sHz", selected_stream.strip()) + filtered_audio_samplerate = re.findall( + r",\s[0-9]+\sHz", selected_stream.strip() + ) # get audio bitrate metadata if filtered_audio_bitrate: filtered = filtered_audio_bitrate[0].split(" ")[1:3] @@ -823,7 +871,9 @@ def __extract_audio_bitrate_nd_samplerate(self, default_stream: int = 0) -> dict result["bitrate"] = "" # get audio samplerate metadata result["samplerate"] = ( - filtered_audio_samplerate[0].split(", ")[1] if filtered_audio_samplerate else "" + filtered_audio_samplerate[0].split(", ")[1] + if filtered_audio_samplerate + else "" ) return result if result and (len(result) == 2) else {} @@ -873,21 +923,33 @@ def __extract_resolution_framerate( result = {} if meta_text: selected_stream = meta_text[ - (default_stream if default_stream > 0 and default_stream < len(meta_text) else 0) + ( + default_stream + if default_stream > 0 and default_stream < len(meta_text) + else 0 + ) ] # filter data - filtered_resolution = re.findall(r"([1-9]\d+)x([1-9]\d+)", selected_stream.strip()) - filtered_framerate = re.findall(r"\d+(?:\.\d+)?\sfps", selected_stream.strip()) + filtered_resolution = re.findall( + r"([1-9]\d+)x([1-9]\d+)", selected_stream.strip() + ) + filtered_framerate = re.findall( + r"\d+(?:\.\d+)?\sfps", selected_stream.strip() + ) filtered_tbr = re.findall(r"\d+(?:\.\d+)?\stbr", selected_stream.strip()) # extract framerate metadata if filtered_framerate: # calculate actual framerate - result["framerate"] = float(re.findall(r"[\d\.\d]+", filtered_framerate[0])[0]) + result["framerate"] = float( + re.findall(r"[\d\.\d]+", filtered_framerate[0])[0] + ) elif filtered_tbr: # guess from TBR(if fps unavailable) - result["framerate"] = float(re.findall(r"[\d\.\d]+", filtered_tbr[0])[0]) + result["framerate"] = float( + re.findall(r"[\d\.\d]+", filtered_tbr[0])[0] + ) # extract resolution metadata if filtered_resolution: @@ -902,7 +964,9 @@ def __extract_resolution_framerate( else 0 ) ] - filtered_orientation = re.findall(r"[-]?\d+\.\d+", selected_stream.strip()) + filtered_orientation = re.findall( + r"[-]?\d+\.\d+", selected_stream.strip() + ) result["orientation"] = float(filtered_orientation[0]) else: result["orientation"] = 0.0 @@ -931,7 +995,10 @@ def __extract_duration(self, inseconds: bool = True) -> float | list[str]: ) if t_duration: return ( - sum(float(x) * 60**i for i, x in enumerate(reversed(t_duration[0].split(":")))) + sum( + float(x) * 60**i + for i, x in enumerate(reversed(t_duration[0].split(":"))) + ) if inseconds else t_duration ) diff --git a/tests/test_ffhelper.py b/tests/test_ffhelper.py index b572727..f5fc861 100644 --- a/tests/test_ffhelper.py +++ b/tests/test_ffhelper.py @@ -145,9 +145,72 @@ def test_get_valid_ffmpeg_path(paths: str, ffmpeg_download_paths: str, results: @pytest.mark.xfail(raises=Exception) def test_check_sp_output() -> None: """ - Testing check_sp_output method + Testing check_sp_output method raises CalledProcessError on invalid flags. """ - check_sp_output(["ffmpeg", "-Vv"]) + check_sp_output(["ffmpeg", "-Vv"], timeout=2.0) + + +def test_check_sp_output_happy_path() -> None: + """ + Testing check_sp_output returns stdout bytes for a successful command. + """ + output = check_sp_output([return_static_ffmpeg(), "-version"], timeout=5.0) + assert isinstance(output, bytes) and len(output) > 0, "Expected non-empty stdout" + assert b"ffmpeg version" in output, "Expected ffmpeg version banner in stdout" + + +def test_check_sp_output_force_retrieve_stderr_bypass() -> None: + """ + Testing that `force_retrieve_stderr=True` suppresses CalledProcessError on + non-zero exit and returns stderr bytes. Mirrors how `extract_device_n_demuxer` + consumes `ffmpeg -list_devices` output (which exits non-zero by design). + """ + # `-list_devices true -i dummy` exits non-zero on every platform + cmd = [ + return_static_ffmpeg(), + "-hide_banner", + "-list_devices", + "-i", + "dummy", + ] + stderr = check_sp_output(cmd, force_retrieve_stderr=True, timeout=5.0) + assert isinstance(stderr, bytes), "Expected bytes return on stderr bypass" + + +def test_check_sp_output_graceful_timeout() -> None: + """ + Testing two-step timeout handling: a long-running FFmpeg process must be + terminated within roughly `timeout + grace_period` seconds and must NOT + raise CalledProcessError (timeout kills are bypassed by design). + """ + import time + + # null source → null sink loop runs forever; ensures we hit the timeout + cmd = [ + return_static_ffmpeg(), + "-hide_banner", + "-loglevel", + "error", + "-f", + "lavfi", + "-i", + "nullsrc", + "-t", + "60", + "-f", + "null", + "-", + ] + timeout = 1.0 + start = time.monotonic() + output = check_sp_output(cmd, timeout=timeout) + elapsed = time.monotonic() - start + # must return bytes (possibly empty), no exception + assert isinstance(output, bytes), "Expected bytes return on graceful timeout" + # must complete within timeout + 2s grace + reasonable overhead + assert elapsed < timeout + 5.0, ( + f"Process took {elapsed:.2f}s — graceful shutdown likely stuck" + ) @pytest.mark.parametrize(