Python for Engineers : 2025

Smart Code Review Assistant

smart_code_review/

│

├── app.py

├── templates/

│ ├── index.html

│ └── result.html

└── uploads/

app.py

import os

from flask import Flask, render_template, request

from pylint import epylint as lint

import openai

app = Flask(__name__)

app.config['UPLOAD_FOLDER'] = 'uploads'

os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)

# 🔑 Set your OpenAI API key

openai.api_key = "YOUR_OPENAI_API_KEY"

def analyze_with_pylint(file_path):

"""Run pylint analysis and return report."""

(pylint_stdout, _) = lint.py_run(file_path + " --disable=R,C", return_std=True)

return pylint_stdout.getvalue()

def analyze_with_ai(code):

"""Send code to OpenAI for smart suggestions."""

prompt = f"""

You are a senior Python reviewer.

Review the following code and provide:

1. Code quality feedback

2. Security issues

3. Performance suggestions

4. Refactoring tips

Code:

{code}

"""

response = openai.ChatCompletion.create(

model="gpt-4o-mini", # or 'gpt-4' if available

messages=[{"role": "user", "content": prompt}],

max_tokens=600,

temperature=0.4,

)

return response.choices[0].message["content"]

@app.route("/", methods=["GET", "POST"])

def index():

if request.method == "POST":

if 'file' not in request.files:

return "No file uploaded"

file = request.files['file']

if file.filename == "":

return "No selected file"

file_path = os.path.join(app.config['UPLOAD_FOLDER'], file.filename)

file.save(file_path)

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

code_content = f.read()

pylint_result = analyze_with_pylint(file_path)

ai_feedback = analyze_with_ai(code_content)

return render_template("result.html",

ai_feedback=ai_feedback,

pylint_result=pylint_result)

return render_template("index.html")

if __name__ == "__main__":

app.run(debug=True)

templates/index.html

<!DOCTYPE html>

<head>

<title>Smart Code Review Assistant</title>

<style>

