Python for Engineers : 2025

Local ML Model Trainer Interface

import streamlit as st

import pandas as pd

import numpy as np

import matplotlib.pyplot as plt

import seaborn as sns

from sklearn.model_selection import train_test_split

from sklearn.metrics import (

accuracy_score, precision_score, recall_score, f1_score,

mean_squared_error, confusion_matrix

)

from sklearn.preprocessing import StandardScaler

from sklearn.linear_model import LogisticRegression, LinearRegression

from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor

from sklearn.svm import SVC, SVR

from sklearn.neighbors import KNeighborsClassifier, KNeighborsRegressor

st.set_page_config(page_title="Local ML Model Trainer", layout="wide")

st.title("Local ML Model Trainer Interface")

st.write("Upload a dataset → choose an algorithm → train → view results")

# ────────────────────────────────────────────────

# Upload Dataset

# ────────────────────────────────────────────────

uploaded_file = st.file_uploader("📤 Upload CSV Dataset", type=["csv"])

if uploaded_file:

df = pd.read_csv(uploaded_file)

st.success("Dataset Loaded Successfully!")

st.write("### Data Preview")

st.dataframe(df.head())

st.write("### Dataset Info")

st.write(df.describe())

# Target column selection

target_col = st.selectbox(" Select Target Column (Y)", df.columns)

# Feature columns

X = df.drop(columns=[target_col])

y = df[target_col]

# Auto detect problem type

if df[target_col].dtype == object or df[target_col].nunique() < 15:

problem_type = "classification"

else:

problem_type = "regression"

st.info(f"Detected Problem Type: **{problem_type.upper()}**")

# Choose model based on problem type

if problem_type == "classification":

model_choice = st.selectbox(

"Choose Model",

["Logistic Regression", "Random Forest Classifier", "SVM Classifier", "KNN Classifier"]

)

else:

model_choice = st.selectbox(

"Choose Model",

["Linear Regression", "Random Forest Regressor", "SVM Regressor", "KNN Regressor"]

)

test_size = st.slider("Test Size (Train %)", 0.1, 0.5, 0.2)

# Train button

if st.button(" Train Model"):

# Preprocessing

scaler = StandardScaler()

X_scaled = scaler.fit_transform(X.select_dtypes(include=np.number))

X_train, X_test, y_train, y_test = train_test_split(

X_scaled, y, test_size=test_size, random_state=42

)

# Model Selection

if model_choice == "Logistic Regression":

model = LogisticRegression()

elif model_choice == "Random Forest Classifier":

model = RandomForestClassifier()

elif model_choice == "SVM Classifier":

model = SVC()

elif model_choice == "KNN Classifier":

model = KNeighborsClassifier()

elif model_choice == "Linear Regression":

model = LinearRegression()

elif model_choice == "Random Forest Regressor":

model = RandomForestRegressor()

elif model_choice == "SVM Regressor":

model = SVR()

elif model_choice == "KNN Regressor":

model = KNeighborsRegressor()

# Train

model.fit(X_train, y_train)

y_pred = model.predict(X_test)

st.success("Model Trained Successfully!")

# ────────────────────────────────────────────────

# Show Metrics

# ────────────────────────────────────────────────

st.write("## 📈 Model Performance")

if problem_type == "classification":

st.write("### 🔹 Classification Metrics")

st.write(f"Accuracy: **{accuracy_score(y_test, y_pred):.4f}**")

st.write(f"Precision: **{precision_score(y_test, y_pred, average='weighted'):.4f}**")

st.write(f"Recall: **{recall_score(y_test, y_pred, average='weighted'):.4f}**")

st.write(f"F1 Score: **{f1_score(y_test, y_pred, average='weighted'):.4f}**")

# Confusion Matrix

cm = confusion_matrix(y_test, y_pred)

fig, ax = plt.subplots(figsize=(5, 4))

sns.heatmap(cm, annot=True, fmt="d", cmap="Blues", ax=ax)

st.write("### Confusion Matrix")

st.pyplot(fig)

else:

st.write("### 🔹 Regression Metrics")

rmse = np.sqrt(mean_squared_error(y_test, y_pred))

st.write(f"RMSE: **{rmse:.4f}**")

# ────────────────────────────────────────────────

# Feature Importance (for tree models)

# ────────────────────────────────────────────────

if "Forest" in model_choice:

st.write("## Feature Importance")

importance = model.feature_importances_

fig, ax = plt.subplots(figsize=(6, 4))

sns.barplot(x=importance, y=X.columns, ax=ax)

st.pyplot(fig)

Automatic Dataset Cleaner

#!/usr/bin/env python3

"""

Automatic Dataset Cleaner

Usage:

python automatic_dataset_cleaner.py --file /path/to/data.csv

python automatic_dataset_cleaner.py --file /path/to/data.csv --missing-strategy mean --outliers cap --encode --scale standard

Outputs:

- /path/to/data_cleaned.csv (cleaned dataset)

- /path/to/data_cleaning_report.json (summary of cleaning actions)

"""

import argparse

import pandas as pd

import numpy as np

import json

import os

from datetime import datetime

from sklearn.preprocessing import StandardScaler, MinMaxScaler

# ---------- Helpers ----------

def read_csv(path):

df = pd.read_csv(path)

return df

def save_csv(df, path):

df.to_csv(path, index=False)

def save_json(obj, path):

with open(path, "w", encoding="utf-8") as f:

json.dump(obj, f, indent=2, default=str)

def summary_stats(df):

return {

"rows": int(df.shape[0]),

"columns": int(df.shape[1]),

"missing_per_column": df.isnull().sum().to_dict(),

"dtypes": {c: str(t) for c, t in df.dtypes.items()},

"sample_head": df.head(3).to_dict(orient="records")

}

def auto_cast_columns(df):

"""Try to cast columns to numeric/datetime where appropriate."""

conversions = {}

for col in df.columns:

if df[col].dtype == object:

# try datetime

try:

parsed = pd.to_datetime(df[col], errors="coerce")

non_null = parsed.notnull().sum()

if non_null / max(1, len(parsed)) > 0.6:

df[col] = parsed

conversions[col] = "datetime"

continue

except Exception:

pass

# try numeric

coerced = pd.to_numeric(df[col].str.replace(",", "").replace(" ", ""), errors="coerce")

if coerced.notnull().sum() / max(1, len(coerced)) > 0.6:

df[col] = coerced

conversions[col] = "numeric"

return df, conversions

# ---------- Missing values ----------

def handle_missing(df, strategy="mean", fill_value=None, threshold_drop_col=0.5):

"""

strategy: 'drop-row', 'drop-col', 'mean', 'median', 'mode', 'ffill', 'bfill', 'constant'

threshold_drop_col: if fraction of missing > threshold, drop column

"""

report = {"strategy": strategy, "dropped_columns": [], "details": {}}

# drop columns with too many missing values

missing_frac = df.isnull().mean()

cols_to_drop = missing_frac[missing_frac > threshold_drop_col].index.tolist()

if cols_to_drop:

df = df.drop(columns=cols_to_drop)

report["dropped_columns"] = cols_to_drop

if strategy == "drop-row":

before = len(df)

df = df.dropna(axis=0)

report["rows_dropped"] = before - len(df)

elif strategy == "drop-col":

before_cols = df.shape[1]

df = df.dropna(axis=1)

report["cols_dropped"] = before_cols - df.shape[1]

elif strategy in ("mean", "median", "mode", "ffill", "bfill", "constant"):

for col in df.columns:

if df[col].isnull().any():

if strategy == "mean" and pd.api.types.is_numeric_dtype(df[col]):

val = df[col].mean()

df[col] = df[col].fillna(val)

report["details"][col] = f"filled mean={val}"

elif strategy == "median" and pd.api.types.is_numeric_dtype(df[col]):

val = df[col].median()

df[col] = df[col].fillna(val)

report["details"][col] = f"filled median={val}"

elif strategy == "mode":

mode_val = df[col].mode()

