wavmark
Where is the problem? I added watermarked audio, and as long as I edit it, the watermark will definitely be lost
import time import pika import torch import numpy as np import soundfile as sf import wavmark import json
def log(message): print(f"[LOG] {message}")
def int_to_32bit_binary_array(number): log("Start converting integer to a 32-bit binary array")
Convert integer to a 32-bit unsigned integer's binary string, then split into single characters and convert to a list of integers
binary_str = format(number & 0xFFFFFFFF, '032b') # Convert to a 32-bit binary string log(f"Integer successfully converted to 32-bit binary string: {binary_str}") result = np.array(list(binary_str), dtype=int) # Convert to numpy array log("32-bit binary string has been converted to a numpy array") return result
def binary_array_to_int(binary_array): """Convert a 32-bit binary array to an integer""" log("Starting conversion from a 32-bit binary array to an integer")
Convert binary array to a binary number in string form
binary_str = ''.join([str(bit) for bit in binary_array])
Convert binary string to an integer
result = int(binary_str, 2) log(f"Conversion complete: {result}") return result
def encode_signal(model, device, signal, message_tensor): log("Starting encoding process") signal_tensor = torch.FloatTensor(signal).to(device)[None] log("Audio signal converted to tensor and transferred to computing device") encoded_signal = model.encode(signal_tensor, message_tensor) log("Encoding completed, obtained audio signal embedded with message") return encoded_signal.detach().cpu().numpy().squeeze()
def decode_signal(model, device, encoded_signal): log("Starting decoding process") encoded_signal_tensor = torch.FloatTensor(encoded_signal).to(device).unsqueeze(0) log("Encoded signal converted to tensor and transferred to computing device") decoded_message = (model.decode(encoded_signal_tensor) >= 0.5).int().detach().cpu().numpy().squeeze() log("Decoding completed, original message recovered") return decoded_message
def encode_audio_with_watermark(input_wav_path, number_message): log("Starting audio encoding process")
device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu') model = wavmark.load_model().to(device) message_npy = int_to_32bit_binary_array(number_message)
signal, _ = sf.read(input_wav_path) # Not concerned about the sample rate trunck = signal[0:16000]
with torch.no_grad(): message_tensor = torch.FloatTensor(message_npy).to(device)[None] encoded_signal = encode_signal(model, device, trunck, message_tensor)
log("Audio encoding completed") return encoded_signal # Return encoded audio signal
def decode_audio_from_file(audio_wav_path): log("Starting audio decoding process")
device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu') model = wavmark.load_model().to(device)
Read data from audio file
signal, _ = sf.read(audio_wav_path) # Not concerned about the sample rate trunck = signal[0:16000] # Assuming we only process the first 16000 samples
with torch.no_grad(): decoded_message = decode_signal(model, device, trunck)
log("Audio decoding completed") return decoded_message
Example usage
log("Example call to encode_audio_with_watermark function") input_wav_path = "input.mp3" # Input audio file path number_message = 123456789 # Digital information you want to encode encoded_signal = encode_audio_with_watermark(input_wav_path, number_message)
Assuming we save the encoded signal as a new audio file (if needed)
sf.write("encoded_output.wav", encoded_signal, samplerate=16000) # Save the encoded audio
log("Example call to decode_audio_from_file function") decoded_message = decode_audio_from_file("encoded_output5.mp3")
Convert the decoded binary array back to an integer
recovered_number_message = binary_array_to_int(decoded_message)
print(f"Recovered watermark (integer): {recovered_number_message}")
