dahua_focus_manager.py

#!/usr/bin/env python3
"""
Dahua Focus Manager v9
- Controls day/night mode transitions via schedule (not camera auto-switching)
- Uses astral library to calculate sunrise/sunset times
- Forces IR filter transitions when there's enough ambient light for autofocus
- Preserves your configured Day/Night profile settings (gain, exposure, etc.)

How it works:
1. Cameras are set to forced Day or Night mode (DayNightColor=1 or 2)
2. Script calculates optimal transition times based on sunrise/sunset
3. Before sunset (while still light): forces Night mode + autofocus
4. After sunrise (with good light): forces Day mode + autofocus
5. Your VideoInOptions settings are preserved - only the mode switching is controlled
"""

import requests
from requests.auth import HTTPDigestAuth
import hashlib
import time
import json
import logging
import os
from pathlib import Path
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

from astral import LocationInfo
from astral.sun import sun

# Configuration file paths
CONFIG_DIR = Path(os.environ.get("DAHUA_CONFIG_DIR", Path.home() / ".config" / "dahua-focus-manager"))
CONFIG_FILE = CONFIG_DIR / "config.json"
CREDS_FILE = CONFIG_DIR / "credentials"

# Default configuration
DEFAULT_CONFIG = {
    "cameras": [
        {"name": "camera_1", "ip": "192.168.1.100"},
        {"name": "camera_2", "ip": "192.168.1.101"},
    ],
    "location": {
        "latitude": 45.018647,
        "longitude": -93.4817843,
        "timezone": "America/Chicago"
    },
    # Minutes before sunset to trigger night mode (while there's still light)
    "night_transition_offset": -30,
    # Minutes after sunrise to trigger day mode (when there's good light)
    "day_transition_offset": 30,
    "poll_interval": 60,
    "autofocus_wait": 15,
    "session_refresh_interval": 120,
    "health_check_file": "/tmp/dahua_focus_manager_health"
}

# Logging setup (must be before load_config/load_credentials)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)


def load_config():
    """Load configuration from file or return defaults."""
    if CONFIG_FILE.exists():
        try:
            with open(CONFIG_FILE, "r") as f:
                config = json.load(f)
                logger.info(f"Loaded config from {CONFIG_FILE}")
                return config
        except Exception as e:
            logger.warning(f"Failed to load config: {e}, using defaults")
    return DEFAULT_CONFIG


def load_credentials():
    """Load credentials from file."""
    creds = {}

    # Check multiple locations for credentials file
    cred_locations = [
        CREDS_FILE,
        Path.home() / ".dahua_creds",
        Path("/etc/dahua-focus-manager/credentials"),
    ]

    creds_path = None
    for loc in cred_locations:
        if loc.exists():
            creds_path = loc
            break

    if not creds_path:
        logger.error(f"No credentials file found. Searched: {[str(p) for p in cred_locations]}")
        logger.error("Create a credentials file with DAHUA_USER and DAHUA_PASS")
        raise FileNotFoundError("Credentials file not found")

    with open(creds_path, "r") as f:
        for line in f:
            line = line.strip()
            if "=" in line and not line.startswith("#"):
                key, val = line.split("=", 1)
                creds[key] = val

    logger.info(f"Loaded credentials from {creds_path}")
    return creds.get("DAHUA_USER", "admin"), creds.get("DAHUA_PASS", "")


# Load configuration
CONFIG = load_config()
CAMERAS = CONFIG.get("cameras", DEFAULT_CONFIG["cameras"])
LOCATION = CONFIG.get("location", DEFAULT_CONFIG["location"])
NIGHT_OFFSET = CONFIG.get("night_transition_offset", -30)
DAY_OFFSET = CONFIG.get("day_transition_offset", 30)
POLL_INTERVAL = CONFIG.get("poll_interval", 60)
AUTOFOCUS_WAIT = CONFIG.get("autofocus_wait", 15)
SESSION_REFRESH_INTERVAL = CONFIG.get("session_refresh_interval", 120)
HEALTH_CHECK_FILE = CONFIG.get("health_check_file", "/tmp/dahua_focus_manager_health")
BRIGHTNESS_LOG_FILE = CONFIG.get("brightness_log_file", "/tmp/dahua_brightness_log.csv")

