Python: Batch an Interator

from itertools import islice

def batched(iterator, batch_size):
    """
    Takes an iterator and a batch-size, and returns an iterator that yields batches of the given size.
    
    Parameters:
    - iterator: An iterator from which to generate batches.
    - batch_size: An integer specifying the size of each batch.
    
    Yields:
    - Batches of elements from the input iterator, each batch being a list of elements up to the specified batch_size.
    """
    iterator = iter(iterator)  # Ensure it's an iterator
    while True:
        batch = list(islice(iterator, batch_size))
        if not batch:
            break
        yield batch

Usage:

for batch in batched(range(80), 9):
    print(batch)


d = {str(idx): idx for idx in range(10)}

for batch in batched(d.items(), 3):
    for key, value in batch:
        print(key, value)

Output

[0, 1, 2, 3, 4, 5, 6, 7, 8]
[9, 10, 11, 12, 13, 14, 15, 16, 17]
[18, 19, 20, 21, 22, 23, 24, 25, 26]
[27, 28, 29, 30, 31, 32, 33, 34, 35]
[36, 37, 38, 39, 40, 41, 42, 43, 44]
[45, 46, 47, 48, 49, 50, 51, 52, 53]
[54, 55, 56, 57, 58, 59, 60, 61, 62]
[63, 64, 65, 66, 67, 68, 69, 70, 71]
[72, 73, 74, 75, 76, 77, 78, 79]
0 0
1 1
2 2
3 3
4 4
5 5
6 6
7 7
8 8
9 9

Skill estimation and recommender systems

In a recommender system, you try to predict empty cells in the matrix between users and items.

In skill estimation you try to predict empty cell in the matrix between players and games.

Collaborative filtering also has a nice intuition.

pdict: Persistent dict (python)

from sqlitedict import SqliteDict
import sqlite3

def pdict(db_path:str|Path):
    if not isinstance(db_path, Path):
        db_path = Path(db_path)
    if not db_path.exists():
        db_path.parent.mkdir(exist_ok=True, parents=True)
        conn = sqlite3.connect(db_path)
        conn.execute('PRAGMA journal_mode=WAL;')
        conn.commit()
        conn.close()
    return SqliteDict(db_path, autocommit=True)

Generative Deep Learning, Chapter 1

 Challange

Create an image classifier with this architecture:

• Fully Connected Layer, size=200
• ReLU
• Fully Connected Layer, size=150
• ReLU
• Fully Connected Layer, size=10 (output)

Train it on CIFAR10 dataset (lr=0.001) works. Evaluate it. Plot a view images with the predicted and actual label.

import numpy as np
import torchvision
import torch
from torchvision import datasets, transforms
from torch.utils.data import DataLoader
from torch import nn, optim
from torch.nn import functional as F
import matplotlib.pyplot as plt



NUM_CLASSES = 10
BATCH_SIZE = 64


# load dataset

dataset = torchvision.datasets.CIFAR10('cifar10', download=True)

transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5,), (0.5,))
])

train_data = datasets.CIFAR10(root='./data', train=True, download=True, transform=transform)
test_data = datasets.CIFAR10(root='./data', train=False, download=True, transform=transform)

train_loader = DataLoader(train_data, batch_size=BATCH_SIZE, shuffle=True, num_workers=2)
test_loader = DataLoader(test_data, batch_size=BATCH_SIZE, shuffle=False, num_workers=2)


# define the architecture

class MLP(nn.Module):

  def __init__(self):
    super().__init__()
    self.fc1 = nn.Linear(32*32*3, 200)
    self.fc2 = nn.Linear(200,150)
    self.fc3 = nn.Linear(150,10)
    self.relu = nn.ReLU()

  def forward(self, x):
    out = x.view(x.size(0), -1)
    out = self.fc1(out)
    out = self.relu(out)
    out = self.fc2(out)
    out = self.relu(out)
    out = self.fc3(out)
    return out
    
    
# train it

from tqdm import tqdm


model = MLP()
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

num_epochs = 10

