HiFi Decode fixes and improvements

vhsdecode/hifi/HiFiDecode.py

@@ -394,10 +394,10 @@ def __init__(self, options=None):
         if options is None:
             options = dict()
         self.options = options
-        self.sample_rate = options['input_rate']
+        self.sample_rate: int = options['input_rate']
         self.options = options
-        self.if_rate = 8388608
-        self.audio_rate = 192000
+        self.if_rate: int = 8388608
+        self.audio_rate: int = 192000
 
         # main deemphasis time constant
         self.tau = 56e-6
@@ -408,14 +408,15 @@ def __init__(self, options=None):
         self.audioRes_denominator = self.getResamplingRatios()
 
         # block overlap and edge discard
-        self.blocks_second = 8
-        self.block_size = int(self.sample_rate / self.blocks_second)
-        self.block_audio_size = int(self.audio_rate / self.blocks_second)
-        self.block_overlap_audio = 192 if not self.options['preview'] else 0
-        self.block_overlap = round(
-            self.block_overlap_audio *
-            self.audioRes_denominator * self.ifresample_denominator /
-            (self.audioRes_numerator * self.ifresample_numerator)
+        self.blocks_second: int = 8
+        self.block_size: int = int(self.sample_rate / self.blocks_second)
+        self.block_audio_size: int = int(self.audio_rate / self.blocks_second)
+        self.block_overlap_audio: int = int(self.audio_rate / 5e2)
+        audio_final_rate = ((self.options['audio_rate'] / self.audio_rate) *
+                            (self.audioRes_numerator / self.audioRes_denominator))
+        self.block_overlap: int = round(
+            self.block_overlap_audio * self.ifresample_denominator /
+            (self.ifresample_numerator * audio_final_rate)
         )
 
         # start of filter design stuff
@@ -528,8 +529,8 @@ def demodblock(self, data):
         preL = samplerate_resample(preAudioResampleL, self.audioRes_numerator, self.audioRes_denominator)
         preR = samplerate_resample(preAudioResampleR, self.audioRes_numerator, self.audioRes_denominator)
 
-        dcL = self.dcCancelL.work(np.mean(audioL))
-        dcR = self.dcCancelR.work(np.mean(audioR))
+        dcL = np.mean(audioL)
+        dcR = np.mean(audioR)
 
         return dcL, dcR, (audioL + preL) / 2, (audioR + preR) / 2, preL, preR
 
@@ -558,7 +559,8 @@ def notchFreq(self) -> float:
         return self.standard.Hfreq
 
     @staticmethod
-    def cancelDC(c, dc):
+    @njit(cache=True, fastmath=True, nogil=True)
+    def cancelDC(c: np.array, dc: np.array):
         return c - dc
 
     @staticmethod

vhsdecode/hifi/main.py

@@ -79,6 +79,14 @@
     f"Range (20~100): 100 being a hard gate effect",
 )
 
+parser.add_argument(
+    "--gain",
+    dest="gain",
+    type=float,
+    default=1.0,
+    help="Sets the gain/volume of the output audio (default is 1.0)",
+)
+
 parser.add_argument(
     "--h8", dest="H8", action="store_true", default=False, help="Video8/Hi8, 8mm tape format"
 )