body { font-family: Arial; margin: 50px; background-color: #f9f9f9; }

h1 { color: #333; }

.upload-box { background: white; padding: 30px; border-radius: 10px; width: 400px; }

input[type=file] { margin: 20px 0; }

button { padding: 10px 20px; background: #4CAF50; color: white; border: none; border-radius: 5px; }

</style>

</head>

<body>

<h1>🤖 Smart Code Review Assistant</h1>

<label>Upload your Python file (.py):</label><br>

<button type="submit">Analyze Code</button>

</form>

</div>

</body>

</html>

templates/result.html

<!DOCTYPE html>

<head>

<title>Review Result</title>

<style>

body { font-family: Arial; margin: 40px; background-color: #f4f4f4; }

pre { background: white; padding: 20px; border-radius: 8px; overflow-x: auto; }

h2 { color: #444; }

</style>

</head>

<body>

<h1>✅ Code Review Results</h1>

<h2>🧠 AI Feedback</h2>

<pre>{{ ai_feedback }}</pre>

<h2>🧩 Pylint Static Analysis</h2>

<pre>{{ pylint_result }}</pre>

</body>

</html>

Data Structure Visualizer

import tkinter as tk

from tkinter import ttk, messagebox

import networkx as nx

import matplotlib

matplotlib.use("TkAgg")

import matplotlib.pyplot as plt

from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg

import time

import threading

# Small sleep used to allow GUI to update between steps (very short)

STEP_DELAY = 0.15

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

# Data structure backends

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

class Stack:

def __init__(self):

self.items = []

def push(self, v):

self.items.append(v)

def pop(self):

if self.items:

return self.items.pop()

return None

def to_list(self):

# top at right

return list(self.items)

class Queue:

def __init__(self):

self.items = []

def enqueue(self, v):

self.items.append(v)

def dequeue(self):

if self.items:

return self.items.pop(0)

return None

def to_list(self):

# front at left

return list(self.items)

class LinkedListNode:

def __init__(self, val):

self.val = val

self.next = None

class LinkedList:

def __init__(self):

self.head = None

def to_list(self):

out = []

cur = self.head

while cur:

out.append(cur.val)

cur = cur.next

return out

def insert_at(self, index, value):

node = LinkedListNode(value)

if index <= 0 or not self.head:

# insert at head

node.next = self.head

self.head = node

return

cur = self.head

i = 0

while cur.next and i < index-1:

cur = cur.next

i += 1

node.next = cur.next

cur.next = node

def delete_at(self, index):

if not self.head:

return None

if index <= 0:

removed = self.head

self.head = self.head.next

return removed.val

cur = self.head

i = 0

while cur.next and i < index-1:

cur = cur.next

i += 1

if cur.next:

removed = cur.next

cur.next = removed.next

return removed.val

return None

class BSTNode:

def __init__(self, val):

self.val = val

self.left = None

self.right = None

class BST:

def __init__(self):

self.root = None

def insert(self, val):

if self.root is None:

self.root = BSTNode(val)

return

cur = self.root

while True:

if val == cur.val:

# ignore duplicates

return

if val < cur.val:

if cur.left:

cur = cur.left

else:

cur.left = BSTNode(val)

return

else:

if cur.right:

cur = cur.right

else:

cur.right = BSTNode(val)

return

def search(self, val):

cur = self.root

while cur:

if val == cur.val:

return True

elif val < cur.val:

cur = cur.left

else:

cur = cur.right

return False

def delete(self, val):

# standard BST delete (recursively)

def _delete(node, key):

if node is None:

return node, None

if key < node.val:

node.left, removed = _delete(node.left, key)

return node, removed

elif key > node.val:

node.right, removed = _delete(node.right, key)

return node, removed

else:

# found node

removed_val = node.val

if node.left is None:

return node.right, removed_val

elif node.right is None:

return node.left, removed_val

else:

# find inorder successor (smallest in right subtree)

succ_parent = node

succ = node.right

while succ.left:

succ_parent = succ

succ = succ.left

node.val = succ.val

node.right, _ = _delete(node.right, succ.val)

return node, removed_val

self.root, removed = _delete(self.root, val)

return removed

def to_graph(self):

# output nodes and edges for visualization

g = nx.DiGraph()

def add_nodes(n):

if not n:

return

g.add_node(str(n.val), label=str(n.val))

if n.left:

g.add_node(str(n.left.val), label=str(n.left.val))

g.add_edge(str(n.val), str(n.left.val))

add_nodes(n.left)

if n.right:

g.add_node(str(n.right.val), label=str(n.right.val))

g.add_edge(str(n.val), str(n.right.val))

add_nodes(n.right)

add_nodes(self.root)

return g

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

# Visualization helpers

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

def draw_stack(ax, stack: Stack):

ax.clear()

items = stack.to_list()

g = nx.DiGraph()

# nodes left->right, top is rightmost

for i, it in enumerate(items):

g.add_node(f"{i}:{it}", label=str(it))

if i > 0:

g.add_edge(f"{i-1}:{items[i-1]}", f"{i}:{it}")

pos = {}

# horizontal positions

for i in range(len(items)):

pos[f"{i}:{items[i]}"] = (i, 0)

nx.draw_networkx_nodes(g, pos, ax=ax, node_color='skyblue', node_size=1800)

labels = {n: g.nodes[n]['label'] for n in g.nodes()}

nx.draw_networkx_labels(g, pos, labels, ax=ax, font_size=12)

nx.draw_networkx_edges(g, pos, ax=ax, arrows=False)

ax.set_title("Stack (top on the right)")

ax.set_axis_off()

def draw_queue(ax, queue: Queue):

ax.clear()

items = queue.to_list()

g = nx.DiGraph()

for i, it in enumerate(items):

g.add_node(f"{i}:{it}", label=str(it))

if i > 0:

g.add_edge(f"{i-1}:{items[i-1]}", f"{i}:{it}")

pos = {f"{i}:{items[i]}": (i, 0) for i in range(len(items))}

nx.draw_networkx_nodes(g, pos, ax=ax, node_color='lightgreen', node_size=1800)

labels = {n: g.nodes[n]['label'] for n in g.nodes()}

nx.draw_networkx_labels(g, pos, labels, ax=ax, font_size=12)

nx.draw_networkx_edges(g, pos, ax=ax, arrows=False)

ax.set_title("Queue (front on the left)")

ax.set_axis_off()

def draw_linked_list(ax, ll: LinkedList):

ax.clear()

items = ll.to_list()

g = nx.DiGraph()

for i, it in enumerate(items):

g.add_node(f"{i}:{it}", label=str(it))

if i > 0:

g.add_edge(f"{i-1}:{items[i-1]}", f"{i}:{it}")

pos = {f"{i}:{items[i]}": (i, 0) for i in range(len(items))}

nx.draw_networkx_nodes(g, pos, ax=ax, node_color='lightyellow', node_size=1600)

labels = {n: g.nodes[n]['label'] for n in g.nodes()}

nx.draw_networkx_labels(g, pos, labels, ax=ax, font_size=12)

nx.draw_networkx_edges(g, pos, ax=ax, arrows=True, arrowstyle='-|>', arrowsize=20)

ax.set_title("Singly Linked List (head on left)")

ax.set_axis_off()

def hierarchical_pos(G, root=None, width=1.0, vert_gap=0.2, vert_loc=0, xcenter=0.5, pos=None, parent=None):

"""

Create a hierarchical position layout for a tree (recursive).

G: networkx graph

root: root node

returns dict node->(x,y)

"""

if pos is None:

pos = {root: (xcenter, vert_loc)}

children = list(G.successors(root))

if len(children) != 0:

dx = width / len(children)

nextx = xcenter - width/2 - dx/2

for child in children:

nextx += dx

pos[child] = (nextx, vert_loc - vert_gap)

hierarchical_pos(G, root=child, width=dx, vert_gap=vert_gap, vert_loc=vert_loc-vert_gap, xcenter=nextx, pos=pos, parent=root)

return pos

def draw_bst(ax, bst: BST):

ax.clear()

g = bst.to_graph()

if g.number_of_nodes() == 0:

ax.text(0.5, 0.5, "Empty BST", horizontalalignment='center')

ax.set_axis_off()

return

# pick root

root = list(g.nodes())[0]

pos = hierarchical_pos(g, root=root, width=1.0, vert_gap=0.2, vert_loc=0.9, xcenter=0.5)

nx.draw_networkx_nodes(g, pos, ax=ax, node_size=1400, node_color='lightcoral')

labels = {n: g.nodes[n]['label'] for n in g.nodes()}

nx.draw_networkx_labels(g, pos, labels, ax=ax, font_size=12)

nx.draw_networkx_edges(g, pos, ax=ax, arrows=True)

ax.set_title("Binary Search Tree (root at top)")

ax.set_axis_off()

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

# GUI Application

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

class DSVisualizerApp:

def __init__(self, root):

self.root = root

self.root.title("Interactive Data Structure Visualizer")

self.root.geometry("1000x700")

self.notebook = ttk.Notebook(root)

self.notebook.pack(fill='both', expand=True)

# Data structures

self.stack = Stack()

self.queue = Queue()

self.ll = LinkedList()

self.bst = BST()

# Create tabs

self.create_stack_tab()

self.create_queue_tab()

self.create_linkedlist_tab()

self.create_bst_tab()

def make_canvas_frame(self, parent):

frame = ttk.Frame(parent)

# matplotlib figure

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

canvas = FigureCanvasTkAgg(fig, master=frame)

canvas_widget = canvas.get_tk_widget()

canvas_widget.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)

return frame, fig, ax, canvas

def add_log(self, text_widget, message):

text_widget.config(state='normal')

text_widget.insert('end', message + "\n")

text_widget.see('end')

text_widget.config(state='disabled')

# ---- Stack Tab ----

def create_stack_tab(self):

tab = ttk.Frame(self.notebook)

self.notebook.add(tab, text="Stack")

left = ttk.Frame(tab)

left.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)

right = ttk.Frame(tab, width=300)

right.pack(side=tk.RIGHT, fill=tk.Y)

canvas_frame, fig, ax, canvas = self.make_canvas_frame(left)

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

self.stack_fig = fig; self.stack_ax = ax; self.stack_canvas = canvas

# Controls

ttk.Label(right, text="Push value:").pack(pady=6)

val_entry = ttk.Entry(right); val_entry.pack(pady=6)

def on_push():

v = val_entry.get().strip()

if not v:

messagebox.showwarning("Input", "Enter a value to push")

return

self.stack.push(v)

self.add_log(stack_log, f"PUSH {v}")

val_entry.delete(0, 'end')

self.redraw_stack()

ttk.Button(right, text="Push", command=on_push).pack(pady=6)

def on_pop():

val = self.stack.pop()

if val is None:

messagebox.showinfo("Stack", "Stack is empty")

else:

self.add_log(stack_log, f"POP {val}")

self.redraw_stack()

ttk.Button(right, text="Pop", command=on_pop).pack(pady=6)

ttk.Button(right, text="Clear", command=lambda: (self.stack.items.clear(), self.redraw_stack())).pack(pady=6)

# Log

ttk.Label(right, text="Operations:").pack(pady=6)

stack_log = tk.Text(right, height=20, width=30, state='disabled')

stack_log.pack(pady=6)

self.redraw_stack()

def redraw_stack(self):

draw_stack(self.stack_ax, self.stack)

self.stack_canvas.draw_idle()

# ---- Queue Tab ----

def create_queue_tab(self):

tab = ttk.Frame(self.notebook)

self.notebook.add(tab, text="Queue")

left = ttk.Frame(tab)

left.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)

right = ttk.Frame(tab, width=300)

right.pack(side=tk.RIGHT, fill=tk.Y)

canvas_frame, fig, ax, canvas = self.make_canvas_frame(left)

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

self.queue_fig = fig; self.queue_ax = ax; self.queue_canvas = canvas

ttk.Label(right, text="Enqueue value:").pack(pady=6)

val_entry = ttk.Entry(right); val_entry.pack(pady=6)

def on_enqueue():

v = val_entry.get().strip()

if not v:

messagebox.showwarning("Input", "Enter a value to enqueue")

return

self.queue.enqueue(v)

self.add_log(queue_log, f"ENQUEUE {v}")

val_entry.delete(0, 'end')

self.redraw_queue()

ttk.Button(right, text="Enqueue", command=on_enqueue).pack(pady=6)

def on_dequeue():

v = self.queue.dequeue()

if v is None:

messagebox.showinfo("Queue", "Queue is empty")

else:

self.add_log(queue_log, f"DEQUEUE {v}")

self.redraw_queue()

ttk.Button(right, text="Dequeue", command=on_dequeue).pack(pady=6)

ttk.Button(right, text="Clear", command=lambda: (self.queue.items.clear(), self.redraw_queue())).pack(pady=6)

ttk.Label(right, text="Operations:").pack(pady=6)

queue_log = tk.Text(right, height=20, width=30, state='disabled')

queue_log.pack(pady=6)

self.redraw_queue()

def redraw_queue(self):

draw_queue(self.queue_ax, self.queue)

self.queue_canvas.draw_idle()

# ---- Linked List Tab ----

def create_linkedlist_tab(self):

tab = ttk.Frame(self.notebook)

self.notebook.add(tab, text="Linked List")

left = ttk.Frame(tab)

left.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)

right = ttk.Frame(tab, width=320)

right.pack(side=tk.RIGHT, fill=tk.Y)

canvas_frame, fig, ax, canvas = self.make_canvas_frame(left)

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

self.ll_fig = fig; self.ll_ax = ax; self.ll_canvas = canvas

ttk.Label(right, text="Value:").pack(pady=6)

val_entry = ttk.Entry(right); val_entry.pack(pady=6)

ttk.Label(right, text="Index (0-based):").pack(pady=6)

idx_entry = ttk.Entry(right); idx_entry.pack(pady=6)

def on_insert():

v = val_entry.get().strip()

idx = idx_entry.get().strip()

if not v:

messagebox.showwarning("Input", "Enter a value")

return

try:

idxi = int(idx) if idx != "" else 0

except:

messagebox.showwarning("Input", "Index must be integer")

return

self.ll.insert_at(idxi, v)

self.add_log(ll_log, f"INSERT {v} at {idxi}")

val_entry.delete(0, 'end'); idx_entry.delete(0,'end')

self.redraw_ll()

ttk.Button(right, text="Insert", command=on_insert).pack(pady=6)

def on_delete():

idx = idx_entry.get().strip()

try:

idxi = int(idx) if idx != "" else 0

except:

messagebox.showwarning("Input", "Index must be integer")

return

removed = self.ll.delete_at(idxi)

if removed is None:

messagebox.showinfo("LinkedList", "No node at that index")

else:

self.add_log(ll_log, f"DELETE {removed} from {idxi}")

val_entry.delete(0, 'end'); idx_entry.delete(0,'end')

self.redraw_ll()

ttk.Button(right, text="Delete", command=on_delete).pack(pady=6)

ttk.Button(right, text="Clear", command=lambda: (self.ll.__init__(), self.redraw_ll())).pack(pady=6)

ttk.Label(right, text="Operations:").pack(pady=6)

ll_log = tk.Text(right, height=20, width=36, state='disabled')

ll_log.pack(pady=6)

self.redraw_ll()

def redraw_ll(self):

draw_linked_list(self.ll_ax, self.ll)

self.ll_canvas.draw_idle()

# ---- BST Tab ----

def create_bst_tab(self):

tab = ttk.Frame(self.notebook)

self.notebook.add(tab, text="Binary Search Tree")

left = ttk.Frame(tab)

left.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)

