ambilight-python.py

import subprocess
import time
from serial import Serial
from PIL import Image
import logging
import os
import signal
import sys

# ===== LOGGING =====
logging.basicConfig(
    level=logging.INFO,  # DEBUG/INFO/WARNING/ERROR/CRITICAL
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('ambilight.log')
    ]
)
logger = logging.getLogger(__name__)


# ===== SETTINGS =====
# Temp directory for screenshot path
SCREENSHOT_TEMP_PATH = "/dev/shm/ambilight"  # In this example, a ramdisk is used.
os.makedirs(SCREENSHOT_TEMP_PATH, exist_ok=True)  # Create dir, if it not exist

# LED Conf (120 LEDs)
TOTAL_LEDS = 120
TOP_LEDS = 42       # Top
BOTTOM_LEDS = 42    # Bottom
SIDE_LEDS = 18      # Left and right

# Screen settings (3440x1440, 21:9)
SCREEN_WIDTH = 3440
SCREEN_HEIGHT = 1440
PIXELS_PER_LED = 30      # Zone for capture color for 1 pixel
PIXEL_SAMPLE_STEP = 8    # Pixel sampling step, if flickering is observed, increase it

# Scaling (True/False)
ENABLE_SCALING = False # May load CPU if turn on
SCREENSHOT_SCALE_FACTOR = 20  # Scaling factor (active if only ENABLE_SCALING=True)

# Timeout between LED update (in seconds)
SLEEP_TIMEOUT = 0.5

#Arduino port
ser = Serial('/dev/ttyUSB0', 115200) 