# Load credentials
USERNAME, PASSWORD = load_credentials()

# Set up location for sun calculations
TIMEZONE = ZoneInfo(LOCATION.get("timezone", "America/Chicago"))
LOCATION_INFO = LocationInfo(
    "Camera Location",
    "Region",
    LOCATION.get("timezone", "America/Chicago"),
    LOCATION.get("latitude", 45.018647),
    LOCATION.get("longitude", -93.4817843)
)


def get_sun_times(date=None):
    """Get sunrise and sunset times for a given date."""
    if date is None:
        date = datetime.now(TIMEZONE).date()
    s = sun(LOCATION_INFO.observer, date=date, tzinfo=TIMEZONE)
    return s["sunrise"], s["sunset"]


def get_transition_times():
    """Get the scheduled day and night transition times for today."""
    sunrise, sunset = get_sun_times()

    # Day transition: after sunrise + offset
    day_transition = sunrise + timedelta(minutes=DAY_OFFSET)

    # Night transition: before sunset + offset (offset is negative)
    night_transition = sunset + timedelta(minutes=NIGHT_OFFSET)

    return day_transition, night_transition


def write_health_check(status_data):
    """Write health check file with current status."""
    try:
        with open(HEALTH_CHECK_FILE, "w") as f:
            json.dump(status_data, f, indent=2, default=str)
    except Exception as e:
        logger.debug(f"Failed to write health check: {e}")


def log_brightness(camera_data):
    """Append brightness data to CSV log file for analysis."""
    try:
        import os
        file_exists = os.path.exists(BRIGHTNESS_LOG_FILE)
        with open(BRIGHTNESS_LOG_FILE, "a") as f:
            if not file_exists:
                # Write header
                f.write("timestamp,camera,brightness_0,brightness_1,gain_0,gain_1,exposure_0,exposure_1,dnc_mode,need_fill_light\n")
            for cam in camera_data:
                f.write(f"{cam['timestamp']},{cam['name']},{cam['brightness_0']},{cam['brightness_1']},"
                        f"{cam['gain_0']},{cam['gain_1']},{cam['exposure_0']},{cam['exposure_1']},"
                        f"{cam['dnc_mode']},{cam['need_fill_light']}\n")
    except Exception as e:
        logger.debug(f"Failed to log brightness: {e}")


