Convert an integer (0 to 65,535) to a 16-bit binary message:
def int_to_message(value: int, batch_size: int = 1) -> torch.Tensor: """ Convert an integer to a 16-bit binary message. Args: value: Integer between 0 and 65,535 batch_size: Number of copies in the batch Returns: Binary tensor of shape [batch_size, 16] """ assert 0 <= value < 2**16, f"Value must be between 0 and {2**16-1}" # Convert to binary string and pad to 16 bits binary_str = format(value, '016b') # Convert to tensor bits = [int(b) for b in binary_str] message = torch.tensor([bits], dtype=torch.int32) # Repeat for batch if batch_size > 1: message = message.repeat(batch_size, 1) return message# Example: Encode model version IDmodel_version = 42message = int_to_message(model_version)print(f"Model version {model_version}: {message}")# Output: Model version 42: tensor([[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 1, 0, 1, 0]])
# Get detailed detection resultsresult, decoded_message = detector(watermarked_audio)# result shape: [batch, 2, frames]# result[:, 1, :] contains watermark probabilities per frame# decoded_message shape: [batch, 16]# Each value is the probability of that bit being 1print(f"Message probabilities: {decoded_message}")# Convert to binary (threshold at 0.5)binary_message = (decoded_message > 0.5).int()print(f"Binary message: {binary_message}")
If the detector does not find a watermark, the decoded message will be random. Always check the detection probability before trusting the message.
# Test detection on clean (non-watermarked) audioclean_audio = torch.randn(1, 1, 48000)detect_prob, message = detector.detect_watermark(clean_audio)print(f"Detection probability: {detect_prob.item():.3f}") # ~0.0print(f"Message: {message}") # Random values!
When no watermark is detected (low detection probability), the message will be random. Always validate the detection probability before using the decoded message.
Use the same message across all chunks in streaming mode:
model = AudioSeal.load_generator("audioseal_wm_streaming")model.eval()# Create a consistent message for all chunkssecret_message = int_to_message(123)audio_chunks = [...] # Your chunkswatermarked_chunks = []with model.streaming(batch_size=1): for chunk in audio_chunks: # Use same message for all chunks watermarked = model( chunk, message=secret_message, # Same message alpha=1.0 ) watermarked_chunks.append(watermarked)full_audio = torch.cat(watermarked_chunks, dim=-1)# Decode from full audiodetect_prob, decoded = detector.detect_watermark(full_audio)print(f"Decoded ID: {message_to_int(decoded)}")