battery-charging-optimizer/validate_pyscript.py

#!/usr/bin/env python3
"""
PyScript Validation Tool for Battery Charging Optimizer
Checks for common issues without needing Home Assistant runtime
"""

import ast
import re
from pathlib import Path

def check_file(filepath):
    """Analyze a PyScript file for potential issues"""
    print(f"\n{'='*60}")
    print(f"Analyzing: {filepath.name}")
    print('='*60)

    with open(filepath, 'r') as f:
        content = f.read()

    issues = []
    warnings = []

    # 1. Check imports
    print("\n1. Checking imports...")
    imports = re.findall(r'^(?:from|import)\s+(\S+)', content, re.MULTILINE)

    critical_imports = ['zoneinfo', 'requests', 'json', 'datetime']
    for imp in critical_imports:
        if any(imp in i for i in imports):
            print(f"   ✓ Found import: {imp}")
        else:
            if imp == 'zoneinfo':
                issues.append(f"CRITICAL: zoneinfo might not be available in PyScript")

    # 2. Check for PyScript-specific constructs
    print("\n2. Checking PyScript constructs...")

    # @service decorators
    services = re.findall(r'@service\s+def\s+(\w+)', content)
    if services:
        print(f"   ✓ Found {len(services)} services: {', '.join(services)}")
    else:
        warnings.append("No @service decorators found")

    # @time_trigger decorators
    triggers = re.findall(r'@time_trigger\("([^"]+)"\)', content)
    if triggers:
        print(f"   ✓ Found {len(triggers)} time triggers:")
        for t in triggers:
            print(f"     - {t}")

    # Check cron syntax
    for trigger in triggers:
        if not trigger.startswith('cron('):
            issues.append(f"Invalid time_trigger syntax: {trigger}")

    # 3. Check state access patterns
    print("\n3. Checking state access patterns...")

    # state.get() calls
    state_gets = re.findall(r"state\.get\(['\"]([^'\"]+)['\"]\)", content)
    if state_gets:
        print(f"   Found {len(state_gets)} state.get() calls")
        unique_entities = set(state_gets)
        print(f"   Unique entities accessed: {len(unique_entities)}")
        for entity in sorted(unique_entities)[:5]:
            print(f"     - {entity}")
        if len(unique_entities) > 5:
            print(f"     ... and {len(unique_entities) - 5} more")

    # state.getattr() calls
    state_getattrs = re.findall(r"state\.getattr\(['\"]?([^'\"]+)['\"]\)", content)
    if state_getattrs:
        print(f"   Found {len(state_getattrs)} state.getattr() calls")

    # Check for missing null checks after getattr
    getattr_lines = [i for i, line in enumerate(content.split('\n'), 1)
                     if 'state.getattr' in line]
    for line_num in getattr_lines:
        lines = content.split('\n')
        next_line = lines[line_num] if line_num < len(lines) else ""
        if '.get(' in next_line and 'or {}' not in lines[line_num-1]:
            warnings.append(f"Line {line_num}: state.getattr() without null check before .get()")

    # 4. Check for datetime handling
    print("\n4. Checking datetime handling...")

    if 'datetime.now()' in content:
        if 'datetime.now().astimezone()' in content or 'datetime.now(TIMEZONE)' in content:
            print("   ✓ Timezone-aware datetime usage found")
        else:
            warnings.append("datetime.now() without timezone - may cause UTC/local issues")

    if 'fromisoformat' in content:
        print("   ✓ ISO format datetime parsing found")
        if 'tzinfo' not in content:
            warnings.append("fromisoformat without timezone handling")

    # 5. Check for task.executor usage
    print("\n5. Checking async task patterns...")

    if 'task.executor' in content:
        print("   ✓ task.executor found (for blocking I/O)")
        if 'requests.get' in content:
            if 'task.executor(requests.get' in content:
                print("   ✓ Correctly wrapped requests.get in task.executor")
            else:
                issues.append("CRITICAL: requests.get not wrapped in task.executor")

    # 6. Check state.set() calls
    print("\n6. Checking state.set() patterns...")

    state_sets = re.findall(r"state\.set\(['\"]([^'\"]+)['\"]", content)
    if state_sets:
        print(f"   Found {len(state_sets)} state.set() calls:")
        for entity in set(state_sets):
            print(f"     - {entity}")

    # Check for large attribute sets
    if 'new_attributes' in content:
        print("   ✓ Using new_attributes in state.set()")
        if "'schedule':" in content:
            warnings.append("Schedule in attributes might be large - monitor state size")

    # 7. Check service calls
    print("\n7. Checking service calls...")

    service_calls = re.findall(r'pyscript\.(\w+)\(', content)
    if service_calls:
        print(f"   Found service calls: {', '.join(set(service_calls))}")
        warnings.append("Service-to-service calls in PyScript - verify syntax")

    # 8. Check error handling
    print("\n8. Checking error handling...")

    try_blocks = content.count('try:')
    except_blocks = content.count('except')

    print(f"   Found {try_blocks} try blocks, {except_blocks} except handlers")

    if 'traceback' in content:
        print("   ✓ traceback module usage found")

    # 9. Syntax validation
    print("\n9. Validating Python syntax...")
    try:
        ast.parse(content)
        print("   ✓ Python syntax is valid")
    except SyntaxError as e:
        issues.append(f"CRITICAL: Syntax error at line {e.lineno}: {e.msg}")

    # Summary
    print("\n" + "="*60)
    print("SUMMARY")
    print("="*60)

    if not issues and not warnings:
        print("✅ No issues found!")
    else:
        if issues:
            print(f"\n🔴 CRITICAL ISSUES ({len(issues)}):")
            for i, issue in enumerate(issues, 1):
                print(f"   {i}. {issue}")

        if warnings:
            print(f"\n⚠️  WARNINGS ({len(warnings)}):")
            for i, warning in enumerate(warnings, 1):
                print(f"   {i}. {warning}")

    return issues, warnings