if not mode_val.empty:

val = mode_val.iloc[0]

df[col] = df[col].fillna(val)

report["details"][col] = f"filled mode={val}"

else:

df[col] = df[col].fillna(fill_value)

report["details"][col] = f"filled mode_empty used const={fill_value}"

elif strategy == "ffill":

df[col] = df[col].fillna(method="ffill").fillna(method="bfill")

report["details"][col] = "filled forward/backward"

elif strategy == "bfill":

df[col] = df[col].fillna(method="bfill").fillna(method="ffill")

report["details"][col] = "filled backward/forward"

else: # constant

df[col] = df[col].fillna(fill_value)

report["details"][col] = f"filled const={fill_value}"

else:

raise ValueError("Unknown missing strategy")

return df, report

# ---------- Outlier detection/treatment ----------

def iqr_outlier_bounds(series, k=1.5):

q1 = series.quantile(0.25)

q3 = series.quantile(0.75)

iqr = q3 - q1

low = q1 - k * iqr

high = q3 + k * iqr

return low, high

def handle_outliers(df, method="remove", k=1.5, numeric_only=True):

"""

method: 'remove' (drop rows with outlier), 'cap' (clip to bounds), 'mark' (add boolean column)

returns df, report

"""

report = {"method": method, "columns": {}}

numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist() if numeric_only else df.columns.tolist()

rows_before = df.shape[0]

for col in numeric_cols:

series = df[col].dropna()

if series.empty:

continue

low, high = iqr_outlier_bounds(series, k=k)

is_out = (df[col] < low) | (df[col] > high)

out_count = int(is_out.sum())

if out_count == 0:

continue

report["columns"][col] = {"outliers": out_count, "bounds": (float(low), float(high))}

if method == "remove":

df = df.loc[~is_out]

elif method == "cap":

df[col] = df[col].clip(lower=low, upper=high)

elif method == "mark":

df[f"{col}_outlier"] = is_out.astype(int)

else:

raise ValueError("Unknown outlier method")

rows_after = df.shape[0]

report["rows_before"] = int(rows_before)

report["rows_after"] = int(rows_after)

return df, report

# ---------- Duplicates ----------

def handle_duplicates(df, subset=None, keep="first"):

before = df.shape[0]

df2 = df.drop_duplicates(subset=subset, keep=keep)

after = df2.shape[0]

report = {"rows_before": int(before), "rows_after": int(after), "dropped": int(before-after)}

return df2, report

# ---------- Encoding ----------

def encode_categoricals(df, one_hot=False, max_unique_for_onehot=20):

report = {"encoded_columns": {}}

cat_cols = df.select_dtypes(include=["category", "object"]).columns.tolist()

if not cat_cols:

return df, report

for col in cat_cols:

nunique = df[col].nunique(dropna=False)

if one_hot and nunique <= max_unique_for_onehot:

dummies = pd.get_dummies(df[col].astype(str), prefix=col, dummy_na=True)

df = pd.concat([df.drop(columns=[col]), dummies], axis=1)

report["encoded_columns"][col] = {"method": "one_hot", "new_cols": list(dummies.columns)}

else:

# label encoding (simple mapping)

mapping = {val: i for i, val in enumerate(df[col].astype(str).unique())}

df[col] = df[col].astype(str).map(mapping)

report["encoded_columns"][col] = {"method": "label", "mapping_sample": dict(list(mapping.items())[:10])}

return df, report

# ---------- Scaling ----------

def scale_numeric(df, method="standard"):

numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()

report = {"method": method, "scaled_columns": numeric_cols}

if not numeric_cols:

return df, report

arr = df[numeric_cols].values.astype(float)

if method == "standard":

scaler = StandardScaler()

elif method == "minmax":

scaler = MinMaxScaler()

else:

raise ValueError("Unknown scaling method")

scaled = scaler.fit_transform(arr)

df[numeric_cols] = scaled

return df, report

# ---------- Main cleaning pipeline ----------