@@ -160,12 +168,14 @@ def read(self, samples, frames=-1, dtype='float64', always_2d=False,
         out = np.asarray(np.frombuffer(data, dtype=np.int16), dtype=dtype)
         return out
 
-    def _read_next_chunk(self, blocksize, overlap, dtype):
-        data = self.file_path.read(blocksize)
+    def _read_next_chunk(self, blocksize, overlap, dtype) -> np.array:
+        data = self.file_path.read(blocksize - overlap)
         assert len(data) % 2 == 0, "data is misaligned"
         out = np.asarray(np.frombuffer(data, dtype=np.int16), dtype=dtype)
+        self._overlap = np.copy(out[-overlap:]) if np.size(self._overlap) == 0 else self._overlap
+        result = np.concatenate((self._overlap, out))
         self._overlap = np.copy(out[-overlap:])
-        return np.concatenate((self._overlap, out))
+        return result
 
     # yields infinite generator for _read_next_chunk
     def blocks(self, blocksize=None, overlap=0, frames=-1, dtype='float64',
@@ -284,7 +294,7 @@ def seconds_to_str(seconds: float) -> str:
 
 def log_decode(start_time: datetime, frames: int, decode_options: dict):
     elapsed_time: timedelta = datetime.now() - start_time
-    audio_time: float = frames / decode_options["input_rate"]
+    audio_time: float = frames / (2 * decode_options["input_rate"])
     relative_speed: float = audio_time / elapsed_time.total_seconds()
     elapsed_time_format: str = seconds_to_str(elapsed_time.total_seconds())
     audio_time_format: str = seconds_to_str(audio_time)
@@ -296,6 +306,40 @@ def log_decode(start_time: datetime, frames: int, decode_options: dict):
     )
 
 
+def gain_adjust(audio: np.array, gain: float) -> np.array:
+    return np.multiply(audio, gain)
+
+
+def prepare_stereo(l: np.array, r: np.array, noise_reduction: NoiseReduction, decode_options: dict):
+    if decode_options["noise_reduction"]:
+        stereo = noise_reduction.stereo(
+            gain_adjust(l, decode_options["gain"]),
+            gain_adjust(r, decode_options["gain"])
+        )
+    else:
+        stereo = list(map(
+            list,
+            zip(gain_adjust(l, decode_options["gain"]),
+                gain_adjust(r, decode_options["gain"]))
+        ))
+    return stereo
+
+
+def post_process(audioL: np.array, audioR: np.array, audio_rate: int, decode_options: dict):
+    left_audio = samplerate_resample(audioL,
+                                     decode_options["audio_rate"],
+                                     audio_rate,
+                                     "sinc_fastest") \
+        if decode_options["audio_rate"] != audio_rate else audioL
+    right_audio = samplerate_resample(audioR,
+                                      decode_options["audio_rate"],
+                                      audio_rate,
+                                      "sinc_fastest") \
+        if decode_options["audio_rate"] != audio_rate else audioR
+
+    return left_audio, right_audio
+
+
 def decode(decoder, input_file, decode_options, output_file):
     start_time = datetime.now()
     noise_reduction = NoiseReduction(
@@ -316,27 +360,18 @@ def decode(decoder, input_file, decode_options, output_file):
                 current_block, audioL, audioR = decoder.block_decode(
                     block, block_count=current_block
                 )
-                left_audio = samplerate_resample(audioL,
-                                                 decode_options["audio_rate"],
-                                                 decoder.audioRate,
-                                                 "sinc_fastest") \
-                    if decode_options["audio_rate"] != decoder.audioRate else audioL
-                right_audio = samplerate_resample(audioR,
-                                                  decode_options["audio_rate"],
-                                                  decoder.audioRate,
-                                                  "sinc_fastest") \
-                    if decode_options["audio_rate"] != decoder.audioRate else audioR
-
-                if decode_options["noise_reduction"]:
-                    stereo = noise_reduction.stereo(left_audio, right_audio)
-                else:
-                    stereo = list(map(list, zip(left_audio, right_audio)))
+                left_audio, right_audio = post_process(audioL, audioR, decoder.audioRate, decode_options)
+                stereo = prepare_stereo(left_audio, right_audio, noise_reduction, decode_options)
 
                 if decode_options["auto_fine_tune"]:
                     log_bias(decoder)
 
+                current_block += 1
                 log_decode(start_time, f.tell(), decode_options)
-                w.write(stereo)
+                try:
+                    w.write(stereo)
+                except ValueError:
+                    pass
 
         elapsed_time = datetime.now() - start_time
         dt_string = elapsed_time.total_seconds()
@@ -380,33 +415,24 @@ def decode_parallel(decoders: List[HiFiDecode],
                 while len(futures_queue) > threads:
                     future = futures_queue.pop(0)
                     blocknum, audioL, audioR = future.result()
-                    left_audio = samplerate_resample(audioL,
-                                                     decode_options["audio_rate"],
-                                                     decoder.audioRate,
-                                                     "sinc_fastest") \
-                        if decode_options["audio_rate"] != decoder.audioRate else audioL
-                    right_audio = samplerate_resample(audioR,
-                                                      decode_options["audio_rate"],
-                                                      decoder.audioRate,
-                                                      "sinc_fastest") \
-                        if decode_options["audio_rate"] != decoder.audioRate else audioR
-                    if decode_options["noise_reduction"]:
-                        stereo = noise_reduction.stereo(left_audio, right_audio)
-                    else:
-                        stereo = list(map(list, zip(left_audio, right_audio)))
+                    left_audio, right_audio = post_process(audioL, audioR, decoder.audioRate, decode_options)
+                    stereo = prepare_stereo(left_audio, right_audio, noise_reduction, decode_options)
                     log_decode(start_time, f.tell(), decode_options)
-                    w.write(stereo)
+                    try:
+                        w.write(stereo)
+                    except ValueError:
+                        pass
 
             print("Emptying the decode queue ...")
             while len(futures_queue) > 0:
                 future = futures_queue.pop(0)
                 blocknum, audioL, audioR = future.result()
-                if decode_options["noise_reduction"]:
-                    stereo = noise_reduction.stereo(audioL, audioR)
-                else:
-                    stereo = list(map(list, zip(audioL, audioR)))
-                log_decode(start_time, f.tell(), decode_options)
-                w.write(stereo)
+                left_audio, right_audio = post_process(audioL, audioR, decoder.audioRate, decode_options)
+                stereo = prepare_stereo(left_audio, right_audio, noise_reduction, decode_options)
+                try:
+                    w.write(stereo)
+                except ValueError:
+                    pass
 
         elapsed_time = datetime.now() - start_time
         dt_string = elapsed_time.total_seconds()
@@ -445,7 +471,7 @@ def log_bias(decoder: HiFiDecode):
         )
 
 
-def main():
+def main() -> int:
     args = parser.parse_args()
 
     system = select_system(args)
@@ -461,6 +487,7 @@ def main():
         "nr_side_gain": args.NR_side_gain,
         "grc": args.GRC,
         "audio_rate": args.rate,
+        "gain": args.gain,
     }
 
     filename, outname, _, _ = get_basics(args)
@@ -470,7 +497,7 @@ def main():
                 "Existing decode files found, remove them or run command with --overwrite"
             )
             print("\t", outname)
-            sys.exit(1)
+            return 1
 
     print("Initializing ...")
     if decode_options["format"] == "vhs":
@@ -484,7 +511,7 @@ def main():
         decoder = HiFiDecode(decode_options)
         LCRef, RCRef = decoder.standard.LCarrierRef, decoder.standard.RCarrierRef
         if args.BG:
-            LCRef, RCRef = guess_bias(decoder, filename, decoder.blockSize)
+            LCRef, RCRef = guess_bias(decoder, filename, int(decoder.sample_rate))
             decoder.updateAFE(LCRef, RCRef)
 
         if args.threads > 1 and not args.GRC:
@@ -497,10 +524,12 @@ def main():
             )
         else:
             decode(decoder, filename, decode_options, outname)
+        print('Decode finished successfully')
+        return 0
     else:
         print("No sample rate specified")
-        sys.exit(0)
+        return 0
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())