path: root/deepcool-lm

#!/usr/bin/env python3
"""
Deepcool LM360 LCD Driver - Unified CLI Tool
Supports system monitoring, brightness control, and custom image rendering
"""
import usb.core
import usb.util
import sys
import time
import argparse
import psutil
import socket
import json
import os
import threading
from PIL import Image, ImageDraw, ImageFont

VENDOR_ID = 0x3633
PRODUCT_ID = 0x0026
EP_OUT = 0x01
WIDTH = 320
HEIGHT = 240

# Frame header that works
FRAME_HEADER = bytes([0xaa, 0x08, 0x00, 0x00, 0x01, 0x00, 0x58, 0x02, 0x00, 0x2c, 0x01, 0xbc, 0x11])

# IPC socket path
SOCKET_PATH = "/var/run/deepcool-lm.sock"

class DisplayState:
    """Manages the current display state"""
    def __init__(self):
        self.mode = 'monitor'  # 'monitor', 'image', 'solid'
        self.data = None  # Store image framebuffer or color
        self.lock = threading.Lock()

    def set_monitor_mode(self):
        with self.lock:
            self.mode = 'monitor'
            self.data = None

    def set_image_mode(self, framebuffer):
        with self.lock:
            self.mode = 'image'
            self.data = framebuffer

    def set_solid_mode(self, framebuffer):
        with self.lock:
            self.mode = 'solid'
            self.data = framebuffer

    def get_state(self):
        with self.lock:
            return self.mode, self.data

class IPCServer:
    """Unix socket server for IPC communication"""
    def __init__(self, device, display_state):
        self.device = device
        self.display_state = display_state
        self.socket_path = SOCKET_PATH
        self.running = False
        self.server_socket = None

    def start(self):
        """Start the IPC server in a background thread"""
        # Remove old socket if it exists
        try:
            os.unlink(self.socket_path)
        except OSError:
            if os.path.exists(self.socket_path):
                raise

        self.server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.server_socket.bind(self.socket_path)
        os.chmod(self.socket_path, 0o666)  # Allow all users to connect
        self.server_socket.listen(1)
        self.running = True

        # Start server thread
        thread = threading.Thread(target=self._server_loop, daemon=True)
        thread.start()

    def _server_loop(self):
        """Accept and handle client connections"""
        while self.running:
            try:
                conn, _ = self.server_socket.accept()
                self._handle_client(conn)
            except Exception as e:
                if self.running:
                    print(f"IPC server error: {e}", file=sys.stderr)

    def _handle_client(self, conn):
        """Handle a single client request"""
        try:
            # Receive command (max 10MB for images)
            data = b''
            while True:
                chunk = conn.recv(4096)
                if not chunk:
                    break
                data += chunk
                if len(data) > 10 * 1024 * 1024:  # 10MB limit
                    break

            if not data:
                return

            # Parse command
            cmd = json.loads(data.decode('utf-8'))
            action = cmd.get('action')

            # Execute command
            response = {'status': 'ok'}

            if action == 'brightness_up':
                self.device.brightness_up()
            elif action == 'brightness_down':
                self.device.brightness_down()
            elif action == 'monitor':
                # Switch back to monitor mode
                self.display_state.set_monitor_mode()
            elif action == 'image':
                # Image data is base64 encoded in the command
                import base64
                img_data = base64.b64decode(cmd['data'])
                from io import BytesIO
                img = Image.open(BytesIO(img_data))
                framebuffer = rgb_to_framebuffer(img)
                self.display_state.set_image_mode(framebuffer)
            elif action == 'solid':
                color = tuple(cmd['color'])
                img = Image.new('RGB', (WIDTH, HEIGHT), color=color)
                framebuffer = rgb_to_framebuffer(img)
                self.display_state.set_solid_mode(framebuffer)
            else:
                response = {'status': 'error', 'message': f'Unknown action: {action}'}

            # Send response
            conn.sendall(json.dumps(response).encode('utf-8'))

        except Exception as e:
            error_response = {'status': 'error', 'message': str(e)}
            try:
                conn.sendall(json.dumps(error_response).encode('utf-8'))
            except:
                pass
        finally:
            conn.close()

    def stop(self):
        """Stop the IPC server"""
        self.running = False
        if self.server_socket:
            self.server_socket.close()
        try:
            os.unlink(self.socket_path)
        except:
            pass