def clean_dataset(

input_path,

missing_strategy="mean",

missing_constant=None,

missing_drop_threshold=0.5,

outlier_method="remove",

outlier_k=1.5,

outlier_numeric_only=True,

dedupe_subset=None,

encode=False,

one_hot=False,

scale_method=None

# read

df = read_csv(input_path)

report = {"input_path": input_path, "start_time": str(datetime.now()), "initial_summary": summary_stats(df), "steps": {}}

# auto-cast columns

df, convs = auto_cast_columns(df)

report["steps"]["auto_cast"] = convs

# missing

df, missing_report = handle_missing(df, strategy=missing_strategy, fill_value=missing_constant, threshold_drop_col=missing_drop_threshold)

report["steps"]["missing"] = missing_report

# duplicates

df, dup_report = handle_duplicates(df, subset=dedupe_subset, keep="first")

report["steps"]["duplicates"] = dup_report

# outliers

df, out_report = handle_outliers(df, method=outlier_method, k=outlier_k, numeric_only=outlier_numeric_only)

report["steps"]["outliers"] = out_report

# encode

if encode:

df, enc_report = encode_categoricals(df, one_hot=one_hot)

report["steps"]["encoding"] = enc_report

# scale

if scale_method:

df, scale_report = scale_numeric(df, method=scale_method)

report["steps"]["scaling"] = scale_report

report["final_summary"] = summary_stats(df)

report["end_time"] = str(datetime.now())

# output

base, ext = os.path.splitext(input_path)

cleaned_path = f"{base}_cleaned{ext}"

report_path = f"{base}_cleaning_report.json"

save_csv(df, cleaned_path)

save_json(report, report_path)

return cleaned_path, report_path, report

# ---------- CLI ----------

def parse_args():

p = argparse.ArgumentParser(description="Automatic Dataset Cleaner")

p.add_argument("--file", "-f", required=True, help="Path to CSV file")

p.add_argument("--missing-strategy", default="mean", choices=["drop-row","drop-col","mean","median","mode","ffill","bfill","constant"], help="Missing value strategy")

p.add_argument("--missing-constant", default=None, help="Constant value to fill missing when strategy=constant")

p.add_argument("--missing-drop-threshold", type=float, default=0.5, help="Drop columns with missing fraction > threshold")

p.add_argument("--outliers", default="remove", choices=["remove","cap","mark","none"], help="Outlier handling method")

p.add_argument("--outlier-k", type=float, default=1.5, help="IQR multiplier for outlier detection")

p.add_argument("--dedupe-subset", default=None, help="Comma separated columns to consider for duplicates (default=all columns)")

p.add_argument("--encode", action="store_true", help="Encode categorical columns")

p.add_argument("--one-hot", action="store_true", help="One-hot encode small cardinality categoricals (used with --encode)")

p.add_argument("--scale", default=None, choices=["standard","minmax"], help="Scale numeric columns")

return p.parse_args()

def main_cli():

args = parse_args()

dedupe_subset = args.dedupe_subset.split(",") if args.dedupe_subset else None

outlier_method = args.outliers if args.outliers != "none" else None

cleaned_path, report_path, report = clean_dataset(

input_path=args.file,

missing_strategy=args.missing_strategy,

missing_constant=args.missing_constant,

missing_drop_threshold=args.missing_drop_threshold,

outlier_method=outlier_method,

outlier_k=args.outlier_k,

outlier_numeric_only=True,

dedupe_subset=dedupe_subset,

encode=args.encode,

one_hot=args.one_hot,

scale_method=args.scale

)

print("Cleaning complete.")

print("Cleaned file:", cleaned_path)

print("Report saved to:", report_path)

# Also print a short summary

print(json.dumps({

"rows_before": report["initial_summary"]["rows"],

"rows_after": report["final_summary"]["rows"],

"columns_before": report["initial_summary"]["columns"],

"columns_after": report["final_summary"]["columns"],

"missing_info": report["initial_summary"]["missing_per_column"]

}, indent=2))

if __name__ == "__main__":

main_cli()

Local Chat App Over LAN (Socket-Based)

chat_server.py

import socket

import threading

HOST = "0.0.0.0" # Accept connections from LAN

PORT = 5000

clients = []

usernames = {}

# Broadcast message to all connected clients

def broadcast(msg, sender_conn=None):

for client in clients:

if client != sender_conn:

try:

client.send(msg.encode())

except:

pass

def handle_client(conn, addr):

print(f"[NEW CONNECTION] {addr}")

try:

username = conn.recv(1024).decode()

usernames[conn] = username

broadcast(f" {username} joined the chat!")

print(f"User '{username}' connected.")

while True:

msg = conn.recv(1024).decode()

if not msg:

break

full_msg = f"{username}: {msg}"

print(full_msg)

broadcast(full_msg, conn)

except Exception as e:

print("Error:", e)

finally:

print(f"{addr} disconnected")

clients.remove(conn)

broadcast(f"{usernames[conn]} left the chat.")

conn.close()

def start_server():

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

server.bind((HOST, PORT))

server.listen()

print(f"Server started on {socket.gethostbyname(socket.gethostname())}:{PORT}")

while True:

conn, addr = server.accept()

clients.append(conn)

thread = threading.Thread(target=handle_client, args=(conn, addr))

thread.daemon = True

thread.start()

if __name__ == "__main__":

start_server()

chat_client.py

import socket

import threading

import tkinter as tk

from tkinter import scrolledtext, messagebox

class ChatClient:

def __init__(self, root):

self.root = root

root.title("LAN Chat Client")

root.geometry("400x500")

tk.Label(root, text="Server IP:").pack()

self.ip_entry = tk.Entry(root)

self.ip_entry.pack()

tk.Label(root, text="Username:").pack()

self.username_entry = tk.Entry(root)

self.username_entry.pack()

tk.Button(root, text="Connect", command=self.connect_server).pack(pady=10)

self.chat_area = scrolledtext.ScrolledText(root, state="disabled")

self.chat_area.pack(expand=True, fill="both", pady=10)

self.msg_entry = tk.Entry(root)

self.msg_entry.pack(fill="x")

tk.Button(root, text="Send", command=self.send_message).pack(pady=5)

self.sock = None

self.running = False

def connect_server(self):

ip = self.ip_entry.get()

username = self.username_entry.get()

if not ip or not username:

messagebox.showerror("Error", "Enter IP and Username!")

return

try:

self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

self.sock.connect((ip, 5000))

self.sock.send(username.encode())

except:

messagebox.showerror("Error", "Cannot connect to server!")

return

self.running = True

threading.Thread(target=self.receive_messages, daemon=True).start()

self.chat_area.config(state="normal")

self.chat_area.insert("end", "Connected!\n")

self.chat_area.config(state="disabled")

def receive_messages(self):

while self.running:

try:

msg = self.sock.recv(1024).decode()

if msg:

self.chat_area.config(state="normal")

self.chat_area.insert("end", msg + "\n")

self.chat_area.yview("end")

self.chat_area.config(state="disabled")

except:

break

def send_message(self):

msg = self.msg_entry.get()

if msg and self.sock:

try:

self.sock.send(msg.encode())

self.msg_entry.delete(0, tk.END)

except:

pass

def on_close(self):

self.running = False

if self.sock:

self.sock.close()

self.root.destroy()

if __name__ == "__main__":

root = tk.Tk()

client = ChatClient(root)

root.protocol("WM_DELETE_WINDOW", client.on_close)

root.mainloop()

Library Book Borrowing System

"""

Library Book Borrowing System (Tkinter + SQLite)

Features:

- User Login / Register

- Admin Login

- Borrow / Return Books

- Due Date Tracking

- Overdue Alerts

"""

import sqlite3

import tkinter as tk

from tkinter import ttk, messagebox

from datetime import datetime, timedelta

DB = "library.db"

# ----------------- DATABASE SETUP -----------------

def init_db():

conn = sqlite3.connect(DB)

c = conn.cursor()

# Users Table

c.execute("""

CREATE TABLE IF NOT EXISTS users (

id INTEGER PRIMARY KEY AUTOINCREMENT,

username TEXT UNIQUE,

password TEXT,

role TEXT

)

""")

# Books Table

c.execute("""

CREATE TABLE IF NOT EXISTS books (

id INTEGER PRIMARY KEY AUTOINCREMENT,

title TEXT,

author TEXT,

available INTEGER DEFAULT 1

)

""")

# Borrow Table

c.execute("""

CREATE TABLE IF NOT EXISTS borrowed (

id INTEGER PRIMARY KEY AUTOINCREMENT,

user_id INTEGER,

book_id INTEGER,

borrowed_date TEXT,

due_date TEXT,

FOREIGN KEY(user_id) REFERENCES users(id),

FOREIGN KEY(book_id) REFERENCES books(id)

)

""")

# Default admin

c.execute("SELECT * FROM users WHERE role='admin'")

if not c.fetchone():

c.execute("INSERT INTO users(username, password, role) VALUES('admin','admin','admin')")

print("Default admin created: admin / admin")

conn.commit()

conn.close()

# ----------------- LOGIN WINDOW -----------------

class LoginWindow:

def __init__(self, root):

self.root = root

root.title("Library System - Login")

root.geometry("350x250")

tk.Label(root, text="Username:", font=("Arial", 12)).pack(pady=5)

self.username_entry = tk.Entry(root)

self.username_entry.pack()

tk.Label(root, text="Password:", font=("Arial", 12)).pack(pady=5)

self.password_entry = tk.Entry(root, show="*")

self.password_entry.pack()

tk.Button(root, text="Login", command=self.login).pack(pady=10)

tk.Button(root, text="Register", command=self.register).pack()

def login(self):

username = self.username_entry.get()

password = self.password_entry.get()

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute("SELECT * FROM users WHERE username=? AND password=?", (username, password))

user = c.fetchone()

conn.close()

if user:

role = user[3]

if role == "admin":

AdminDashboard(tk.Toplevel(), user)

else:

UserDashboard(tk.Toplevel(), user)

else:

messagebox.showerror("Error", "Invalid credentials!")

def register(self):

RegisterWindow(tk.Toplevel())

# ----------------- REGISTER WINDOW -----------------

class RegisterWindow:

def __init__(self, root):

self.root = root

root.title("Register")

root.geometry("300x250")

tk.Label(root, text="Create Username").pack(pady=5)

self.user_entry = tk.Entry(root)

self.user_entry.pack()

tk.Label(root, text="Create Password").pack(pady=5)

self.pass_entry = tk.Entry(root, show="*")

self.pass_entry.pack()

tk.Button(root, text="Register", command=self.register_user).pack(pady=10)

def register_user(self):

username = self.user_entry.get()

password = self.pass_entry.get()

conn = sqlite3.connect(DB)

c = conn.cursor()

try:

c.execute("INSERT INTO users(username,password,role) VALUES(?,?,?)",

(username, password, "user"))

conn.commit()

messagebox.showinfo("Success", "Registration complete!")

self.root.destroy()

except:

messagebox.showerror("Error", "Username already exists.")

conn.close()

# ----------------- ADMIN DASHBOARD -----------------

class AdminDashboard:

def __init__(self, root, user):

self.root = root

root.title("Admin Dashboard")

root.geometry("600x500")

tk.Label(root, text="Admin Dashboard", font=("Arial", 16)).pack(pady=10)

tk.Button(root, text="Add Book", width=20, command=self.add_book_window).pack(pady=5)

tk.Button(root, text="Remove Book", width=20, command=self.remove_book_window).pack(pady=5)

tk.Button(root, text="View All Books", width=20, command=self.view_books).pack(pady=5)

tk.Button(root, text="View Users", width=20, command=self.view_users).pack(pady=5)

def add_book_window(self):

win = tk.Toplevel()

win.title("Add Book")

win.geometry("300x200")

tk.Label(win, text="Title").pack()

title = tk.Entry(win)

title.pack()

tk.Label(win, text="Author").pack()

author = tk.Entry(win)

author.pack()

def save():

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute("INSERT INTO books(title, author) VALUES(?,?)", (title.get(), author.get()))

conn.commit()

conn.close()

messagebox.showinfo("Success", "Book Added!")

win.destroy()

tk.Button(win, text="Save", command=save).pack(pady=10)

def remove_book_window(self):

win = tk.Toplevel()

win.title("Remove Book")

win.geometry("300x200")

tk.Label(win, text="Book ID").pack()

book_id = tk.Entry(win)

book_id.pack()

def delete():

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute("DELETE FROM books WHERE id=?", (book_id.get(),))

conn.commit()

conn.close()

messagebox.showinfo("Removed", "Book deleted!")

win.destroy()

tk.Button(win, text="Delete", command=delete).pack(pady=10)

def view_books(self):

BookListWindow(tk.Toplevel(), admin=True)

def view_users(self):

UsersListWindow(tk.Toplevel())

# ----------------- USER DASHBOARD -----------------

class UserDashboard:

def __init__(self, root, user):

self.root = root

self.user = user

root.title("User Dashboard")

root.geometry("600x500")

tk.Label(root, text=f"Welcome {user[1]}", font=("Arial", 16)).pack(pady=10)

tk.Button(root, text="Borrow Book", width=20, command=self.borrow_window).pack(pady=10)

tk.Button(root, text="Return Book", width=20, command=self.return_window).pack(pady=10)

tk.Button(root, text="My Borrowed Books", width=20, command=self.my_books).pack(pady=10)

self.check_due_alerts()

def borrow_window(self):

BookListWindow(tk.Toplevel(), user=self.user)

def return_window(self):

ReturnBookWindow(tk.Toplevel(), self.user)

def my_books(self):

UserBorrowedBooks(tk.Toplevel(), self.user)

def check_due_alerts(self):

conn = sqlite3.connect(DB)

c = conn.cursor()

today = datetime.now()

c.execute("""SELECT books.title, borrowed.due_date

FROM borrowed

JOIN books ON books.id = borrowed.book_id

WHERE borrowed.user_id=?""",

(self.user[0],))

rows = c.fetchall()

conn.close()

alerts = []

for title, due_date in rows:

due = datetime.strptime(due_date, "%Y-%m-%d")

days_left = (due - today).days

if days_left < 0:

alerts.append(f"OVERDUE: {title} (Due {due_date})")

elif days_left <= 2:

alerts.append(f"Due Soon: {title} (Due {due_date})")

if alerts:

messagebox.showwarning("Due Date Alerts", "\n".join(alerts))

# ----------------- BOOK LIST WINDOW -----------------

class BookListWindow:

def __init__(self, root, admin=False, user=None):

self.root = root

self.user = user

root.title("Books List")

root.geometry("600x400")

columns = ("ID","Title","Author","Available")

tree = ttk.Treeview(root, columns=columns, show="headings")

for col in columns:

tree.heading(col, text=col)

tree.pack(fill="both", expand=True)

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute("SELECT * FROM books")

rows = c.fetchall()

conn.close()

for r in rows:

tree.insert("", tk.END, values=r)

if user:

tk.Button(root, text="Borrow Selected", command=lambda: self.borrow(tree)).pack(pady=10)

def borrow(self, tree):

selected = tree.focus()

if not selected:

messagebox.showwarning("Select", "Select a book first!")

return

values = tree.item(selected)["values"]

book_id, title, author, available = values

if available == 0:

messagebox.showerror("Unavailable", "Book already borrowed!")

return

due = (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d")

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute("UPDATE books SET available=0 WHERE id=?", (book_id,))

c.execute("INSERT INTO borrowed(user_id, book_id, borrowed_date, due_date) VALUES(?,?,?,?)",

(self.user[0], book_id, datetime.now().strftime("%Y-%m-%d"), due))

conn.commit()

conn.close()

messagebox.showinfo("Success", f"Book borrowed! Due on {due}")

self.root.destroy()

# ----------------- RETURN BOOK WINDOW -----------------

class ReturnBookWindow:

def __init__(self, root, user):

self.root = root

self.user = user

root.title("Return Book")

root.geometry("500x350")

columns = ("Borrow ID","Book Title","Due Date")

self.tree = ttk.Treeview(root, columns=columns, show="headings")

for col in columns:

self.tree.heading(col, text=col)

self.tree.pack(fill="both", expand=True)

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute("""SELECT borrowed.id, books.title, borrowed.due_date

FROM borrowed

JOIN books ON books.id = borrowed.book_id

WHERE borrowed.user_id=?""",

(user[0],))

rows = c.fetchall()

conn.close()

for r in rows:

self.tree.insert("", tk.END, values=r)

tk.Button(root, text="Return Selected", command=self.return_book).pack(pady=10)

def return_book(self):

selected = self.tree.focus()

if not selected:

messagebox.showwarning("Select", "Select a book!")

return

borrow_id, title, due = self.tree.item(selected)["values"]

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute("SELECT book_id FROM borrowed WHERE id=?", (borrow_id,))

book_id = c.fetchone()[0]

c.execute("DELETE FROM borrowed WHERE id=?", (borrow_id,))

c.execute("UPDATE books SET available=1 WHERE id=?", (book_id,))

conn.commit()

conn.close()

messagebox.showinfo("Returned", f"{title} returned successfully!")

self.root.destroy()

# ----------------- USER BORROWED LIST -----------------

class UserBorrowedBooks:

def __init__(self, root, user):

self.root = root

root.title("My Borrowed Books")

root.geometry("600x350")

columns = ("Book Title","Borrowed Date","Due Date")

tree = ttk.Treeview(root, columns=columns, show="headings")

for col in columns:

tree.heading(col, text=col)

tree.pack(fill="both", expand=True)

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute("""SELECT books.title, borrowed.borrowed_date, borrowed.due_date

FROM borrowed

JOIN books ON books.id = borrowed.book_id

WHERE borrowed.user_id=?""",

(user[0],))

rows = c.fetchall()

conn.close()

for r in rows:

tree.insert("", tk.END, values=r)

# ----------------- USERS LIST WINDOW (ADMIN) -----------------

class UsersListWindow:

def __init__(self, root):

root.title("All Users")

root.geometry("600x300")

columns = ("ID","Username","Role")

tree = ttk.Treeview(root, columns=columns, show="headings")

for col in columns:

tree.heading(col, text=col)

tree.pack(fill="both", expand=True)

conn = sqlite3.connect(DB)

c = conn.cursor()

c.execute("SELECT id, username, role FROM users")

rows = c.fetchall()

conn.close()

for r in rows:

tree.insert("", tk.END, values=r)

# ----------------- MAIN APP -----------------

if __name__ == "__main__":

init_db()

root = tk.Tk()

LoginWindow(root)

root.mainloop()

Motion Detection Security Recorder

"""

Motion Detection Security Recorder

---------------------------------

Usage:

python motion_recorder.py

python motion_recorder.py --source 0 # default webcam

python motion_recorder.py --source "video.mp4"

python motion_recorder.py --source "rtsp://...."

Outputs:

./recordings/YYYYMMDD_HHMMSS_motion.mp4

./recordings/recording_log.csv (optional)

Notes:

- Adjust MIN_AREA and SENSITIVITY for your camera / scene.

- Pre-buffer keeps recent frames so clip contains seconds before detection.

"""

import cv2

import numpy as np

import argparse

import time

from datetime import datetime

from collections import deque

import os

import csv

# -------------------------

# Configuration (tweakable)

# -------------------------

OUTPUT_DIR = "recordings"

LOG_CSV = True # write a CSV log of recorded clips

FPS = 20 # expected framerate for recording (adjust to your camera)

FRAME_WIDTH = 640 # resize frames for faster processing

FRAME_HEIGHT = 480

MIN_AREA = 1200 # minimum contour area to be considered motion (tweak)

SENSITIVITY = 25 # how much difference triggers motion (lower => more sensitive)

PRE_BUFFER_SECONDS = 3 # include 3 seconds before motion started

POST_RECORD_SECONDS = 4 # record N seconds after motion stops

CODEC = "mp4v" # codec fourcc (try 'XVID' or 'avc1' if 'mp4v' not available)

# -------------------------

# Helpers

# -------------------------

def ensure_dir(path):

if not os.path.exists(path):

os.makedirs(path, exist_ok=True)

def timestamp_str():

return datetime.now().strftime("%Y%m%d_%H%M%S")

def make_output_filename():

return f"{timestamp_str()}_motion.mp4"

def write_log(csv_path, row):

header = ["filename","start_time","end_time","duration_s","frames","source"]

exists = os.path.exists(csv_path)

with open(csv_path, "a", newline="", encoding="utf-8") as f:

w = csv.writer(f)

if not exists:

w.writerow(header)

w.writerow(row)

# -------------------------

# Motion Recorder class

# -------------------------

class MotionRecorder:

def __init__(self, source=0, output_dir=OUTPUT_DIR):

self.source = source

self.output_dir = output_dir

ensure_dir(self.output_dir)

self.log_path = os.path.join(self.output_dir, "recording_log.csv") if LOG_CSV else None

self.cap = cv2.VideoCapture(self.source)

if not self.cap.isOpened():

raise RuntimeError(f"Cannot open source: {source}")

# If camera provides FPS, override

native_fps = self.cap.get(cv2.CAP_PROP_FPS)

if native_fps and native_fps > 0:

self.fps = native_fps

else:

self.fps = FPS

# Use resize dims

self.width = FRAME_WIDTH

self.height = FRAME_HEIGHT

# background subtractor (more robust) + simple diff fallback

self.bg_sub = cv2.createBackgroundSubtractorMOG2(history=500, varThreshold=16, detectShadows=True)

self.pre_buffer = deque(maxlen=int(self.fps * PRE_BUFFER_SECONDS))

self.is_recording = False

self.writer = None

self.record_start_time = None

self.frames_recorded = 0

self.last_motion_time = None

def release(self):

if self.cap:

self.cap.release()

if self.writer:

self.writer.release()

cv2.destroyAllWindows()

def start(self):

print("Starting motion detection. Press 'q' to quit.")

try:

while True:

ret, frame = self.cap.read()

if not ret:

print("Stream ended or cannot fetch frame.")

break

# resize for consistent processing

frame = cv2.resize(frame, (self.width, self.height))

gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

gray_blur = cv2.GaussianBlur(gray, (5,5), 0)

# background subtraction mask

fgmask = self.bg_sub.apply(gray_blur)

# threshold to reduce noise

_, thresh = cv2.threshold(fgmask, SENSITIVITY, 255, cv2.THRESH_BINARY)

# morphological operations to reduce small noise

kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5,5))

