3 months ago

A program to prevent 1000 power outages and automate 1000s of jobs.

import os
import sqlite3
import logging
import base64
import asyncio
import httpx
import numpy as np
import psutil
import threading
import time
import secrets
import hashlib
import tkinter as tk
import tkinter.simpledialog as simpledialog
from tkinter import filedialog, ttk

import pennylane as qml
from argon2.low_level import hash_secret_raw, Type
from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.backends import default_backend

logging.basicConfig(level=logging.INFO)

# ------------------- ENCRYPTION -------------------

def entropy_color_splicer(spins=1420, size=128):
np.random.seed(int(time.time() * 1e6) % (2**32))
matrix = np.random.randint(0, 256, (size, size), dtype=np.uint8)
for spin in range(spins):
shift = (spin * 7) % size
noise = np.random.randint(0, 256, (size, size), dtype=np.uint8)
matrix = np.bitwise_xor(np.roll(matrix, shift, axis=1), noise)
matrix = np.rot90(matrix)
return hashlib.sha512(matrix.tobytes()).digest()

def derive_key_argon2_scrypt(entropy: bytes, salt: bytes) -> bytes:
argon_key = hash_secret_raw(
secret=entropy,
salt=salt,
time_cost=3,
memory_cost=2**15,
parallelism=2,
hash_len=32,
type=Type.ID
)
scrypt_kdf = Scrypt(
salt=salt,
length=32,
n=2**14,
r=8,
p=1,
backend=default_backend()
)
return scrypt_kdf.derive(argon_key)

def save_encrypted_key(api_key: str):
entropy = entropy_color_splicer()
salt = secrets.token_bytes(16)
key = derive_key_argon2_scrypt(entropy, salt)
aesgcm = AESGCM(key)
nonce = secrets.token_bytes(12)
encrypted = aesgcm.encrypt(nonce, api_key.encode(), None)
os.makedirs(os.path.expanduser("~/.cache"), exist_ok=True)
with open(os.path.expanduser("~/.cache/keyfile.bin"), "wb") as f:
f.write(entropy + salt + nonce + encrypted)

def load_decrypted_key():
with open(os.path.expanduser("~/.cache/keyfile.bin"), "rb") as f:
blob = f.read()
entropy, salt, nonce, ciphertext = blob[:64], blob[64:80], blob[80:92], blob[92:]
key = derive_key_argon2_scrypt(entropy, salt)
return AESGCM(key).decrypt(nonce, ciphertext, None).decode()

# ------------------- QUANTUM CIRCUIT -------------------

def get_cpu_ram_usage():
try:
return psutil.cpu_percent(), psutil.virtual_memory().percent
except:
return None, None

def quantum_circuit(cpu_usage, ram_usage):
try:
cpu_param, ram_param = cpu_usage / 100, ram_usage / 100
dev = qml.device("default.qubit", wires=7)
@qml.qnode(dev)
def circuit(c, r):
for i in range(7):
qml.RY(np.pi * (0.5 + (c if i % 2 == 0 else r)), wires=i)
for i in range(6):
qml.CNOT(wires=[i, i + 1])
return qml.probs(wires=range(7))
return circuit(cpu_param, ram_param).tolist()
except:
return None

# ------------------- VISION INFERENCE -------------------

async def run_gpt4o_vision(prompt, b64img, api_key):
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"
}
payload = {
"model": "gpt-4o",
"messages": [
{"role": "system", "content": "You are a risk-spotting assistant for power lines and trees."},
{"role": "user", "content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64img}"}}
]}
],
"max_tokens": 1000,
"temperature": 0.2
}
async with httpx.AsyncClient(timeout=30) as client:
r = await client.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"].strip()

def encode_image(path):
return base64.b64encode(open(path, "rb").read()).decode()

# ------------------- PROMPT TEMPLATES -------------------