def main():
    """Main validation routine"""
    print("PyScript Battery Optimizer Validation Tool")
    print("="*60)

    base_path = Path(__file__).parent / "openems"

    files_to_check = [
        base_path / "battery_charging_optimizer.py",
        base_path / "hastrom_flex_extended.py"
    ]

    all_issues = []
    all_warnings = []

    for filepath in files_to_check:
        if not filepath.exists():
            print(f"\n❌ File not found: {filepath}")
            continue

        issues, warnings = check_file(filepath)
        all_issues.extend(issues)
        all_warnings.extend(warnings)

    # Final summary
    print("\n" + "="*60)
    print("OVERALL SUMMARY")
    print("="*60)

    print(f"\nTotal files checked: {len(files_to_check)}")
    print(f"Critical issues: {len(all_issues)}")
    print(f"Warnings: {len(all_warnings)}")

    if all_issues:
        print("\n🔴 ACTION REQUIRED: Fix critical issues before deploying")
    elif all_warnings:
        print("\n⚠️  REVIEW RECOMMENDED: Check warnings")
    else:
        print("\n✅ All checks passed!")

    print("\n" + "="*60)
    print("NEXT STEPS")
    print("="*60)
    print("""
1. Copy files to Home Assistant:
   - /config/pyscript/battery_charging_optimizer.py
   - /config/pyscript/hastrom_flex_extended.py

2. Reload PyScript:
   - Home Assistant → Developer Tools → Services
   - Call: pyscript.reload

3. Check logs:
   - Settings → System → Logs
   - Look for errors from pyscript or your scripts

4. Test manually:
   - Call service: pyscript.getprices_extended
   - Call service: pyscript.calculate_charging_schedule
   - Check states: pyscript.battery_charging_schedule

5. Monitor execution:
   - Enable debug logging for pyscript in configuration.yaml:
     logger:
       default: info
       logs:
         custom_components.pyscript: debug
    """)


if __name__ == "__main__":
    main()