def send_ipc_command(cmd):
    """Send a command to the running service via IPC"""
    try:
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.connect(SOCKET_PATH)
        sock.sendall(json.dumps(cmd).encode('utf-8'))
        sock.shutdown(socket.SHUT_WR)

        # Receive response
        response = b''
        while True:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response += chunk

        sock.close()
        return json.loads(response.decode('utf-8'))
    except FileNotFoundError:
        return None  # Service not running
    except Exception as e:
        return {'status': 'error', 'message': str(e)}

class LM360:
    def __init__(self):
        self.dev = None
        self.interface = 0

    def connect(self):
        self.dev = usb.core.find(idVendor=VENDOR_ID, idProduct=PRODUCT_ID)
        if self.dev is None:
            return False
        if self.dev.is_kernel_driver_active(self.interface):
            try:
                self.dev.detach_kernel_driver(self.interface)
            except:
                pass
        try:
            self.dev.set_configuration()
            usb.util.claim_interface(self.dev, self.interface)
        except:
            return False
        return True

    def write(self, data):
        try:
            return self.dev.write(EP_OUT, data, timeout=5000)
        except Exception as e:
            print(f"Write error: {e}")
            return 0

    def send_frame(self, framebuffer):
        """Send a frame to the display"""
        self.write(FRAME_HEADER)
        self.write(framebuffer)

    def brightness_up(self):
        """Increase brightness"""
        cmd = bytes([0xaa, 0x04, 0x00, 0x06, 0x03, 0x61, 0x00, 0xd2, 0x46])
        self.write(cmd)

    def brightness_down(self):
        """Decrease brightness"""
        cmd = bytes([0xaa, 0x04, 0x00, 0x06, 0x03, 0x1d, 0x00, 0xe6, 0x0b])
        self.write(cmd)

    def zen_mode_toggle(self):
        """Toggle Zen mode (screen off/on)"""
        cmd = bytes([0xaa, 0x04, 0x00, 0x03, 0x00, 0x00, 0x00, 0xdc, 0x9b])
        self.write(cmd)

    def init_query(self):
        """Send init/query command"""
        cmd = bytes([0xaa, 0x01, 0x00, 0x09, 0x29, 0x91])
        self.write(cmd)

    def usb_reset(self):
        """Perform USB reset"""
        if self.dev:
            try:
                self.dev.reset()
            except:
                pass

    def disconnect(self):
        if self.dev:
            try:
                usb.util.release_interface(self.dev, self.interface)
                usb.util.dispose_resources(self.dev)
            except:
                pass

# ============================================================================
# System Info & Rendering Functions
# ============================================================================

def get_temp_color(temp):
    """Return color gradient based on temperature"""
    if temp < 40:
        return (100, 200, 255)  # Cool blue
    elif temp < 60:
        return (100, 255, 100)  # Green
    elif temp < 75:
        return (255, 220, 50)   # Yellow
    elif temp < 85:
        return (255, 140, 0)    # Orange
    else:
        return (255, 50, 50)    # Red

def load_fonts():
    """Load fonts with fallback"""
    try:
        return {
            'medium': ImageFont.truetype("/usr/share/fonts/TTF/DejaVuSans-Bold.ttf", 40),
            'normal': ImageFont.truetype("/usr/share/fonts/TTF/DejaVuSans.ttf", 26),
            'normal_bold': ImageFont.truetype("/usr/share/fonts/TTF/DejaVuSans-Bold.ttf", 26),
            'small': ImageFont.truetype("/usr/share/fonts/TTF/DejaVuSans.ttf", 20),
        }
    except:
        default = ImageFont.load_default()
        return {k: default for k in ['medium', 'normal', 'normal_bold', 'small']}