right = ttk.Frame(tab, width=350)

right.pack(side=tk.RIGHT, fill=tk.Y)

canvas_frame, fig, ax, canvas = self.make_canvas_frame(left)

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

self.bst_fig = fig; self.bst_ax = ax; self.bst_canvas = canvas

ttk.Label(right, text="Value (integer):").pack(pady=6)

val_entry = ttk.Entry(right); val_entry.pack(pady=6)

def on_insert():

v = val_entry.get().strip()

if not v:

messagebox.showwarning("Input", "Enter a value")

return

try:

vi = int(v)

except:

messagebox.showwarning("Input", "BST requires integer values")

return

self.bst.insert(vi)

self.add_log(bst_log, f"INSERT {vi}")

val_entry.delete(0,'end')

self.redraw_bst()

ttk.Button(right, text="Insert", command=on_insert).pack(pady=6)

def on_delete():

v = val_entry.get().strip()

if not v:

messagebox.showwarning("Input", "Enter a value")

return

try:

vi = int(v)

except:

messagebox.showwarning("Input", "BST requires integer values")

return

removed = self.bst.delete(vi)

if removed is None:

messagebox.showinfo("BST", f"Value {vi} not found")

else:

self.add_log(bst_log, f"DELETE {vi}")

val_entry.delete(0,'end')

self.redraw_bst()

ttk.Button(right, text="Delete", command=on_delete).pack(pady=6)

def on_search():

v = val_entry.get().strip()

if not v:

messagebox.showwarning("Input", "Enter a value")

return

try:

vi = int(v)

except:

messagebox.showwarning("Input", "BST requires integer values")

return

found = self.bst.search(vi)

self.add_log(bst_log, f"SEARCH {vi} -> {'Found' if found else 'Not found'}")

messagebox.showinfo("Search", f"{vi} {'found' if found else 'not found'} in BST")

val_entry.delete(0,'end')

# small highlight trick: we can redraw quickly (no highlight implemented)

self.redraw_bst()

ttk.Button(right, text="Search", command=on_search).pack(pady=6)

ttk.Button(right, text="Clear", command=lambda: (self.bst.__init__(), self.redraw_bst())).pack(pady=6)

ttk.Label(right, text="Operations:").pack(pady=6)

bst_log = tk.Text(right, height=18, width=36, state='disabled')

bst_log.pack(pady=6)

self.redraw_bst()

def redraw_bst(self):

draw_bst(self.bst_ax, self.bst)

self.bst_canvas.draw_idle()

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

# Run the app

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

def main():

root = tk.Tk()

app = DSVisualizerApp(root)

root.mainloop()

if __name__ == "__main__":

main()

Expense Forecasting Tool

import pandas as pd

import numpy as np

import matplotlib.pyplot as plt

from sklearn.linear_model import LinearRegression

from sklearn.metrics import mean_absolute_error, r2_score

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

# STEP 1: Create Sample Expense Data

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

# Example: Monthly total expenses (can be replaced with CSV input)