clean = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, kernel, iterations=1)

clean = cv2.morphologyEx(clean, cv2.MORPH_DILATE, kernel, iterations=2)

# find contours

contours, _ = cv2.findContours(clean, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

motion_detected = False

for cnt in contours:

if cv2.contourArea(cnt) >= MIN_AREA:

motion_detected = True

(x,y,w,h) = cv2.boundingRect(cnt)

# draw rectangle for preview

cv2.rectangle(frame, (x,y), (x+w, y+h), (0,255,0), 2)

# push frame into pre-buffer (store color frames)

self.pre_buffer.append(frame.copy())

# Recording logic

now = time.time()

if motion_detected:

self.last_motion_time = now

if not self.is_recording:

# start new recording using pre-buffer

fname = make_output_filename()

out_path = os.path.join(self.output_dir, fname)

fourcc = cv2.VideoWriter_fourcc(*CODEC)

self.writer = cv2.VideoWriter(out_path, fourcc, self.fps, (self.width, self.height))

if not self.writer.isOpened():

print("Warning: VideoWriter failed to open. Check codec availability.")

# flush pre-buffer to writer

for bf in list(self.pre_buffer):

if self.writer:

self.writer.write(bf)

self.is_recording = True

self.record_start_time = datetime.now()

self.frames_recorded = len(self.pre_buffer)

self.current_out_path = out_path

print(f"[{self.record_start_time}] Motion started -> Recording to {out_path}")

# If recording, write current frame and manage stop condition

if self.is_recording:

if self.writer:

self.writer.write(frame)

self.frames_recorded += 1

# stop if no motion for POST_RECORD_SECONDS

if self.last_motion_time and (now - self.last_motion_time) > POST_RECORD_SECONDS:

# finalize

record_end = datetime.now()

duration = (record_end - self.record_start_time).total_seconds()

print(f"[{record_end}] Motion ended. Duration: {duration:.2f}s, Frames: {self.frames_recorded}")

# close writer

if self.writer:

self.writer.release()

self.writer = None

# write log

if LOG_CSV and self.log_path:

write_log(self.log_path, [

os.path.basename(self.current_out_path),

self.record_start_time.strftime("%Y-%m-%d %H:%M:%S"),

record_end.strftime("%Y-%m-%d %H:%M:%S"),

f"{duration:.2f}",

str(self.frames_recorded),

str(self.source)

])

self.is_recording = False

self.frames_recorded = 0

# clear pre_buffer so next record begins clean

self.pre_buffer.clear()

# Show simple preview window (optional)

preview = frame.copy()

status_text = f"REC" if self.is_recording else "Idle"

cv2.putText(preview, f"Status: {status_text}", (10,20), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0,0,255) if self.is_recording else (0,255,0), 2)