def get_system_info():
    """Get all system information"""
    info = {
        'cpu_temp': 0.0,
        'gpu_temp': 0.0,
        'cpu_percent': 0.0,
        'cpu_freq': 0.0,
    }

    # CPU temperature
    try:
        temps = psutil.sensors_temperatures()
        if 'coretemp' in temps and len(temps['coretemp']) > 0:
            info['cpu_temp'] = round(temps['coretemp'][0].current, 1)
        if 'nvme' in temps and len(temps['nvme']) > 0:
            info['gpu_temp'] = round(temps['nvme'][0].current, 1)
    except:
        pass

    # CPU usage and frequency
    try:
        info['cpu_percent'] = round(psutil.cpu_percent(interval=0.1), 1)
        freq = psutil.cpu_freq()
        if freq:
            info['cpu_freq'] = round(freq.current / 1000, 2)
    except:
        pass

    return info

def rgb_to_framebuffer(img):
    """Convert PIL RGB image to RGB565 framebuffer"""
    if img.mode != 'RGB':
        img = img.convert('RGB')
    if img.size != (WIDTH, HEIGHT):
        img = img.resize((WIDTH, HEIGHT), Image.Resampling.LANCZOS)

    framebuffer = bytearray()
    pixels = img.load()

    for y in range(HEIGHT):
        for x in range(WIDTH):
            r, g, b = pixels[x, y]
            # Convert to RGB565
            r5 = (r >> 3) & 0x1F
            g6 = (g >> 2) & 0x3F
            b5 = (b >> 3) & 0x1F
            rgb565 = (r5 << 11) | (g6 << 5) | b5
            # Append as little-endian
            framebuffer.append(rgb565 & 0xFF)
            framebuffer.append((rgb565 >> 8) & 0xFF)

    return bytes(framebuffer)

def draw_rounded_rect(draw, xy, radius, fill=None, outline=None, width=1):
    """Draw a rounded rectangle using PIL's built-in method"""
    draw.rounded_rectangle(xy, radius=radius, fill=fill, outline=outline, width=width)

def render_monitor_display(info, fonts):
    """Main monitoring layout with CPU and GPU as 2 equal rows"""
    img = Image.new('RGB', (WIDTH, HEIGHT), color=(14, 14, 18))
    draw = ImageDraw.Draw(img)

    # === CPU SECTION (Top Half) ===
    cpu_color = get_temp_color(info['cpu_temp'])

    # CPU rounded border box
    draw_rounded_rect(draw, [12, 10, 308, 125], radius=8, outline=(40, 40, 50), width=2)

    # CPU Label (BOLD) with icon and temperature
    draw.text((20, 20), "⚙ CPU", fill=(110, 110, 130), font=fonts['normal_bold'])

    cpu_temp_text = f"{info['cpu_temp']:.0f}°"
    bbox = draw.textbbox((0, 0), cpu_temp_text, font=fonts['medium'])
    text_w = bbox[2] - bbox[0]
    draw.text((WIDTH - 20 - text_w, 20), cpu_temp_text, fill=cpu_color, font=fonts['medium'])

    # CPU Usage details
    usage_detail = f"Usage: {info['cpu_percent']:.0f}% • {info['cpu_freq']:.2f} GHz"
    draw.text((20, 68), usage_detail, fill=(130, 130, 150), font=fonts['small'])

    # CPU Progress bar - full width with rounded corners
    bar_x, bar_y, bar_w, bar_h = 23, 93, WIDTH - 46, 18
    bar_radius = bar_h // 2  # Fully rounded (half the height)
    draw_rounded_rect(draw, [bar_x, bar_y, bar_x + bar_w, bar_y + bar_h],
                      radius=bar_radius, fill=(30, 30, 38), outline=None)
    fill_w = int(bar_w * (info['cpu_percent'] / 100.0))
    if fill_w >= bar_radius * 2:  # Only draw if wide enough for rounded corners
        bar_color = (100, 180, 255) if info['cpu_percent'] < 75 else (255, 150, 50)
        draw_rounded_rect(draw, [bar_x, bar_y, bar_x + fill_w, bar_y + bar_h],
                          radius=bar_radius, fill=bar_color, outline=None)

    # === GPU SECTION (Bottom Half) ===
    gpu_color = get_temp_color(info['gpu_temp'])

    # GPU rounded border box
    draw_rounded_rect(draw, [12, 135, 308, 230], radius=8, outline=(40, 40, 50), width=2)

    # GPU Label (BOLD) with icon and temperature
    draw.text((20, 145), "▣ GPU", fill=(110, 110, 130), font=fonts['normal_bold'])

    gpu_temp_text = f"{info['gpu_temp']:.0f}°"
    bbox = draw.textbbox((0, 0), gpu_temp_text, font=fonts['medium'])
    text_w = bbox[2] - bbox[0]
    draw.text((WIDTH - 20 - text_w, 145), gpu_temp_text, fill=gpu_color, font=fonts['medium'])

    # GPU Temp Progress bar - full width with rounded corners
    bar_x, bar_y, bar_w, bar_h = 23, 200, WIDTH - 46, 18
    bar_radius = bar_h // 2  # Fully rounded (half the height)
    draw_rounded_rect(draw, [bar_x, bar_y, bar_x + bar_w, bar_y + bar_h],
                      radius=bar_radius, fill=(30, 30, 38), outline=None)
    fill_w = int(bar_w * min(info['gpu_temp'] / 100.0, 1.0))
    if fill_w >= bar_radius * 2:  # Only draw if wide enough for rounded corners
        draw_rounded_rect(draw, [bar_x, bar_y, bar_x + fill_w, bar_y + bar_h],
                          radius=bar_radius, fill=gpu_color, outline=None)

    return rgb_to_framebuffer(img)