# ===== AUTO SETTINGS =====
if ENABLE_SCALING:
    SCALED_WIDTH = SCREEN_WIDTH // SCREENSHOT_SCALE_FACTOR
    SCALED_HEIGHT = SCREEN_HEIGHT // SCREENSHOT_SCALE_FACTOR
    SCALED_PIXELS_PER_LED = max(1, PIXELS_PER_LED // SCREENSHOT_SCALE_FACTOR)
else:
    # Using original width and height
    SCALED_WIDTH = SCREEN_WIDTH
    SCALED_HEIGHT = SCREEN_HEIGHT
    SCALED_PIXELS_PER_LED = PIXELS_PER_LED


# Data arrays
right_leds = [0] * (SIDE_LEDS * 3)
top_leds = [0] * (TOP_LEDS * 3)
left_leds = [0] * (SIDE_LEDS * 3)
bottom_leds = [0] * (BOTTOM_LEDS * 3)

def turn_off_leds():
    """Sending turn of command"""
    try:
        off_data = [0] * (TOTAL_LEDS * 3)
        ser.write(off_data)
        logger.info("All leds are off")
    except Exception as e:
        logger.error(f"Error while turn off LEDs: {str(e)}")

def signal_handler(sig, frame):
    """Shutdown handlers"""
    logger.info("Termination signal received. Disabling LEDs...")
    turn_off_leds()
    ser.close()
    sys.exit(0)

# Registering signal handlers
signal.signal(signal.SIGINT, signal_handler)  # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler)  # Process termination signal

def timed_operation(label):
    """Logging decorator"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            start_time = time.time()
            result = func(*args, **kwargs)
            elapsed = (time.time() - start_time) * 1000
            logger.debug(f"{label}: {elapsed:.1f} ms")
            return result
        return wrapper
    return decorator

@timed_operation("Screen capture")
def capture_screen():

    try:
        #Making screenshot
        screenshot_path = f"{SCREENSHOT_TEMP_PATH}/screen.jpg"
        subprocess.run(
            ["gpu-screen-recorder", "-w", "screen", "-o", screenshot_path],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL
        )

        img = Image.open(f"{SCREENSHOT_TEMP_PATH}/screen.jpg").convert("RGB")

        if ENABLE_SCALING:
            img = img.resize((SCALED_WIDTH, SCALED_HEIGHT), Image.LANCZOS)

        # Removing file
        os.unlink(screenshot_path)

        return img

    except Exception as e:
        logger.error(f"Screen capture error: {str(e)}")
        raise


@timed_operation("Processing right LED")
def process_right():
    for i in range(SIDE_LEDS):
        mirror_i = SIDE_LEDS - 1 - i
        y1 = int(mirror_i * SCALED_HEIGHT / SIDE_LEDS)
        y2 = int((mirror_i + 1) * SCALED_HEIGHT / SIDE_LEDS)
        region = screen.crop((SCALED_WIDTH - SCALED_PIXELS_PER_LED, y1, SCALED_WIDTH, y2))
        r, g, b = get_region_color(region)
        right_leds[i*3 : i*3+3] = [r, g, b]

@timed_operation("Processing top LED")
def process_top():
    for i in range(TOP_LEDS):
        mirror_i = TOP_LEDS - 1 - i
        x1 = int(mirror_i * SCALED_WIDTH / TOP_LEDS)
        x2 = int((mirror_i + 1) * SCALED_WIDTH / TOP_LEDS)
        region = screen.crop((x1, 0, x2, SCALED_PIXELS_PER_LED))
        r, g, b = get_region_color(region)
        top_leds[i*3 : i*3+3] = [r, g, b]

@timed_operation("Processing left LED")
def process_left():
    for i in range(SIDE_LEDS):
        y1 = int(i * SCALED_HEIGHT / SIDE_LEDS)
        y2 = int((i + 1) * SCALED_HEIGHT / SIDE_LEDS)
        region = screen.crop((0, y1, SCALED_PIXELS_PER_LED, y2))
        r, g, b = get_region_color(region)
        left_leds[i*3 : i*3+3] = [r, g, b]

@timed_operation("Processing bottom LED")
def process_bottom():
    for i in range(BOTTOM_LEDS):
        x1 = int(i * SCALED_WIDTH / BOTTOM_LEDS)
        x2 = int((i + 1) * SCALED_WIDTH / BOTTOM_LEDS)
        region = screen.crop((x1, SCALED_HEIGHT - SCALED_PIXELS_PER_LED, x2, SCALED_HEIGHT))
        r, g, b = get_region_color(region)
        bottom_leds[i*3 : i*3+3] = [r, g, b]


def get_region_color(region):
    pixels = region.load()
    width, height = region.size
    total_r = total_g = total_b = count = 0
    
    for y in range(0, height, PIXEL_SAMPLE_STEP):
        for x in range(0, width, PIXEL_SAMPLE_STEP):
            r, g, b = pixels[x, y][:3]
            total_r += r
            total_g += g
            total_b += b
            count += 1
    
    return (total_r//count, total_g//count, total_b//count) if count else (0, 0, 0)

while True:
    try:
        cycle_start = time.time()
        
        # Main cycle
        screen = capture_screen()
        process_right()
        process_top()
        process_left()
        process_bottom()
        
        # Sending data
        send_start = time.time()
        ser.write(right_leds + top_leds + left_leds + bottom_leds)
        logger.debug(f"Sending data: {(time.time() - send_start)*1000:.1f} ms")
        
        # Statistic
        total_time = time.time() - cycle_start
        logger.info(f"Cycle ended. {total_time*1000:.0f} ms FPS: {1/total_time:.1f}")

        # FPS stabilization
        sleep_time = max(0, SLEEP_TIMEOUT - total_time)
        if sleep_time > 0:
            logger.debug(f"Pause {sleep_time*1000:.0f} ms")
            time.sleep(sleep_time)
        else:
            logger.debug(f"Pause between updates less than cycle time")
            logger.debug(f"Pause {SLEEP_TIMEOUT*1000:.0f} ms, Cycle {total_time*1000:.0f} ms")
        
    except Exception as e:
        logger.error(f"Error: {str(e)}", exc_info=True)
        time.sleep(1)  # Pause at error
        
    except KeyboardInterrupt:
        logger.info("Application stopped by user")
        turn_off_leds()
        ser.close()

    except Exception as e:
        logger.error(f"Critical error: {str(e)}", exc_info=True)
        turn_off_leds()
        ser.close()
        raise