mirror of
https://github.com/JamesTheGiblet/BuddAI.git
synced 2026-01-08 21:58:40 +00:00
Add unit tests for analytics, fallback client, and refactored validators
- Implemented comprehensive unit tests for the BuddAI Analytics module, covering fallback statistics calculations. - Created tests for the FallbackClient to ensure proper escalation to various AI models and handling of missing API keys. - Developed unit tests for the refactored validator system, validating various hardware and coding standards. - Established a base validator interface and implemented specific validators for ESP32, Arduino, motor control, memory safety, and more. - Enhanced the validator registry to auto-discover and manage validators effectively. - Included detailed validation logic for common issues in embedded systems programming, such as unused variables, safety timeouts, and coding style violations.
This commit is contained in:
JamesTheGiblet
parent
99ef8f5592
commit
d4e09f6d13
43 changed files with 5036 additions and 622 deletions
|
|
@ -35,7 +35,34 @@ class LearningMetrics:
|
|||
"correction_rate": correction_rate,
|
||||
"improvement": self.calculate_trend()
|
||||
}
|
||||
|
||||
|
||||
def get_fallback_stats(self):
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 1. Total Assistant Responses & Escalations
|
||||
cursor.execute("SELECT COUNT(*) FROM messages WHERE role = 'assistant'")
|
||||
total_responses = cursor.fetchone()[0] or 0
|
||||
|
||||
cursor.execute("SELECT COUNT(*) FROM messages WHERE role = 'assistant' AND content LIKE '%Fallback Triggered%'")
|
||||
total_escalations = cursor.fetchone()[0] or 0
|
||||
|
||||
# 2. Learned Rules from Fallback
|
||||
cursor.execute("SELECT COUNT(*) FROM code_rules WHERE learned_from LIKE 'fallback_%'")
|
||||
learned_rules_count = cursor.fetchone()[0] or 0
|
||||
|
||||
conn.close()
|
||||
|
||||
fallback_rate = (total_escalations / total_responses * 100) if total_responses > 0 else 0.0
|
||||
learning_success = (learned_rules_count / total_escalations * 100) if total_escalations > 0 else 0.0
|
||||
|
||||
return {
|
||||
"total_escalations": total_escalations,
|
||||
"fallback_rate": round(fallback_rate, 1),
|
||||
"learning_success": round(learning_success, 1),
|
||||
"most_escalated_topics": []
|
||||
}
|
||||
|
||||
def calculate_trend(self):
|
||||
"""Is BuddAI getting better over time?"""
|
||||
# Compare last 7 days vs previous 7 days
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import os
|
||||
import logging
|
||||
import difflib
|
||||
|
||||
# Optional import for Google Generative AI
|
||||
try:
|
||||
|
|
@ -15,6 +16,13 @@ try:
|
|||
except ImportError:
|
||||
HAS_OPENAI = False
|
||||
|
||||
# Optional import for Anthropic
|
||||
try:
|
||||
import anthropic
|
||||
HAS_CLAUDE = True
|
||||
except ImportError:
|
||||
HAS_CLAUDE = False
|
||||
|
||||
class FallbackClient:
|
||||
"""
|
||||
Handles escalation to external AI models (Gemini, OpenAI) when local confidence is low.
|
||||
|
|
@ -22,9 +30,11 @@ class FallbackClient:
|
|||
def __init__(self):
|
||||
self.gemini_key = os.getenv("GEMINI_API_KEY")
|
||||
self.openai_key = os.getenv("OPENAI_API_KEY")
|
||||
self.claude_key = os.getenv("ANTHROPIC_API_KEY")
|
||||
|
||||
self.gemini_client = None
|
||||
self.openai_client = None
|
||||
self.claude_client = None
|
||||
|
||||
# Initialize Gemini
|
||||
if self.gemini_key and HAS_GEMINI:
|
||||
|
|
@ -41,29 +51,54 @@ class FallbackClient:
|
|||
except Exception as e:
|
||||
print(f"⚠️ Failed to initialize OpenAI client: {e}")
|
||||
|
||||
def escalate(self, model_alias: str, original_prompt: str, buddai_attempt: str, confidence: int) -> str:
|
||||
# Initialize Claude
|
||||
if self.claude_key and HAS_CLAUDE:
|
||||
try:
|
||||
self.claude_client = anthropic.Anthropic(api_key=self.claude_key)
|
||||
except Exception as e:
|
||||
print(f"⚠️ Failed to initialize Claude client: {e}")
|
||||
|
||||
def is_available(self, model_alias: str) -> bool:
|
||||
"""Check if a specific model client is configured and available"""
|
||||
if model_alias == 'gemini':
|
||||
return self.gemini_client is not None
|
||||
elif model_alias in ['gpt4', 'chatgpt']:
|
||||
return self.openai_client is not None
|
||||
elif model_alias == 'claude':
|
||||
return self.claude_client is not None
|
||||
return False
|
||||
|
||||
def escalate(self, model_alias: str, original_prompt: str, buddai_attempt: str, confidence: int, **kwargs) -> str:
|
||||
"""
|
||||
Routes the escalation request to the appropriate provider.
|
||||
"""
|
||||
validation_issues = kwargs.get('validation_issues')
|
||||
|
||||
# Context injection for prompt builder
|
||||
self.hardware_profile = kwargs.get('hardware_profile', 'Generic')
|
||||
self.style_preferences = kwargs.get('style_preferences', 'Standard')
|
||||
|
||||
if model_alias == 'gemini':
|
||||
return self._call_gemini(original_prompt, buddai_attempt, confidence)
|
||||
return self._call_gemini(original_prompt, buddai_attempt, confidence, validation_issues)
|
||||
elif model_alias in ['gpt4', 'chatgpt']:
|
||||
return self._call_openai(model_alias, original_prompt, buddai_attempt, confidence)
|
||||
return self._call_openai(model_alias, original_prompt, buddai_attempt, confidence, validation_issues)
|
||||
elif model_alias == 'claude':
|
||||
return self._call_claude(original_prompt, buddai_attempt, confidence, validation_issues)
|
||||
|
||||
return f"⚠️ Fallback model '{model_alias}' not supported for active escalation."
|
||||
|
||||
def _call_gemini(self, original_prompt: str, buddai_attempt: str, confidence: int) -> str:
|
||||
def _call_gemini(self, original_prompt: str, buddai_attempt: str, confidence: int, validation_issues: list = None) -> str:
|
||||
if not self.gemini_client:
|
||||
return f"⚠️ Gemini fallback unavailable (Key missing or init failed)."
|
||||
|
||||
try:
|
||||
prompt = self._build_prompt(original_prompt, buddai_attempt, confidence)
|
||||
prompt = self.build_fallback_prompt(original_prompt, buddai_attempt, confidence, validation_issues)
|
||||
response = self.gemini_client.generate_content(prompt)
|
||||
return f"✨ **Gemini Fallback (Confidence: {confidence}%)**\n\n{response.text}"
|
||||
except Exception as e:
|
||||
return f"❌ Error calling Gemini API: {str(e)}"
|
||||
|
||||
def _call_openai(self, model_alias: str, original_prompt: str, buddai_attempt: str, confidence: int) -> str:
|
||||
def _call_openai(self, model_alias: str, original_prompt: str, buddai_attempt: str, confidence: int, validation_issues: list = None) -> str:
|
||||
if not self.openai_client:
|
||||
return f"⚠️ OpenAI fallback unavailable (Key missing or init failed)."
|
||||
|
||||
|
|
@ -74,7 +109,7 @@ class FallbackClient:
|
|||
target_model = model_map.get(model_alias, 'gpt-3.5-turbo')
|
||||
|
||||
try:
|
||||
prompt = self._build_prompt(original_prompt, buddai_attempt, confidence)
|
||||
prompt = self.build_fallback_prompt(original_prompt, buddai_attempt, confidence, validation_issues)
|
||||
response = self.openai_client.chat.completions.create(
|
||||
model=target_model,
|
||||
messages=[
|
||||
|
|
@ -86,9 +121,60 @@ class FallbackClient:
|
|||
except Exception as e:
|
||||
return f"❌ Error calling OpenAI API: {str(e)}"
|
||||
|
||||
def _build_prompt(self, original, attempt, confidence):
|
||||
def _call_claude(self, original_prompt: str, buddai_attempt: str, confidence: int, validation_issues: list = None) -> str:
|
||||
if not hasattr(self, 'claude_client') or not self.claude_client:
|
||||
return f"⚠️ Claude fallback unavailable (Key missing or init failed)."
|
||||
|
||||
try:
|
||||
prompt = self.build_fallback_prompt(original_prompt, buddai_attempt, confidence, validation_issues)
|
||||
message = self.claude_client.messages.create(
|
||||
model="claude-3-sonnet-20240229",
|
||||
max_tokens=4096,
|
||||
messages=[{"role": "user", "content": prompt}]
|
||||
)
|
||||
return f"✨ **Claude Fallback (Confidence: {confidence}%)**\n\n{message.content[0].text}"
|
||||
except Exception as e:
|
||||
return f"❌ Error calling Claude API: {str(e)}"
|
||||
|
||||
def build_fallback_prompt(self, user_request, buddai_code, confidence, validation_issues):
|
||||
issues_str = "None"
|
||||
if validation_issues:
|
||||
issues_str = "\n".join([f"- {i.get('message', str(i))}" for i in validation_issues])
|
||||
|
||||
return f"""
|
||||
[USER REQUEST]: {original}
|
||||
[LOCAL ATTEMPT ({confidence}% confidence)]: {attempt}
|
||||
[TASK]: Fix issues, apply best practices, and return corrected code.
|
||||
"""
|
||||
BuddAI attempted this request but confidence is low ({confidence}%).
|
||||
|
||||
Original request: {user_request}
|
||||
|
||||
BuddAI's attempt:
|
||||
{buddai_code}
|
||||
|
||||
Validation issues:
|
||||
{issues_str}
|
||||
|
||||
Please provide improved solution considering:
|
||||
- Hardware: {self.hardware_profile}
|
||||
- User's style: {self.style_preferences}
|
||||
- Forge Theory: Use exponential smoothing where applicable
|
||||
"""
|
||||
|
||||
def extract_learning_patterns(self, buddai_code: str, fallback_code: str) -> list:
|
||||
"""
|
||||
Compare what BuddAI tried vs what Claude provided
|
||||
Extract the key differences
|
||||
"""
|
||||
# Diff the code
|
||||
diff = difflib.unified_diff(
|
||||
buddai_code.splitlines(),
|
||||
fallback_code.splitlines(),
|
||||
lineterm=''
|
||||
)
|
||||
|
||||
patterns = []
|
||||
# Identify new patterns
|
||||
for line in diff:
|
||||
if line.startswith('+') and not line.startswith('+++'):
|
||||
patterns.append(line[1:].strip())
|
||||
|
||||
# Return learnable rules
|
||||
return patterns
|
||||
|
|
@ -349,6 +349,20 @@ class SmartLearner:
|
|||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def store_rule(self, pattern: str, confidence: float, source: str):
|
||||
"""Store a single rule from fallback or other sources"""
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""
|
||||
INSERT INTO code_rules
|
||||
(rule_text, pattern_find, pattern_replace, confidence, learned_from)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
""", (pattern, "", "", confidence, source))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def diff_code(self, original: str, corrected: str) -> str:
|
||||
"""Generate a simple diff"""
|
||||
return "\n".join(difflib.unified_diff(
|
||||
|
|
|
|||
|
|
@ -112,6 +112,20 @@ class StorageManager:
|
|||
)
|
||||
""")
|
||||
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS fallback_solutions (
|
||||
id INTEGER PRIMARY KEY,
|
||||
timestamp TEXT,
|
||||
user_request TEXT,
|
||||
buddai_attempt TEXT,
|
||||
buddai_confidence REAL,
|
||||
fallback_model TEXT,
|
||||
fallback_solution TEXT,
|
||||
validation_improved BOOLEAN,
|
||||
learned_pattern TEXT
|
||||
)
|
||||
""")
|
||||
|
||||
# Migrations (Idempotent)
|
||||
try: cursor.execute("ALTER TABLE sessions ADD COLUMN title TEXT")
|
||||
except: pass
|
||||
|
|
|
|||
|
|
@ -1,457 +1,31 @@
|
|||
import re
|
||||
from typing import List, Dict, Tuple, Optional
|
||||
from validators import (
|
||||
ESP32Validator, MotorValidator, ServoValidator, MemoryValidator,
|
||||
ForgeTheoryValidator, TimingValidator, ArduinoValidator, StyleValidator
|
||||
)
|
||||
|
||||
class CodeValidator:
|
||||
"""Validate generated code before showing to user"""
|
||||
|
||||
def find_line(self, code: str, substring: str) -> int:
|
||||
for i, line in enumerate(code.splitlines(), 1):
|
||||
if substring in line:
|
||||
return i
|
||||
return -1
|
||||
|
||||
def has_safety_timeout(self, code: str) -> bool:
|
||||
# Simple heuristic: needs millis, subtraction, and a comparison to a value/constant
|
||||
# We want to avoid matching debounce logic (usually < 100ms)
|
||||
if "millis()" not in code: return False
|
||||
|
||||
# Check for constants like SAFETY_TIMEOUT, MOTOR_TIMEOUT
|
||||
if re.search(r'>\s*[A-Z_]*TIMEOUT', code):
|
||||
return True
|
||||
|
||||
# Check for state machine timeout (Combat Protocol)
|
||||
if "DISARM" in code and "millis" in code and ">" in code:
|
||||
return True
|
||||
|
||||
# Check for numeric literals > 500 (Debounce is usually 50)
|
||||
comparisons = re.findall(r'>\s*(\d+)', code)
|
||||
return any(int(val) > 500 for val in comparisons)
|
||||
|
||||
def matches_style(self, code: str) -> bool:
|
||||
# Placeholder for style matching logic
|
||||
return True
|
||||
|
||||
def apply_style(self, code: str) -> str:
|
||||
# Placeholder for style application
|
||||
return code
|
||||
|
||||
def refactor_loop_to_function(self, code: str) -> str:
|
||||
"""Extract loop body into runSystemLogic()"""
|
||||
loop_match = re.search(r'void\s+loop\s*\(\s*\)\s*\{', code)
|
||||
if not loop_match: return code
|
||||
|
||||
start_idx = loop_match.end()
|
||||
brace_count = 1
|
||||
loop_body_end = -1
|
||||
|
||||
for i, char in enumerate(code[start_idx:], start=start_idx):
|
||||
if char == '{': brace_count += 1
|
||||
elif char == '}': brace_count -= 1
|
||||
|
||||
if brace_count == 0:
|
||||
loop_body_end = i
|
||||
break
|
||||
|
||||
if loop_body_end == -1: return code
|
||||
|
||||
body = code[start_idx:loop_body_end]
|
||||
new_code = code[:start_idx] + "\n runSystemLogic();\n" + code[loop_body_end:]
|
||||
new_code += "\n\nvoid runSystemLogic() {" + body + "}\n"
|
||||
return new_code
|
||||
def __init__(self):
|
||||
self.validators = [
|
||||
ESP32Validator(),
|
||||
MotorValidator(),
|
||||
ServoValidator(),
|
||||
MemoryValidator(),
|
||||
ForgeTheoryValidator(),
|
||||
TimingValidator(),
|
||||
ArduinoValidator(),
|
||||
StyleValidator()
|
||||
]
|
||||
|
||||
def validate(self, code: str, hardware: str, user_message: str = "") -> Tuple[bool, List[Dict]]:
|
||||
"""Check code against known rules"""
|
||||
issues = []
|
||||
for validator in self.validators:
|
||||
issues.extend(validator.validate(code, hardware, user_message))
|
||||
|
||||
# Check 1: ESP32 PWM
|
||||
if "ESP32" in hardware.upper():
|
||||
if "analogWrite" in code:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, "analogWrite"),
|
||||
"message": "ESP32 doesn't support analogWrite(). Use ledcWrite()",
|
||||
"fix": lambda c: c.replace("analogWrite", "ledcWrite")
|
||||
})
|
||||
|
||||
# Check 2: Non-blocking code
|
||||
if "delay(" in code and "motor" in code.lower():
|
||||
issues.append({
|
||||
"severity": "warning",
|
||||
"line": self.find_line(code, "delay"),
|
||||
"message": "Using delay() in motor code blocks safety checks",
|
||||
"fix": lambda c: c # No auto-fix
|
||||
})
|
||||
|
||||
# Check 3: Safety timeout
|
||||
if ("motor" in code.lower() or "servo" in code.lower()):
|
||||
if not self.has_safety_timeout(code):
|
||||
# Context-aware stop logic
|
||||
is_servo = "Servo" in code and "L298N" not in code
|
||||
stop_logic = " // STOP MOTORS\n ledcWrite(0, 0);\n ledcWrite(1, 0);"
|
||||
if is_servo:
|
||||
stop_logic = " // STOP SERVO\n // Implement safe position (e.g. myServo.write(90));"
|
||||
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "Critical: No safety timeout detected (must be > 500ms).",
|
||||
"fix": lambda c, sl=stop_logic: "#define SAFETY_TIMEOUT 5000\nunsigned long lastCommand = 0;\n" + \
|
||||
re.sub(r'(void\s+loop\s*\(\s*\)\s*\{)', \
|
||||
rf'\1\n // [AUTO-FIX] Safety Timeout\n if (millis() - lastCommand > SAFETY_TIMEOUT) {{\n{sl}\n }}\n', c)
|
||||
})
|
||||
|
||||
# Check 4: L298N PWM Pin Misuse
|
||||
pwm_pins = re.findall(r'ledcAttachPin\s*\(\s*(\w+)\s*,', code)
|
||||
for pin in pwm_pins:
|
||||
# Check if digitalWrite is used on this pin
|
||||
if re.search(r'digitalWrite\s*\(\s*' + re.escape(pin) + r'\s*,', code):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, f"digitalWrite({pin}"),
|
||||
"message": f"Conflict: PWM pin '{pin}' used with digitalWrite(). Use ledcWrite() for speed control.",
|
||||
"fix": lambda c, p=pin: re.sub(r'digitalWrite\s*\(\s*' + re.escape(p) + r'\s*,\s*[^)]+\);?', f'// [Fixed] Removed conflicting digitalWrite on PWM pin {p}', c)
|
||||
})
|
||||
|
||||
# Check 5: Broken Debounce Logic (Type Mismatch)
|
||||
# Example: if (buttonState != lastDebounceTime)
|
||||
bad_debounce = re.search(r'if\s*\(\s*\w+\s*[!=]=\s*\w*DebounceTime\s*\)', code)
|
||||
if bad_debounce:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, bad_debounce.group(0)),
|
||||
"message": "Type Mismatch: Comparing button state (int) with time (long).",
|
||||
"fix": lambda c: c.replace(bad_debounce.group(0), "if ((millis() - lastDebounceTime) > debounceDelay)")
|
||||
})
|
||||
|
||||
# Check 6: Safety Timeout Value
|
||||
timeout_match = re.search(r'#define\s+SAFETY_TIMEOUT\s+(\d+)', code)
|
||||
if timeout_match and int(timeout_match.group(1)) > 5000:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, timeout_match.group(0)),
|
||||
"message": f"Safety timeout {timeout_match.group(1)}ms is too long (Max: 5000ms).",
|
||||
"fix": lambda c: re.sub(r'(#define\s+SAFETY_TIMEOUT\s+)\d+', r'\g<1>5000', c)
|
||||
})
|
||||
|
||||
# Check 7: Broken Safety Timer Logic (Static Init)
|
||||
bad_static = re.search(r'static\s+unsigned\s+long\s+(\w+)\s*=\s*millis\(\);', code)
|
||||
if bad_static:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, bad_static.group(0)),
|
||||
"message": "Static timer initialized with millis() prevents reset. Initialize to 0.",
|
||||
"fix": lambda c: c.replace(bad_static.group(0), f"static unsigned long {bad_static.group(1)} = 0;")
|
||||
})
|
||||
|
||||
# Check 8: Incomplete Motor Logic (L298N Validation)
|
||||
# If user explicitly asks for L298N or DC Motor, OR asks for 'motor' without 'servo'
|
||||
is_l298n_request = "l298n" in user_message.lower() or "dc motor" in user_message.lower() or ("motor" in user_message.lower() and "servo" not in user_message.lower())
|
||||
|
||||
if is_l298n_request:
|
||||
# 1. Check for Direction Pins (IN1/IN2)
|
||||
if not re.search(r'(?:#define|const\s+int)\s+\w*(?:IN1|IN2|DIR)\w*', code, re.IGNORECASE):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "Missing L298N Direction Pins (IN1/IN2).",
|
||||
"fix": lambda c: "// [AUTO-FIX] L298N Definitions\n#define IN1 18\n#define IN2 19\n" + c
|
||||
})
|
||||
|
||||
# 2. Check for PWM Pin (ENA)
|
||||
if not re.search(r'(?:#define|const\s+int)\s+\w*(?:ENA|ENB|PWM)\w*', code, re.IGNORECASE):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "Missing L298N PWM Pin (ENA).",
|
||||
"fix": lambda c: "#define ENA 21 // [AUTO-FIX] Missing PWM Pin\n" + c
|
||||
})
|
||||
|
||||
# 3. Check for Direction Control (digitalWrite)
|
||||
if "digitalWrite" not in code:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "L298N requires digitalWrite() for direction control.",
|
||||
"fix": lambda c: re.sub(r'(void\s+loop\s*\(\s*\)\s*\{)', r'\1\n // [AUTO-FIX] Set Direction\n digitalWrite(IN1, HIGH);\n digitalWrite(IN2, LOW);\n', c)
|
||||
})
|
||||
|
||||
# Check 9: Unnecessary Wire.h
|
||||
wire_include = re.search(r'#include\s+[<"]Wire\.h[>"]', code)
|
||||
if wire_include:
|
||||
# Check if Wire is actually used (excluding the include itself)
|
||||
rest_of_code = code.replace(wire_include.group(0), "")
|
||||
if not re.search(r'\bWire\b', rest_of_code):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, wire_include.group(0)),
|
||||
"message": "Unnecessary #include <Wire.h> detected.",
|
||||
"fix": lambda c: re.sub(r'#include\s+[<"]Wire\.h[>"]', '// [Auto-Fix] Removed unnecessary Wire.h', c)
|
||||
})
|
||||
|
||||
# Check 10: High-Frequency Serial Logging
|
||||
if ("Serial.print" in code or "Serial.write" in code) and \
|
||||
("motor" in code.lower() or "servo" in code.lower()):
|
||||
# Check for throttling pattern (simple heuristic for timer variables)
|
||||
if not re.search(r'(print|log|debug|serial)\s*Timer', code, re.IGNORECASE) and \
|
||||
not re.search(r'last\s*(Print|Log|Debug)', code, re.IGNORECASE):
|
||||
issues.append({
|
||||
"severity": "warning",
|
||||
"line": self.find_line(code, "Serial.print"),
|
||||
"message": "Serial logging in motor loops causes jitter. Ensure it's throttled (e.g. every 100ms).",
|
||||
"fix": lambda c: c + "\n// [Performance] Warning: Serial.print() inside loops can interrupt motor timing."
|
||||
})
|
||||
|
||||
# Check 11: Feature Bloat (Unrequested Button)
|
||||
if user_message:
|
||||
msg_lower = user_message.lower()
|
||||
# If user didn't ask for inputs/buttons
|
||||
if not any(w in msg_lower for w in ['button', 'switch', 'input', 'trigger']):
|
||||
# Pattern 1: Variable assignment (int btn = digitalRead(...))
|
||||
for match in re.finditer(r'(?:int|bool|byte)\s+(\w*(?:button|btn|switch)\w*)\s*=\s*digitalRead\s*\([^;]+;', code, re.IGNORECASE):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, match.group(0)),
|
||||
"message": f"Feature Bloat: Unrequested button code detected ('{match.group(1)}').",
|
||||
"fix": lambda c, m=match.group(0): c.replace(m, "")
|
||||
})
|
||||
|
||||
# Pattern 2: Direct usage in conditions (if (digitalRead(BUTTON_PIN)...))
|
||||
for match in re.finditer(r'digitalRead\s*\(\s*(\w*(?:BUTTON|BTN|SWITCH)\w*)\s*\)', code, re.IGNORECASE):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, match.group(0)),
|
||||
"message": f"Feature Bloat: Unrequested button check detected ('{match.group(1)}').",
|
||||
"fix": lambda c, m=match.group(0): c.replace(m, "0")
|
||||
})
|
||||
|
||||
# Pattern 3: pinMode(..., INPUT)
|
||||
for match in re.finditer(r'pinMode\s*\(\s*\w+\s*,\s*INPUT(?:_PULLUP)?\s*\);', code):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, match.group(0)),
|
||||
"message": "Feature Bloat: Unrequested input pin configuration.",
|
||||
"fix": lambda c, m=match.group(0): c.replace(m, "")
|
||||
})
|
||||
|
||||
# Pattern 4: Unused button variable initialization (int btn = LOW;)
|
||||
for match in re.finditer(r'(?:int|bool|byte)\s+(\w*(?:button|btn|switch)\w*)\s*=\s*(?:LOW|HIGH|0|1|false|true)\s*;', code, re.IGNORECASE):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, match.group(0)),
|
||||
"message": f"Feature Bloat: Unused button variable '{match.group(1)}'.",
|
||||
"fix": lambda c, m=match.group(0): c.replace(m, "")
|
||||
})
|
||||
|
||||
# Check 14: State Machine for Weapons (Combat Protocol)
|
||||
if "weapon" in user_message.lower() or "combat" in user_message.lower() or "state machine" in user_message.lower():
|
||||
if "enum" not in code and "bool isArmed" not in code:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "Combat code requires a State Machine (enum State or bool isArmed).",
|
||||
"fix": lambda c: c.replace("void setup", "\n// [AUTO-FIX] State Machine\nenum State { DISARMED, ARMING, ARMED, FIRING };\nState currentState = DISARMED;\nunsigned long stateTimer = 0;\n\nvoid setup") if "void setup" in c else "// [AUTO-FIX] State Machine\nenum State { DISARMED, ARMING, ARMED, FIRING };\nState currentState = DISARMED;\n" + c
|
||||
})
|
||||
|
||||
if "Serial.read" not in code and "Serial.available" not in code:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "Missing Serial Command handling (e.g., 'A' to Arm).",
|
||||
"fix": lambda c: c.replace("void loop() {", "void loop() {\n if (Serial.available()) {\n char cmd = Serial.read();\n // Handle commands\n }\n")
|
||||
})
|
||||
|
||||
# Check 15: Function Naming Conventions (camelCase)
|
||||
# Exclude standard Arduino functions
|
||||
func_defs = re.finditer(r'\b(void|int|bool|float|double|String|char|long|unsigned(?:\s+long)?)\s+([a-zA-Z0-9_]+)\s*\(', code)
|
||||
for match in func_defs:
|
||||
func_name = match.group(2)
|
||||
if func_name in ['setup', 'loop', 'main']: continue
|
||||
|
||||
# Check if camelCase (starts with lowercase, no underscores unless specific style)
|
||||
if not re.match(r'^[a-z][a-zA-Z0-9]*$', func_name):
|
||||
# Check if it's snake_case or PascalCase
|
||||
suggestion = func_name
|
||||
if '_' in func_name: # snake_case -> camelCase
|
||||
components = func_name.split('_')
|
||||
suggestion = components[0].lower() + ''.join(x.title() for x in components[1:])
|
||||
elif func_name[0].isupper(): # PascalCase -> camelCase
|
||||
suggestion = func_name[0].lower() + func_name[1:]
|
||||
|
||||
issues.append({
|
||||
"severity": "warning",
|
||||
"line": self.find_line(code, match.group(0)),
|
||||
"message": f"Style: Function '{func_name}' should be camelCase (e.g., '{suggestion}').",
|
||||
"fix": lambda c, old=func_name, new=suggestion: c.replace(old, new)
|
||||
})
|
||||
|
||||
# Check 16: Monolithic Code Structure
|
||||
if "function" in user_message.lower() or "naming" in user_message.lower() or "modular" in user_message.lower():
|
||||
has_custom_funcs = False
|
||||
for match in re.finditer(r'\b(void|int|bool|float|double|String|char|long|unsigned(?:\s+long)?)\s+([a-zA-Z0-9_]+)\s*\(', code):
|
||||
if match.group(2) not in ['setup', 'loop', 'main']:
|
||||
has_custom_funcs = True
|
||||
break
|
||||
|
||||
if not has_custom_funcs:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "Structure Violation: Request asked for functions but code is monolithic.",
|
||||
"fix": lambda c: c.replace("void loop() {", "void loop() {\n runSystemLogic();\n}\n\nvoid runSystemLogic() {") + "\n}"
|
||||
})
|
||||
|
||||
# Check 17: Loop Length (Modularity)
|
||||
if "function" in user_message.lower() or "naming" in user_message.lower() or "modular" in user_message.lower():
|
||||
loop_match = re.search(r'void\s+loop\s*\(\s*\)\s*\{', code)
|
||||
if loop_match:
|
||||
start_idx = loop_match.end()
|
||||
brace_count = 1
|
||||
loop_body = ""
|
||||
|
||||
for char in code[start_idx:]:
|
||||
if char == '{': brace_count += 1
|
||||
elif char == '}': brace_count -= 1
|
||||
|
||||
if brace_count == 0:
|
||||
break
|
||||
loop_body += char
|
||||
|
||||
# Count significant lines
|
||||
lines = [line.strip() for line in loop_body.split('\n')]
|
||||
significant_lines = [l for l in lines if l and not l.startswith('//') and not l.startswith('/*') and l != '']
|
||||
|
||||
if len(significant_lines) >= 10:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": f"Modularity Violation: loop() has {len(significant_lines)} lines (limit 10). Move logic to functions.",
|
||||
"fix": lambda c: self.refactor_loop_to_function(c)
|
||||
})
|
||||
|
||||
# Check 18: ADC Resolution (ESP32)
|
||||
if "ESP32" in hardware.upper():
|
||||
adc_res_match = re.search(r'#define\s+(\w*ADC\w*RES\w*)\s+(\d+)', code, re.IGNORECASE)
|
||||
if adc_res_match:
|
||||
val = int(adc_res_match.group(2))
|
||||
if val not in [4095, 4096]:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, adc_res_match.group(0)),
|
||||
"message": f"Hardware Mismatch: ESP32 ADC is 12-bit (4095), not {val}.",
|
||||
"fix": lambda c, old=adc_res_match.group(0), name=adc_res_match.group(1): c.replace(old, f"#define {name} 4095")
|
||||
})
|
||||
|
||||
# Check 20: Hardcoded 10-bit ADC math
|
||||
# Matches / 1023, / 1023.0, / 1024.0 (avoiding / 1024 int for bytes)
|
||||
for match in re.finditer(r'/\s*(1023(?:\.0?)?f?|1024(?:\.0)f?)', code):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, match.group(0)),
|
||||
"message": "Hardware Mismatch: ESP32 ADC is 12-bit. Use 4095.0, not 1023/1024.",
|
||||
"fix": lambda c, m=match.group(0): c.replace(m, "/ 4095.0")
|
||||
})
|
||||
|
||||
# Check 21: Status LED Pattern
|
||||
if "status" in user_message.lower() and ("led" in user_message.lower() or "indicator" in user_message.lower()):
|
||||
# Detect breathing logic (incrementing duty cycle in loop)
|
||||
breathing_match = re.search(r'(?:dutyCycle|brightness)\s*(\+=|\+\+|\-=|\-\-)', code)
|
||||
if breathing_match:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, breathing_match.group(0)),
|
||||
"message": "Wrong Pattern: Status indicators should use Blink Patterns (States), not Breathing/Fading.",
|
||||
"fix": lambda c: c + "\n// [Fix Required] Implement setStatusLED(LEDStatus state) instead of fading."
|
||||
})
|
||||
|
||||
# Check for missing Enum
|
||||
if not re.search(r'enum\s+(?:StatusState|LEDStatus)\s*\{', code):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "Missing Status Enum: Status LEDs require a state machine (enum LEDStatus {OFF, IDLE, ACTIVE, ERROR}).",
|
||||
"fix": lambda c: c.replace("void setup", "\n// [AUTO-FIX] Status Enum\nenum LEDStatus { OFF, IDLE, ACTIVE, ERROR };\nLEDStatus currentStatus = IDLE;\nunsigned long lastBlink = 0;\n\nvoid setup") if "void setup" in c else "// [AUTO-FIX] Status Enum\nenum LEDStatus { OFF, IDLE, ACTIVE, ERROR };\nLEDStatus currentStatus = IDLE;\nunsigned long lastBlink = 0;\n" + c
|
||||
})
|
||||
|
||||
# Check 19: Unnecessary Debouncing (Analog/Battery)
|
||||
if "battery" in user_message.lower() or "voltage" in user_message.lower() or "analog" in user_message.lower():
|
||||
if "button" not in user_message.lower():
|
||||
debounce_match = re.search(r'(?:debounce|lastDebounceTime)', code, re.IGNORECASE)
|
||||
if debounce_match:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, debounce_match.group(0)),
|
||||
"message": "Logic Error: Debouncing detected in analog/battery code. Analog sensors don't need debouncing.",
|
||||
"fix": lambda c: re.sub(r'.*debounce.*', '// [Fixed] Removed unnecessary debounce logic', c, flags=re.IGNORECASE)
|
||||
})
|
||||
|
||||
# Check 12: Undefined Pin Constants
|
||||
pin_vars = set(re.findall(r'(?:digitalRead|digitalWrite|pinMode|ledcAttachPin)\s*\(\s*([a-zA-Z_]\w+)', code))
|
||||
for var in pin_vars:
|
||||
if var in ['LED_BUILTIN', 'HIGH', 'LOW', 'INPUT', 'OUTPUT', 'INPUT_PULLUP', 'true', 'false']:
|
||||
continue
|
||||
|
||||
# Check if defined
|
||||
is_defined = re.search(r'#define\s+' + re.escape(var) + r'\b', code) or \
|
||||
re.search(r'\b(?:const\s+)?(?:int|byte|uint8_t|short)\s+' + re.escape(var) + r'\s*=', code)
|
||||
|
||||
if not is_defined:
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": f"Undefined variable '{var}' used in pin operation.",
|
||||
"fix": lambda c, v=var: f"#define {v} 2 // [Auto-Fix] Defined missing pin\n" + c
|
||||
})
|
||||
|
||||
# Check 22: Misused Debouncing (Animation Timing)
|
||||
if "brightness" in code or "fade" in code:
|
||||
misused_debounce = re.search(r'if\s*\(\s*\(?\s*millis\(\)\s*-\s*\w+\s*\)?\s*>\s*(\w*DEBOUNCE\w*)\s*\)\s*\{', code, re.IGNORECASE)
|
||||
if misused_debounce:
|
||||
var_name = misused_debounce.group(1)
|
||||
# Check if the block actually modifies brightness (simple heuristic lookahead)
|
||||
start_index = misused_debounce.end()
|
||||
snippet = code[start_index:start_index+200]
|
||||
if any(x in snippet for x in ['brightness', 'fade', 'dutyCycle', 'ledcWrite']):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"line": self.find_line(code, var_name),
|
||||
"message": f"Semantic Error: Using {var_name} for animation/fading. Use UPDATE_INTERVAL or FADE_SPEED.",
|
||||
"fix": lambda c, v=var_name: c.replace(v, "FADE_SPEED" if v.isupper() else "fadeSpeed")
|
||||
})
|
||||
|
||||
# Check 24: Unused Variables in Setup
|
||||
setup_match = re.search(r'void\s+setup\s*\(\s*\)\s*\{', code)
|
||||
if setup_match:
|
||||
start_idx = setup_match.end()
|
||||
brace_count = 1
|
||||
setup_body = ""
|
||||
for char in code[start_idx:]:
|
||||
if char == '{': brace_count += 1
|
||||
elif char == '}': brace_count -= 1
|
||||
if brace_count == 0: break
|
||||
setup_body += char
|
||||
|
||||
clean_body = re.sub(r'//.*', '', setup_body)
|
||||
clean_body = re.sub(r'/\*.*?\*/', '', clean_body, flags=re.DOTALL)
|
||||
|
||||
local_vars = re.finditer(r'\b((?:static\s+)?(?:const\s+)?(?:int|float|bool|char|String|long|double|byte|uint8_t|unsigned(?:\s+long)?))\s+([a-zA-Z_]\w*)\s*(?:=|;)', clean_body)
|
||||
|
||||
for match in local_vars:
|
||||
var_type = match.group(1)
|
||||
var_name = match.group(2)
|
||||
if len(re.findall(r'\b' + re.escape(var_name) + r'\b', clean_body)) == 1:
|
||||
issues.append({
|
||||
"severity": "warning",
|
||||
"line": self.find_line(code, f"{var_type} {var_name}"),
|
||||
"message": f"Unused variable '{var_name}' in setup().",
|
||||
"fix": lambda c, v=var_name, t=var_type: re.sub(r'\b' + re.escape(t) + r'\s+' + re.escape(v) + r'[^;]*;\s*', '', c)
|
||||
})
|
||||
|
||||
# Check 25: Missing Serial.begin
|
||||
if re.search(r'Serial\.(?:print|write|println|printf)', code) and not re.search(r'Serial\.begin\s*\(', code):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "Missing Serial.begin() initialization.",
|
||||
"fix": lambda c: re.sub(r'void\s+setup\s*\(\s*\)\s*\{', r'void setup() {\n Serial.begin(115200);', c, count=1)
|
||||
})
|
||||
|
||||
# Check 26: Missing Wire.begin
|
||||
if re.search(r'Wire\.(?!h\b|begin\b)', code) and not re.search(r'Wire\.begin\s*\(', code):
|
||||
issues.append({
|
||||
"severity": "error",
|
||||
"message": "Missing Wire.begin() initialization for I2C.",
|
||||
"fix": lambda c: re.sub(r'void\s+setup\s*\(\s*\)\s*\{', r'void setup() {\n Wire.begin();', c, count=1)
|
||||
})
|
||||
|
||||
return len([i for i in issues if i['severity'] == 'error']) == 0, issues
|
||||
|
||||
def auto_fix(self, code: str, issues: List[Dict]) -> str:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue