"""Core DeepDream algorithm, adapted from Chollet's Deep Learning with Python (Ch. 12) to use MobileNetV2 instead of InceptionV3 for lighter, faster inference on free-tier CPU hardware. """ import numpy as np import tensorflow as tf from tensorflow import keras from tensorflow.keras.applications import mobilenet_v2 # These 4 layers give a shallow -> deep progression of feature complexity, # mirroring the spread of the book's InceptionV3 "mixed4..mixed7" choices. # Verified MobileNetV2 layer names (also used by the official TF image # segmentation tutorial's encoder feature pyramid). LAYER_NAMES = [ "block_1_expand_relu", "block_3_expand_relu", "block_6_expand_relu", "block_13_expand_relu", ] MAX_DIMENSION = 512 _model = None _feature_extractor = None def get_feature_extractor(): """Lazily build the MobileNetV2 multi-output feature extractor once, and reuse it across requests -- rebuilding it per request would reload ImageNet weights every time, which is unnecessarily slow. """ global _model, _feature_extractor if _feature_extractor is None: _model = keras.applications.MobileNetV2(weights="imagenet", include_top=False) outputs_dict = {name: _model.get_layer(name).output for name in LAYER_NAMES} _feature_extractor = keras.Model(inputs=_model.inputs, outputs=outputs_dict) return _feature_extractor def compute_loss(input_image, layer_settings): feature_extractor = get_feature_extractor() features = feature_extractor(input_image) loss = tf.zeros(shape=()) for name in features.keys(): coeff = layer_settings[name] activation = features[name] # Only involve non-border pixels in the loss, to avoid border artifacts. # Normalize by activation size so layers with different spatial # resolutions contribute comparably to the total loss. scale = tf.cast(tf.size(activation[:, 2:-2, 2:-2, :]), "float32") loss += coeff * tf.reduce_sum(tf.square(activation[:, 2:-2, 2:-2, :])) / scale return loss def make_gradient_ascent_step(layer_settings): """Build a fresh @tf.function closure over this run's layer_settings. layer_settings changes per request (from the UI sliders), and a plain dict of Python floats can't be passed cleanly as a tf.function argument -- capturing it via closure and retracing once per request is simple and cheap relative to the cost of the gradient-ascent loop itself. """ @tf.function def gradient_ascent_step(image, learning_rate): with tf.GradientTape() as tape: tape.watch(image) loss = compute_loss(image, layer_settings) grads = tape.gradient(loss, image) grads = tf.math.l2_normalize(grads) image += learning_rate * grads return loss, image return gradient_ascent_step def gradient_ascent_loop(image, iterations, learning_rate, max_loss, layer_settings, octave_index=0, num_octaves=1, progress_cb=None): step_fn = make_gradient_ascent_step(layer_settings) for i in range(iterations): loss, image = step_fn(image, learning_rate) if progress_cb is not None: progress_cb(octave_index, num_octaves, i, iterations, float(loss)) if max_loss is not None and loss > max_loss: break return image def preprocess_image(pil_image, max_dimension=MAX_DIMENSION): """Resize (capping the longest side, keeping aspect ratio) and prepare a PIL image for MobileNetV2. Free-tier CPU inference is what we're optimizing for, so oversized uploads are silently downscaled rather than processed at full resolution. """ pil_image = pil_image.convert("RGB") width, height = pil_image.size longest_side = max(width, height) if longest_side > max_dimension: scale = max_dimension / longest_side pil_image = pil_image.resize((int(width * scale), int(height * scale))) img = keras.utils.img_to_array(pil_image) img = np.expand_dims(img, axis=0) img = mobilenet_v2.preprocess_input(img) return img, pil_image def deprocess_image(img): img = img.reshape((img.shape[1], img.shape[2], 3)) # MobileNetV2's preprocess_input scales pixels to [-1, 1]; undo that here. img /= 2.0 img += 0.5 img *= 255.0 img = np.clip(img, 0, 255).astype("uint8") return img def run_deepdream( pil_image, layer_settings, step_size=20.0, num_octave=3, octave_scale=1.4, iterations=12, max_loss=15.0, progress_cb=None, ): """Runs the multi-octave DeepDream gradient-ascent loop and returns a PIL image. `layer_settings` maps each name in LAYER_NAMES to its weight (float); `progress_cb(octave_index, num_octaves, iteration, iterations, loss)` is called after every gradient-ascent step, for UI progress reporting. """ original_img, _resized_pil = preprocess_image(pil_image) original_shape = original_img.shape[1:3] successive_shapes = [original_shape] for i in range(1, num_octave): shape = tuple(int(dim / (octave_scale ** i)) for dim in original_shape) successive_shapes.append(shape) successive_shapes = successive_shapes[::-1] shrunk_original_img = tf.image.resize(original_img, successive_shapes[0]) img = tf.identity(original_img) for octave_index, shape in enumerate(successive_shapes): img = tf.image.resize(img, shape) img = gradient_ascent_loop( img, iterations=iterations, learning_rate=step_size, max_loss=max_loss, layer_settings=layer_settings, octave_index=octave_index, num_octaves=len(successive_shapes), progress_cb=progress_cb, ) upscaled_shrunk_original_img = tf.image.resize(shrunk_original_img, shape) same_size_original = tf.image.resize(original_img, shape) lost_detail = same_size_original - upscaled_shrunk_original_img img += lost_detail shrunk_original_img = tf.image.resize(original_img, shape) result = deprocess_image(img.numpy()) return keras.utils.array_to_img(result)