class DahuaCamera:
    def __init__(self, name, ip, username, password):
        self.name = name
        self.ip = ip
        self.username = username
        self.password = password
        self.session = requests.Session()
        self.session_id = None
        self.current_mode = None  # 'day' or 'night'
        self.last_autofocus_time = 0
        self.last_session_refresh = 0
        self.digest_auth = HTTPDigestAuth(username, password)
        self.consecutive_failures = 0

    def login(self):
        """Login to camera via RPC2 with challenge-response auth."""
        self.session = requests.Session()
        self.digest_auth = HTTPDigestAuth(self.username, self.password)

        login_url = f"http://{self.ip}/RPC2_Login"

        login_req = {
            "method": "global.login",
            "params": {
                "userName": self.username,
                "password": "",
                "clientType": "Web3.0",
                "loginType": "Direct"
            },
            "id": 1
        }

        try:
            resp = self.session.post(login_url, json=login_req, timeout=10)
            data = resp.json()
        except Exception as e:
            logger.error(f"{self.name}: Login request failed: {e}")
            return False

        if data.get("result"):
            self.session_id = data.get("session")
            self.last_session_refresh = time.time()
            self.consecutive_failures = 0
            return True

        if "params" not in data:
            logger.error(f"{self.name}: No params in login response")
            return False

        params = data.get("params", {})
        realm = params.get("realm", "")
        random_val = params.get("random", "")
        session_from_challenge = data.get("session", "")

        ha1 = hashlib.md5(f"{self.username}:{realm}:{self.password}".encode()).hexdigest().upper()
        password_hash = hashlib.md5(f"{self.username}:{random_val}:{ha1}".encode()).hexdigest().upper()

        login_req2 = {
            "method": "global.login",
            "params": {
                "userName": self.username,
                "password": password_hash,
                "clientType": "Web3.0",
                "loginType": "Direct",
                "authorityType": "Default"
            },
            "id": 2,
            "session": session_from_challenge
        }

        try:
            resp = self.session.post(login_url, json=login_req2, timeout=10)
            data = resp.json()
        except Exception as e:
            logger.error(f"{self.name}: Login auth failed: {e}")
            return False

        if data.get("result"):
            self.session_id = data.get("session")
            self.last_session_refresh = time.time()
            self.consecutive_failures = 0
            logger.debug(f"{self.name}: Logged in successfully")
            return True
        else:
            logger.error(f"{self.name}: Login failed: {data}")
            return False

    def keepalive(self):
        """Send keepalive to maintain session."""
        if not self.session_id:
            return self.login()

        url = f"http://{self.ip}/RPC2"
        req = {
            "method": "global.keepAlive",
            "params": {"timeout": 300},
            "id": 99,
            "session": self.session_id
        }

        try:
            resp = self.session.post(url, json=req, timeout=10)
            data = resp.json()
            if data.get("result"):
                self.last_session_refresh = time.time()
                logger.debug(f"{self.name}: Session keepalive successful")
                return True
        except Exception as e:
            logger.debug(f"{self.name}: Keepalive failed: {e}")

        self.session_id = None
        return self.login()

    def ensure_session(self):
        """Ensure we have a valid session, refreshing if needed."""
        now = time.time()

        if not self.session_id:
            return self.login()

        if now - self.last_session_refresh > SESSION_REFRESH_INTERVAL:
            return self.keepalive()

        return True

    def get_video_status(self):
        """Get full video input status including brightness, gain, exposure, DncMode."""
        if not self.ensure_session():
            self.consecutive_failures += 1
            return None

        url = f"http://{self.ip}/RPC2"
        req = {
            "method": "devVideoInput.getVideoInStatus",
            "params": {"channel": 0},
            "id": 10,
            "session": self.session_id
        }

        try:
            resp = self.session.post(url, json=req, timeout=10)
            data = resp.json()
            if data.get("result"):
                self.consecutive_failures = 0
                return data.get("params", {}).get("status", {})
        except Exception as e:
            logger.error(f"{self.name}: Failed to get video status: {e}")
            self.session_id = None

        self.consecutive_failures += 1
        return None

    def get_dnc_mode(self):
        """Get current DncMode value (1=Day, 0=Night) from camera status."""
        status = self.get_video_status()
        if status:
            return status.get("DncMode")
        return None

    def get_brightness_data(self):
        """Get brightness and related sensor data for logging."""
        status = self.get_video_status()
        if not status:
            return None

        brightness = status.get("Brightness", [0, 0])
        gain = status.get("Gain", [0, 0])
        exposure = status.get("ExposureValue", [0, 0])

        return {
            "timestamp": datetime.now(TIMEZONE).isoformat(),
            "name": self.name,
            "brightness_0": brightness[0] if len(brightness) > 0 else 0,
            "brightness_1": brightness[1] if len(brightness) > 1 else 0,
            "gain_0": gain[0] if len(gain) > 0 else 0,
            "gain_1": gain[1] if len(gain) > 1 else 0,
            "exposure_0": exposure[0] if len(exposure) > 0 else 0,
            "exposure_1": exposure[1] if len(exposure) > 1 else 0,
            "dnc_mode": status.get("DncMode", -1),
            "need_fill_light": status.get("NeedFillLight", -1)
        }

    def set_day_night_color(self, mode):
        """Force day/night mode. mode: 1=Day(Color), 2=Night(B&W with IR)

        This controls the IR cut filter state while preserving your
        configured VideoInOptions settings (gain, exposure, etc.).
        """
        url = f"http://{self.ip}/cgi-bin/configManager.cgi"
        params = {
            "action": "setConfig",
            "VideoInOptions[0].DayNightColor": mode
        }
        try:
            resp = self.session.get(url, params=params, auth=self.digest_auth, timeout=10)
            if "OK" in resp.text:
                mode_name = "Day" if mode == 1 else "Night"
                logger.info(f"{self.name}: Forced DayNightColor to {mode_name} ({mode})")
                return True
            else:
                logger.error(f"{self.name}: Failed to set DayNightColor: {resp.text}")
        except Exception as e:
            logger.error(f"{self.name}: Failed to set DayNightColor: {e}")
        return False

    def trigger_autofocus(self):
        """Trigger autofocus via CGI with digest auth."""
        url = f"http://{self.ip}/cgi-bin/devVideoInput.cgi"
        params = {"action": "autoFocus"}
        try:
            resp = self.session.get(url, params=params, auth=self.digest_auth, timeout=30)
            if "OK" in resp.text or resp.status_code == 200:
                logger.info(f"{self.name}: Autofocus triggered")
                return True
            else:
                logger.error(f"{self.name}: Autofocus failed: {resp.text}")
        except Exception as e:
            logger.error(f"{self.name}: Autofocus failed: {e}")
        return False

    def get_focus_status(self):
        """Get current focus status."""
        url = f"http://{self.ip}/cgi-bin/devVideoInput.cgi"
        params = {"action": "getFocusStatus"}
        try:
            resp = self.session.get(url, params=params, auth=self.digest_auth, timeout=10)
            # Parse focus value
            for line in resp.text.split('\n'):
                if 'status.Focus=' in line:
                    focus = line.split('=')[1]
                    return float(focus)
        except Exception as e:
            logger.error(f"{self.name}: Failed to get focus status: {e}")
        return None

    def transition_to_mode(self, target_mode):
        """Transition camera to day or night mode with autofocus.

        Args:
            target_mode: 'day' or 'night'
        """
        if self.current_mode == target_mode:
            logger.debug(f"{self.name}: Already in {target_mode} mode, skipping")
            return True

        logger.info(f"{self.name}: Transitioning to {target_mode} mode...")

        # Set the forced day/night mode
        # DayNightColor: 1=Day(Color), 2=Night(B&W/IR)
        color_mode = 1 if target_mode == 'day' else 2

        if not self.set_day_night_color(color_mode):
            logger.error(f"{self.name}: Failed to set DayNightColor")
            return False

        # Wait for IR filter to switch and image to stabilize
        logger.info(f"{self.name}: Waiting for IR filter and image to stabilize...")
        time.sleep(5)

        # Get focus before
        focus_before = self.get_focus_status()
        logger.info(f"{self.name}: Focus before autofocus: {focus_before}")

        # Trigger autofocus
        if not self.trigger_autofocus():
            logger.error(f"{self.name}: Autofocus command failed")
            self.current_mode = target_mode  # Still update mode
            return False

        # Wait for autofocus to complete
        logger.info(f"{self.name}: Waiting {AUTOFOCUS_WAIT}s for autofocus to complete...")
        time.sleep(AUTOFOCUS_WAIT)

        # Get focus after
        focus_after = self.get_focus_status()
        logger.info(f"{self.name}: Focus after autofocus: {focus_after}")

        self.current_mode = target_mode
        self.last_autofocus_time = time.time()

        logger.info(f"{self.name}: Successfully transitioned to {target_mode} mode")
        return True

    def get_status_dict(self):
        """Return current status as a dictionary for health check."""
        return {
            "name": self.name,
            "ip": self.ip,
            "current_mode": self.current_mode,
            "session_active": self.session_id is not None,
            "consecutive_failures": self.consecutive_failures,
            "last_autofocus": datetime.fromtimestamp(self.last_autofocus_time).isoformat() if self.last_autofocus_time > 0 else None
        }


