USER
```python
import os
import sys
import threading
import logging
from io import BytesIO
from typing import Optional, Union
import gi
import requests
from PIL import Image
from fastai.vision.all import load_learner, PILImage, Learner, vision_learner, untar_data, URLs, ImageDataLoaders
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, GdkPixbuf, Gdk, GLib
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s:%(name)s: %(message)s',
handlers=[
logging.FileHandler("image_predictor_app.log"),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
class ImagePredictorApp(Gtk.Window):
def __init__(self):
super().__init__(title="Image Predictor with Model Training")
self.set_border_width(10)
self.set_default_size(1200, 800) # Increased size to accommodate training settings
# Initialize model attribute
self.model: Optional[Learner] = None
# Initialize UI components
self.init_ui()
def init_ui(self):
"""Initialize and arrange all UI components."""
# Main container using Paned for resizable sections
paned = Gtk.Paned.new(Gtk.Orientation.HORIZONTAL)
self.add(paned)
# --- Left Pane: Controls and Training Settings ---
left_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
paned.pack1(left_box, resize=True, shrink=False)
# --- Controls Section ---
controls_frame = Gtk.Frame(label="Prediction Controls")
controls_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
controls_box.set_margin_bottom(10)
controls_box.set_margin_top(10)
controls_box.set_margin_start(10)
controls_box.set_margin_end(10)
controls_frame.add(controls_box)
left_box.pack_start(controls_frame, False, False, 0)
# Load Model Button
self.load_model_button = Gtk.Button(label="Load Model (.pkl / .pth)")
self.load_model_button.connect("clicked", self.on_load_model)
controls_box.pack_start(self.load_model_button, False, False, 0)
# Image Selection
image_select_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=5)
self.image_entry = Gtk.Entry()
self.image_entry.set_placeholder_text("Enter image path or URL")
self.image_entry.connect("activate", self.on_enter_image_path)
self.image_button = Gtk.Button(label="Browse Image")
self.image_button.connect("clicked", self.on_select_image)
image_select_box.pack_start(self.image_entry, True, True, 0)
image_select_box.pack_start(self.image_button, False, False, 0)
controls_box.pack_start(image_select_box, False, False, 0)
# Threshold Slider
threshold_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=5)
self.threshold_label = Gtk.Label(label="Prediction Threshold: 50%")
self.threshold_label.set_xalign(0)
threshold_box.pack_start(self.threshold_label, False, False, 0)
self.threshold_adjustment = Gtk.Adjustment(value=50, lower=0, upper=100, step_increment=1, page_increment=10)
self.threshold_slider = Gtk.Scale(orientation=Gtk.Orientation.HORIZONTAL, adjustment=self.threshold_adjustment)
self.threshold_slider.set_digits(0)
self.threshold_slider.connect("value-changed", self.on_threshold_changed)
threshold_box.pack_start(self.threshold_slider, False, False, 0)
controls_box.pack_start(threshold_box, False, False, 0)
# Predict Button
self.predict_button = Gtk.Button(label="Predict")
self.predict_button.connect("clicked", self.on_predict)
self.predict_button.set_sensitive(False)
controls_box.pack_start(self.predict_button, False, False, 0)
# Output Display
self.output_label = Gtk.Label(label="Output will be displayed here.")
self.output_label.set_line_wrap(True)
self.output_label.set_xalign(0)
self.output_label.set_justify(Gtk.Justification.LEFT)
controls_box.pack_start(self.output_label, False, False, 0)
# Progress Bar
self.progress_bar = Gtk.ProgressBar()
self.progress_bar.set_show_text(True)
self.progress_bar.set_visible(False)
controls_box.pack_start(self.progress_bar, False, False, 0)
# --- Training Settings Section ---
training_frame = Gtk.Frame(label="Model Training Settings")
training_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
training_box.set_margin_bottom(10)
training_box.set_margin_top(10)
training_box.set_margin_start(10)
training_box.set_margin_end(10)
training_frame.add(training_box)
left_box.pack_start(training_frame, False, False, 0)
# Dataset Selection
dataset_select_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=5)
self.dataset_entry = Gtk.Entry()
self.dataset_entry.set_placeholder_text("Enter dataset path or URL")
self.dataset_button = Gtk.Button(label="Browse Dataset")
self.dataset_button.connect("clicked", self.on_select_dataset)
dataset_select_box.pack_start(self.dataset_entry, True, True, 0)
dataset_select_box.pack_start(self.dataset_button, False, False, 0)
training_box.pack_start(dataset_select_box, False, False, 0)
# Model Architecture Selection
architecture_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=5)
architecture_label = Gtk.Label(label="Model Architecture:")
architecture_label.set_xalign(0)
architecture_box.pack_start(architecture_label, False, False, 0)
self.architecture_combo = Gtk.ComboBoxText()
for arch in ["resnet18", "resnet34", "resnet50", "resnet101", "resnet152"]:
self.architecture_combo.append_text(arch)
self.architecture_combo.set_active(1) # Default to resnet34
architecture_box.pack_start(self.architecture_combo, False, False, 0)
training_box.pack_start(architecture_box, False, False, 0)
# Hyperparameters
hyperparams_grid = Gtk.Grid(column_spacing=10, row_spacing=10)
# Learning Rate
lr_label = Gtk.Label(label="Learning Rate:")
lr_label.set_halign(Gtk.Align.END)
hyperparams_grid.attach(lr_label, 0, 0, 1, 1)
self.lr_entry = Gtk.Entry()
self.lr_entry.set_text("1e-3")
hyperparams_grid.attach(self.lr_entry, 1, 0, 1, 1)
# Batch Size
bs_label = Gtk.Label(label="Batch Size:")
bs_label.set_halign(Gtk.Align.END)
hyperparams_grid.attach(bs_label, 0, 1, 1, 1)
self.bs_entry = Gtk.Entry()
self.bs_entry.set_text("32")
hyperparams_grid.attach(self.bs_entry, 1, 1, 1, 1)
# Number of Epochs
epochs_label = Gtk.Label(label="Number of Epochs:")
epochs_label.set_halign(Gtk.Align.END)
hyperparams_grid.attach(epochs_label, 0, 2, 1, 1)
self.epochs_entry = Gtk.Entry()
self.epochs_entry.set_text("10")
hyperparams_grid.attach(self.epochs_entry, 1, 2, 1, 1)
# Save Model Name
save_label = Gtk.Label(label="Save Model As:")
save_label.set_halign(Gtk.Align.END)
hyperparams_grid.attach(save_label, 0, 3, 1, 1)
self.save_entry = Gtk.Entry()
self.save_entry.set_text("trained_model.pkl")
hyperparams_grid.attach(self.save_entry, 1, 3, 1, 1)
training_box.pack_start(hyperparams_grid, False, False, 0)
# Train Model Button
self.train_button = Gtk.Button(label="Train Model")
self.train_button.connect("clicked", self.on_train_model)
training_box.pack_start(self.train_button, False, False, 0)
# Training Output Display
self.training_output = Gtk.TextView()
self.training_output.set_editable(False)
self.training_output.set_wrap_mode(Gtk.WrapMode.WORD)
training_output_scrolled = Gtk.ScrolledWindow()
training_output_scrolled.set_vexpand(True)
training_output_scrolled.add(self.training_output)
training_box.pack_start(training_output_scrolled, True, True, 0)
# --- Right Pane: Image Preview ---
preview_frame = Gtk.Frame(label="Image Preview")
preview_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=5)
preview_box.set_margin_bottom(10)
preview_box.set_margin_top(10)
preview_box.set_margin_start(10)
preview_box.set_margin_end(10)
preview_frame.add(preview_box)
self.image_preview = Gtk.Image()
self.image_preview.set_valign(Gtk.Align.START)
preview_box.pack_start(self.image_preview, True, True, 0)
# Drag and Drop Support for Image Preview
self.image_preview.drag_dest_set(Gtk.DestDefaults.ALL, [], Gdk.DragAction.COPY) # Corrected this line
self.image_preview.drag_dest_add_uri_targets()
self.image_preview.connect("drag-data-received", self.on_drag_data_received)
paned.pack2(preview_frame, resize=True, shrink=False)
# ------------------ Prediction Controls ------------------
def on_drag_data_received(self, widget, drag_context, x, y, data, info, time):
"""Handle drag-and-drop of image files or URLs."""
uris = data.get_uris()
if uris:
uri = uris[0]
path_or_url = self.parse_uri(uri)
if path_or_url:
self.image_entry.set_text(path_or_url)
self.display_image(path_or_url)
Gtk.drag_finish(drag_context, True, False, time)
@staticmethod
def parse_uri(uri: str) -> Optional[str]:
"""Parse URI to get local path or URL."""
if uri.startswith("file://"):
return uri[7:]
elif uri.startswith(("http://", "https://", "ftp://")):
return uri
else:
return None
def on_enter_image_path(self, widget):
"""Handle pressing Enter in the image entry."""
image_input = self.image_entry.get_text().strip()
if image_input:
self.display_image(image_input)
def on_load_model(self, button: Gtk.Button):
"""Handle the Load Model button click."""
dialog = Gtk.FileChooserDialog(
title="Select a Model File",
parent=self,
action=Gtk.FileChooserAction.OPEN
)
dialog.add_buttons(
Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL,
Gtk.STOCK_OPEN, Gtk.ResponseType.OK
)
dialog.set_filter(self.create_model_filter())
response = dialog.run()
if response == Gtk.ResponseType.OK:
model_path = dialog.get_filename()
logger.info(f"Selected model file: {model_path}")
self.load_model_async(model_path)
dialog.destroy()
@staticmethod
def create_model_filter() -> Gtk.FileFilter:
"""Create a file filter for model files."""
file_filter = Gtk.FileFilter()
file_filter.set_name("Model Files (*.pkl, *.pth)")
file_filter.add_pattern("*.pkl")
file_filter.add_pattern("*.pth")
return file_filter
def load_model_async(self, model_path: str):
"""Load the model in a separate thread to keep UI responsive."""
self.set_controls_sensitive(False)
self.show_progress("Loading model...")
thread = threading.Thread(target=self.load_model, args=(model_path,), daemon=True)
thread.start()
def load_model(self, model_path: str):
"""Load the machine learning model from the specified path."""
try:
learner = load_learner(model_path)
GLib.idle_add(self.on_model_loaded, learner, model_path)
logger.info(f"Model loaded successfully from: {model_path}")
except Exception as e:
logger.error(f"Failed to load model: {e}", exc_info=True)
GLib.idle_add(self.on_model_load_failed, str(e))
def on_model_loaded(self, learner: Learner, model_path: str):
"""Callback when the model is successfully loaded."""
self.model = learner
self.predict_button.set_sensitive(True)
self.output_label.set_markup(f"<span foreground='green'>Model loaded successfully from:</span>\n{model_path}")
self.hide_progress()
self.set_controls_sensitive(True)
def on_model_load_failed(self, error_message: str):
"""Callback when the model fails to load."""
self.model = None
self.predict_button.set_sensitive(False)
self.output_label.set_markup(f"<span foreground='red'>Failed to load model:</span>\n{error_message}")
self.hide_progress()
self.set_controls_sensitive(True)
def on_select_image(self, button: Gtk.Button):
"""Handle the Select Image button click."""
dialog = Gtk.FileChooserDialog(
title="Select an Image",
parent=self,
action=Gtk.FileChooserAction.OPEN
)
dialog.add_buttons(
Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL,
Gtk.STOCK_OPEN, Gtk.ResponseType.OK
)
dialog.set_filter(self.create_image_filter())
response = dialog.run()
if response == Gtk.ResponseType.OK:
image_path = dialog.get_filename()
self.image_entry.set_text(image_path)
self.display_image(image_path)
logger.info(f"Selected image file: {image_path}")
dialog.destroy()
@staticmethod
def create_image_filter() -> Gtk.FileFilter:
"""Create a file filter for image files."""
img_filter = Gtk.FileFilter()
img_filter.set_name("Image Files (*.png, *.jpg, *.jpeg, *.webp)")
img_filter.add_pattern("*.png")
img_filter.add_pattern("*.jpg")
img_filter.add_pattern("*.jpeg")
img_filter.add_pattern("*.webp")
img_filter.add_mime_type("image/png")
img_filter.add_mime_type("image/jpeg")
img_filter.add_mime_type("image/webp")
return img_filter
def on_threshold_changed(self, scale: Gtk.Scale):
"""Update the threshold label when the slider value changes."""
value = self.threshold_slider.get_value()
self.threshold_label.set_text(f"Prediction Threshold: {int(value)}%")
def on_predict(self, button: Gtk.Button):
"""Handle the Predict button click."""
if not self.model:
self.output_label.set_markup("<span foreground='red'>Please load a model first.</span>")
return
image_input = self.image_entry.get_text().strip()
if not image_input:
self.output_label.set_markup("<span foreground='red'>Please provide an image path or URL.</span>")
return
self.set_controls_sensitive(False)
self.show_progress("Performing prediction...")
thread = threading.Thread(target=self.predict_image, args=(image_input,), daemon=True)
thread.start()
def predict_image(self, image_input: str):
"""Load the image and perform prediction in a separate thread."""
try:
image = self.load_image(image_input)
GLib.idle_add(self.display_image, image_input)
pred, pred_idx, outputs = self.model.predict(image)
probabilities = outputs.numpy() * 100 # Convert to percentage
classes = self.model.dls.vocab # Get class names
threshold = self.threshold_slider.get_value() # Get the threshold value
# Prepare formatted prediction results
output_lines = ["<b>Predictions:</b>"]
for class_name, probability in zip(classes, probabilities):
if probability >= threshold:
line = f"• {class_name}: {probability:.2f}%"
output_lines.append(line)
output_lines.append(f"\n<b>Overall Prediction:</b> {pred}")
# Update the output label using Pango Markup for better formatting
markup_text = "\n".join(output_lines)
GLib.idle_add(self.update_output, markup_text)
logger.info(f"Prediction successful for image: {image_input}")
except Exception as e:
logger.error(f"Error during prediction: {e}", exc_info=True)
GLib.idle_add(self.update_output, f"<span foreground='red'>Error during prediction:</span>\n{e}")
finally:
GLib.idle_add(self.hide_progress)
GLib.idle_add(self.set_controls_sensitive, True)
def load_image(self, image_input: str) -> PILImage:
"""
Load an image from a local path or a URL.
Args:
image_input (str): The path or URL of the image.
Returns:
PILImage: The loaded image.
"""
logger.debug(f"Loading image from: {image_input}")
if image_input.startswith(('http://', 'https://', 'ftp://')):
response = requests.get(image_input, timeout=10)
response.raise_for_status()
image = PILImage.create(BytesIO(response.content))
logger.debug("Image loaded from URL.")
return image
elif os.path.isfile(image_input):
image = PILImage.create(image_input)
logger.debug("Image loaded from local file.")
return image
else:
logger.error("Invalid image path or URL.")
raise ValueError("Invalid image path or URL.")
def display_image(self, image_input: str):
"""
Display the selected image in the image preview widget.
Args:
image_input (str): The path or URL of the image.
"""
try:
if image_input.startswith(('http://', 'https://', 'ftp://')):
response = requests.get(image_input, timeout=10)
response.raise_for_status()
loader = GdkPixbuf.PixbufLoader.new()
loader.write(response.content)
loader.close()
pixbuf = loader.get_pixbuf()
else:
pixbuf = GdkPixbuf.Pixbuf.new_from_file(image_input)
# Calculate scaling while preserving aspect ratio
max_width, max_height = 600, 600
width = pixbuf.get_width()
height = pixbuf.get_height()
scaling_factor = min(max_width / width, max_height / height, 1)
new_width = int(width * scaling_factor)
new_height = int(height * scaling_factor)
scaled_pixbuf = pixbuf.scale_simple(new_width, new_height, GdkPixbuf.InterpType.BILINEAR)
self.image_preview.set_from_pixbuf(scaled_pixbuf)
logger.debug("Image displayed successfully.")
except Exception as e:
logger.error(f"Failed to display image: {e}", exc_info=True)
self.image_preview.set_from_icon_name("image-missing", Gtk.IconSize.DIALOG)
self.update_output(f"<span foreground='red'>Failed to display image preview:</span>\n{e}")
def update_output(self, text: str):
"""Update the output label with the given text."""
self.output_label.set_markup(text)
def set_controls_sensitive(self, sensitive: bool):
"""Enable or disable controls based on the sensitive flag."""
self.load_model_button.set_sensitive(sensitive)
self.image_button.set_sensitive(sensitive)
self.image_entry.set_sensitive(sensitive)
self.threshold_slider.set_sensitive(sensitive)
self.predict_button.set_sensitive(sensitive and self.model is not None)
self.train_button.set_sensitive(sensitive)
self.dataset_button.set_sensitive(sensitive)
self.dataset_entry.set_sensitive(sensitive)
self.architecture_combo.set_sensitive(sensitive)
self.lr_entry.set_sensitive(sensitive)
self.bs_entry.set_sensitive(sensitive)
self.epochs_entry.set_sensitive(sensitive)
self.save_entry.set_sensitive(sensitive)
def show_progress(self, message: str):
"""Display the progress bar with a message."""
self.progress_bar.set_visible(True)
self.progress_bar.set_fraction(0.0)
self.progress_bar.set_text(message)
self.progress_bar.pulse()
# Start a timeout to animate the progress bar
GLib.timeout_add(100, self.animate_progress)
def animate_progress(self) -> bool:
"""Animate the progress bar."""
self.progress_bar.pulse()
return True # Continue calling
def hide_progress(self):
"""Hide the progress bar."""
self.progress_bar.set_visible(False)
self.progress_bar.set_text("")
# ------------------ Model Training Controls ------------------
def on_select_dataset(self, button: Gtk.Button):
"""Handle the Browse Dataset button click."""
dialog = Gtk.FileChooserDialog(
title="Select a Dataset Directory",
parent=self,
action=Gtk.FileChooserAction.SELECT_FOLDER
)
dialog.add_buttons(
Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL,
Gtk.STOCK_OPEN, Gtk.ResponseType.OK
)
response = dialog.run()
if response == Gtk.ResponseType.OK:
dataset_path = dialog.get_filename()
self.dataset_entry.set_text(dataset_path)
logger.info(f"Selected dataset directory: {dataset_path}")
dialog.destroy()
def on_train_model(self, button: Gtk.Button):
"""Handle the Train Model button click."""
dataset_input = self.dataset_entry.get_text().strip()
architecture = self.architecture_combo.get_active_text()
lr = self.lr_entry.get_text().strip()
bs = self.bs_entry.get_text().strip()
epochs = self.epochs_entry.get_text().strip()
save_model_name = self.save_entry.get_text().strip()
# Validate inputs
if not dataset_input or not os.path.isdir(dataset_input):
self.append_training_output("Invalid dataset path.", error=True)
return
if not architecture:
self.append_training_output("Please select a model architecture.", error=True)
return
try:
lr = float(lr)
bs = int(bs)
epochs = int(epochs)
except ValueError:
self.append_training_output("Learning rate must be a float, Batch size and Epochs must be integers.", error=True)
return
if not save_model_name:
self.append_training_output("Please specify a name to save the trained model.", error=True)
return
# Start training in a separate thread
self.set_controls_sensitive(False)
self.show_training_progress("Starting model training...")
thread = threading.Thread(
target=self.train_model,
args=(dataset_input, architecture, lr, bs, epochs, save_model_name),
daemon=True
)
thread.start()
def train_model(self, dataset_path: str, architecture: str, lr: float, bs: int, epochs: int, save_model_name: str):
"""Train the model with the specified settings."""
try:
logger.info(f"Starting training with architecture: {architecture}, LR: {lr}, BS: {bs}, Epochs: {epochs}")
GLib.idle_add(self.append_training_output, f"Loading dataset from: {dataset_path}")
# Create ImageDataLoaders
data = ImageDataLoaders.from_folder(
dataset_path,
valid_pct=0.2,
item_tfms=Resize(224),
batch_tfms=aug_transforms(),
bs=bs
)
# Initialize the learner
learner = vision_learner(data, arch=getattr(models, architecture)(), metrics=accuracy)
# Start training
GLib.idle_add(self.append_training_output, "Starting training...")
learner.fine_tune(epochs, base_lr=lr, callbacks=[TrainingCallback(self)])
# Save the trained model
learner.export(save_model_name)
logger.info(f"Model trained and saved as: {save_model_name}")
GLib.idle_add(self.append_training_output, f"Model trained and saved as: {save_model_name}", success=True)
# Optionally, load the newly trained model
GLib.idle_add(self.load_model_async, save_model_name)
except Exception as e:
logger.error(f"Training failed: {e}", exc_info=True)
GLib.idle_add(self.append_training_output, f"Training failed: {e}", error=True)
finally:
GLib.idle_add(self.hide_training_progress)
GLib.idle_add(self.set_controls_sensitive, True)
def show_training_progress(self, message: str):
"""Display the training progress bar with a message."""
self.progress_bar.set_visible(True)
self.progress_bar.set_fraction(0.0)
self.progress_bar.set_text(message)
self.progress_bar.pulse()
# Start a timeout to animate the progress bar
GLib.timeout_add(100, self.animate_progress)
def hide_training_progress(self):
"""Hide the training progress bar."""
self.progress_bar.set_visible(False)
self.progress_bar.set_text("")
def append_training_output(self, message: str, error: bool = False, success: bool = False):
"""Append messages to the training output TextView."""
buffer = self.training_output.get_buffer()
end_iter = buffer.get_end_iter()
if error:
formatted_message = f"<span foreground='red'>{message}</span>\n"
elif success:
formatted_message = f"<span foreground='green'>{message}</span>\n"
else:
formatted_message = f"{message}\n"
buffer.insert_markup(end_iter, formatted_message)
# ------------------ Model Training Callbacks ------------------
class TrainingCallback(Callback):
"""A callback to monitor training progress and output."""
def __init__(self, app):
self.app = app
def after_epoch(self):
epoch = self.learn.epoch
loss = self.learn.recorder.losses[-1]
acc = self.learn.recorder.metrics[-1]
message = f"Epoch {epoch + 1}: Loss={loss:.4f}, Accuracy={acc * 100:.2f}%"
GLib.idle_add(self.app.append_training_output, message)
def after_fit(self):
GLib.idle_add(self.app.append_training_output, "Training completed successfully.", success=True)
# ------------------ Prediction Functions ------------------
def load_image(self, image_input: str) -> PILImage:
"""
Load an image from a local path or a URL.
Args:
image_input (str): The path or URL of the image.
Returns:
PILImage: The loaded image.
"""
logger.debug(f"Loading image from: {image_input}")
if image_input.startswith(('http://', 'https://', 'ftp://')):
response = requests.get(image_input, timeout=10)
response.raise_for_status()
image = PILImage.create(BytesIO(response.content))
logger.debug("Image loaded from URL.")
return image
elif os.path.isfile(image_input):
image = PILImage.create(image_input)
logger.debug("Image loaded from local file.")
return image
else:
logger.error("Invalid image path or URL.")
raise ValueError("Invalid image path or URL.")
def display_image(self, image_input: str):
"""
Display the selected image in the image preview widget.
Args:
image_input (str): The path or URL of the image.
"""
try:
if image_input.startswith(('http://', 'https://', 'ftp://')):
response = requests.get(image_input, timeout=10)
response.raise_for_status()
loader = GdkPixbuf.PixbufLoader.new()
loader.write(response.content)
loader.close()
pixbuf = loader.get_pixbuf()
else:
pixbuf = GdkPixbuf.Pixbuf.new_from_file(image_input)
# Calculate scaling while preserving aspect ratio
max_width, max_height = 600, 600
width = pixbuf.get_width()
height = pixbuf.get_height()
scaling_factor = min(max_width / width, max_height / height, 1)
new_width = int(width * scaling_factor)
new_height = int(height * scaling_factor)
scaled_pixbuf = pixbuf.scale_simple(new_width, new_height, GdkPixbuf.InterpType.BILINEAR)
self.image_preview.set_from_pixbuf(scaled_pixbuf)
logger.debug("Image displayed successfully.")
except Exception as e:
logger.error(f"Failed to display image: {e}", exc_info=True)
self.image_preview.set_from_icon_name("image-missing", Gtk.IconSize.DIALOG)
self.update_output(f"<span foreground='red'>Failed to display image preview:</span>\n{e}")
def update_output(self, text: str):
"""Update the output label with the given text."""
self.output_label.set_markup(text)
# ------------------ Utility Functions ------------------
def set_controls_sensitive(self, sensitive: bool):
"""Enable or disable controls based on the sensitive flag."""
# Prediction Controls
self.load_model_button.set_sensitive(sensitive)
self.image_button.set_sensitive(sensitive)
self.image_entry.set_sensitive(sensitive)
self.threshold_slider.set_sensitive(sensitive)
self.predict_button.set_sensitive(sensitive and self.model is not None)
# Training Controls
self.train_button.set_sensitive(sensitive)
self.dataset_button.set_sensitive(sensitive)
self.dataset_entry.set_sensitive(sensitive)
self.architecture_combo.set_sensitive(sensitive)
self.lr_entry.set_sensitive(sensitive)
self.bs_entry.set_sensitive(sensitive)
self.epochs_entry.set_sensitive(sensitive)
self.save_entry.set_sensitive(sensitive)
# ------------------ Model Training Functions ------------------
def on_train_model(self, button: Gtk.Button):
"""Handle the Train Model button click."""
dataset_input = self.dataset_entry.get_text().strip()
architecture = self.architecture_combo.get_active_text()
lr = self.lr_entry.get_text().strip()
bs = self.bs_entry.get_text().strip()
epochs = self.epochs_entry.get_text().strip()
save_model_name = self.save_entry.get_text().strip()
# Validate inputs
if not dataset_input or not os.path.isdir(dataset_input):
self.append_training_output("Invalid dataset path.", error=True)
return
if not architecture:
self.append_training_output("Please select a model architecture.", error=True)
return
try:
lr = float(lr)
bs = int(bs)
epochs = int(epochs)
except ValueError:
self.append_training_output("Learning rate must be a float, Batch size and Epochs must be integers.", error=True)
return
if not save_model_name:
self.append_training_output("Please specify a name to save the trained model.", error=True)
return
# Start training in a separate thread
self.set_controls_sensitive(False)
self.show_training_progress("Starting model training...")
thread = threading.Thread(
target=self.train_model,
args=(dataset_input, architecture, lr, bs, epochs, save_model_name),
daemon=True
)
thread.start()
def train_model(self, dataset_path: str, architecture: str, lr: float, bs: int, epochs: int, save_model_name: str):
"""Train the model with the specified settings."""
try:
logger.info(f"Starting training with architecture: {architecture}, LR: {lr}, BS: {bs}, Epochs: {epochs}")
GLib.idle_add(self.append_training_output, f"Loading dataset from: {dataset_path}")
from fastai.vision.all import models, accuracy, Callback, Resize, aug_transforms
# Create ImageDataLoaders
data = ImageDataLoaders.from_folder(
dataset_path,
valid_pct=0.2,
item_tfms=Resize(224),
batch_tfms=aug_transforms(),
bs=bs
)
logger.info("ImageDataLoaders created successfully.")
GLib.idle_add(self.append_training_output, "Dataset loaded successfully.")
# Initialize the learner
learner = vision_learner(data, arch=getattr(models, architecture)(), metrics=accuracy)
logger.info("Learner initialized successfully.")
GLib.idle_add(self.append_training_output, f"Initialized learner with architecture: {architecture}")
# Start training with fine-tuning
GLib.idle_add(self.append_training_output, "Starting training...")
learner.fine_tune(epochs, base_lr=lr, callbacks=[self.TrainingCallback(self)])
# Save the trained model
learner.export(save_model_name)
logger.info(f"Model trained and saved as: {save_model_name}")
GLib.idle_add(self.append_training_output, f"Model trained and saved as: {save_model_name}", success=True)
# Automatically load the newly trained model
GLib.idle_add(self.load_model_async, save_model_name)
except Exception as e:
logger.error(f"Training failed: {e}", exc_info=True)
GLib.idle_add(self.append_training_output, f"Training failed: {e}", error=True)
finally:
GLib.idle_add(self.hide_training_progress)
GLib.idle_add(self.set_controls_sensitive, True)
def show_training_progress(self, message: str):
"""Display the training progress bar with a message."""
self.progress_bar.set_visible(True)
self.progress_bar.set_fraction(0.0)
self.progress_bar.set_text(message)
self.progress_bar.pulse()
# Start a timeout to animate the progress bar
GLib.timeout_add(100, self.animate_progress)
def hide_training_progress(self):
"""Hide the training progress bar."""
self.progress_bar.set_visible(False)
self.progress_bar.set_text("")
def append_training_output(self, message: str, error: bool = False, success: bool = False):
"""Append messages to the training output TextView."""
buffer = self.training_output.get_buffer()
end_iter = buffer.get_end_iter()
if error:
formatted_message = f"<span foreground='red'>{message}</span>\n"
elif success:
formatted_message = f"<span foreground='green'>{message}</span>\n"
else:
formatted_message = f"{message}\n"
buffer.insert_markup(end_iter, formatted_message)
# ------------------ Model Training Callback Class ------------------
from fastai.callback.core import Callback
class TrainingCallback(Callback):
"""A callback to monitor training progress and output."""
def __init__(self, app):
self.app = app
def after_epoch(self):
epoch = self.learn.epoch
loss = self.learn.recorder.losses[-1]
acc = self.learn.recorder.metrics[-1]
message = f"Epoch {epoch + 1}: Loss={loss:.4f}, Accuracy={acc * 100:.2f}%"
GLib.idle_add(self.app.append_training_output, message)
def after_fit(self):
GLib.idle_add(self.app.append_training_output, "Training completed successfully.", success=True)
# ------------------ Main Function ------------------
def main():
"""Initialize and run the GTK application."""
app = ImagePredictorApp()
app.connect("destroy", Gtk.main_quit)
app.show_all()
Gtk.main()
if __name__ == "__main__":
main()
```
Add support for other model types, rather than just Vision models.ASSISTANT
To extend your `ImagePredictorApp` to support other model types beyond vision models, you'll need to:
1. **Introduce Model Type Selection:** Allow users to select the type of model they wish to work with (e.g., Vision, Text, Tabular, etc.).
2. **Handle Different Model Loading Mechanisms:** Depending on the selected model type, use appropriate libraries and methods to load models (e.g., FastAI for vision and text, Scikit-learn for tabular data, PyTorch for custom models).
3. **Adapt the Training Pipeline:** Ensure that the training process accommodates different data types and model architectures based on the selected model type.
4. **Update the Prediction Logic:** Modify the prediction functionality to handle various input types (images, text, tabular data) accordingly.
Below is a revised version of your application that incorporates support for both **Vision** and **Text** models using FastAI. This example can be further extended to include more model types like Tabular or custom PyTorch models.
### Key Changes:
1. **Model Type Selection:** Added a `Gtk.ComboBoxText` to allow users to select the model type (`Vision` or `Text`).
2. **Dynamic UI Adjustments:** Based on the selected model type, the application adjusts available options and input fields.
3. **Separate Training Pipelines:** Implemented distinct training workflows for Vision and Text models.
4. **Unified Model Loading and Prediction:** Updated model loading and prediction functions to handle different model types seamlessly.
### Updated Code:
```python
import os
import sys
import threading
import logging
from io import BytesIO
from typing import Optional, Union
import gi
import requests
from PIL import Image
from fastai.vision.all import (
load_learner as load_vision_learner,
PILImage,
Learner,
vision_learner,
untar_data,
URLs,
ImageDataLoaders,
models as vision_models,
accuracy,
Callback,
Resize,
aug_transforms,
)
from fastai.text.all import (
load_learner as load_text_learner,
TextDataLoaders,
text_classifier_learner,
AWD_LSTM,
)
from sklearn.externals import joblib # For scikit-learn models
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, GdkPixbuf, Gdk, GLib
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s:%(name)s: %(message)s',
handlers=[
logging.FileHandler("model_predictor_app.log"),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
class ModelPredictorApp(Gtk.Window):
def __init__(self):
super().__init__(title="Model Predictor with Training Support")
self.set_border_width(10)
self.set_default_size(1200, 800)
# Initialize model attributes
self.vision_model: Optional[Learner] = None
self.text_model: Optional[Learner] = None
self.sk_model: Optional[object] = None # For sklearn models
# Initialize UI components
self.init_ui()
def init_ui(self):
"""Initialize and arrange all UI components."""
# Main container using Paned for resizable sections
paned = Gtk.Paned.new(Gtk.Orientation.HORIZONTAL)
self.add(paned)
# --- Left Pane: Controls and Training Settings ---
left_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
paned.pack1(left_box, resize=True, shrink=False)
# --- Model Type Selection ---
model_type_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=5)
model_type_label = Gtk.Label(label="Model Type:")
model_type_label.set_xalign(0)
model_type_box.pack_start(model_type_label, False, False, 0)
self.model_type_combo = Gtk.ComboBoxText()
for model_type in ["Vision", "Text", "Tabular", "Custom PyTorch"]:
self.model_type_combo.append_text(model_type)
self.model_type_combo.set_active(0) # Default to Vision
self.model_type_combo.connect("changed", self.on_model_type_changed)
model_type_box.pack_start(self.model_type_combo, False, False, 0)
left_box.pack_start(model_type_box, False, False, 0)
# --- Controls Section ---
controls_frame = Gtk.Frame(label="Prediction Controls")
controls_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
controls_box.set_margin_bottom(10)
controls_box.set_margin_top(10)
controls_box.set_margin_start(10)
controls_box.set_margin_end(10)
controls_frame.add(controls_box)
left_box.pack_start(controls_frame, False, False, 0)
# Load Model Button
self.load_model_button = Gtk.Button(label="Load Model")
self.load_model_button.connect("clicked", self.on_load_model)
controls_box.pack_start(self.load_model_button, False, False, 0)
# Dynamic Input Selection based on Model Type
self.input_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=5)
self.input_entry = Gtk.Entry()
self.input_entry.set_placeholder_text("Enter input (path/URL/text)")
self.input_entry.connect("activate", self.on_enter_input)
self.input_button = Gtk.Button(label="Browse/Input")
self.input_button.connect("clicked", self.on_select_input)
self.input_box.pack_start(self.input_entry, True, True, 0)
self.input_box.pack_start(self.input_button, False, False, 0)
controls_box.pack_start(self.input_box, False, False, 0)
# Threshold Slider (Applicable for Classification)
self.threshold_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=5)
self.threshold_label = Gtk.Label(label="Prediction Threshold: 50%")
self.threshold_label.set_xalign(0)
self.threshold_box.pack_start(self.threshold_label, False, False, 0)
self.threshold_adjustment = Gtk.Adjustment(value=50, lower=0, upper=100, step_increment=1, page_increment=10)
self.threshold_slider = Gtk.Scale(orientation=Gtk.Orientation.HORIZONTAL, adjustment=self.threshold_adjustment)
self.threshold_slider.set_digits(0)
self.threshold_slider.connect("value-changed", self.on_threshold_changed)
self.threshold_box.pack_start(self.threshold_slider, False, False, 0)
controls_box.pack_start(self.threshold_box, False, False, 0)
# Predict Button
self.predict_button = Gtk.Button(label="Predict")
self.predict_button.connect("clicked", self.on_predict)
self.predict_button.set_sensitive(False)
controls_box.pack_start(self.predict_button, False, False, 0)
# Output Display
self.output_label = Gtk.Label(label="Output will be displayed here.")
self.output_label.set_line_wrap(True)
self.output_label.set_xalign(0)
self.output_label.set_justify(Gtk.Justification.LEFT)
controls_box.pack_start(self.output_label, False, False, 0)
# Progress Bar
self.progress_bar = Gtk.ProgressBar()
self.progress_bar.set_show_text(True)
self.progress_bar.set_visible(False)
controls_box.pack_start(self.progress_bar, False, False, 0)
# --- Training Settings Section ---
training_frame = Gtk.Frame(label="Model Training Settings")
training_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
training_box.set_margin_bottom(10)
training_box.set_margin_top(10)
training_box.set_margin_start(10)
training_box.set_margin_end(10)
training_frame.add(training_box)
left_box.pack_start(training_frame, False, False, 0)
# Dataset Selection
dataset_select_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=5)
self.dataset_entry = Gtk.Entry()
self.dataset_entry.set_placeholder_text("Enter dataset path or URL")
self.dataset_button = Gtk.Button(label="Browse Dataset")
self.dataset_button.connect("clicked", self.on_select_dataset)
dataset_select_box.pack_start(self.dataset_entry, True, True, 0)
dataset_select_box.pack_start(self.dataset_button, False, False, 0)
training_box.pack_start(dataset_select_box, False, False, 0)
# Model Architecture Selection (Dynamic based on Model Type)
self.architecture_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=5)
architecture_label = Gtk.Label(label="Model Architecture:")
architecture_label.set_xalign(0)
self.architecture_box.pack_start(architecture_label, False, False, 0)
self.architecture_combo = Gtk.ComboBoxText()
# Initially populate with Vision models
self.populate_architecture_options("Vision")
self.architecture_combo.set_active(0)
self.architecture_box.pack_start(self.architecture_combo, False, False, 0)
training_box.pack_start(self.architecture_box, False, False, 0)
# Hyperparameters
self.hyperparams_grid = Gtk.Grid(column_spacing=10, row_spacing=10)
# Learning Rate
lr_label = Gtk.Label(label="Learning Rate:")
lr_label.set_halign(Gtk.Align.END)
self.hyperparams_grid.attach(lr_label, 0, 0, 1, 1)
self.lr_entry = Gtk.Entry()
self.lr_entry.set_text("1e-3")
self.hyperparams_grid.attach(self.lr_entry, 1, 0, 1, 1)
# Batch Size
bs_label = Gtk.Label(label="Batch Size:")
bs_label.set_halign(Gtk.Align.END)
self.hyperparams_grid.attach(bs_label, 0, 1, 1, 1)
self.bs_entry = Gtk.Entry()
self.bs_entry.set_text("32")
self.hyperparams_grid.attach(self.bs_entry, 1, 1, 1, 1)
# Number of Epochs
epochs_label = Gtk.Label(label="Number of Epochs:")
epochs_label.set_halign(Gtk.Align.END)
self.hyperparams_grid.attach(epochs_label, 0, 2, 1, 1)
self.epochs_entry = Gtk.Entry()
self.epochs_entry.set_text("10")
self.hyperparams_grid.attach(self.epochs_entry, 1, 2, 1, 1)
# Save Model Name
save_label = Gtk.Label(label="Save Model As:")
save_label.set_halign(Gtk.Align.END)
self.hyperparams_grid.attach(save_label, 0, 3, 1, 1)
self.save_entry = Gtk.Entry()
self.save_entry.set_text("trained_model.pkl")
self.hyperparams_grid.attach(self.save_entry, 1, 3, 1, 1)
training_box.pack_start(self.hyperparams_grid, False, False, 0)
# Train Model Button
self.train_button = Gtk.Button(label="Train Model")
self.train_button.connect("clicked", self.on_train_model)
training_box.pack_start(self.train_button, False, False, 0)
# Training Output Display
self.training_output = Gtk.TextView()
self.training_output.set_editable(False)
self.training_output.set_wrap_mode(Gtk.WrapMode.WORD)
training_output_scrolled = Gtk.ScrolledWindow()
training_output_scrolled.set_vexpand(True)
training_output_scrolled.add(self.training_output)
training_box.pack_start(training_output_scrolled, True, True, 0)
# --- Right Pane: Input Preview ---
preview_frame = Gtk.Frame(label="Input Preview")
preview_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=5)
preview_box.set_margin_bottom(10)
preview_box.set_margin_top(10)
preview_box.set_margin_start(10)
preview_box.set_margin_end(10)
preview_frame.add(preview_box)
self.input_preview = Gtk.Image()
self.input_preview.set_valign(Gtk.Align.START)
preview_box.pack_start(self.input_preview, True, True, 0)
# Drag and Drop Support for Input Preview (Image and Text)
self.input_preview.drag_dest_set(Gtk.DestDefaults.ALL, [], Gdk.DragAction.COPY)
# Add more targets based on model type
self.input_preview.drag_dest_add_uri_targets()
self.input_preview.connect("drag-data-received", self.on_drag_data_received)
paned.pack2(preview_frame, resize=True, shrink=False)
# Initial UI state
self.on_model_type_changed(None)
# ------------------ UI Update Based on Model Type ------------------
def on_model_type_changed(self, combo):
"""Adjust UI components based on the selected model type."""
model_type = self.model_type_combo.get_active_text()
logger.info(f"Model type selected: {model_type}")
# Update Architecture Options
self.populate_architecture_options(model_type)
# Adjust Input Placeholder and Browse Button
self.input_entry.set_placeholder_text(self.get_input_placeholder(model_type))
self.input_button.set_label(self.get_input_button_label(model_type))
# Clear Input Preview
self.clear_input_preview()
def populate_architecture_options(self, model_type: str):
"""Populate architecture_combo based on model type."""
self.architecture_combo.remove_all()
if model_type == "Vision":
for arch in ["resnet18", "resnet34", "resnet50", "resnet101", "resnet152"]:
self.architecture_combo.append_text(arch)
elif model_type == "Text":
for arch in ["awd_lstm"]:
self.architecture_combo.append_text(arch)
elif model_type == "Tabular":
for arch in ["tabular_resnet"]:
self.architecture_combo.append_text(arch)
elif model_type == "Custom PyTorch":
self.architecture_combo.append_text("Custom")
else:
self.architecture_combo.append_text("Unknown")
self.architecture_combo.set_active(0)
def get_input_placeholder(self, model_type: str) -> str:
"""Get placeholder text based on model type."""
if model_type == "Vision":
return "Enter image path or URL"
elif model_type == "Text":
return "Enter text input"
elif model_type == "Tabular":
return "Enter data file path or parameters"
elif model_type == "Custom PyTorch":
return "Enter input data path or text"
else:
return "Enter input"
def get_input_button_label(self, model_type: str) -> str:
"""Get browse/input button label based on model type."""
if model_type == "Vision":
return "Browse Image"
elif model_type == "Text":
return "Input Text"
elif model_type == "Tabular":
return "Browse Dataset"
elif model_type == "Custom PyTorch":
return "Browse/Input"
else:
return "Browse/Input"
def clear_input_preview(self):
"""Clear the input preview area."""
self.input_preview.set_from_icon_name("image-missing", Gtk.IconSize.DIALOG)
# ------------------ Prediction Controls ------------------
def on_drag_data_received(self, widget, drag_context, x, y, data, info, time):
"""Handle drag-and-drop of files or URLs based on model type."""
uris = data.get_uris()
if uris:
uri = uris[0]
model_type = self.model_type_combo.get_active_text()
path_or_url = self.parse_uri(uri)
if path_or_url:
if model_type in ["Vision", "Custom PyTorch"]:
self.input_entry.set_text(path_or_url)
self.display_input(path_or_url)
elif model_type == "Text":
# For text, drag-and-drop might not be straightforward; handle accordingly
self.input_entry.set_text(path_or_url)
self.display_input_text(path_or_url)
elif model_type == "Tabular":
self.input_entry.set_text(path_or_url)
# Potentially load a sample from the dataset
else:
self.output_label.set_markup("<span foreground='red'>Unsupported input type.</span>")
Gtk.drag_finish(drag_context, True, False, time)
@staticmethod
def parse_uri(uri: str) -> Optional[str]:
"""Parse URI to get local path or URL."""
if uri.startswith("file://"):
return uri[7:]
elif uri.startswith(("http://", "https://", "ftp://")):
return uri
else:
return None
def on_enter_input(self, widget):
"""Handle pressing Enter in the input entry."""
model_type = self.model_type_combo.get_active_text()
user_input = self.input_entry.get_text().strip()
if user_input:
if model_type in ["Vision", "Custom PyTorch"]:
self.display_input(user_input)
elif model_type == "Text":
self.display_input_text(user_input)
elif model_type == "Tabular":
# Handle tabular data preview if necessary
self.display_input_tabular(user_input)
def on_select_input(self, button):
"""Handle the Select/Input button click based on model type."""
model_type = self.model_type_combo.get_active_text()
if model_type in ["Vision", "Custom PyTorch", "Tabular"]:
dialog = Gtk.FileChooserDialog(
title="Select Input",
parent=self,
action=Gtk.FileChooserAction.OPEN
)
dialog.add_buttons(
Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL,
Gtk.STOCK_OPEN, Gtk.ResponseType.OK
)
if model_type == "Vision" or model_type == "Custom PyTorch":
dialog.set_filter(self.create_image_filter())
elif model_type == "Tabular":
dialog.set_filter(self.create_dataset_filter())
response = dialog.run()
if response == Gtk.ResponseType.OK:
input_path = dialog.get_filename()
self.input_entry.set_text(input_path)
if model_type in ["Vision", "Custom PyTorch"]:
self.display_input(input_path)
elif model_type == "Tabular":
self.display_input_tabular(input_path)
dialog.destroy()
elif model_type == "Text":
# For text input, open a dialog to enter text
self.open_text_input_dialog()
def open_text_input_dialog(self):
"""Open a dialog to input text for Text models."""
dialog = Gtk.Dialog(title="Enter Text", parent=self, flags=0)
dialog.add_buttons(
Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL,
Gtk.STOCK_OK, Gtk.ResponseType.OK
)
box = dialog.get_content_area()
text_view = Gtk.TextView()
text_view.set_wrap_mode(Gtk.WrapMode.WORD)
box.add(text_view)
dialog.show_all()
response = dialog.run()
if response == Gtk.ResponseType.OK:
buffer = text_view.get_buffer()
start_iter = buffer.get_start_iter()
end_iter = buffer.get_end_iter()
user_text = buffer.get_text(start_iter, end_iter, True)
self.input_entry.set_text(user_text)
self.display_input_text(user_text)
dialog.destroy()
@staticmethod
def create_image_filter() -> Gtk.FileFilter:
"""Create a file filter for image files."""
img_filter = Gtk.FileFilter()
img_filter.set_name("Image Files (*.png, *.jpg, *.jpeg, *.webp)")
img_filter.add_pattern("*.png")
img_filter.add_pattern("*.jpg")
img_filter.add_pattern("*.jpeg")
img_filter.add_pattern("*.webp")
img_filter.add_mime_type("image/png")
img_filter.add_mime_type("image/jpeg")
img_filter.add_mime_type("image/webp")
return img_filter
@staticmethod
def create_dataset_filter() -> Gtk.FileFilter:
"""Create a file filter for dataset files."""
data_filter = Gtk.FileFilter()
data_filter.set_name("CSV Files (*.csv)")
data_filter.add_pattern("*.csv")
return data_filter
def on_threshold_changed(self, scale: Gtk.Scale):
"""Update the threshold label when the slider value changes."""
value = self.threshold_slider.get_value()
self.threshold_label.set_text(f"Prediction Threshold: {int(value)}%")
def on_predict(self, button: Gtk.Button):
"""Handle the Predict button click."""
model_type = self.model_type_combo.get_active_text()
threshold = self.threshold_slider.get_value()
if model_type == "Vision" and not self.vision_model:
self.output_label.set_markup("<span foreground='red'>Please load a Vision model first.</span>")
return
elif model_type == "Text" and not self.text_model:
self.output_label.set_markup("<span foreground='red'>Please load a Text model first.</span>")
return
elif model_type == "Tabular" and not self.sk_model:
self.output_label.set_markup("<span foreground='red'>Please load a Scikit-learn model first.</span>")
return
elif model_type == "Custom PyTorch" and not self.vision_model:
self.output_label.set_markup("<span foreground='red'>Please load a PyTorch model first.</span>")
return
user_input = self.input_entry.get_text().strip()
if not user_input:
self.output_label.set_markup("<span foreground='red'>Please provide the necessary input.</span>")
return
self.set_controls_sensitive(False)
self.show_progress("Performing prediction...")
thread = threading.Thread(target=self.predict_input, args=(model_type, user_input, threshold), daemon=True)
thread.start()
def predict_input(self, model_type: str, user_input: str, threshold: float):
"""Perform prediction based on the model type and input."""
try:
if model_type == "Vision":
image = self.load_image(user_input)
GLib.idle_add(self.display_input, user_input)
pred, pred_idx, outputs = self.vision_model.predict(image)
probabilities = outputs.numpy() * 100 # Convert to percentage
classes = self.vision_model.dls.vocab # Get class names
# Filter predictions based on threshold
output_lines = ["<b>Predictions:</b>"]
for class_name, probability in zip(classes, probabilities):
if probability >= threshold:
line = f"• {class_name}: {probability:.2f}%"
output_lines.append(line)
output_lines.append(f"\n<b>Top Prediction:</b> {pred}")
markup_text = "\n".join(output_lines)
GLib.idle_add(self.update_output, markup_text)
elif model_type == "Text":
GLib.idle_add(self.display_input_text, user_input)
pred, pred_idx, outputs = self.text_model.predict(user_input)
probabilities = outputs.numpy() * 100
classes = self.text_model.dls.vocab
# Filter predictions based on threshold
output_lines = ["<b>Predictions:</b>"]
for class_name, probability in zip(classes, probabilities):
if probability >= threshold:
line = f"• {class_name}: {probability:.2f}%"
output_lines.append(line)
output_lines.append(f"\n<b>Top Prediction:</b> {pred}")
markup_text = "\n".join(output_lines)
GLib.idle_add(self.update_output, markup_text)
elif model_type == "Tabular":
# For simplicity, assume the input is a CSV file path for batch prediction
if not os.path.isfile(user_input):
raise ValueError("Invalid dataset path for tabular prediction.")
import pandas as pd
data = pd.read_csv(user_input)
predictions = self.sk_model.predict_proba(data)
classes = self.sk_model.classes_
# Process predictions
output_lines = ["<b>Batch Predictions:</b>"]
for idx, row in data.iterrows():
output_lines.append(f"Record {idx + 1}:")
for class_name, prob in zip(classes, row):
if prob * 100 >= threshold:
output_lines.append(f" • {class_name}: {prob * 100:.2f}%")
# Assuming binary classification, add top prediction
top_pred = self.sk_model.predict(data.iloc[[idx]])[0]
output_lines.append(f" <b>Top Prediction:</b> {top_pred}\n")
markup_text = "\n".join(output_lines)
GLib.idle_add(self.update_output, markup_text)
elif model_type == "Custom PyTorch":
# Assuming similar to Vision models; customize as needed
image = self.load_image(user_input)
GLib.idle_add(self.display_input, user_input)
pred, pred_idx, outputs = self.vision_model.predict(image)
probabilities = outputs.detach().numpy() * 100
classes = [f"Class {i}" for i in range(len(predictions))] # Placeholder
# Filter predictions based on threshold
output_lines = ["<b>Predictions:</b>"]
for class_name, probability in zip(classes, probabilities):
if probability >= threshold:
line = f"• {class_name}: {probability:.2f}%"
output_lines.append(line)
output_lines.append(f"\n<b>Top Prediction:</b> {pred}")
markup_text = "\n".join(output_lines)
GLib.idle_add(self.update_output, markup_text)
logger.info(f"Prediction successful for input: {user_input}")
except Exception as e:
logger.error(f"Error during prediction: {e}", exc_info=True)
GLib.idle_add(self.update_output, f"<span foreground='red'>Error during prediction:</span>\n{e}")
finally:
GLib.idle_add(self.hide_progress)
GLib.idle_add(self.set_controls_sensitive, True)
def load_image(self, image_input: str) -> PILImage:
"""
Load an image from a local path or a URL.
Args:
image_input (str): The path or URL of the image.
Returns:
PILImage: The loaded image.
"""
logger.debug(f"Loading image from: {image_input}")
if image_input.startswith(('http://', 'https://', 'ftp://')):
response = requests.get(image_input, timeout=10)
response.raise_for_status()
image = PILImage.create(BytesIO(response.content))
logger.debug("Image loaded from URL.")
return image
elif os.path.isfile(image_input):
image = PILImage.create(image_input)
logger.debug("Image loaded from local file.")
return image
else:
logger.error("Invalid image path or URL.")
raise ValueError("Invalid image path or URL.")
def display_input(self, image_input: str):
"""
Display the selected image in the input preview widget.
Args:
image_input (str): The path or URL of the image.
"""
try:
if image_input.startswith(('http://', 'https://', 'ftp://')):
response = requests.get(image_input, timeout=10)
response.raise_for_status()
loader = GdkPixbuf.PixbufLoader.new()
loader.write(response.content)
loader.close()
pixbuf = loader.get_pixbuf()
else:
pixbuf = GdkPixbuf.Pixbuf.new_from_file(image_input)
# Calculate scaling while preserving aspect ratio
max_width, max_height = 600, 600
width = pixbuf.get_width()
height = pixbuf.get_height()
scaling_factor = min(max_width / width, max_height / height, 1)
new_width = int(width * scaling_factor)
new_height = int(height * scaling_factor)
scaled_pixbuf = pixbuf.scale_simple(new_width, new_height, GdkPixbuf.InterpType.BILINEAR)
self.input_preview.set_from_pixbuf(scaled_pixbuf)
logger.debug("Image displayed successfully.")
except Exception as e:
logger.error(f"Failed to display input image: {e}", exc_info=True)
self.input_preview.set_from_icon_name("image-missing", Gtk.IconSize.DIALOG)
self.update_output(f"<span foreground='red'>Failed to display input preview:</span>\n{e}")
def display_input_text(self, text: str):
"""Display text input in the preview area."""
self.input_preview.set_from_icon_name("text-x-generic", Gtk.IconSize.DIALOG)
# Optionally, display the text in a separate label or text view
def display_input_tabular(self, path: str):
"""Display a preview of the tabular data."""
try:
import pandas as pd
data = pd.read_csv(path)
preview = data.head().to_string()
# Display preview in a separate text view or label
self.input_preview.set_from_icon_name("text-x-generic", Gtk.IconSize.DIALOG)
self.update_output(f"<b>Data Preview:</b>\n{preview}")
except Exception as e:
logger.error(f"Failed to display tabular data: {e}", exc_info=True)
self.update_output(f"<span foreground='red'>Failed to display data preview:</span>\n{e}")
def update_output(self, text: str):
"""Update the output label with the given text."""
self.output_label.set_markup(text)
def set_controls_sensitive(self, sensitive: bool):
"""Enable or disable controls based on the sensitive flag."""
# Prediction Controls
self.load_model_button.set_sensitive(sensitive)
self.input_button.set_sensitive(sensitive)
self.input_entry.set_sensitive(sensitive)
self.threshold_slider.set_sensitive(sensitive)
self.predict_button.set_sensitive(sensitive and (
(self.model_type_combo.get_active_text() == "Vision" and self.vision_model) or
(self.model_type_combo.get_active_text() == "Text" and self.text_model) or
(self.model_type_combo.get_active_text() == "Tabular" and self.sk_model) or
(self.model_type_combo.get_active_text() == "Custom PyTorch" and self.vision_model)
))
# Training Controls
self.train_button.set_sensitive(sensitive)
self.dataset_button.set_sensitive(sensitive)
self.dataset_entry.set_sensitive(sensitive)
self.architecture_combo.set_sensitive(sensitive)
self.lr_entry.set_sensitive(sensitive)
self.bs_entry.set_sensitive(sensitive)
self.epochs_entry.set_sensitive(sensitive)
self.save_entry.set_sensitive(sensitive)
def show_progress(self, message: str):
"""Display the progress bar with a message."""
self.progress_bar.set_visible(True)
self.progress_bar.set_fraction(0.0)
self.progress_bar.set_text(message)
self.progress_bar.pulse()
# Start a timeout to animate the progress bar
GLib.timeout_add(100, self.animate_progress)
def animate_progress(self) -> bool:
"""Animate the progress bar."""
self.progress_bar.pulse()
return True # Continue calling
def hide_progress(self):
"""Hide the progress bar."""
self.progress_bar.set_visible(False)
self.progress_bar.set_text("")
# ------------------ Model Loading ------------------
def on_load_model(self, button: Gtk.Button):
"""Handle the Load Model button click."""
model_type = self.model_type_combo.get_active_text()
dialog = Gtk.FileChooserDialog(
title="Select a Model File",
parent=self,
action=Gtk.FileChooserAction.OPEN
)
dialog.add_buttons(
Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL,
Gtk.STOCK_OPEN, Gtk.ResponseType.OK
)
dialog.set_filter(self.create_model_filter(model_type))
response = dialog.run()
if response == Gtk.ResponseType.OK:
model_path = dialog.get_filename()
logger.info(f"Selected model file: {model_path}")
self.load_model_async(model_type, model_path)
dialog.destroy()
def create_model_filter(self, model_type: str) -> Gtk.FileFilter:
"""Create a file filter for model files based on model type."""
file_filter = Gtk.FileFilter()
file_filter.set_name("Model Files")
if model_type in ["Vision", "Text", "Custom PyTorch"]:
file_filter.add_pattern("*.pkl")
file_filter.add_pattern("*.pth")
file_filter.add_mime_type("application/octet-stream")
elif model_type == "Tabular":
file_filter.add_pattern("*.pkl")
file_filter.add_pattern("*.joblib")
file_filter.add_mime_type("application/octet-stream")
else:
file_filter.add_pattern("*")
return file_filter
def load_model_async(self, model_type: str, model_path: str):
"""Load the model in a separate thread to keep UI responsive."""
self.set_controls_sensitive(False)
self.show_progress("Loading model...")
thread = threading.Thread(target=self.load_model, args=(model_type, model_path), daemon=True)
thread.start()
def load_model(self, model_type: str, model_path: str):
"""Load the machine learning model from the specified path."""
try:
if model_type == "Vision":
learner = load_vision_learner(model_path)
GLib.idle_add(self.on_model_loaded, model_type, learner, model_path)
logger.info(f"Vision model loaded successfully from: {model_path}")
elif model_type == "Text":
learner = load_text_learner(model_path)
GLib.idle_add(self.on_model_loaded, model_type, learner, model_path)
logger.info(f"Text model loaded successfully from: {model_path}")
elif model_type == "Tabular":
sk_model = joblib.load(model_path)
GLib.idle_add(self.on_model_loaded, model_type, sk_model, model_path)
logger.info(f"Scikit-learn model loaded successfully from: {model_path}")
elif model_type == "Custom PyTorch":
# Load custom PyTorch model (assuming it's a FastAI Learner)
learner = load_vision_learner(model_path) # Adjust as per your custom model
GLib.idle_add(self.on_model_loaded, model_type, learner, model_path)
logger.info(f"Custom PyTorch model loaded successfully from: {model_path}")
else:
raise ValueError("Unsupported model type.")
except Exception as e:
logger.error(f"Failed to load model: {e}", exc_info=True)
GLib.idle_add(self.on_model_load_failed, model_type, str(e))
def on_model_loaded(self, model_type: str, model, model_path: str):
"""Callback when the model is successfully loaded."""
if model_type == "Vision":
self.vision_model = model
elif model_type == "Text":
self.text_model = model
elif model_type == "Tabular":
self.sk_model = model
elif model_type == "Custom PyTorch":
self.vision_model = model # Adjust based on implementation
self.update_output(f"<span foreground='green'>Model loaded successfully from:</span>\n{model_path}")
self.hide_progress()
self.set_controls_sensitive(True)
def on_model_load_failed(self, model_type: str, error_message: str):
"""Callback when the model fails to load."""
if model_type == "Vision":
self.vision_model = None
elif model_type == "Text":
self.text_model = None
elif model_type == "Tabular":
self.sk_model = None
elif model_type == "Custom PyTorch":
self.vision_model = None # Adjust based on implementation
self.update_output(f"<span foreground='red'>Failed to load model:</span>\n{error_message}")
self.hide_progress()
self.set_controls_sensitive(True)
# ------------------ Model Training Controls ------------------
def on_train_model(self, button: Gtk.Button):
"""Handle the Train Model button click."""
model_type = self.model_type_combo.get_active_text()
dataset_input = self.dataset_entry.get_text().strip()
architecture = self.architecture_combo.get_active_text()
lr = self.lr_entry.get_text().strip()
bs = self.bs_entry.get_text().strip()
epochs = self.epochs_entry.get_text().strip()
save_model_name = self.save_entry.get_text().strip()
# Validate inputs
if not dataset_input or not os.path.isdir(dataset_input):
self.append_training_output("Invalid dataset path.", error=True)
return
if not architecture:
self.append_training_output("Please select a model architecture.", error=True)
return
try:
lr = float(lr)
bs = int(bs)
epochs = int(epochs)
except ValueError:
self.append_training_output("Learning rate must be a float, Batch size and Epochs must be integers.", error=True)
return
if not save_model_name:
self.append_training_output("Please specify a name to save the trained model.", error=True)
return
# Start training in a separate thread
self.set_controls_sensitive(False)
self.show_training_progress("Starting model training...")
thread = threading.Thread(
target=self.train_model,
args=(model_type, dataset_input, architecture, lr, bs, epochs, save_model_name),
daemon=True
)
thread.start()
def train_model(self, model_type: str, dataset_path: str, architecture: str, lr: float, bs: int, epochs: int, save_model_name: str):
"""Train the model with the specified settings."""
try:
logger.info(f"Starting training with model type: {model_type}, architecture: {architecture}, LR: {lr}, BS: {bs}, Epochs: {epochs}")
GLib.idle_add(self.append_training_output, f"Loading dataset from: {dataset_path}")
if model_type == "Vision":
# Create ImageDataLoaders
data = ImageDataLoaders.from_folder(
dataset_path,
valid_pct=0.2,
item_tfms=Resize(224),
batch_tfms=aug_transforms(),
bs=bs
)
logger.info("ImageDataLoaders created successfully.")
GLib.idle_add(self.append_training_output, "Dataset loaded successfully.")
# Initialize the learner
learner = vision_learner(data, arch=getattr(vision_models, architecture)(), metrics=accuracy)
logger.info("Learner initialized successfully.")
GLib.idle_add(self.append_training_output, f"Initialized learner with architecture: {architecture}")
# Start training with fine-tuning
GLib.idle_add(self.append_training_output, "Starting training...")
learner.fine_tune(epochs, base_lr=lr, callbacks=[self.TrainingCallback(self)])
# Save the trained model
learner.export(save_model_name)
logger.info(f"Vision model trained and saved as: {save_model_name}")
GLib.idle_add(self.append_training_output, f"Model trained and saved as: {save_model_name}", success=True)
# Automatically load the newly trained model
GLib.idle_add(self.load_model_async, model_type, save_model_name)
elif model_type == "Text":
from fastai.text.all import TextDataLoaders, text_classifier_learner, AWD_LSTM
# Create TextDataLoaders
data = TextDataLoaders.from_folder(
dataset_path,
valid_pct=0.2,
text_vocab=None,
bs=bs
)
logger.info("TextDataLoaders created successfully.")
GLib.idle_add(self.append_training_output, "Dataset loaded successfully.")
# Initialize the learner
learner = text_classifier_learner(data, AWD_LSTM, metrics=accuracy)
logger.info("Text learner initialized successfully.")
GLib.idle_add(self.append_training_output, f"Initialized text learner with architecture: {architecture}")
# Start training with fine-tuning
GLib.idle_add(self.append_training_output, "Starting training...")
learner.fine_tune(epochs, base_lr=lr, callbacks=[self.TrainingCallback(self)])
# Save the trained model
learner.export(save_model_name)
logger.info(f"Text model trained and saved as: {save_model_name}")
GLib.idle_add(self.append_training_output, f"Model trained and saved as: {save_model_name}", success=True)
# Automatically load the newly trained model
GLib.idle_add(self.load_model_async, model_type, save_model_name)
elif model_type == "Tabular":
# Example using Scikit-learn (e.g., RandomForestClassifier)
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
data = pd.read_csv(dataset_path)
if 'target' not in data.columns:
raise ValueError("Dataset must contain a 'target' column.")
X = data.drop('target', axis=1)
y = data['target']
X_train, X_valid, y_train, y_valid = train_test_split(X, y, test_size=0.2)
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
preds = model.predict(X_valid)
acc = accuracy_score(y_valid, preds)
GLib.idle_add(self.append_training_output, f"Validation Accuracy: {acc * 100:.2f}%")
# Save the model
joblib.dump(model, save_model_name)
logger.info(f"Scikit-learn model trained and saved as: {save_model_name}")
GLib.idle_add(self.append_training_output, f"Model trained and saved as: {save_model_name}", success=True)
# Automatically load the newly trained model
GLib.idle_add(self.load_model_async, model_type, save_model_name)
elif model_type == "Custom PyTorch":
# Placeholder for custom PyTorch model training
# Implement as per your custom requirements
GLib.idle_add(self.append_training_output, "Custom PyTorch training not implemented yet.", error=True)
else:
raise ValueError("Unsupported model type for training.")
except Exception as e:
logger.error(f"Training failed: {e}", exc_info=True)
GLib.idle_add(self.append_training_output, f"Training failed: {e}", error=True)
finally:
GLib.idle_add(self.hide_training_progress)
GLib.idle_add(self.set_controls_sensitive, True)
def show_training_progress(self, message: str):
"""Display the training progress bar with a message."""
self.progress_bar.set_visible(True)
self.progress_bar.set_fraction(0.0)
self.progress_bar.set_text(message)
self.progress_bar.pulse()
# Start a timeout to animate the progress bar
GLib.timeout_add(100, self.animate_progress)
def hide_training_progress(self):
"""Hide the training progress bar."""
self.progress_bar.set_visible(False)
self.progress_bar.set_text("")
def append_training_output(self, message: str, error: bool = False, success: bool = False):
"""Append messages to the training output TextView."""
buffer = self.training_output.get_buffer()
end_iter = buffer.get_end_iter()
if error:
formatted_message = f"<span foreground='red'>{message}</span>\n"
elif success:
formatted_message = f"<span foreground='green'>{message}</span>\n"
else:
formatted_message = f"{message}\n"
buffer.insert_markup(end_iter, formatted_message)
# ------------------ Model Training Callback Class ------------------
class TrainingCallback(Callback):
"""A callback to monitor training progress and output."""
def __init__(self, app):
self.app = app
def after_epoch(self):
epoch = self.learn.epoch
loss = self.learn.recorder.losses[-1]
metrics = self.learn.recorder.metrics[-1]
metric_name = 'Accuracy' if 'accuracy' in self.learn.metrics else 'Metric'
message = f"Epoch {epoch + 1}: Loss={loss:.4f}, {metric_name}={metrics * 100:.2f}%"
GLib.idle_add(self.app.append_training_output, message)
def after_fit(self):
GLib.idle_add(self.app.append_training_output, "Training completed successfully.", success=True)
# ------------------ Prediction Functions ------------------
# load_image and display_input methods are already defined above
# ------------------ Utility Functions ------------------
# set_controls_sensitive is already defined above
# ------------------ Main Function ------------------
def main():
"""Initialize and run the GTK application."""
app = ModelPredictorApp()
app.connect("destroy", Gtk.main_quit)
app.show_all()
Gtk.main()
if __name__ == "__main__":
ModelPredictorApp.main()
```
### Detailed Explanation of Changes:
1. **Model Type Selection:**
- Added a `Gtk.ComboBoxText` named `model_type_combo` to allow users to select the model type (`Vision`, `Text`, `Tabular`, `Custom PyTorch`).
- Connected the `changed` signal to `on_model_type_changed` to dynamically adjust the UI based on the selected model type.
2. **Dynamic Architecture Options:**
- The `architecture_combo` now populates different architectures based on the selected model type. For example, Vision models have `resnet` variants, Text models use architectures like `awd_lstm`, etc.
3. **Input Handling:**
- Adjusted the input fields to vary based on the model type. For Vision and Custom PyTorch models, users can select images. For Text models, users can input text directly. For Tabular models, users can select dataset files (e.g., CSV).
4. **Model Loading:**
- Updated the `load_model_async` and `load_model` methods to handle different model types. For Vision and Text models, FastAI's `load_learner` is used. For Tabular models, Scikit-learn's `joblib.load` is utilized.
- Ensured that models are stored in separate attributes (`vision_model`, `text_model`, `sk_model`) based on their type.
5. **Prediction Logic:**
- Enhanced the `predict_input` method to handle predictions differently based on model type:
- **Vision:** Processes images and displays top predictions.
- **Text:** Processes text inputs and displays classification results.
- **Tabular:** Performs batch predictions on datasets and displays results.
- **Custom PyTorch:** Placeholder for custom implementations; adjust as needed.
6. **Training Pipeline:**
- Expanded the `train_model` method to support training for Vision, Text, and Tabular models:
- **Vision:** Utilizes FastAI's `ImageDataLoaders` and trains using `vision_learner`.
- **Text:** Utilizes FastAI's `TextDataLoaders` and trains using `text_classifier_learner`.
- **Tabular:** Demonstrates a simple Scikit-learn training pipeline with `RandomForestClassifier`.
- **Custom PyTorch:** Placeholder for custom training logic; implement as per requirements.
7. **UI Enhancements:**
- Adjusted various UI components to be dynamic and respond to the selected model type.
- Enhanced error handling and output messaging to inform users of the application's state.
8. **Logging:**
- Enhanced logging to provide detailed insights into the application's operations, making debugging easier.
### Extending to More Model Types:
To further extend the application to support additional model types (e.g., **Tabular** using FastAI or **Custom PyTorch** models), follow these steps:
1. **Add the Model Type to the ComboBox:**
- Update the `model_type_combo` with new model types.
2. **Implement Model Loading Logic:**
- Define how each new model type should be loaded. For example, use `torch.load` for custom PyTorch models or extend FastAI's capabilities for other data types.
3. **Design Training Pipelines:**
- Customize the `train_model` method to handle the specific requirements of training the new model types, including data preprocessing, model architecture selection, and evaluation metrics.
4. **Handle Different Input Types:**
- Ensure that the input handling and prediction logic cater to the data formats required by the new model types.
5. **Update the UI Accordingly:**
- Modify the UI to provide necessary options and feedback based on the new model types.
### Dependencies:
Ensure that all required libraries are installed. You can install missing dependencies using `pip`:
```bash
pip install fastai scikit-learn pillow requests pygobject
```
**Note:** The above code assumes familiarity with FastAI and Scikit-learn. For integrating other frameworks like TensorFlow or custom models, additional adjustments will be necessary.
### Final Remarks:
This enhanced application now supports multiple model types, providing a more versatile tool for users working with different kinds of machine learning models. By following the structured approach outlined above, you can continue to expand the application's capabilities to encompass a broader range of model types and functionalities.