cv2.imshow("Motion Recorder - Preview", preview)

# keyboard quit

key = cv2.waitKey(1) & 0xFF

if key == ord('q'):

print("Quit requested by user.")

break

except KeyboardInterrupt:

print("Interrupted by user.")

finally:

# cleanup

if self.writer:

self.writer.release()

if LOG_CSV and self.log_path:

# if we were recording when interrupted, log end

end_time = datetime.now()

duration = (end_time - self.record_start_time).total_seconds() if self.record_start_time else 0

write_log(self.log_path, [

os.path.basename(self.current_out_path),

self.record_start_time.strftime("%Y-%m-%d %H:%M:%S") if self.record_start_time else "",

end_time.strftime("%Y-%m-%d %H:%M:%S"),

f"{duration:.2f}",

str(self.frames_recorded),

str(self.source)

])

self.release()

print("Released resources. Exiting.")

# -------------------------

# CLI arg parsing

# -------------------------

def parse_args():

parser = argparse.ArgumentParser(description="Motion Detection Security Recorder")

parser.add_argument("--source", type=str, default="0", help="Video source: 0 (webcam), file path, or RTSP URL")

parser.add_argument("--out", type=str, default=OUTPUT_DIR, help="Output recordings folder")

parser.add_argument("--fps", type=int, default=FPS, help="Recording FPS fallback")

parser.add_argument("--w", type=int, default=FRAME_WIDTH, help="Frame width (resize)")

parser.add_argument("--h", type=int, default=FRAME_HEIGHT, help="Frame height (resize)")

parser.add_argument("--min-area", type=int, default=MIN_AREA, help="Min contour area to detect motion")

parser.add_argument("--sensitivity", type=int, default=SENSITIVITY, help="Threshold sensitivity for mask")

args = parser.parse_args()

return args

# -------------------------

# Entrypoint

# -------------------------

def main():

args = parse_args()

source = args.source

# convert "0" -> 0 for webcam

if source.isdigit():

source = int(source)

global FPS, FRAME_WIDTH, FRAME_HEIGHT, MIN_AREA, SENSITIVITY, OUTPUT_DIR

FPS = args.fps

FRAME_WIDTH = args.w

FRAME_HEIGHT = args.h

MIN_AREA = args.min_area

SENSITIVITY = args.sensitivity

OUTPUT_DIR = args.out

ensure_dir(OUTPUT_DIR)