data = {

"Month": [

"Jan", "Feb", "Mar", "Apr", "May", "Jun",

"Jul", "Aug", "Sep", "Oct", "Nov", "Dec"

"Expense": [21000, 19500, 22000, 25000, 24500, 26000, 27500, 28500, 29500, 31000, 32500, 34000]

}

df = pd.DataFrame(data)

df["Month_Number"] = np.arange(1, len(df) + 1)

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

# STEP 2: Train the Model

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

X = df[["Month_Number"]]

y = df["Expense"]

model = LinearRegression()

model.fit(X, y)

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

# STEP 3: Forecast Next 3 Months

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

future_months = np.arange(len(df) + 1, len(df) + 4).reshape(-1, 1)

future_predictions = model.predict(future_months)

# Combine results

forecast_df = pd.DataFrame({

"Month_Number": future_months.flatten(),

"Predicted_Expense": future_predictions

})

print("\n🔮 Expense Forecast:")

print(forecast_df)

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

# STEP 4: Visualize Results

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

plt.figure(figsize=(10, 6))

plt.plot(df["Month_Number"], df["Expense"], marker='o', label="Actual Expenses")

plt.plot(forecast_df["Month_Number"], forecast_df["Predicted_Expense"], marker='o', linestyle='--', color='orange', label="Predicted")

plt.title("Expense Forecasting Tool")

plt.xlabel("Month")

plt.ylabel("Expense (₹)")

plt.legend()

plt.grid(True)

plt.show()

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

# STEP 5: Evaluate Model

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

predictions = model.predict(X)

print("\n📊 Model Evaluation:")

print("MAE:", round(mean_absolute_error(y, predictions), 2))

print("R² Score:", round(r2_score(y, predictions), 3))

AI Flash Fiction Generator

import streamlit as st

from openai import OpenAI

import textwrap

# Initialize API

client = OpenAI(api_key="YOUR_OPENAI_API_KEY")

st.set_page_config(page_title="AI Flash Fiction Generator", page_icon="📖", layout="centered")

st.title("📖 AI Flash Fiction Generator")

st.caption("Enter a theme or mood → Get a 100-word short story instantly!")

theme = st.text_input("✨ Enter a theme (e.g., hope, mystery, space, love):")

if st.button("Generate Story"):

if theme.strip() == "":

st.warning("Please enter a theme!")

else:

with st.spinner("Generating your story..."):

prompt = f"Write a 100-word flash fiction story about '{theme}'. The story should be complete, emotional, and end with a twist."

response = client.chat.completions.create(

model="gpt-3.5-turbo",

messages=[

{"role": "system", "content": "You are a creative storyteller who writes short fiction."},

{"role": "user", "content": prompt}

temperature=0.9,

max_tokens=200

)

story = response.choices[0].message.content.strip()

story = textwrap.fill(story, width=80)

st.subheader("🪄 Your 100-Word Story:")

st.write(story)

st.success("Done! You can regenerate by changing the theme.")

The Top Programming Languages 2025 - IEEE Spectrum

In the “Spectrum” default ranking, which is weighted with the interests of IEEE members in mind, we see that once again Python has the top spot, with the biggest change in the top five being JavaScript’s drop from third place last year to sixth place this year.

Source: https://spectrum.ieee.org/top-programming-languages-2025

Smart Image Color Palette Extractor

import cv2

import numpy as np

from sklearn.cluster import KMeans

from tkinter import Tk, filedialog

from PIL import Image

import matplotlib.pyplot as plt

def rgb_to_hex(rgb):

"""Convert RGB tuple to HEX string."""

return "#{:02x}{:02x}{:02x}".format(int(rgb[0]), int(rgb[1]), int(rgb[2]))

def extract_colors(image_path, num_colors=6):

"""Extract dominant colors from an image using KMeans clustering."""

# Read and convert image

image = cv2.imread(image_path)

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

# Reshape image data for clustering

pixels = image.reshape((-1, 3))

pixels = np.float32(pixels)

# Apply K-Means

kmeans = KMeans(n_clusters=num_colors, random_state=42)

kmeans.fit(pixels)

colors = np.round(kmeans.cluster_centers_).astype(int)

return colors

def display_palette(colors):

"""Display color palette using matplotlib."""

plt.figure(figsize=(10, 2))

plt.axis("off")

# Create palette blocks

for i, color in enumerate(colors):

plt.subplot(1, len(colors), i + 1)

plt.imshow(np.ones((100, 100, 3), dtype=np.uint8) * np.uint8(color))

plt.title(rgb_to_hex(color), fontsize=10)

plt.axis("off")

plt.tight_layout()

plt.show()

def main():

# GUI File Picker

root = Tk()

root.withdraw()

image_path = filedialog.askopenfilename(title="Select an Image",

filetypes=[("Image Files", "*.jpg *.jpeg *.png *.bmp")])

if not image_path:

print("❌ No file selected.")

return

print(f"📸 Analyzing: {image_path}")

colors = extract_colors(image_path, num_colors=6)

print("\n🎨 Dominant Colors:")

for color in colors:

print(f"RGB: {tuple(color)} | HEX: {rgb_to_hex(color)}")

# Show palette visually

display_palette(colors)

if __name__ == "__main__":

main()

Offline Python compiler apps for Android

Pydroid 3:
Features: A complete offline Python 3 IDE with a built-in interpreter, pip package manager, and a custom repository for pre-built packages of scientific libraries like NumPy, SciPy, and TensorFlow.
Other tools: Includes a C, C++, and Fortran compiler, a PDB debugger, and support for Tkinter, Kivy, and PySide6.
Availability: Free on the Google Play Store.
pyIDE:
Features: A free, user-friendly offline Python compiler for writing and running code on your device without an internet connection.
Best for: Beginners looking to practice Python on the go.
Availability: Available for download on the Softonic website.
QPython 3L:
Features: An app that allows you to run Python 3 on your Android device without an internet connection.
Availability: Found on the Google Play Store.

Medical Symptom Checker

APP.PY

import os

import pandas as pd

from flask import Flask, render_template, request

from openai import OpenAI

app = Flask(__name__)

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# Load demo dataset

conditions_df = pd.read_csv("conditions.csv")

@app.route("/", methods=["GET", "POST"])

def index():

suggestions = None

disclaimer = "⚠️ This tool is for educational/demo purposes only. It does not provide medical advice. Please consult a doctor for any health concerns."

if request.method == "POST":

user_symptoms = request.form.get("symptoms").lower()

matched = conditions_df[conditions_df["symptom"].isin(user_symptoms.split(","))]

# Aggregate suggestions from dataset

dataset_suggestions = matched[["possible_condition", "severity"]].drop_duplicates()

# AI-generated reasoning

try:

ai_prompt = f"""

You are a medical assistant (for demo purposes only).

Based on the symptoms: {user_symptoms}, suggest a few possible conditions.

Provide only educational hints, and include a note advising to consult a doctor.

"""

response = client.chat.completions.create(

model="gpt-4o-mini",

messages=[{"role": "user", "content": ai_prompt}],

temperature=0.6,

)

ai_output = response.choices[0].message.content

except Exception as e:

ai_output = f"(AI service unavailable: {e})"

suggestions = {

"user_symptoms": user_symptoms,

"dataset": dataset_suggestions.to_dict(orient="records"),

"ai_output": ai_output,

"disclaimer": disclaimer

}

return render_template("index.html", suggestions=suggestions)

if __name__ == "__main__":

app.run(debug=True)

index.html

<!DOCTYPE html>

<head>

<title>Medical Symptom Checker (Demo)</title>

<style>

body { background: #f8f9fa; }

.container { max-width: 700px; margin-top: 50px; }

.card { box-shadow: 0 0 10px rgba(0,0,0,0.1); }

</style>

</head>

<body>

<h2 class="text-center mb-4">🩺 Medical Symptom Checker (Demo)</h2>

<label for="symptoms" class="form-label">Enter your symptoms (comma-separated):</label>

<button type="submit" class="btn btn-primary mt-3 w-100">Check Possible Conditions</button>

</form>

{% if suggestions %}

<h5>Entered Symptoms:</h5>

<p>{{ suggestions.user_symptoms }}</p>

{% if suggestions.dataset %}

<h5>Dataset-based Suggestions:</h5>

<ul>

{% for item in suggestions.dataset %}

<li>{{ item.possible_condition }} — <strong>{{ item.severity }}</strong> severity</li>

{% endfor %}

</ul>

{% endif %}

<h5>AI Analysis:</h5>

<p>{{ suggestions.ai_output }}</p>

</div>

{% endif %}

</div>

</body>

</html>

Virtual Stock Trading Game

"""

Virtual Stock Trading Game (Streamlit)

Features:

- Simple username registration/login

- Buy / Sell simulated orders at current market price (via yfinance)

- Portfolio view, transaction history

- Leaderboard by total portfolio value

- SQLite persistence

Run:

streamlit run virtual_trading_app.py

"""

import streamlit as st

import yfinance as yf

import pandas as pd

import sqlite3

from datetime import datetime

import altair as alt

import os

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

# Config

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

DB_FILE = "trading.db"

STARTING_CASH = 100000.0 # default starting cash for new users

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

# Database helpers

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

def get_conn():

conn = sqlite3.connect(DB_FILE, check_same_thread=False)

return conn

def init_db():

conn = get_conn()

cur = conn.cursor()

cur.execute("""

CREATE TABLE IF NOT EXISTS users (

id INTEGER PRIMARY KEY AUTOINCREMENT,

username TEXT UNIQUE,

cash REAL,

created_at TEXT

)

""")

cur.execute("""

CREATE TABLE IF NOT EXISTS holdings (

id INTEGER PRIMARY KEY AUTOINCREMENT,

user_id INTEGER,

ticker TEXT,

quantity REAL,

avg_price REAL,

FOREIGN KEY(user_id) REFERENCES users(id)

)

""")

cur.execute("""

CREATE TABLE IF NOT EXISTS transactions (

id INTEGER PRIMARY KEY AUTOINCREMENT,

user_id INTEGER,

ticker TEXT,

quantity REAL,

price REAL,

side TEXT, -- 'BUY' or 'SELL'

timestamp TEXT,

FOREIGN KEY(user_id) REFERENCES users(id)

)

""")

conn.commit()

conn.close()

def create_user(username, starting_cash=STARTING_CASH):

conn = get_conn()

cur = conn.cursor()

now = datetime.utcnow().isoformat()

try:

cur.execute("INSERT INTO users (username, cash, created_at) VALUES (?, ?, ?)",

(username, float(starting_cash), now))

conn.commit()

except sqlite3.IntegrityError:

pass

conn.close()

def get_user(username):

conn = get_conn()

cur = conn.cursor()

cur.execute("SELECT id, username, cash, created_at FROM users WHERE username=?", (username,))

row = cur.fetchone()

conn.close()

if row:

return {"id": row[0], "username": row[1], "cash": row[2], "created_at": row[3]}

return None

def update_cash(user_id, new_cash):

conn = get_conn()

cur = conn.cursor()

cur.execute("UPDATE users SET cash=? WHERE id=?", (new_cash, user_id))

conn.commit()

conn.close()

def get_holdings(user_id):

conn = get_conn()

cur = conn.cursor()

cur.execute("SELECT ticker, quantity, avg_price FROM holdings WHERE user_id=?", (user_id,))

rows = cur.fetchall()

conn.close()

df = pd.DataFrame(rows, columns=["ticker", "quantity", "avg_price"])

if df.empty:

return pd.DataFrame(columns=["ticker", "quantity", "avg_price"])

return df

def upsert_holding(user_id, ticker, qty_delta, trade_price):

"""

Add or update holdings:

- If buying: qty_delta positive -> update quantity and avg_price

- If selling: qty_delta negative -> reduce quantity; if qty becomes 0 remove row

"""

conn = get_conn()

cur = conn.cursor()

cur.execute("SELECT id, quantity, avg_price FROM holdings WHERE user_id=? AND ticker=?", (user_id, ticker))

row = cur.fetchone()

if row:

hid, qty, avg = row

new_qty = qty + qty_delta

if new_qty <= 0.000001:

cur.execute("DELETE FROM holdings WHERE id=?", (hid,))

else:

if qty_delta > 0:

# new weighted avg: (qty*avg + qty_delta*trade_price) / (qty+qty_delta)

new_avg = (qty * avg + qty_delta * trade_price) / (qty + qty_delta)

else:

new_avg = avg

cur.execute("UPDATE holdings SET quantity=?, avg_price=? WHERE id=?", (new_qty, new_avg, hid))

else:

if qty_delta > 0:

cur.execute("INSERT INTO holdings (user_id, ticker, quantity, avg_price) VALUES (?,?,?,?)",

(user_id, ticker, qty_delta, trade_price))

conn.commit()

conn.close()

def record_transaction(user_id, ticker, quantity, price, side):

conn = get_conn()

cur = conn.cursor()

now = datetime.utcnow().isoformat()

cur.execute("INSERT INTO transactions (user_id, ticker, quantity, price, side, timestamp) VALUES (?, ?, ?, ?, ?, ?)",

(user_id, ticker, quantity, price, side, now))

conn.commit()

conn.close()

def get_transactions(user_id, limit=200):

conn = get_conn()

cur = conn.cursor()

cur.execute("SELECT ticker, quantity, price, side, timestamp FROM transactions WHERE user_id=? ORDER BY id DESC LIMIT ?",

(user_id, limit))

rows = cur.fetchall()

conn.close()

df = pd.DataFrame(rows, columns=["ticker", "quantity", "price", "side", "timestamp"])

if df.empty:

return pd.DataFrame(columns=["ticker", "quantity", "price", "side", "timestamp"])

return df

def get_leaderboard(top_n=20):

conn = get_conn()

cur = conn.cursor()

cur.execute("SELECT id, username, cash FROM users")

users = cur.fetchall()

leaderboard = []

for uid, username, cash in users:

# compute portfolio market value

holdings = get_holdings(uid)

total = float(cash)

if not holdings.empty:

tickers = list(holdings["ticker"].unique())

market = fetch_market_prices(tickers)

for _, row in holdings.iterrows():

t = row["ticker"]

q = float(row["quantity"])

price = market.get(t, 0.0)

total += q * price

leaderboard.append({"username": username, "total": total})

conn.close()

lb = pd.DataFrame(leaderboard).sort_values("total", ascending=False).reset_index(drop=True)

return lb.head(top_n)

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

# Market helpers (yfinance)

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

def fetch_price(ticker):

"""

Return latest price (use fast yfinance call). If ticker invalid, raises.

"""

try:

t = yf.Ticker(ticker)

# use fast info if available

price = None

# prefer real-time quote

quote = t.history(period="1d", interval="1m")

if not quote.empty:

price = quote["Close"].iloc[-1]

else:

info = t.info

price = info.get("regularMarketPrice")

if price is None:

raise ValueError("Price not available")

return float(price)

except Exception as e:

raise ValueError(f"Could not fetch price for {ticker}: {e}")

def fetch_market_prices(tickers):

"""

Bulk fetch latest closing prices for a list of tickers using yfinance download -> faster.

Returns dict ticker -> price

"""

out = {}

if not tickers:

return out

# yfinance can handle list

try:

df = yf.download(tickers, period="1d", interval="1m", progress=False)

# df['Close'] may be single or multi-column

if isinstance(df.columns, pd.MultiIndex):

closes = df['Close'].iloc[-1]

for t in tickers:

try:

out[t] = float(closes[t])

except Exception:

out[t] = 0.0

else:

out[tickers[0]] = float(df['Close'].iloc[-1])

except Exception:

# fallback to single fetch

for t in tickers:

try:

out[t] = fetch_price(t)

except Exception:

out[t] = 0.0

return out

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

# Trading logic

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

def attempt_buy(user, ticker, quantity):

"""

Attempt to buy `quantity` shares at current price. Returns (success, message).

"""

try:

qty = float(quantity)

if qty <= 0:

return False, "Quantity must be > 0"

except:

return False, "Invalid quantity"

try:

price = fetch_price(ticker)

except Exception as e:

return False, f"Price fetch error: {e}"

cost = qty * price

if cost > user["cash"] + 1e-9:

return False, f"Insufficient funds: need {cost:.2f}, available {user['cash']:.2f}"

# perform transaction

new_cash = float(user["cash"]) - cost

update_cash(user["id"], new_cash)

upsert_holding(user["id"], ticker.upper(), qty, price)

record_transaction(user["id"], ticker.upper(), qty, price, "BUY")

# refresh user

return True, f"Bought {qty} shares of {ticker.upper()} at {price:.2f} (cost {cost:.2f})"

def attempt_sell(user, ticker, quantity):

try:

qty = float(quantity)

if qty <= 0:

return False, "Quantity must be > 0"

except:

return False, "Invalid quantity"

holdings = get_holdings(user["id"])

if holdings.empty or ticker.upper() not in list(holdings["ticker"].str.upper()):

return False, "No holdings for this ticker"

row = holdings[holdings["ticker"].str.upper() == ticker.upper()].iloc[0]

owned = float(row["quantity"])

if qty > owned + 1e-9:

return False, f"Not enough shares to sell (owned {owned})"

try:

price = fetch_price(ticker)

except Exception as e:

return False, f"Price fetch error: {e}"

proceeds = qty * price

new_cash = float(get_user(user["username"])["cash"]) + proceeds

update_cash(user["id"], new_cash)

upsert_holding(user["id"], ticker.upper(), -qty, price)

record_transaction(user["id"], ticker.upper(), qty, price, "SELL")

return True, f"Sold {qty} shares of {ticker.upper()} at {price:.2f} (proceeds {proceeds:.2f})"

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

# UI

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

def login_ui():

st.sidebar.header("Player Login / Register")

username = st.sidebar.text_input("Enter username", key="login_username")

if st.sidebar.button("Login / Register"):

if not username.strip():

st.sidebar.error("Please enter a username")

return None

create_user(username.strip())

user = get_user(username.strip())

st.session_state["user"] = user

st.sidebar.success(f"Logged in as {user['username']}")

return user

return None

def main_app(user):

st.title("📈 Virtual Stock Trading Game")

st.write("**Simulation only — not financial advice.**")

st.markdown("---")

# show user summary

col1, col2 = st.columns([2,1])

with col1:

st.subheader(f"Hello, {user['username']} 👋")

st.write(f"**Cash:** ${user['cash']:.2f}")

holdings = get_holdings(user["id"])

if holdings.empty:

st.info("You have no holdings yet. Search a ticker and buy to get started.")

else:

st.write("Your holdings:")

# Fetch market prices for tickers

tickers = list(holdings["ticker"].unique())

market = fetch_market_prices([t for t in tickers])

holdings_display = holdings.copy()

holdings_display["market_price"] = holdings_display["ticker"].apply(lambda t: market.get(t, 0.0))

holdings_display["market_value"] = holdings_display["quantity"] * holdings_display["market_price"]

holdings_display["unreal_pnl"] = holdings_display["market_value"] - holdings_display["quantity"] * holdings_display["avg_price"]

st.dataframe(holdings_display.style.format({"quantity":"{:.3f}", "avg_price":"{:.2f}", "market_price":"{:.2f}", "market_value":"{:.2f}", "unreal_pnl":"{:.2f}"}), use_container_width=True)

total_market = holdings_display["market_value"].sum()

st.write(f"Total holdings market value: ${total_market:.2f}")

with col2:

st.subheader("Leaderboard")

lb = get_leaderboard()

if lb.empty:

st.write("No players yet.")

else:

st.table(lb.style.format({"total":"${:,.2f}"}).head(10))

st.markdown("---")

# Trading panel

st.header("Trade")

tcol1, tcol2 = st.columns(2)

with tcol1:

ticker = st.text_input("Ticker (e.g., AAPL)", key="trade_ticker")

qty = st.number_input("Quantity", min_value=0.0, value=1.0, step=1.0, key="trade_qty")

with tcol2:

if st.button("Fetch Price"):

try:

price = fetch_price(ticker)

st.success(f"Price for {ticker.upper()}: ${price:.2f}")

except Exception as e:

st.error(str(e))

if st.button("Buy"):

if not ticker:

st.error("Enter ticker")

else:

ok, msg = attempt_buy(user, ticker, qty)

if ok:

st.success(msg)

# refresh user object

st.session_state["user"] = get_user(user["username"])

else:

st.error(msg)

if st.button("Sell"):

if not ticker:

st.error("Enter ticker")

else:

ok, msg = attempt_sell(user, ticker, qty)

if ok:

st.success(msg)

st.session_state["user"] = get_user(user["username"])

else:

st.error(msg)

st.markdown("---")

# Transaction history and portfolio chart

st.header("Transaction History & Portfolio Value")

tx = get_transactions(user["id"], limit=500)

st.subheader("Recent Transactions")

if tx.empty:

st.info("No transactions yet.")

else:

st.dataframe(tx, use_container_width=True)

# Portfolio value over time (reconstruct from transactions)

st.subheader("Portfolio Value (by re-using transactions)")

# basic reconstruction: assume each transaction timestamp, compute cash and holdings snapshot

# We will create a simple time series from transactions for demo

conn = get_conn()

q = conn.cursor()

q.execute("SELECT timestamp, ticker, quantity, price, side FROM transactions WHERE user_id=? ORDER BY id ASC", (user["id"],))

rows = q.fetchall()

conn.close()

if rows:

df_tx = pd.DataFrame(rows, columns=["timestamp","ticker","quantity","price","side"])

df_tx["timestamp"] = pd.to_datetime(df_tx["timestamp"])

# sample points: we compute portfolio value at each tx time using latest market prices (this is approximate)

records = []

cash = get_user(user["username"])["cash"]

# Instead compute forward: start with starting cash and apply transactions in order to track cash (we need starting cash)

start_user = get_user(user["username"])

# To compute portfolio value over time properly we'd need historic prices at each tx time — skip heavy calls; instead show current portfolio snapshot vs time by trade counts

# So we'll create a simple chart: cumulative invested vs current market value

holdings_now = get_holdings(user["id"])

if not holdings_now.empty:

prices = fetch_market_prices(list(holdings_now["ticker"].unique()))

holdings_now["market_price"] = holdings_now["ticker"].apply(lambda t: prices.get(t, 0.0))

holdings_now["market_value"] = holdings_now["quantity"] * holdings_now["market_price"]

chart_df = holdings_now[["ticker","market_value"]]

chart_df = chart_df.rename(columns={"market_value":"value"})

st.write("Current holdings market values:")

st.dataframe(holdings_now)

chart = alt.Chart(chart_df).mark_bar().encode(x="ticker", y="value")

st.altair_chart(chart, use_container_width=True)

st.markdown("---")

st.sidebar.markdown("## Player Actions")

if st.sidebar.button("Refresh Data"):

st.session_state["user"] = get_user(user["username"])

st.experimental_rerun()

if st.sidebar.button("Log out"):

st.session_state.pop("user", None)

st.experimental_rerun()

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

# App entrypoint

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

def main():

st.set_page_config(page_title="Virtual Stock Trading Game", layout="wide")

init_db()

st.sidebar.title("Virtual Trading")

user = st.session_state.get("user", None)

if not user:

ui_user = login_ui()

if ui_user:

user = ui_user

else:

# refresh user data from DB

user = get_user(user["username"])

st.session_state["user"] = user

if user:

main_app(user)

else:

st.title("Welcome to the Virtual Stock Trading Game")

st.write("Create a username in the left panel to start. You'll receive some starting cash to practice trading.")

st.info("This app uses real market prices via yfinance but only simulates trades with fake money.")

if __name__ == "__main__":

main()

AI Meeting Scheduler Bot

import os

import json

from datetime import datetime, timedelta, time as dtime

from dateutil import parser as dateparse

import pytz

import re

from flask import Flask, redirect, url_for, session, request, render_template_string, flash

from google.oauth2.credentials import Credentials

from google_auth_oauthlib.flow import Flow

from googleapiclient.discovery import build

import nltk

nltk.download("punkt") # ensure tokens are available

# ---------- Config ----------

CLIENT_SECRETS_FILE = "credentials.json" # downloaded from Google Cloud

SCOPES = ["https://www.googleapis.com/auth/calendar.readonly"]

TOKEN_FOLDER = "tokens"

if not os.path.exists(TOKEN_FOLDER):

os.makedirs(TOKEN_FOLDER)

# Flask config

app = Flask(__name__)

app.secret_key = os.environ.get("FLASK_SECRET", "dev-secret") # change in prod

# Make sure redirect URI in Cloud Console matches this

os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1" # only for local dev

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

def token_path_for_email(email):

safe = email.replace("@", "_at_").replace(".", "_dot_")

return os.path.join(TOKEN_FOLDER, f"token_{safe}.json")

def save_credentials(creds: Credentials, email: str):

p = token_path_for_email(email)

with open(p, "w") as f:

f.write(creds.to_json())

def load_credentials(email: str):

p = token_path_for_email(email)

if not os.path.exists(p):

return None

with open(p, "r") as f:

data = json.load(f)

return Credentials.from_authorized_user_info(data, SCOPES)

def create_flow(state=None):

return Flow.from_client_secrets_file(

CLIENT_SECRETS_FILE,

scopes=SCOPES,

redirect_uri=url_for("oauth2callback", _external=True)

)

# ---------- NLP for simple preference parsing ----------

def parse_natural_preferences(text):

"""

Very lightweight preference extraction:

- looks for 'morning', 'afternoon', 'evening', 'tomorrow', 'next week', 'this week'

- returns bias window (start_hour, end_hour) and date-range hints

"""

text = text.lower()

prefs = {"hours": None, "date_hint": None}

if re.search(r"\bmorn(ing)?\b", text):

prefs["hours"] = (8, 12)

elif re.search(r"\bafternoon\b", text):

prefs["hours"] = (13, 17)

elif re.search(r"\bevening\b", text):

prefs["hours"] = (17, 21)

# dates

if "tomorrow" in text:

prefs["date_hint"] = ("tomorrow", 1)

elif "next week" in text:

prefs["date_hint"] = ("next_week", 7)

elif "this week" in text:

prefs["date_hint"] = ("this_week", 0)

# specific dates (try parsing)

found_dates = re.findall(r"\b(?:on\s)?([A-Za-z]{3,9}\s+\d{1,2}(?:st|nd|rd|th)?(?:,\s*\d{4})?)\b", text)

if found_dates:

# take first parseable date

try:

d = dateparse.parse(found_dates[0])

prefs["explicit_date"] = d.date().isoformat()

except Exception:

pass

return prefs

# ---------- Availability logic ----------

def query_freebusy(service, calendar_ids, start_dt, end_dt, timezone="UTC"):

body = {

"timeMin": start_dt.isoformat(),

"timeMax": end_dt.isoformat(),

"items": [{"id": cid} for cid in calendar_ids]

}

resp = service.freebusy().query(body=body).execute()

busy = {}

for cal_id, cal_data in resp["calendars"].items():

busy[cal_id] = cal_data.get("busy", [])

return busy

def invert_busy_to_free(busy_intervals, start_dt, end_dt, min_slot_minutes=30):

"""

Given busy intervals (list of {"start": iso, "end": iso"}), return free intervals between start_dt and end_dt.

"""

tz = pytz.UTC

# Merge busy and compute free windows

intervals = []

for b in busy_intervals:

s = dateparse.parse(b["start"]).astimezone(tz)

e = dateparse.parse(b["end"]).astimezone(tz)

intervals.append((s, e))

# sort and merge overlaps

intervals.sort()

merged = []

for s,e in intervals:

if not merged:

merged.append([s,e])

else:

if s <= merged[-1][1]:

if e > merged[-1][1]:

merged[-1][1] = e

else:

merged.append([s,e])

free_windows = []

cur = start_dt

for s,e in merged:

if s > cur:

if (s - cur).total_seconds() / 60 >= min_slot_minutes:

free_windows.append((cur, s))

if e > cur:

cur = e

if end_dt > cur:

if (end_dt - cur).total_seconds() / 60 >= min_slot_minutes:

free_windows.append((cur, end_dt))

return free_windows

def intersect_free_lists(list_of_free_lists, meeting_duration_minutes=30):

"""

Each free list is a list of (start, end) windows. We want intersections across calendars and then break into slots of meeting_duration_minutes.

Very simple sweep approach.

"""

# flatten all interval endpoints with tags

events = []

for free_list in list_of_free_lists:

for s,e in free_list:

events.append((s, 1))

events.append((e, -1))

# sort by time

events.sort()

needed = len(list_of_free_lists)

cur_count = 0

last_time = None

intersections = []

for t, delta in events:

prev = cur_count

cur_count += delta

if prev < needed and cur_count == needed:

# interval started

last_time = t

elif prev == needed and cur_count < needed and last_time is not None:

# interval ended at t

intersections.append((last_time, t))

last_time = None

# Break intersections into meeting_duration-sized slots

slots = []

for s,e in intersections:

start = s

while start + timedelta(minutes=meeting_duration_minutes) <= e:

slot_end = start + timedelta(minutes=meeting_duration_minutes)

slots.append((start, slot_end))

start = start + timedelta(minutes=meeting_duration_minutes) # non-overlapping contiguous slots

return slots

# ---------- Flask routes ----------

INDEX_HTML = """

<!doctype html>

<title>AI Meeting Scheduler Bot</title>

<h2>AI Meeting Scheduler Bot — Demo</h2>

<p>1) Authorize yourself (and any other calendar accounts you own) via Google OAuth.</p>

<p>2) Add participant calendar emails (must have given access or be your own authorized accounts).</p>

<label>Participant emails (comma separated):</label><br>

<label>Meeting duration (minutes):</label>

<label>Search days ahead (default 7):</label>

<label>Optional email/preference text (paste):</label><br>

<button type="submit">Suggest slots</button>

</form>

<hr>

<p>To authorize a calendar, go to <a href="/authorize">/authorize</a>, sign in and allow calendar access. The app will remember your token locally.</p>

"""

@app.route("/")

def index():

return render_template_string(INDEX_HTML)

@app.route("/authorize")

def authorize():

# start OAuth flow - will ask user for email after consent

flow = create_flow()

auth_url, state = flow.authorization_url(prompt="consent", access_type="offline", include_granted_scopes="true")

session["flow_state"] = state

return redirect(auth_url)

@app.route("/oauth2callback")

def oauth2callback():

state = session.get("flow_state", None)

flow = create_flow(state=state)

flow.fetch_token(authorization_response=request.url)

creds = flow.credentials

# get email of the authenticated user via token info

service = build("oauth2", "v2", credentials=creds)

try:

info = service.userinfo().get().execute()

email = info.get("email")

except Exception:

# fallback: ask user to input an identifier; but for demo we assume success

email = creds.token_uri or "unknown"

# save credentials

save_credentials(creds, email)

return f"Authorized for {email}. You can now close this tab and return to the app (Home)."

@app.route("/suggest", methods=["POST"])

def suggest():

emails_raw = request.form.get("emails", "")

duration = int(request.form.get("duration", "30"))

days = int(request.form.get("days", "7"))

pref_text = request.form.get("pref", "")

# parse emails

emails = [e.strip() for e in emails_raw.split(",") if e.strip()]

if not emails:

return "Please provide at least one participant email (your authorized account or someone who shared calendar)."

# load credentials for each email (must have tokens saved)

creds_for = {}

for e in emails:

creds = load_credentials(e)

if creds is None:

return f"No token found for {e}. Please authorize that account (visit /authorize and sign in with that email)."

creds_for[e] = creds

# timezone & date range (use UTC for simplicity, better: detect user's tz)

tz = pytz.UTC

now = datetime.now(tz)

start_dt = now + timedelta(hours=1) # start searching from +1 hour

end_dt = now + timedelta(days=days)

# parse preferences

prefs = parse_natural_preferences(pref_text)

# adjust hours if prefs provided

if prefs.get("hours"):

pref_start_hour, pref_end_hour = prefs["hours"]

else:

pref_start_hour, pref_end_hour = 9, 17 # default business hours

# Build calendar service for freebusy queries: we can reuse the first user's creds to call freebusy for multiple calendars

# But Google freebusy requires a service with credentials that have access to calendars queried.

# We'll use each user's own service to fetch busy; however freebusy can accept many items in single query if the caller has access.

# For demo: call freebusy per account, query that account's own calendar id primary.

list_free_lists = []

for e, creds in creds_for.items():

service = build("calendar", "v3", credentials=creds)

# use 'primary' for that account

cal_id = "primary"

# Query busy for that calendar

body = {

"timeMin": start_dt.isoformat(),

"timeMax": end_dt.isoformat(),

"items": [{"id": cal_id}]

}

resp = service.freebusy().query(body=body).execute()

busy = resp["calendars"][cal_id].get("busy", [])

free = invert_busy_to_free(busy, start_dt, end_dt, min_slot_minutes=duration)

# apply daily hours restriction: cut free windows to business hours or prefs

filtered_free = []

for s,e in free:

# slice s..e into days and keep only time within pref hours

ptr = s

while ptr < e:

day_end = (ptr.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1))

seg_end = min(e, day_end)

# define allowed window for this day in tz

allowed_start = ptr.replace(hour=pref_start_hour, minute=0, second=0, microsecond=0)

allowed_end = ptr.replace(hour=pref_end_hour, minute=0, second=0, microsecond=0)

# clamp allowed_start to ptr/day start/...

window_start = max(ptr, allowed_start)

window_end = min(seg_end, allowed_end)

if window_end > window_start and (window_end - window_start).total_seconds()/60 >= duration:

filtered_free.append((window_start, window_end))

ptr = seg_end

list_free_lists.append(filtered_free)

# intersect free lists

candidate_slots = intersect_free_lists(list_free_lists, meeting_duration_minutes=duration)

# Format result: show top 20 slots

candidate_slots = sorted(candidate_slots)[:20]

# Render simple HTML response

out = "<h2>Suggested Meeting Slots (UTC)</h2><ol>"

for s,e in candidate_slots:

out += f"<li>{s.isoformat()} → {e.isoformat()}</li>"

out += "</ol>"

if not candidate_slots:

out += "<p><b>No common slots found in that range & preferences. Try increasing days or changing hours.</b></p>"

out += '<p><a href="/">Back</a></p>'

return out

if __name__ == "__main__":

app.run(debug=True)

Blog Pages