import dearpygui.dearpygui as dpg import easygui import threading import libxheopus import os import subprocess import tempfile import wave import pyaudio import ctypes class App: def __init__(self): self.inputfilepath = None self.outputpath = None self.deinputfilepath = None self.deoutputpath = None self.derender = None self.delen = 0 self.depausepos = 0 self.decurrentplay = 0 self.deplay = False self.deopened = False self.deplayframeskip = 0 portaudio = pyaudio.PyAudio() self.decoder = None self.streamoutput = portaudio.open(format=pyaudio.paInt16, channels=2, rate=48000, output=True) def selectinputfile(self, sender, data): file_path = easygui.fileopenbox("Select Video/Audio File") dpg.set_value("inpathshow", f"input: {file_path}") self.inputfilepath = file_path def selectoutputpath(self, sender, data): file_path = easygui.diropenbox(title="Select Output Folder") dpg.set_value("outpathshow", f"output: {file_path}") self.outputpath = file_path def changeversionopus(self, sender, data): if data == "hev2": dpg.configure_item("opusframesize", items=["120", "100", "80", "60", "40", "20", "10", "5"]) else: dpg.configure_item("opusframesize", items=["60", "40", "20", "10", "5"]) if int(dpg.get_value("opusframesize")) > 60: dpg.configure_item("opusframesize", default_value="60") def changeprofileopus(self, sender, data): if data == "xHE-Opus v2": dpg.configure_item("opusstereomode", show=False) dpg.configure_item("opusbitrate", min_value=2.5, max_value=510, min_clamped=True, max_clamped=True, step_fast=1, default_value=64) else: dpg.configure_item("opusstereomode", show=True) dpg.configure_item("opusbitrate", min_value=5, max_value=1020, min_clamped=True, max_clamped=True, step_fast=1, default_value=64) def selectdeoutputpath(self, sender, data): file_path = easygui.diropenbox() dpg.set_value("deoutpathshow", f"output: {file_path}") self.deoutputpath = file_path if file_path != None or file_path != "" and self.deinputfilepath != None or self.deinputfilepath != "": dpg.configure_item("deplayconvert", show=True) else: dpg.configure_item("deplayconvert", show=False) def selectdeinputfile(self, sender, data): file_path = easygui.fileopenbox("Select Xopus File", filetypes=["*.xopus"], default="*.xopus") dpg.set_value("deinpathshow", f"input: {file_path}") self.deinputfilepath = file_path if file_path != None or file_path != "None" or file_path != "": if self.deopened: self.derender.close() self.derender = libxheopus.XopusReader(file_path) self.deopened = True self.stopaudio(None, None) self.decurrentplay = 0 self.depausepos = 0 self.deplayframeskip = 0 self.delen = 0 dpg.configure_item("deinfo", show=True) thread = threading.Thread(target=self.readmetadatathread, daemon=True) thread.start() else: dpg.configure_item("deinfo", show=False) dpg.configure_item("deplayconvert", show=False) def convert(self): signaltype = str(dpg.get_value("opussignaltype")).lower() profile = str(dpg.get_value("opusprofile")).strip().lower() stereomode = str(dpg.get_value("opusstereomode")).lower() if stereomode == "stereo l/r": stereomode = 1 elif stereomode == "stereo mid/side": stereomode = 2 else: stereomode = 2 if signaltype == "music": signalauto = False signalvoice = False elif signaltype == "voice": signalauto = False signalvoice = True else: signalauto = True signalvoice = False try: total = 0 current = 0 filename = os.path.splitext(os.path.basename(self.inputfilepath))[0] dpg.set_value("convertstatus", "init encoder...") print(profile) if profile == "xhe-opus v1": encoder = libxheopus.DualOpusEncoder(dpg.get_value("opusapp"), 48000, dpg.get_value("opusversion")) else: encoder = libxheopus.PSOpusEncoder(dpg.get_value("opusapp"), 48000, dpg.get_value("opusversion")) encoder.set_bitrate_mode(dpg.get_value("opusbitmode")) encoder.set_bandwidth(dpg.get_value("opusbandwidth")) encoder.set_bitrates(int(dpg.get_value("opusbitrate")*1000)) encoder.set_compression(dpg.get_value("opuscompression")) encoder.set_packet_loss(dpg.get_value("opuspacketloss")) if profile != "xhe-opus v2": encoder.set_stereo_mode(stereomode, dpg.get_value("opusenajoint")) encoder.set_feature(dpg.get_value("opusenapred"), False, dpg.get_value("opusenadtx")) encoder.enable_voice_mode(signalvoice, signalauto) desired_frame_size = encoder.set_frame_size(int(dpg.get_value("opusframesize"))) dpg.set_value("convertstatus", "init writer...") writer = libxheopus.XopusWriter(f"{self.outputpath}/{filename}.xopus", encoder) dpg.set_value("convertstatus", "converting to wav int16 with ffmpeg...") temp_dir = tempfile.mkdtemp() temp_wave_file = os.path.join(temp_dir, filename + ".wav") subprocess.run(["ffmpeg", "-i", self.inputfilepath, "-vn", "-acodec", "pcm_s16le", "-ac", "2", "-ar", "48000", temp_wave_file], check=True) dpg.set_value("convertstatus", "reading temp wav...") wav_file = wave.open(temp_wave_file, 'rb') while True: frames = wav_file.readframes(desired_frame_size) if not frames: break # Break the loop when all frames have been read total += len(frames) wav_file.rewind() dpg.set_value("convertstatus", "Encoding...") while True: frames = wav_file.readframes(desired_frame_size) if not frames: break # Break the loop when all frames have been read writer.write(frames) current += len(frames) dpg.set_value("convertprogbar", min(1.0, max(0.0, current / total))) # show percentage writer.close() wav_file.close() os.remove(temp_wave_file) except Exception as e: dpg.set_value("convertstatus", str(e)) raise e else: dpg.set_value("convertstatus", "Converted") dpg.configure_item("okconvertbutton", show=True) def startconvert(self, sender, data): dpg.configure_item("okconvertbutton", show=False) dpg.configure_item("convertingwindow", show=True) if self.outputpath is None or self.outputpath == "" or self.inputfilepath is None or self.inputfilepath == "": dpg.set_value("convertstatus", "Please check input file and output file") dpg.configure_item("okconvertbutton", show=True) else: thread = threading.Thread(target=self.convert, daemon=True) thread.start() def readmetadatathread(self): metadata = self.derender.readmetadata() dpg.set_value("deloudness", f'Loudness: {int(metadata["footer"]["contentloudness"])} DBFS') dpg.set_value("demetadata", f'Metadata: {metadata["header"]}') self.delen = metadata["footer"]["length"] def playaudiothread(self): self.decoder = libxheopus.xOpusDecoder() for data in self.derender.decode(self.decoder, True, self.depausepos): if self.deplay: if data != b"": self.streamoutput.write(data) else: self.deplayframeskip += 1 dpg.set_value("destatusfs", f"Frame Skip: {self.deplayframeskip}") self.decurrentplay += 1 try: dpg.set_value("deplayingprog", min(1.0, max(0.0, self.decurrentplay / self.delen))) except: dpg.set_value("deplayingprog", 0) else: if self.decurrentplay != 0: self.depausepos = self.decurrentplay break if dpg.get_value("deplayingprog") != 1: dpg.set_value("destatus", "Paused") else: dpg.set_value("destatus", "Stopped") def playpauseaudio(self, sender, data): dpg.configure_item("destatusfs", show=True) self.deplay = not self.deplay if self.deplay: if self.depausepos != 0: self.decurrentplay = self.depausepos dpg.set_value("destatus", "Playing") dpg.configure_item("deplaybutton", label="Pause") thread = threading.Thread(target=self.playaudiothread, daemon=True) thread.start() else: dpg.set_value("destatus", "Paused") dpg.configure_item("deplaybutton", label="Play") def stopaudio(self, sender, data): dpg.configure_item("destatusfs", show=False) dpg.set_value("destatusfs", "Frame Skip: 0") self.decurrentplay = 0 self.depausepos = 0 self.deplayframeskip = 0 self.deplay = False dpg.set_value("deplayingprog", 0) dpg.set_value("destatus", "Stopped") dpg.configure_item("deplaybutton", label="Play") def deconvertthread(self): outwav = wave.open(self.deoutputpath + "/" + os.path.splitext(os.path.basename(self.deinputfilepath))[0] + ".wav", "w") # Set the parameters of the WAV file outwav.setnchannels(2) # Stereo outwav.setsampwidth(2) # 2 bytes (16 bits) per sample outwav.setframerate(48000) self.decoder = libxheopus.xOpusDecoder() for data in self.derender.decode(self.decoder, True, self.depausepos): self.decurrentplay += 1 # Write the audio data to the file if data != b"": outwav.writeframes(data) else: self.deplayframeskip += 1 dpg.set_value("destatusfs", f"Frame Skip: {self.deplayframeskip}") dpg.set_value("deplayingprog", min(1.0, max(0.0, self.decurrentplay / self.delen))) outwav.close() self.decurrentplay = 0 dpg.set_value("destatus", "Converted") dpg.configure_item("destatusfs", show=False) def startdeconvert(self, sender, data): self.stopaudio(None, None) dpg.configure_item("destatusfs", show=True) dpg.set_value("destatus", "Converting") thread = threading.Thread(target=self.deconvertthread, daemon=True) thread.start() def window(self): with dpg.window(label="Encoder", width=420, no_close=True): dpg.add_text("input:", tag="inpathshow") dpg.add_text("output:", tag="outpathshow") dpg.add_button(label="Select Input File", callback=self.selectinputfile) dpg.add_button(label="Select Output Path", callback=self.selectoutputpath) dpg.add_combo(["xHE-Opus v1", "xHE-Opus v2"], label="Profile", default_value="xHE-Opus v1", tag="opusprofile", callback=self.changeprofileopus) dpg.add_combo(["hev2", "exper", "stable", "old"], label="Version", default_value="hev2", tag="opusversion", callback=self.changeversionopus) dpg.add_combo(["120", "100", "80", "60", "40", "20", "10", "5"], label="Frame Size (ms)", tag="opusframesize", default_value="120") dpg.add_combo(["voip", "audio", "restricted_lowdelay"], label="Application", default_value="restricted_lowdelay", tag="opusapp") dpg.add_combo(["VBR", "CVBR", "CBR"], label="Bitrate Mode", default_value="CVBR", tag="opusbitmode") dpg.add_combo(["auto", "fullband", "superwideband", "wideband", "mediumband", "narrowband"], label="Bandwidth", tag="opusbandwidth", default_value="fullband") dpg.add_combo(["Stereo L/R", "Stereo Mid/Side"], label="Stereo Mode", tag="opusstereomode", default_value="Stereo L/R") dpg.add_combo(["Auto", "Voice", "Music"], label="Signal Type", tag="opussignaltype", default_value="Auto") dpg.add_input_float(label="Bitrates", min_value=5, max_value=1020, min_clamped=True, max_clamped=True, step_fast=1, default_value=64, tag="opusbitrate") dpg.add_input_int(label="Compression Level", max_clamped=True, min_clamped=True, min_value=0, max_value=10, default_value=10, tag="opuscompression") dpg.add_input_int(label="Packet Loss", max_clamped=True, min_clamped=True, min_value=0, max_value=100, default_value=0, tag="opuspacketloss") dpg.add_checkbox(label="Prediction", tag="opusenapred") dpg.add_checkbox(label="DTX", tag="opusenadtx") dpg.add_checkbox(label="Auto Mono (Mid/Side Encoding)", tag="opusenajoint") dpg.add_button(label="Convert", callback=self.startconvert) with dpg.window(label="converting", show=False, tag="convertingwindow", modal=True, no_resize=True, no_move=True, no_title_bar=True, width=320): dpg.add_text("converting...", tag="convertstatus") dpg.add_progress_bar(tag="convertprogbar") dpg.add_button(label="OK", callback=lambda: dpg.configure_item("convertingwindow", show=False), tag="okconvertbutton", show=False) with dpg.window(label="Player/Decoder", height=320, width=420, pos=(500, 0), no_close=True): dpg.add_text("input:", tag="deinpathshow") dpg.add_text("output:", tag="deoutpathshow") dpg.add_button(label="Select Input File", callback=self.selectdeinputfile) dpg.add_button(label="Select Output Path", callback=self.selectdeoutputpath) with dpg.group(tag="deinfo", show=False): dpg.add_text("Loudness: ? DBFS", tag="deloudness") dpg.add_text("Metadata: ?", tag="demetadata", wrap=400) dpg.add_progress_bar(tag="deplayingprog") dpg.add_text("Frame Skip: 0", tag="destatusfs", show=False) dpg.add_text("Stopped", tag="destatus") dpg.add_button(label="Play", tag="deplaybutton", callback=self.playpauseaudio) dpg.add_button(label="Stop", tag="destopbutton", callback=self.stopaudio) dpg.add_button(label="Convert", tag="deplayconvert", show=False, callback=self.startdeconvert) def init(self): dpg.create_context() dpg.create_viewport(title='xHE-Opus GUI', width=1280, height=720) # set viewport window dpg.setup_dearpygui() # -------------- add code here -------------- self.window() with dpg.texture_registry(): width, height, channels, data = dpg.load_image("xHE-Opus.png") dpg.add_static_texture(width=512, height=192, default_value=data, tag="app_logo_background") with dpg.window(no_background=True, no_title_bar=True, no_move=True, no_resize=True, tag="backgroundviewportlogo"): dpg.add_image("app_logo_background") dpg.add_text("ThaiSDR Solutions", pos=(230, 230)) # ------------------------------------------- dpg.show_viewport() ctypes.windll.user32.ShowWindow(ctypes.windll.kernel32.GetConsoleWindow(), 0) while dpg.is_dearpygui_running(): self.render() dpg.render_dearpygui_frame() dpg.destroy_context() def render(self): # insert here any code you would like to run in the render loop # you can manually stop by using stop_dearpygui() or self.exit() dpg.configure_item("backgroundviewportlogo", pos=(dpg.get_viewport_width() - 550, dpg.get_viewport_height() - 300)) def exit(self): dpg.destroy_context() app = App() app.init()