for epoch in range(num_epochs):
  running_loss = 0.0
  for images, labels in tqdm(train_loader):
    # Flatten the images for the MLP
    images = images.view(images.size(0), -1)
    optimizer.zero_grad()
    outputs = model(images)
    loss = criterion(outputs, labels)
    loss.backward()
    optimizer.step()
    running_loss += loss.item()
  print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {running_loss/len(train_loader):.4f}')
  
  
  
# evaluate it
 
model.eval()
correct = 0
total = 0
with torch.no_grad():
    for images, labels in tqdm(test_loader, desc='Evaluating'):
        images = images.view(images.size(0), -1)
        outputs = model(images)
        _, predicted = torch.max(outputs.data, 1)
        total += labels.size(0)
        correct += (predicted == labels).sum().item()
accuracy = 100 * correct / total
print(f'Accuracy of the network on the 10000 test images: {accuracy:.2f}%') # 53.19%


# predict the classes

import numpy as np

classes = np.array(['airplane', 'automobile', 'bird', 'cat', 'deer', 'dog', 'frog', 'horse', 'ship', 'truck'])

model.eval()

preds_list = []
actuals_list = []
with torch.no_grad():
    for images, labels in test_loader:
        images = images.view(images.size(0), -1)
        outputs = model(images)
        _, preds = torch.max(outputs, 1)
        preds_list.extend(preds.cpu().numpy())
        actuals_list.extend(labels.cpu().numpy())

preds_array = np.array(preds_list)
actuals_array = np.array(actuals_list)
preds_single = classes[preds_array]
actual_single = classes[actuals_array]


# visualize it

def get_image(idx, loader):
  for i, (images, _) in enumerate(loader):
    if i == idx // BATCH_SIZE:  # Find the batch containing the 54th image
        image = images[idx % BATCH_SIZE]  # Get the 54th image from the batch
        return image
  return None


n_to_show = 10
indices = np.random.choice(range(1000), n_to_show)

fig = plt.figure(figsize=(15, 3))
fig.subplots_adjust(hspace=0.4, wspace=0.4)

for i, idx in enumerate(indices):
  img = get_image(idx, test_loader)
  ax = fig.add_subplot(1, n_to_show, i+1)
  ax.axis('off')
  ax.text(0.5, -0.35, 'pred='+str(preds_single[idx]), fontsize=10, ha='center', transform=ax.transAxes)
  ax.text(0.5, -0.7, 'actu='+str(actual_single[idx]), fontsize=10, ha='center', transform=ax.transAxes)
  img = img.numpy().transpose(1,2,0)
  img = img * 0.5 + 0.5
  img = np.clip(img, 0, 1)
  ax.imshow(img)

Notes:

transforms.Normalize((0.5,), (0.5,))

transform standardizes the pixel values of the images in the dataset.

Normalization typically changes the range of pixel intensity values. The Normalize transform does this by applying the following transformation to each channel of the image:

$\text{normalized\_channel}=\frac{\text{channel}-\text{mean}}{\text{std}}$

For the CIFAR-10 dataset, images are in RGB format, meaning they have three channels (Red, Green, and Blue), each with pixel values in the range [0, 1] after applying transforms.ToTensor().

The Normalize transform here is called with (0.5,) for both the mean and std (standard deviation) parameters, but since CIFAR-10 images have three channels, and you provided a single value, it implicitly applies these values to all three channels.

The choice of (0.5, 0.5, 0.5) for both mean and standard deviation effectively shifts the input images' pixel value range from [0, 1] to [-1, 1] (after applying ToTensor() which scales images to [0, 1]). This is because subtracting 0.5 centers the pixel values around 0, and dividing by 0.5 scales them to a [-1, 1] range.

Operating in a [-1, 1] range can make the training process more stable and efficient for many models by ensuring that the inputs start in a more uniform and centered distribution. This is particularly beneficial for activation functions and optimization algorithms, making it easier to tune hyperparameters and achieve better performance.

Naive skill estimation

Naive Skill Estimation. As simple as it gets.

500 different seeds lead to this distribution:

Mean: 0.5333787274909965

Std: 0.15858765760754445

...................................