recorder = MotionRecorder(source=source, output_dir=OUTPUT_DIR)

recorder.start()

if __name__ == "__main__":

main()

Reverse Image Search (Local Only)

import os

import cv2

import numpy as np

import pickle

from PIL import Image, ImageTk

import tkinter as tk

from tkinter import ttk, filedialog, messagebox

from pathlib import Path

from math import ceil

# ----------------------

# Config

# ----------------------

SUPPORTED_EXT = (".jpg", ".jpeg", ".png", ".bmp", ".tiff")

CACHE_FILE = "image_features_cache.pkl" # optional cache to speed up indexing

THUMB_SIZE = (200, 150) # thumbnail size for display

# ----------------------

# Feature detector factory (SIFT preferred, fallback to ORB)

# ----------------------

def make_feature_detector():

try:

# try SIFT (requires opencv-contrib)

sift = cv2.SIFT_create()

print("Using SIFT detector")

return ("SIFT", sift)

except Exception:

# fallback to ORB

orb = cv2.ORB_create(nfeatures=1500)

print("SIFT not available — falling back to ORB")

return ("ORB", orb)

# ----------------------

# Matcher factory

# ----------------------

def make_matcher(detector_name):

if detector_name == "SIFT":

# FLANN parameters for SIFT (float descriptors)

FLANN_INDEX_KDTREE = 1

index_params = dict(algorithm=FLANN_INDEX_KDTREE, trees=5)

search_params = dict(checks=50)

matcher = cv2.FlannBasedMatcher(index_params, search_params)

return matcher

else:

# ORB uses Hamming distance (binary descriptors)

matcher = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=False)

return matcher

# ----------------------

# Compute descriptors for one image

# ----------------------

def compute_descriptors(img_path, detector_tuple):

"""

Returns: keypoints, descriptors

"""

detector_name, detector = detector_tuple

img = cv2.imdecode(np.fromfile(str(img_path), dtype=np.uint8), cv2.IMREAD_GRAYSCALE)

if img is None:

raise ValueError(f"Failed to read image: {img_path}")

# optional resize to speed up (keep aspect)

h, w = img.shape

max_dim = 1024

if max(h, w) > max_dim:

scale = max_dim / max(h, w)

img = cv2.resize(img, (int(w*scale), int(h*scale)), interpolation=cv2.INTER_AREA)

kp, des = detector.detectAndCompute(img, None)

if des is None:

# no descriptors found — return empty

des = np.array([], dtype=np.float32).reshape(0, 128 if detector_name == "SIFT" else 32)

return kp, des

# ----------------------

# Indexing folder of images

# ----------------------

def index_folder(folder_path, detector_tuple, cache_enabled=True):

"""

Scans folder for images, computes descriptors, returns list of records:

[ {"path": Path(...), "kp": keypoints, "des": descriptors} ... ]

"""

folder = Path(folder_path)

if not folder.exists() or not folder.is_dir():

raise ValueError("Folder path invalid")

# Try load cache (only if detector matches)

cache = {}

if cache_enabled and os.path.exists(CACHE_FILE):

try:

with open(CACHE_FILE, "rb") as f:

cache = pickle.load(f)

except Exception:

cache = {}

records = []

for p in sorted(folder.iterdir()):

if p.suffix.lower() not in SUPPORTED_EXT:

continue

key = str(p.resolve())

# cached item must match detector name to be reused

cached = cache.get(key)

use_cache = cached and cached.get("detector") == detector_tuple[0]

if use_cache:

rec = {"path": Path(key), "kp": None, "des": cached["descriptors"]}

print("Cache hit:", p.name)

else:

try:

kp, des = compute_descriptors(p, detector_tuple)

except Exception as e:

print("Failed to compute:", p, e)

kp, des = [], np.array([])

rec = {"path": p, "kp": kp, "des": des}

# store to cache

cache[key] = {"detector": detector_tuple[0], "descriptors": des}

records.append(rec)

if cache_enabled:

try:

with open(CACHE_FILE, "wb") as f:

pickle.dump(cache, f)

except Exception as e:

print("Could not write cache:", e)

return records

# ----------------------

# Matching logic

# ----------------------

def match_descriptors(query_des, target_des, matcher, detector_name, ratio_thresh=0.75):

"""

Returns number of good matches using Lowe ratio test for k=2 neighbors.

For ORB (binary), ratio test still works with BFMatcher and knn.

"""

if query_des is None or target_des is None or len(query_des) == 0 or len(target_des) == 0:

return 0, []

# For FLANN with SIFT, descriptors must be float32

if detector_name == "SIFT":

if query_des.dtype != np.float32:

query_des = query_des.astype(np.float32)

if target_des.dtype != np.float32:

target_des = target_des.astype(np.float32)

try:

matches = matcher.knnMatch(query_des, target_des, k=2)

except Exception:

# fallback: use BFMatcher with crossCheck off

bf = cv2.BFMatcher()

raw = bf.match(query_des, target_des)

# treat each as good match (not optimal)

good = raw

return len(good), good

# Apply ratio test

good = []

for m_n in matches:

if len(m_n) < 2:

continue

m, n = m_n

if m.distance < ratio_thresh * n.distance:

good.append(m)

return len(good), good

# ----------------------

# Query: find top K similar

# ----------------------

def find_similar(query_path, records, detector_tuple, top_k=6):

detector_name, _ = detector_tuple

matcher = make_matcher(detector_name)

# compute descriptors for query

qkp, qdes = compute_descriptors(query_path, detector_tuple)

results = []

for rec in records:

tdes = rec["des"]

count, good_matches = match_descriptors(qdes, tdes, matcher, detector_name)

# normalized score: matches / sqrt(size_query * size_target) to penalize huge images

denom = max(1.0, np.sqrt(max(1,len(qdes)) * max(1,len(tdes))))

score = count / denom

results.append({"path": rec["path"], "matches": count, "score": score, "good": good_matches})

# sort by score descending

results = sorted(results, key=lambda r: r["score"], reverse=True)

return results[:top_k]

# ----------------------

# Utilities: load thumbnail for Tkinter display

# ----------------------

def pil_image_from_path(p):

# handle non-ascii paths by reading bytes then PIL

arr = np.fromfile(str(p), dtype=np.uint8)

img = cv2.imdecode(arr, cv2.IMREAD_COLOR)

if img is None:

return None

img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)

pil = Image.fromarray(img)

return pil

# ----------------------

# Tkinter GUI

# ----------------------

class ReverseImageSearchGUI:

def __init__(self, master):

self.master = master

master.title("Reverse Image Search — Local (SIFT/ORB)")

master.geometry("1000x700")

self.detector_tuple = make_feature_detector()

self.records = []

self.indexed_folder = None

# Top controls

top = ttk.Frame(master)

top.pack(side=tk.TOP, fill=tk.X, padx=8, pady=8)

ttk.Button(top, text="Choose Images Folder", command=self.choose_folder).pack(side=tk.LEFT, padx=4)

self.folder_label = ttk.Label(top, text="No folder chosen")

self.folder_label.pack(side=tk.LEFT, padx=6)

ttk.Button(top, text="Index Folder", command=self.index_folder).pack(side=tk.LEFT, padx=6)

ttk.Button(top, text="Choose Query Image", command=self.choose_query).pack(side=tk.LEFT, padx=6)

ttk.Label(top, text="Top K:").pack(side=tk.LEFT, padx=(10,0))

self.topk_var = tk.IntVar(value=6)

ttk.Entry(top, textvariable=self.topk_var, width=4).pack(side=tk.LEFT)

# Query preview + results area

mid = ttk.Frame(master)

mid.pack(fill=tk.BOTH, expand=True, padx=8, pady=6)

left = ttk.Frame(mid, width=300)

left.pack(side=tk.LEFT, fill=tk.Y)

ttk.Label(left, text="Query Image:").pack(anchor="w")

self.query_canvas = tk.Label(left, text="No query selected", width=40, height=12, bg="#ddd")