# ============================================================================
# Command Functions
# ============================================================================

def cmd_monitor(device, args):
    """Display system monitoring continuously"""
    # Send init command first
    device.init_query()
    time.sleep(0.5)

    print("Displaying system monitor (Ctrl+C to stop)...\n")

    # Load fonts
    fonts = load_fonts()

    # Create display state
    display_state = DisplayState()

    # Start IPC server
    ipc_server = IPCServer(device, display_state)
    ipc_server.start()
    print("IPC server started on", SOCKET_PATH)

    try:
        frame_count = 0
        while True:
            # Check current display mode
            mode, data = display_state.get_state()

            if mode == 'monitor':
                # Generate and display monitoring frame
                info = get_system_info()
                framebuffer = render_monitor_display(info, fonts)
                device.send_frame(framebuffer)

                frame_count += 1
                print(f"[MONITOR] Frame {frame_count:4d} | CPU: {info['cpu_temp']:5.1f}°C ({info['cpu_percent']:4.1f}%) | GPU: {info['gpu_temp']:5.1f}°C | {info['cpu_freq']:.2f}GHz", end='\r')
            elif mode in ['image', 'solid']:
                # Display static content
                if data:
                    device.send_frame(data)
                    print(f"[{mode.upper()}] Displaying static content...                                                                ", end='\r')

            time.sleep(args.interval)
    except KeyboardInterrupt:
        print(f"\n\nStopped after {frame_count} frames")
    finally:
        ipc_server.stop()

def cmd_image(device, args):
    """Display a custom image"""
    # Try IPC first if service is running
    if device is None:
        import base64
        with open(args.image, 'rb') as f:
            img_data = base64.b64encode(f.read()).decode('utf-8')
        response = send_ipc_command({'action': 'image', 'data': img_data})
        if response and response.get('status') == 'ok':
            print(f"✓ Displayed image: {args.image}")
            return
        elif response:
            print(f"✗ Error: {response.get('message', 'Unknown error')}")
            sys.exit(1)

    # Direct USB access
    device.init_query()
    time.sleep(0.5)

    try:
        img = Image.open(args.image).convert('RGB')
        framebuffer = rgb_to_framebuffer(img)
        device.send_frame(framebuffer)
        print(f"✓ Displayed image: {args.image}")
    except Exception as e:
        print(f"✗ Error loading image: {e}")
        sys.exit(1)

def cmd_solid(device, args):
    """Display solid color"""
    # Try IPC first if service is running
    if device is None:
        response = send_ipc_command({'action': 'solid', 'color': args.color})
        if response and response.get('status') == 'ok':
            print(f"✓ Displayed solid color: {tuple(args.color)}")
            return
        elif response:
            print(f"✗ Error: {response.get('message', 'Unknown error')}")
            sys.exit(1)

    # Direct USB access
    device.init_query()
    time.sleep(0.5)

    img = Image.new('RGB', (WIDTH, HEIGHT), color=tuple(args.color))
    framebuffer = rgb_to_framebuffer(img)
    device.send_frame(framebuffer)
    print(f"✓ Displayed solid color: {tuple(args.color)}")