from dataclasses import dataclass
import random
import trueskill
from scipy.stats import spearmanr


# set seed 
random.seed(0)

def run(n_games):

    @dataclass
    class Player:
        name: str
        skill: int
        estimated_skill: trueskill.Rating = None

    n_players = 100
    skill_n = 100
    players = [Player(f"Player {i}", random.randint(1, skill_n)) for i in range(n_players)]
    ground_truth = [player for player in players]
    ground_truth.sort(key=lambda x: x.skill, reverse=True)


    players.sort(key=lambda x: x.skill, reverse=True)

    random.shuffle(players)


    @dataclass 
    class GameScore:
        player: Player
        score: float


    @dataclass 
    class Game:
        name: str
        players: list[Player]
        scores: list[GameScore]

    games:list[Game] = []

    for i in range(n_games):
        n_players_in_this_game = random.randint(2, n_players)
        game_players = random.sample(players, n_players_in_this_game)

        sigma = random.randint(1, int(skill_n * 0.2))

        scores = []
        for player in game_players:
            in_game_skill = player.skill + random.gauss(0, sigma)
            scores.append(GameScore(player, in_game_skill))

        scores.sort(key=lambda x: x.score, reverse=True)

        games.append(Game(f'Game {i}', game_players, scores))



    def get_score(ground_truth: list[Player], players: list[Player]) -> float:
        expected = [player.name for player in ground_truth]
        actual = [player.name for player in players]
        return spearmanr(expected, actual).correlation


    for player in players:
        player.estimated_skill = 0


    for game in games:

        for i, score in enumerate(game.scores):
            score.player.estimated_skill += len(game.scores) - i


    players.sort(key=lambda x: x.estimated_skill, reverse=True)


    return get_score(ground_truth, players)


from tqdm import tqdm
x = list(range(1, 2000, 5))
y = [run(n) for n in tqdm(x)]

import matplotlib.pyplot as plt
plt.plot(x, y)
plt.show()

Historgram created via:

from tqdm import tqdm
import numpy as np
import matplotlib.pyplot as plt

results = []
for i in tqdm(range(500)):
    random.seed(i)
    results.append(run(1000))

results = np.array(results)

# histogram with 50 bins
plt.hist(results, bins=30)
plt.show()

print(f"Mean: {results.mean()}")
print(f"Std: {results.std()}")

Develop tooling for jupyter notebooks

•  Create a setuptools project
• create a folder where you place your notebook
• start your notebook and do a `pip3 install -e ..`
• Paste this into a cell:

  %load_ext autoreload
  %autoreload 2
  

Intercept tqdm

For example to track the progress in a database or stream it via http

from tqdm import tqdm
import time

class TqdmContextManager:
    """
    A context manager to temporarily override the default tqdm class with a custom one.
    This allows for custom behavior of tqdm progress bars globally within its context.
    """
    def __init__(self):
        # Save a reference to the original tqdm class
        self.original_tqdm = tqdm

    def __enter__(self):
        # Override the global tqdm reference with our custom class upon entering the context
        global tqdm
        tqdm = CustomTqdm
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Restore the original tqdm class upon exiting the context
        global tqdm
        tqdm = self.original_tqdm


class CustomTqdm(tqdm):
    """
    A custom tqdm subclass that limits update frequency of progress output.
    """
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.last_print_time = time.time()  # Record the start time
        self.update_every = 0.1  # Limit to printing every 0.1 seconds (10 times per second)

    def update(self, n=1):
        """
        Override the update method to control the frequency of progress output.
        """
        current_time = time.time()
        if current_time - self.last_print_time >= self.update_every:
            # Print progress if enough time has passed since the last print
            print(f"Progress: {self.n}/{self.total}", end='\r', flush=True)
            self.last_print_time = current_time  # Update the last print time


# Example usage of the custom tqdm with a context manager
def some_function():
    """
    A sample function to demonstrate the use of the custom tqdm progress bar.
    """
    for i in tqdm(range(50)):
        time.sleep(0.05)  # Simulate work with a delay

# Use the custom tqdm progress bar within the context manager
with TqdmContextManager():
    some_function()