self.query_canvas.pack(padx=6, pady=6)

ttk.Button(left, text="Clear Cache", command=self.clear_cache).pack(pady=6)

ttk.Button(left, text="Re-index", command=self.reindex).pack(pady=6)

right = ttk.Frame(mid)

right.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True)

ttk.Label(right, text="Top Matches:").pack(anchor="w")

self.results_frame = ttk.Frame(right)

self.results_frame.pack(fill=tk.BOTH, expand=True)

# status

self.status_var = tk.StringVar(value="Ready")

ttk.Label(master, textvariable=self.status_var).pack(side=tk.BOTTOM, fill=tk.X)

def choose_folder(self):

folder = filedialog.askdirectory()

if not folder:

return

self.indexed_folder = folder

self.folder_label.config(text=folder)

self.status_var.set(f"Selected folder: {folder}")

def index_folder(self):

if not self.indexed_folder:

messagebox.showwarning("Pick folder", "Choose images folder first")

return

self.status_var.set("Indexing folder (computing descriptors)...")

self.master.update_idletasks()

try:

self.records = index_folder(self.indexed_folder, self.detector_tuple, cache_enabled=True)

self.status_var.set(f"Indexed {len(self.records)} images")

messagebox.showinfo("Indexed", f"Indexed {len(self.records)} images.")

except Exception as e:

messagebox.showerror("Indexing failed", str(e))

self.status_var.set("Indexing failed")

def reindex(self):

if not self.indexed_folder:

messagebox.showwarning("Pick folder", "Choose images folder first")

return

# delete cache and re-index

try:

if os.path.exists(CACHE_FILE):

os.remove(CACHE_FILE)

except:

pass

self.index_folder()

def choose_query(self):

q = filedialog.askopenfilename(filetypes=[("Images", "*.jpg *.jpeg *.png *.bmp *.tiff")])

if not q:

return

self.query_path = Path(q)

pil = pil_image_from_path(q)

if pil is None:

messagebox.showerror("Error", "Could not load image")

return

thumb = pil.copy()

thumb.thumbnail(THUMB_SIZE)

tkimg = ImageTk.PhotoImage(thumb)

self.query_canvas.image = tkimg

self.query_canvas.config(image=tkimg, text="")

self.status_var.set(f"Query: {os.path.basename(q)}")

# Run search if indexed

if not self.records:

if messagebox.askyesno("Not indexed", "Folder not indexed yet. Index now?"):

self.index_folder()

else:

return

self.search_query(q)

def search_query(self, qpath):

self.status_var.set("Searching for similar images...")

self.master.update_idletasks()

try:

topk = max(1, int(self.topk_var.get()))

except:

topk = 6

results = find_similar(qpath, self.records, self.detector_tuple, top_k=topk)

# Clear previous results

for w in self.results_frame.winfo_children():

w.destroy()

# Display results in grid

cols = min(3, topk)

r = 0; c = 0

for idx, res in enumerate(results):

path = res["path"]

score = res["score"]

matches = res["matches"]

pil = pil_image_from_path(path)

if pil is None:

continue

thumb = pil.copy()

thumb.thumbnail(THUMB_SIZE)

tkimg = ImageTk.PhotoImage(thumb)

panel = ttk.Frame(self.results_frame, relief=tk.RIDGE, borderwidth=1)

panel.grid(row=r, column=c, padx=6, pady=6, sticky="nsew")

lbl = tk.Label(panel, image=tkimg)

lbl.image = tkimg

lbl.pack()

info = ttk.Label(panel, text=f"{path.name}\nScore:{score:.3f}\nMatches:{matches}", anchor="center")

info.pack()

# click to open full image in default viewer

def make_open(p=path):

return lambda e=None: os.startfile(str(p)) if os.name == 'nt' else os.system(f'xdg-open "{p}"')

lbl.bind("<Button-1>", make_open(path))

c += 1

if c >= cols:

c = 0

r += 1

self.status_var.set("Search complete")

def clear_cache(self):

if os.path.exists(CACHE_FILE):

try:

os.remove(CACHE_FILE)

messagebox.showinfo("Cache", "Cache file removed")

self.status_var.set("Cache cleared")

except Exception as e:

messagebox.showerror("Error", f"Could not remove cache: {e}")

else:

messagebox.showinfo("Cache", "No cache file present")

# ----------------------

# Run

# ----------------------

def main():

root = tk.Tk()

app = ReverseImageSearchGUI(root)

root.mainloop()

if __name__ == "__main__":

main()

Smart Typing Speed Trainer

import tkinter as tk

from tkinter import ttk, messagebox, simpledialog

import time

import random

import json

import os

STATS_FILE = "typing_stats.json"

