swap NUL to os.devnull

Author: slhck

ffmpeg_normalize/_cmd_utils.py

@@ -5,17 +5,15 @@
 import re
 import shlex
 import subprocess
-from platform import system
 from shutil import which
-from typing import Iterator, Any
+from typing import Any, Iterator
 
 from ffmpeg_progress_yield import FfmpegProgress
 
 from ._errors import FFmpegNormalizeError
 
 _logger = logging.getLogger(__name__)
 
-NUL = "NUL" if system() in ("Windows", "cli") else "/dev/null"
 DUR_REGEX = re.compile(
     r"Duration: (?P<hour>\d{2}):(?P<min>\d{2}):(?P<sec>\d{2})\.(?P<ms>\d{2})"
 )

ffmpeg_normalize/_media_file.py

@@ -10,7 +10,7 @@
 
 from tqdm import tqdm
 
-from ._cmd_utils import DUR_REGEX, NUL, CommandRunner
+from ._cmd_utils import DUR_REGEX, CommandRunner
 from ._errors import FFmpegNormalizeError
 from ._streams import (
     AudioStream,
@@ -69,7 +69,7 @@ def __init__(
         self.output_file = output_file
         current_ext = os.path.splitext(output_file)[1][1:]
         # we need to check if it's empty, e.g. /dev/null or NUL
-        if current_ext == "" or self.output_file == NUL:
+        if current_ext == "" or self.output_file == os.devnull:
             self.output_ext = self.ffmpeg_normalize.extension
         else:
             self.output_ext = current_ext
@@ -114,7 +114,7 @@ def parse_streams(self) -> None:
             "0",
             "-f",
             "null",
-            NUL,
+            os.devnull,
         ]
 
         output = CommandRunner().run_command(cmd).get_output()
@@ -435,7 +435,7 @@ def _second_pass(self) -> Iterator[float]:
             return
 
         # special case: if output is a null device, write directly to it
-        if self.output_file == NUL:
+        if self.output_file == os.devnull:
             cmd.append(self.output_file)
         else:
             temp_dir = mkdtemp()
@@ -452,14 +452,14 @@ def _second_pass(self) -> Iterator[float]:
                 )
                 raise e
             else:
-                if self.output_file != NUL:
+                if self.output_file != os.devnull:
                     _logger.debug(
                         f"Moving temporary file from {temp_file} to {self.output_file}"
                     )
                     move(temp_file, self.output_file)
                     rmtree(temp_dir, ignore_errors=True)
         except Exception as e:
-            if self.output_file != NUL:
+            if self.output_file != os.devnull:
                 rmtree(temp_dir, ignore_errors=True)
             raise e
 

ffmpeg_normalize/_streams.py

@@ -4,9 +4,9 @@
 import logging
 import os
 import re
-from typing import TYPE_CHECKING, Iterator, List, Literal, Optional, TypedDict, cast
+from typing import TYPE_CHECKING, Iterator, Literal, TypedDict, cast
 
-from ._cmd_utils import NUL, CommandRunner, dict_to_filter_opts
+from ._cmd_utils import CommandRunner, dict_to_filter_opts
 from ._errors import FFmpegNormalizeError
 
 if TYPE_CHECKING:
@@ -17,6 +17,7 @@
 
 _loudnorm_pattern = re.compile(r"\[Parsed_loudnorm_(\d+)")
 
+
 class EbuLoudnessStatistics(TypedDict):
     input_i: float
     input_tp: float
@@ -239,7 +240,7 @@ def parse_astats(self) -> Iterator[float]:
             "-sn",
             "-f",
             "null",
-            NUL,
+            os.devnull,
         ]
 
         cmd_runner = CommandRunner()
@@ -310,7 +311,7 @@ def parse_loudnorm_stats(self) -> Iterator[float]:
             "-sn",
             "-f",
             "null",
-            NUL,
+            os.devnull,
         ]
 
         cmd_runner = CommandRunner()
@@ -322,11 +323,13 @@ def parse_loudnorm_stats(self) -> Iterator[float]:
         )
 
         # only one stream
-        self.loudness_statistics["ebu_pass1"] = next(iter(AudioStream.prune_and_parse_loudnorm_output(output).values()))
+        self.loudness_statistics["ebu_pass1"] = next(
+            iter(AudioStream.prune_and_parse_loudnorm_output(output).values())
+        )
 
     @staticmethod
     def prune_and_parse_loudnorm_output(
-        output: str
+        output: str,
     ) -> dict[int, EbuLoudnessStatistics]:
         """
         Prune ffmpeg progress lines from output and parse the loudnorm filter output.
@@ -344,7 +347,7 @@ def prune_and_parse_loudnorm_output(
 
     @staticmethod
     def _parse_loudnorm_output(
-        output_lines: list[str]
+        output_lines: list[str],
     ) -> dict[int, EbuLoudnessStatistics]:
         """
         Parse the output of a loudnorm filter to get the EBU loudness statistics.
@@ -403,7 +406,9 @@ def _parse_loudnorm_output(
                                 # convert to floats
                                 loudnorm_stats[key] = float(loudnorm_stats[key])
 
-                        result[stream_index] = cast(EbuLoudnessStatistics, loudnorm_stats)
+                        result[stream_index] = cast(
+                            EbuLoudnessStatistics, loudnorm_stats
+                        )
                         stream_index = -1
                     except Exception as e:
                         raise FFmpegNormalizeError(
@@ -504,15 +509,11 @@ def get_second_pass_opts_ebu(self) -> str:
             "offset": self._constrain(
                 stats["target_offset"], -99, 99, name="target_offset"
             ),
-            "measured_i": self._constrain(
-                stats["input_i"], -99, 0, name="input_i"
-            ),
+            "measured_i": self._constrain(stats["input_i"], -99, 0, name="input_i"),
             "measured_lra": self._constrain(
                 stats["input_lra"], 0, 99, name="input_lra"
             ),
-            "measured_tp": self._constrain(
-                stats["input_tp"], -99, 99, name="input_tp"
-            ),
+            "measured_tp": self._constrain(stats["input_tp"], -99, 99, name="input_tp"),
             "measured_thresh": self._constrain(
                 stats["input_thresh"], -99, 0, name="input_thresh"
             ),