architectures/srgan.py

import tensorflow as tf
import keras
from keras import layers

# import the generator from other modules (or create a custom one)
from .srrn import SRRN

@keras.saving.register_keras_serializable()
class DiscBlock(layers.Layer):
    """Basic discriminator block: Conv -> BatchNorm -> LeakyReLu"""
    def __init__(self, filters, stride=1, name="DiscBlock"):
        super().__init__(name=name)
        self.conv = layers.Conv2D(filters, 3, strides=stride, padding="same")
        self.bn = layers.BatchNormalization()
        self.act = layers.LeakyReLU(0.2)

    def call(self, x, training=False):
        x = self.conv(x)
        x = self.bn(x, training=training)
        return self.act(x)

@keras.saving.register_keras_serializable()
class Discriminator(keras.Model):
    """Discriminator for SRGAN"""
    def __init__(self, name="discriminator", **kwargs):
        super().__init__(name=name, **kwargs)
        
        # Discriminator blocks
        self.blocks = [
            DiscBlock(64, stride=1),
            DiscBlock(64, stride=2),

            DiscBlock(128, stride=1),
            DiscBlock(128, stride=2),

            DiscBlock(256, stride=1),
            DiscBlock(256, stride=2),

            DiscBlock(512, stride=1),
            DiscBlock(512, stride=2)
        ]

        # Final conv
        self.final_conv = layers.Conv2D(1, 3, padding="same")

    def call(self, inputs, training=False):
        x = inputs
        for block in self.blocks:
            x = block(x, training=training)
        x = self.final_conv(x)
        return x
    
    def get_config(self):
        config = super().get_config()
        return config

@keras.saving.register_keras_serializable()
class SRGAN(keras.Model):
    """Super-Resolution Generator Adversarial Network. Includes the generator + discriminator + custom train step"""
    def __init__(self, up_ratio, filters=64, num_blocks=8, lambda_adv=1e-3, name="srgan", **kwargs):
        super().__init__(name=name, **kwargs)
        self.up_ratio = up_ratio

        # Generator (SRResNet)
        self.generator = SRRN(up_ratio=up_ratio, filters=filters, num_blocks=num_blocks,)

        # Discriminator
        self.discriminator = Discriminator()

        # Loss weights
        self.lambda_adv = lambda_adv

        # Loss functions
        self.pixel_loss_fn = tf.keras.losses.MeanAbsoluteError()

        # Optimizers (they are assigned in the compile fase)
        self.gen_optimizer = None
        self.disc_optimizer = None

    def discriminator_loss(self, real_logits, fake_logits):
        # Uses Least Squares instead of Binary Cross-Entropy
        real_loss = tf.reduce_mean((real_logits - 1.0) ** 2)
        fake_loss = tf.reduce_mean((fake_logits - 0.0) ** 2)
        return real_loss + fake_loss

    def _generator_adversarial_loss(self, fake_logits):
        return tf.reduce_mean((fake_logits - 1.0) ** 2)

    def generator_total_loss(self, sr, hr, fake_logits):
        pixel_loss = self.pixel_loss_fn(hr, sr)
        adv_loss = self._generator_adversarial_loss(fake_logits)
        return pixel_loss + self.lambda_adv * adv_loss

    def compile(self, gen_optimizer, disc_optimizer, **kwargs):
        super().compile(**kwargs)

        self.gen_optimizer = gen_optimizer
        self.disc_optimizer = disc_optimizer

    def call(self, inputs):
        return self.generator(inputs)
    
    def get_config(self):
        config = super().get_config()
        config.update({"up_ratio": self.up_ratio, "lambda_adv": self.lambda_adv})
        return config

    @tf.function
    def train_step(self, data):
        lr, hr = data

        with tf.GradientTape() as gen_tape, tf.GradientTape() as disc_tape:
            # Generate SR image
            sr = self.generator(lr, training=True)

            # Discriminator predictions
            real_logits = self.discriminator(hr, training=True)
            fake_logits = self.discriminator(sr, training=True)

            # Losses
            disc_loss = self.discriminator_loss(real_logits, fake_logits)
            gen_loss = self.generator_total_loss(sr, hr, fake_logits)

        # Gradients
        gen_grads = gen_tape.gradient(gen_loss, self.generator.trainable_variables)
        disc_grads = disc_tape.gradient(disc_loss, self.discriminator.trainable_variables)

        # Apply gradients
        self.gen_optimizer.apply_gradients(zip(gen_grads, self.generator.trainable_variables))
        self.disc_optimizer.apply_gradients(zip(disc_grads, self.discriminator.trainable_variables))

        # PSNR y SSIM metri 
        psnr_val = tf.image.psnr(hr, sr, max_val=1.0)
        ssim_val = tf.image.ssim(hr, sr, max_val=1.0)

        return {
            "gen_loss": gen_loss,
            "disc_loss": disc_loss,
            "psnr": tf.reduce_mean(psnr_val),
            "ssim": tf.reduce_mean(ssim_val)
            }