SAMPLE_PARAGRAPHS = {

"Easy": [

"The quick brown fox jumps over the lazy dog.",

"Practice makes perfect. Keep typing every day.",

"Sunrise over the hills gave a golden glow."

"Medium": [

"Learning to type faster requires consistent short sessions and focused practice.",

"Productivity improves when repetitive tasks are automated and simplified.",

"A small daily habit can compound into significant improvement over time."

"Hard": [

"Synthesis of knowledge emerges when creativity meets disciplined experimentation and reflection.",

"Contributing to open-source projects accelerates learning through real-world code review and collaboration.",

"Concurrency issues often surface under load where assumptions about ordering and state break down."

]

}

# -----------------------

# Utilities: stats file

# -----------------------

def load_stats():

if not os.path.exists(STATS_FILE):

return []

try:

with open(STATS_FILE, "r", encoding="utf-8") as f:

return json.load(f)

except Exception:

return []

def save_stats(stats):

with open(STATS_FILE, "w", encoding="utf-8") as f:

json.dump(stats, f, indent=2)

# -----------------------

# Typing logic helpers

# -----------------------

def calc_wpm(char_count, elapsed_seconds):

# Standard WPM uses 5 characters per word

minutes = elapsed_seconds / 60.0 if elapsed_seconds > 0 else 1/60.0

return (char_count / 5.0) / minutes

def calc_accuracy(correct_chars, total_typed):

if total_typed == 0:

return 0.0

return (correct_chars / total_typed) * 100.0

# -----------------------

# Main App

# -----------------------

class TypingTrainerApp:

def __init__(self, root):

self.root = root

self.root.title("Smart Typing Speed Trainer")

self.root.geometry("900x600")

self.difficulty = tk.StringVar(value="Easy")

self.username = tk.StringVar(value="Guest")

self.paragraph = ""

self.start_time = None

self.end_time = None

self.running = False

self.correct_chars = 0

self.total_typed = 0

self.errors = 0

self.stats = load_stats()

self.build_ui()

self.new_paragraph()

def build_ui(self):

top = ttk.Frame(self.root)

top.pack(side=tk.TOP, fill=tk.X, padx=8, pady=8)

ttk.Label(top, text="Name:").pack(side=tk.LEFT)

name_entry = ttk.Entry(top, textvariable=self.username, width=18)

name_entry.pack(side=tk.LEFT, padx=6)

ttk.Label(top, text="Difficulty:").pack(side=tk.LEFT, padx=(12,0))

diff_menu = ttk.OptionMenu(top, self.difficulty, self.difficulty.get(), *SAMPLE_PARAGRAPHS.keys())

diff_menu.pack(side=tk.LEFT, padx=6)

ttk.Button(top, text="Next Paragraph", command=self.new_paragraph).pack(side=tk.LEFT, padx=6)

ttk.Button(top, text="Restart", command=self.restart).pack(side=tk.LEFT, padx=6)

ttk.Button(top, text="Save Result", command=self.save_result).pack(side=tk.LEFT, padx=6)

ttk.Button(top, text="Leaderboard", command=self.show_leaderboard).pack(side=tk.LEFT, padx=6)

ttk.Button(top, text="Clear Stats", command=self.clear_stats).pack(side=tk.RIGHT, padx=6)

# Paragraph display

paragraph_frame = ttk.LabelFrame(self.root, text="Type the text below")

paragraph_frame.pack(fill=tk.BOTH, expand=False, padx=10, pady=8)

self.paragraph_text = tk.Text(paragraph_frame, height=6, wrap="word", font=("Consolas", 14), padx=8, pady=8, state="disabled")

self.paragraph_text.pack(fill=tk.BOTH, expand=True)

# Typing area

typing_frame = ttk.LabelFrame(self.root, text="Your typing")

typing_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=8)

self.typing_text = tk.Text(typing_frame, height=10, wrap="word", font=("Consolas", 14), padx=8, pady=8)

self.typing_text.pack(fill=tk.BOTH, expand=True)

self.typing_text.bind("<Key>", self.on_key_press)

# Disable paste

self.typing_text.bind("<<Paste>>", lambda e: "break")

self.typing_text.bind("<Control-v>", lambda e: "break")

self.typing_text.bind("<Button-2>", lambda e: "break")

# Stats

stats_frame = ttk.Frame(self.root)

stats_frame.pack(fill=tk.X, padx=10, pady=6)

self.wpm_var = tk.StringVar(value="WPM: 0.0")

self.acc_var = tk.StringVar(value="Accuracy: 0.0%")

self.err_var = tk.StringVar(value="Errors: 0")

self.time_var = tk.StringVar(value="Time: 0.0s")

ttk.Label(stats_frame, textvariable=self.wpm_var, font=("Arial", 12)).pack(side=tk.LEFT, padx=8)

ttk.Label(stats_frame, textvariable=self.acc_var, font=("Arial", 12)).pack(side=tk.LEFT, padx=8)

ttk.Label(stats_frame, textvariable=self.err_var, font=("Arial", 12)).pack(side=tk.LEFT, padx=8)

ttk.Label(stats_frame, textvariable=self.time_var, font=("Arial", 12)).pack(side=tk.LEFT, padx=8)

# Progress / highlight correct vs incorrect

lower = ttk.Frame(self.root)

lower.pack(fill=tk.BOTH, expand=False, padx=10, pady=6)

ttk.Label(lower, text="Tips: Start typing to begin the timer. Pasting is disabled.").pack(side=tk.LEFT)

def new_paragraph(self):

self.pause()

diff = self.difficulty.get()

pool = SAMPLE_PARAGRAPHS.get(diff, SAMPLE_PARAGRAPHS["Easy"])

self.paragraph = random.choice(pool)

# show paragraph readonly

self.paragraph_text.config(state="normal")

self.paragraph_text.delete("1.0", tk.END)

self.paragraph_text.insert(tk.END, self.paragraph)

self.paragraph_text.config(state="disabled")

self.restart()

def restart(self):

self.pause()

self.typing_text.delete("1.0", tk.END)

self.start_time = None

self.end_time = None

self.running = False

self.correct_chars = 0

self.total_typed = 0

self.errors = 0

self.update_stats_display(0.0)

self.typing_text.focus_set()

def pause(self):

self.running = False

def on_key_press(self, event):

# ignore non-printing keys that don't change text (Shift, Ctrl, Alt)

if event.keysym in ("Shift_L","Shift_R","Control_L","Control_R","Alt_L","Alt_R","Caps_Lock","Tab","Escape"):

return

# start timer on first real key

if not self.running:

self.start_time = time.time()

self.running = True

# schedule first update

self.root.after(100, self.periodic_update)

# schedule update immediately after Tk has applied the key to the widget

self.root.after(1, self.evaluate_typing)

def evaluate_typing(self):

typed = self.typing_text.get("1.0", tk.END)[:-1] # drop trailing newline Tk adds

target = self.paragraph

# compute per-character correctness up to typed length

total = len(typed)

correct = 0

errors = 0

for i, ch in enumerate(typed):

if i < len(target) and ch == target[i]:

correct += 1

else:

errors += 1

# count omissions beyond target length as errors too if user keeps typing

if total > len(target):

errors += total - len(target)

self.correct_chars = correct

self.total_typed = total

self.errors = errors

# if user finished (typed length equals paragraph length), stop timer

if total >= len(target):

# check final correctness

if correct == len(target):

self.end_time = time.time()

self.running = False

self.update_stats_display(final=True)

messagebox.showinfo("Completed", f"Well done, {self.username.get()}!\nYou finished the paragraph.")

return

# otherwise update live stats

self.update_stats_display(final=False)

def periodic_update(self):

if self.running:

self.update_stats_display(final=False)

self.root.after(200, self.periodic_update)

def update_stats_display(self, final=False):

elapsed = (self.end_time - self.start_time) if (self.start_time and self.end_time) else (time.time() - self.start_time if self.start_time else 0.0)

wpm = calc_wpm(self.total_typed, elapsed) if self.total_typed > 0 else 0.0

acc = calc_accuracy(self.correct_chars, self.total_typed) if self.total_typed > 0 else 0.0

self.wpm_var.set(f"WPM: {wpm:.1f}")

self.acc_var.set(f"Accuracy: {acc:.1f}%")

self.err_var.set(f"Errors: {self.errors}")

self.time_var.set(f"Time: {elapsed:.1f}s")

if final and self.start_time:

# show final summary and auto-save to stats (ask username)

pass

def save_result(self):

if not self.start_time:

messagebox.showwarning("No attempt", "Start typing first to record a result.")

return

# If still running, finalize end time

if self.running:

self.end_time = time.time()

self.running = False

self.evaluate_typing()

elapsed = (self.end_time - self.start_time) if (self.start_time and self.end_time) else 0.0

wpm = calc_wpm(self.total_typed, elapsed) if elapsed > 0 else 0.0

acc = calc_accuracy(self.correct_chars, self.total_typed) if self.total_typed > 0 else 0.0

name = self.username.get().strip() or "Guest"

record = {

"name": name,

"difficulty": self.difficulty.get(),

"wpm": round(wpm, 1),

"accuracy": round(acc, 1),

"errors": int(self.errors),

"chars_typed": int(self.total_typed),

"time_seconds": round(elapsed, 1),

"paragraph": self.paragraph

}

self.stats.append(record)

# keep only latest 200

self.stats = self.stats[-200:]

save_stats(self.stats)

messagebox.showinfo("Saved", f"Result saved for {name}.\nWPM={record['wpm']}, Accuracy={record['accuracy']}%")

# refresh leaderboard

self.show_leaderboard()

def show_leaderboard(self):

lb_win = tk.Toplevel(self.root)

lb_win.title("Leaderboard / Recent Attempts")

lb_win.geometry("700x400")

frame = ttk.Frame(lb_win)

frame.pack(fill=tk.BOTH, expand=True, padx=8, pady=8)

cols = ("name", "difficulty", "wpm", "accuracy", "errors", "time_seconds")

tree = ttk.Treeview(frame, columns=cols, show="headings")

for c in cols:

tree.heading(c, text=c.capitalize())

tree.column(c, width=100, anchor="center")

tree.pack(fill=tk.BOTH, expand=True)

# sort by wpm desc

sorted_stats = sorted(self.stats, key=lambda r: r.get("wpm", 0), reverse=True)

for rec in sorted_stats:

tree.insert("", "end", values=(rec["name"], rec["difficulty"], rec["wpm"], rec["accuracy"], rec["errors"], rec["time_seconds"]))

btn_frame = ttk.Frame(lb_win)

btn_frame.pack(fill=tk.X, pady=6)

ttk.Button(btn_frame, text="Close", command=lb_win.destroy).pack(side=tk.RIGHT, padx=6)

def clear_stats(self):

if messagebox.askyesno("Confirm", "Clear all saved stats? This cannot be undone."):

self.stats = []

save_stats(self.stats)

messagebox.showinfo("Cleared", "All saved stats removed.")

# -----------------------

# Run

# -----------------------

def main():

root = tk.Tk()

app = TypingTrainerApp(root)

root.mainloop()

if __name__ == "__main__":

main()

Blog Pages