Views
No views yet
@inproceedings{jacobellis2024learned,
title={Learned Compression for Compressed Learning},
author={Jacobellis, Dan and Yadwadkar, Neeraja J.},
booktitle={IEEE Data Compression Conference (DCC)},
note={Preprint}
year={2024},
url={http://danjacobellis.net/walloc}
}
pip install walloc PyWavelets pytorch-wavelets1import os
2import torch
3import json
4import matplotlib.pyplot as plt
5import numpy as np
6from types import SimpleNamespace
7from PIL import Image, ImageEnhance
8from IPython.display import display
9from torchvision.transforms import ToPILImage, PILToTensor
10from walloc import walloc
11from walloc.walloc import latent_to_pil, pil_to_latentwget https://hf.co/danjacobellis/walloc/resolve/main/RGB_16x.pthwget https://hf.co/danjacobellis/walloc/resolve/main/RGB_16x.json1device = "cpu"
2codec_config = SimpleNamespace(**json.load(open("RGB_16x.json")))
3checkpoint = torch.load("RGB_16x.pth",map_location="cpu",weights_only=False)
4codec = walloc.Codec2D(
5 channels = codec_config.channels,
6 J = codec_config.J,
7 Ne = codec_config.Ne,
8 Nd = codec_config.Nd,
9 latent_dim = codec_config.latent_dim,
10 latent_bits = codec_config.latent_bits,
11 lightweight_encode = codec_config.lightweight_encode
12)
13codec.load_state_dict(checkpoint['model_state_dict'])
14codec = codec.to(device)
15codec.eval();wget "https://r0k.us/graphics/kodak/kodak/kodim05.png"1img = Image.open("kodim05.png")
2img
codec.eval() is called, the latent is rounded to nearest integer.codec.train() is called, uniform noise is added instead of rounding.1with torch.no_grad():
2 codec.eval()
3 x = PILToTensor()(img).to(torch.float)
4 x = (x/255 - 0.5).unsqueeze(0).to(device)
5 x_hat, _, _ = codec(x)
6ToPILImage()(x_hat[0]+0.5)
1with torch.no_grad():
2 X = codec.wavelet_analysis(x,J=codec.J)
3 z = codec.encoder[0:2](X)
4 z_hat = codec.encoder[2](z)
5 X_hat = codec.decoder(z_hat)
6 x_rec = codec.wavelet_synthesis(X_hat,J=codec.J)
7print(f"dimensionality reduction: {x.numel()/z.numel()}×")dimensionality reduction: 16.0×1plt.figure(figsize=(5,3),dpi=150)
2plt.hist(
3 z.flatten().numpy(),
4 range=(-25,25),
5 bins=151,
6 density=True,
7);
8plt.title("Histogram of latents")
9plt.xlim([-25,25]);
1def scale_for_display(img, n_bits):
2 scale_factor = (2**8 - 1) / (2**n_bits - 1)
3 lut = [int(i * scale_factor) for i in range(2**n_bits)]
4 channels = img.split()
5 scaled_channels = [ch.point(lut * 2**(8-n_bits)) for ch in channels]
6 return Image.merge(img.mode, scaled_channels)1z_padded = torch.nn.functional.pad(z_hat, (0, 0, 0, 0, 0, 4))
2z_pil = latent_to_pil(z_padded,codec.latent_bits,1)
3display(scale_for_display(z_pil[0], codec.latent_bits))
4z_pil[0].save('latent.png')
5png = [Image.open("latent.png")]
6z_rec = pil_to_latent(png,16,codec.latent_bits,1)
7assert(z_rec.equal(z_padded))
8print("compression_ratio: ", x.numel()/os.path.getsize("latent.png"))
compression_ratio: 26.7299918426538561z_pil = latent_to_pil(z_hat,codec.latent_bits,3)
2display(scale_for_display(z_pil[0], codec.latent_bits))
3z_pil[0].save('latent.webp',lossless=True)
4webp = [Image.open("latent.webp")]
5z_rec = pil_to_latent(webp,12,codec.latent_bits,3)
6assert(z_rec.equal(z_hat))
7print("compression_ratio: ", x.numel()/os.path.getsize("latent.webp"))
compression_ratio: 28.8112543962485361z_padded = torch.nn.functional.pad(z_hat, (0, 0, 0, 0, 0, 4))
2z_pil = latent_to_pil(z_padded,codec.latent_bits,4)
3display(scale_for_display(z_pil[0], codec.latent_bits))
4z_pil[0].save('latent.tif',compression="tiff_adobe_deflate")
5tif = [Image.open("latent.tif")]
6z_rec = pil_to_latent(tif,16,codec.latent_bits,4)
7assert(z_rec.equal(z_padded))
8print("compression_ratio: ", x.numel()/os.path.getsize("latent.tif"))
compression_ratio: 21.040345307316381import io
2import os
3import torch
4import torchaudio
5import json
6import matplotlib.pyplot as plt
7from types import SimpleNamespace
8from PIL import Image
9from datasets import load_dataset
10from einops import rearrange
11from IPython.display import Audio
12from walloc import wallocwget https://hf.co/danjacobellis/walloc/resolve/main/stereo_5x.pthwget https://hf.co/danjacobellis/walloc/resolve/main/stereo_5x.json1codec_config = SimpleNamespace(**json.load(open("stereo_5x.json")))
2checkpoint = torch.load("stereo_5x.pth",map_location="cpu",weights_only=False)
3codec = walloc.Codec1D(
4 channels = codec_config.channels,
5 J = codec_config.J,
6 Ne = codec_config.Ne,
7 Nd = codec_config.Nd,
8 latent_dim = codec_config.latent_dim,
9 latent_bits = codec_config.latent_bits,
10 lightweight_encode = codec_config.lightweight_encode,
11 post_filter = codec_config.post_filter
12)
13codec.load_state_dict(checkpoint['model_state_dict'])
14codec.eval();/home/dan/g/lib/python3.12/site-packages/torch/nn/utils/weight_norm.py:143: FutureWarning: `torch.nn.utils.weight_norm` is deprecated in favor of `torch.nn.utils.parametrizations.weight_norm`.
WeightNorm.apply(module, name, dim)1MUSDB = load_dataset("danjacobellis/musdb_segments_val",split='validation')
2audio_buff = io.BytesIO(MUSDB[40]['audio_mix']['bytes'])
3x, fs = torchaudio.load(audio_buff,normalize=False)
4x = x.to(torch.float)
5x = x - x.mean()
6max_abs = x.abs().max()
7x = x / (max_abs + 1e-8)
8x = x/2Audio(x[:,:2**20],rate=44100)codec.eval() is called, the latent is rounded to nearest integer.codec.train() is called, uniform noise is added instead of rounding.1with torch.no_grad():
2 codec.eval()
3 x_hat, _, _ = codec(x.unsqueeze(0))Audio(x_hat[0,:,:2**20],rate=44100)1with torch.no_grad():
2 X = codec.wavelet_analysis(x.unsqueeze(0),J=codec.J)
3 z = codec.encoder[0:2](X)
4 z_hat = codec.encoder[2](z)
5 X_hat = codec.decoder(z_hat)
6 x_rec = codec.wavelet_synthesis(X_hat,J=codec.J)
7print(f"dimensionality reduction: {x.numel()/z.numel():.4g}×")dimensionality reduction: 4.74×1plt.figure(figsize=(5,3),dpi=150)
2plt.hist(
3 z.flatten().numpy(),
4 range=(-25,25),
5 bins=151,
6 density=True,
7);
8plt.title("Histogram of latents")
9plt.xlim([-25,25]);
1def pad(audio, p=2**16):
2 B,C,L = audio.shape
3 padding_size = (p - (L % p)) % p
4 if padding_size > 0:
5 audio = torch.nn.functional.pad(audio, (0, padding_size), mode='constant', value=0)
6 return audio
7with torch.no_grad():
8 L = x.shape[-1]
9 x_padded = pad(x.unsqueeze(0), 2**16)
10 X = codec.wavelet_analysis(x_padded,codec.J)
11 z = codec.encoder(X)
12 ℓ = z.shape[-1]
13 z = pad(z,128)
14 z = rearrange(z, 'b c (w h) -> b c w h', h=128).to("cpu")
15 webp = walloc.latent_to_pil(z,codec.latent_bits,3)[0]
16 buff = io.BytesIO()
17 webp.save(buff, format='WEBP', lossless=True)
18 webp_bytes = buff.getbuffer()1print("compression_ratio: ", x.numel()/len(webp_bytes))
2webpcompression_ratio: 9.83650170496386

1with torch.no_grad():
2 z_hat = walloc.pil_to_latent(
3 [Image.open(buff)],
4 codec.latent_dim,
5 codec.latent_bits,
6 3)
7 X_hat = codec.decoder(rearrange(z_hat, 'b c h w -> b c (h w)')[:,:,:ℓ])
8 x_hat = codec.wavelet_synthesis(X_hat,codec.J)
9 x_hat = codec.post(x_hat)
10 x_hat = codec.clamp(x_hat[0,:,:L])1start, end = 0, 1000
2plt.figure(figsize=(8, 3), dpi=180)
3plt.plot(x[0, start:end], alpha=0.5, c='b', label='Ch.1 (Uncompressed)')
4plt.plot(x_hat[0, start:end], alpha=0.5, c='g', label='Ch.1 (WaLLoC)')
5plt.plot(x[1, start:end], alpha=0.5, c='r', label='Ch.2 (Uncompressed)')
6plt.plot(x_hat[1, start:end], alpha=0.5, c='purple', label='Ch.2 (WaLLoC)')
7
8plt.xlim([400,1000])
9plt.ylim([-0.6,0.3])
10plt.legend(loc='lower center')
11plt.box(False)
12plt.xticks([])
13plt.yticks([]);
!jupyter nbconvert --to markdown README.ipynb[NbConvertApp] Converting notebook README.ipynb to markdown
[NbConvertApp] Support files will be in README_files/
[NbConvertApp] Writing 12900 bytes to README.md!sed -i 's|!\[png](README_files/\(README_[0-9]*_[0-9]*\.png\))||g' README.md!sed -i 's|src="README_files/\(README_[0-9]*\.wav\)"|src="https://huggingface.co/danjacobellis/walloc/resolve/main/README_files/\1"|g' README.md