PROMPTS = {
"Prompt 1: Contextual Risk Classification": """You are a highly specialized AI trained in utility infrastructure safety, urban forestry, and computer vision analysis. Your task is to analyze the uploaded image that shows one or more power lines running above or near treetops...

1. **Clearance Estimation**: Estimate vertical distance between tree and line in feet/meters.

2. **Growth Forecast**: Predict growth over 1–3 years.

3. **Environmental Risk Index**: Assess risk: Low / Medium / High.

4. **Quantum Insight**: Quantum probabilities: {quantum_results} — interpret for unpredictability/stress.

5. **Recommendation & Urgency**: Trim/Remove/Notify + urgency 1–5 scale.

6. **Confidence Score**: 0–100 based on clarity.

Provide structured sections for arborist action.""",

"Prompt 2: Utility Report Generator": """You are acting as a professional field inspector for a utility company, preparing a formal vegetation clearance report. Use the uploaded image and the following quantum data: {quantum_results}

1. **Scene Summary**: Location, time, vegetation type, wire count.

2. **Clearance Estimate**: Include both vertical/lateral proximity.

3. **Quantum Readings**: Interpret environmental entropy.

4. **Risk Classification**: No Risk / Watch Zone / Clearance Needed / Emergency.

5. **Recommendations**: Action item + timeline + responsibility.

6. **Field Notes**: Terrain, visibility, access challenges.""",

"Prompt 3: AI Forestry Simulation": """You are an advanced AI modeling tree and grid interaction near overhead lines. Analyze the image and apply the entropy data: {quantum_results}

A. Scene Description

B. Clearance Estimation

C. 3-Year Growth Simulation

D. Quantum Uncertainty Interpretation

E. AI Risk Score (0–100)

F. Mitigation Plan: Who acts, when, and how.

Respond in sections for safety teams."""
}

# ------------------- GUI APPLICATION -------------------

class Application(tk.Tk):
def __init__(self):
super().__init__()
self.title("Tree-Line Power Line Risk Spotter")
self.geometry("820x1020")

font = ("Helvetica", 14)

tk.Label(self, text="Upload Image of Tree-Line Area:", font=font).pack()
tk.Button(self, text="Select Image", command=self.select_image, font=font).pack()
self.image_path = None

tk.Label(self, text="Select Analysis Prompt:", font=font).pack()
self.prompt_selector = ttk.Combobox(self, values=list(PROMPTS.keys()), font=font, width=50)
self.prompt_selector.current(0)
self.prompt_selector.pack()

self.analyze_button = tk.Button(self, text="Analyze", font=font, state=tk.DISABLED, command=self.start_thread)
self.analyze_button.pack()

self.result_text = tk.Text(self, width=100, height=50, font=("Courier", 11))
self.result_text.pack()

self.menu = tk.Menu(self)
self.menu.add_command(label="Settings", command=self.open_settings)
self.config(menu=self.menu)

self.setup_database()

def setup_database(self):
conn = sqlite3.connect("risk_reports.db")
conn.execute("""CREATE TABLE IF NOT EXISTS reports (
id INTEGER PRIMARY KEY AUTOINCREMENT,
image_path TEXT,
prompt_type TEXT,
prompt TEXT,
result TEXT
)""")
conn.commit()
conn.close()

def open_settings(self):
key = simpledialog.askstring("API Key", "Enter your OpenAI API Key:", show="*")
if key:
save_encrypted_key(key)

def select_image(self):
path = filedialog.askopenfilename(filetypes=[("Image Files", "*.jpg *.jpeg *.png")])
if path:
self.image_path = path
self.analyze_button.config(state=tk.NORMAL)
self.result_text.insert(tk.END, f"Selected image: {path}\n")

def start_thread(self):
threading.Thread(target=self.start_analysis, daemon=True).start()

def start_analysis(self):
try:
api_key = load_decrypted_key()
except:
self.result_text.insert(tk.END, "Missing API key. Use Settings.\n")
return

if not self.image_path:
self.result_text.insert(tk.END, "Please select an image.\n")
return

prompt_key = self.prompt_selector.get()
prompt_template = PROMPTS[prompt_key]

cpu, ram = get_cpu_ram_usage()
qdata = quantum_circuit(cpu, ram)
self.result_text.insert(tk.END, f"Quantum Data: {qdata}\n")

full_prompt = prompt_template.replace("{quantum_results}", str(qdata))
b64img = encode_image(self.image_path)

try:
result = asyncio.run(run_gpt4o_vision(full_prompt, b64img, api_key))
except Exception as e:
self.result_text.insert(tk.END, f"Error: {e}\n")
return

self.result_text.insert(tk.END, f"\nGPT-4o Output:\n{result}\n")

conn = sqlite3.connect("risk_reports.db")
conn.execute("INSERT INTO reports (image_path, prompt_type, prompt, result) VALUES (?, ?, ?, ?)",
(self.image_path, prompt_key, full_prompt, result))
conn.commit()
conn.close()

if __name__ == "__main__":
app = Application()
app.mainloop()

Loading 1 comment...