def determine_current_mode(now, day_transition, night_transition):
    """Determine what mode cameras should be in based on current time."""
    # Handle the case where night_transition is before day_transition (normal case)
    # Timeline: midnight -> day_transition -> night_transition -> midnight

    if day_transition <= now < night_transition:
        return 'day'
    else:
        return 'night'


def main():
    logger.info("=" * 60)
    logger.info("Dahua Focus Manager v9 Starting")
    logger.info("Mode: Scheduled day/night control with autofocus")
    logger.info(f"Location: {LOCATION.get('latitude')}, {LOCATION.get('longitude')}")
    logger.info(f"Timezone: {LOCATION.get('timezone')}")
    logger.info(f"Night transition: {abs(NIGHT_OFFSET)} min before sunset")
    logger.info(f"Day transition: {DAY_OFFSET} min after sunrise")
    logger.info(f"Monitoring {len(CAMERAS)} cameras")
    logger.info(f"Poll interval: {POLL_INTERVAL}s")
    logger.info(f"Health check file: {HEALTH_CHECK_FILE}")
    logger.info("=" * 60)

    # Calculate today's transition times
    day_transition, night_transition = get_transition_times()
    sunrise, sunset = get_sun_times()
    logger.info(f"Today's sunrise: {sunrise.strftime('%H:%M:%S')}")
    logger.info(f"Today's sunset: {sunset.strftime('%H:%M:%S')}")
    logger.info(f"Day mode transition: {day_transition.strftime('%H:%M:%S')}")
    logger.info(f"Night mode transition: {night_transition.strftime('%H:%M:%S')}")

    # Initialize cameras
    cameras = []
    for cam_config in CAMERAS:
        cam = DahuaCamera(cam_config["name"], cam_config["ip"], USERNAME, PASSWORD)
        cameras.append(cam)

        # Check current camera state
        dnc_mode = cam.get_dnc_mode()
        if dnc_mode is not None:
            cam.current_mode = "day" if dnc_mode == 1 else "night"
            logger.info(f"{cam.name}: Current IR state - DncMode={dnc_mode} ({cam.current_mode})")
        else:
            logger.warning(f"{cam.name}: Could not get initial state")

    # Determine what mode we should be in right now
    now = datetime.now(TIMEZONE)
    target_mode = determine_current_mode(now, day_transition, night_transition)
    logger.info(f"Current time: {now.strftime('%H:%M:%S')}, target mode: {target_mode}")

    # Set cameras to correct initial mode
    for cam in cameras:
        if cam.current_mode != target_mode:
            logger.info(f"{cam.name}: Setting initial mode to {target_mode}")
            cam.transition_to_mode(target_mode)
        else:
            # Just force the mode without autofocus if already correct
            color_mode = 1 if target_mode == 'day' else 2
            cam.set_day_night_color(color_mode)

    logger.info("=" * 60)
    logger.info("Monitoring for scheduled transitions...")

    last_date = now.date()
    last_health_check = 0

    try:
        while True:
            now = datetime.now(TIMEZONE)

            # Check if we need to recalculate transition times (new day)
            if now.date() != last_date:
                day_transition, night_transition = get_transition_times()
                sunrise, sunset = get_sun_times()
                logger.info(f"New day - sunrise: {sunrise.strftime('%H:%M:%S')}, sunset: {sunset.strftime('%H:%M:%S')}")
                logger.info(f"Day transition: {day_transition.strftime('%H:%M:%S')}, Night transition: {night_transition.strftime('%H:%M:%S')}")
                last_date = now.date()

            # Determine target mode
            target_mode = determine_current_mode(now, day_transition, night_transition)

            # Check each camera
            for cam in cameras:
                if cam.current_mode != target_mode:
                    logger.info(f"Transition time reached - switching to {target_mode} mode")
                    cam.transition_to_mode(target_mode)

            # Health check and brightness logging every 60 seconds
            if time.time() - last_health_check >= 60:
                status = {
                    "timestamp": now.isoformat(),
                    "sunrise": sunrise.isoformat(),
                    "sunset": sunset.isoformat(),
                    "day_transition": day_transition.isoformat(),
                    "night_transition": night_transition.isoformat(),
                    "current_target_mode": target_mode,
                    "cameras": [cam.get_status_dict() for cam in cameras]
                }
                write_health_check(status)

                # Log brightness data for analysis
                brightness_data = []
                for cam in cameras:
                    data = cam.get_brightness_data()
                    if data:
                        brightness_data.append(data)
                if brightness_data:
                    log_brightness(brightness_data)

                last_health_check = time.time()

            time.sleep(POLL_INTERVAL)

    except KeyboardInterrupt:
        logger.info("Shutting down...")


if __name__ == "__main__":
    main()