def cmd_brightness(device, args):
    """Adjust brightness"""
    # Try IPC first if service is running
    if device is None:
        action = 'brightness_up' if args.direction == 'up' else 'brightness_down'
        response = send_ipc_command({'action': action})
        if response and response.get('status') == 'ok':
            print(f"✓ Brightness {'increased' if args.direction == 'up' else 'decreased'}")
            return
        elif response:
            print(f"✗ Error: {response.get('message', 'Unknown error')}")
            sys.exit(1)

    # Direct USB access
    if args.direction == 'up':
        device.brightness_up()
        print("✓ Brightness increased")
    else:
        device.brightness_down()
        print("✓ Brightness decreased")

# ============================================================================
# Main Entry Point
# ============================================================================

def main():
    parser = argparse.ArgumentParser(
        description='Deepcool LM360 LCD Driver',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  sudo deepcool-lm monitor                  # Display live system monitoring (or switch back to it)
  sudo deepcool-lm monitor --interval 1     # Update every second
  sudo deepcool-lm image photo.jpg          # Display custom image
  sudo deepcool-lm solid --color 255 0 0    # Display red screen
  sudo deepcool-lm brightness up            # Increase brightness
  sudo deepcool-lm brightness down          # Decrease brightness
        """
    )

    subparsers = parser.add_subparsers(dest='command', help='Command to execute')

    # monitor command
    monitor_parser = subparsers.add_parser('monitor', help='Display system monitoring')
    monitor_parser.add_argument('--interval', type=float, default=2.0, help='Update interval in seconds (default: 2.0)')

    # image command
    image_parser = subparsers.add_parser('image', help='Display custom image')
    image_parser.add_argument('image', help='Path to image file (will be resized to 320x240)')

    # solid command
    solid_parser = subparsers.add_parser('solid', help='Display solid color')
    solid_parser.add_argument('--color', type=int, nargs=3, metavar=('R', 'G', 'B'),
                             default=[0, 0, 0], help='RGB color (0-255)')

    # brightness command
    brightness_parser = subparsers.add_parser('brightness', help='Adjust brightness')
    brightness_parser.add_argument('direction', choices=['up', 'down'], help='Increase or decrease')

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    # For commands that can use IPC, try that first
    device = None
    use_ipc_commands = ['image', 'solid', 'brightness', 'monitor']

    # Check if we should use IPC (service is already running)
    service_running = os.path.exists(SOCKET_PATH)

    if args.command in use_ipc_commands and service_running:
        # Service is running, use IPC
        device = None  # Signal to command functions to use IPC

        # Special case: monitor command via IPC just switches back to monitor mode
        if args.command == 'monitor':
            response = send_ipc_command({'action': 'monitor'})
            if response and response.get('status') == 'ok':
                print("✓ Switched to monitoring mode")
                sys.exit(0)
            else:
                print("✗ Failed to switch to monitoring mode")
                sys.exit(1)
    elif args.command == 'monitor' and not service_running:
        # Starting monitor service directly (not via IPC)
        device = LM360()
        if not device.connect():
            print("✗ Failed to connect to LM360")
            print("  Make sure:")
            print("  - Device is plugged in (lsusb | grep 3633)")
            print("  - Running with sudo")
            sys.exit(1)
    else:
        # Need direct USB access for other commands
        device = LM360()
        if not device.connect():
            print("✗ Failed to connect to LM360")
            print("  Make sure:")
            print("  - Device is plugged in (lsusb | grep 3633)")
            print("  - Running with sudo")
            sys.exit(1)

    try:
        # Execute command
        if args.command == 'monitor':
            cmd_monitor(device, args)
        elif args.command == 'image':
            cmd_image(device, args)
        elif args.command == 'solid':
            cmd_solid(device, args)
        elif args.command == 'brightness':
            cmd_brightness(device, args)
    finally:
        if device:
            device.disconnect()

if __name__ == "__main__":
    main()