import pygame import numpy as np from flask import Flask, Response, render_template_string, request, jsonify import time import os import cv2 import subprocess import threading import queue import uuid # ============ CONFIGURATION ============ VIDEO_WIDTH = 1280 VIDEO_HEIGHT = 720 VIDEO_FPS = 30 JPEG_QUALITY = 70 STREAM_PORT = 7860 # Colors (RGB) COLOR_TOP = (255, 100, 100) COLOR_BOTTOM = (100, 255, 100) COLOR_BG = (20, 20, 30) COLOR_GRID = (50, 50, 70) COLOR_FPS = (255, 255, 0) COLOR_CLOCK = (0, 255, 255) # Button colors COLOR_BUTTON_NORMAL = (80, 80, 80) COLOR_BUTTON_HOVER = (100, 100, 200) COLOR_BUTTON_CLICKED = (0, 200, 0) COLOR_BUTTON_BORDER = (200, 200, 200) # Sound source colors COLOR_SOUND_NONE = (255, 100, 100) COLOR_SOUND_PYGAME = (100, 255, 100) COLOR_SOUND_BROWSER = (100, 100, 255) # ======================================= # Initialize Pygame headlessly os.environ['SDL_VIDEODRIVER'] = 'dummy' pygame.init() # Try to initialize mixer, but don't crash if it fails try: pygame.mixer.init(frequency=44100, size=-16, channels=2) print("✅ Audio mixer initialized") except Exception as e: print(f"⚠️ Audio mixer not available: {e}") print(" Continuing without sound...") app = Flask(__name__) class ShaderRenderer: def __init__(self, width=VIDEO_WIDTH, height=VIDEO_HEIGHT): self.width = width self.height = height self.mouse_x = width // 2 self.mouse_y = height // 2 self.start_time = time.time() self.surface = pygame.Surface((width, height)) self.frame_count = 0 self.last_frame_time = time.time() self.fps = 0 self.button_clicked = False # Sound sources self.sound_source = 'none' self.pygame_sound = None self.pygame_playing = False self.sound_amp = 0.0 # Load pygame sound if available if os.path.exists('sound.mp3'): try: self.pygame_sound = pygame.mixer.Sound('sound.mp3') print("✅ Pygame sound loaded") except: print("⚠️ Could not load sound.mp3") def set_mouse(self, x, y): self.mouse_x = max(0, min(self.width, x)) self.mouse_y = max(0, min(self.height, y)) def set_sound_source(self, source): """Change sound source: none, pygame, browser""" self.sound_source = source # Handle pygame sound if source == 'pygame': if self.pygame_sound and not self.pygame_playing: self.pygame_sound.play(loops=-1) self.pygame_playing = True self.sound_amp = 0.5 else: if self.pygame_playing: pygame.mixer.stop() self.pygame_playing = False self.sound_amp = 0.0 def handle_click(self, x, y): """Handle mouse clicks on Pygame surface""" button_rect = pygame.Rect(self.width-250, 120, 220, 50) if button_rect.collidepoint(x, y): self.button_clicked = not self.button_clicked print(f"🎯 Button {'clicked!' if self.button_clicked else 'unclicked!'}") return True return False def render_frame(self): """Render the pygame frame""" t = time.time() - self.start_time # Calculate FPS self.frame_count += 1 if time.time() - self.last_frame_time > 1.0: self.fps = self.frame_count self.frame_count = 0 self.last_frame_time = time.time() # Clear self.surface.fill(COLOR_BG) # Use larger font for 720p font = pygame.font.Font(None, 36) small_font = pygame.font.Font(None, 24) # Draw TOP marker pygame.draw.rect(self.surface, COLOR_TOP, (10, 10, 150, 40)) text = font.render("TOP", True, (255, 255, 255)) self.surface.blit(text, (30, 15)) # Draw BOTTOM marker pygame.draw.rect(self.surface, COLOR_BOTTOM, (10, self.height-50, 150, 40)) text = font.render("BOTTOM", True, (0, 0, 0)) self.surface.blit(text, (20, self.height-45)) # ===== CLOCK DISPLAY ===== current_time = time.time() seconds = int(current_time) % 60 hundredths = int((current_time * 100) % 100) # Clock background pygame.draw.rect(self.surface, (40, 40, 50), (self.width-250, 70, 220, 50)) pygame.draw.rect(self.surface, (100, 100, 150), (self.width-250, 70, 220, 50), 2) # Clock text time_str = f"{seconds:02d}.{hundredths:02d}s" clock_text = font.render(time_str, True, COLOR_CLOCK) self.surface.blit(clock_text, (self.width-230, 80)) # ===== CLICKABLE BUTTON ===== button_rect = pygame.Rect(self.width-250, 140, 220, 50) # Check if mouse is over button mouse_over = button_rect.collidepoint(self.mouse_x, self.mouse_y) # Button color based on state and hover if self.button_clicked: button_color = COLOR_BUTTON_CLICKED elif mouse_over: button_color = COLOR_BUTTON_HOVER else: button_color = COLOR_BUTTON_NORMAL # Draw button pygame.draw.rect(self.surface, button_color, button_rect) pygame.draw.rect(self.surface, COLOR_BUTTON_BORDER, button_rect, 3) # Button text if self.button_clicked: btn_text = "✅ CLICKED!" else: btn_text = "🔘 CLICK ME" text_surf = font.render(btn_text, True, (255, 255, 255)) text_rect = text_surf.get_rect(center=button_rect.center) self.surface.blit(text_surf, text_rect) # ===== MOVING CIRCLE ===== circle_size = 40 + int(30 * np.sin(t * 2)) # Color code based on sound source if self.sound_source == 'pygame': color = COLOR_SOUND_PYGAME elif self.sound_source == 'browser': color = COLOR_SOUND_BROWSER else: color = COLOR_SOUND_NONE pygame.draw.circle(self.surface, color, (self.mouse_x, self.mouse_y), circle_size) # ===== GRID ===== for x in range(0, self.width, 70): alpha = int(40 + 20 * np.sin(x * 0.1 + t)) pygame.draw.line(self.surface, (alpha, alpha, 50), (x, 0), (x, self.height)) for y in range(0, self.height, 70): alpha = int(40 + 20 * np.cos(y * 0.1 + t)) pygame.draw.line(self.surface, (alpha, alpha, 50), (0, y), (self.width, y)) # ===== SOUND METER ===== meter_width = int(250 * self.sound_amp) pygame.draw.rect(self.surface, (60, 60, 60), (self.width-270, 210, 250, 25)) pygame.draw.rect(self.surface, (100, 255, 100), (self.width-270, 210, meter_width, 25)) # ===== FPS COUNTER ===== fps_text = small_font.render(f"FPS: {self.fps}", True, COLOR_FPS) self.surface.blit(fps_text, (self.width-200, self.height-80)) return pygame.image.tostring(self.surface, 'RGB') def get_frame(self): return self.render_frame() def get_frame_jpeg(self, quality=JPEG_QUALITY): """Return frame as JPEG""" frame = self.get_frame() # Convert to numpy array for OpenCV img = np.frombuffer(frame, dtype=np.uint8).reshape((self.height, self.width, 3)) # Convert RGB to BGR for OpenCV img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) # Encode as JPEG _, jpeg = cv2.imencode('.jpg', img, [cv2.IMWRITE_JPEG_QUALITY, quality]) return jpeg.tobytes() renderer = ShaderRenderer() # ============ STREAMING ENDPOINTS ============ @app.route('/video/mjpeg') def video_mjpeg(): """MJPEG streaming endpoint""" def generate(): while True: frame = renderer.get_frame_jpeg() yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n') time.sleep(1/VIDEO_FPS) return Response( generate(), mimetype='multipart/x-mixed-replace; boundary=frame' ) @app.route('/video/webm') def video_webm(): """WebM streaming endpoint""" cmd = [ 'ffmpeg', '-f', 'rawvideo', '-pix_fmt', 'rgb24', '-s', f'{VIDEO_WIDTH}x{VIDEO_HEIGHT}', '-r', str(VIDEO_FPS), '-i', '-', '-c:v', 'libvpx-vp9', '-b:v', '2M', '-cpu-used', '4', '-deadline', 'realtime', '-f', 'webm', '-' ] process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0 ) def generate(): def push_frames(): while True: try: frame = renderer.get_frame() process.stdin.write(frame) except: break threading.Thread(target=push_frames, daemon=True).start() while True: data = process.stdout.read(4096) if not data: break yield data return Response( generate(), mimetype='video/webm', headers={'Cache-Control': 'no-cache', 'Transfer-Encoding': 'chunked'} ) @app.route('/video/mp4') def video_mp4(): """GPU-accelerated MP4 streaming using NVENC""" cmd = [ 'ffmpeg', '-hwaccel', 'cuda', # Use GPU for decoding '-hwaccel_output_format', 'cuda', # Keep data on GPU '-f', 'rawvideo', '-pix_fmt', 'rgb24', '-s', f'{VIDEO_WIDTH}x{VIDEO_HEIGHT}', '-r', '30', '-i', '-', '-c:v', 'h264_nvenc', # NVIDIA's GPU encoder '-preset', 'p4', # p1-p7, p4 is good balance '-tune', 'll', # Low latency tuning '-b:v', '2M', '-bufsize', '4M', '-f', 'mp4', '-movflags', 'frag_keyframe+empty_moov', '-' ] process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0 ) def generate(): def push_frames(): while True: try: frame = renderer.get_frame() process.stdin.write(frame) except: break threading.Thread(target=push_frames, daemon=True).start() while True: data = process.stdout.read(4096) if not data: break yield data return Response( generate(), mimetype='video/mp4', headers={'Cache-Control': 'no-cache', 'Transfer-Encoding': 'chunked'} ) # ============ INTERACTIVITY ENDPOINTS ============ @app.route('/mouse', methods=['POST']) def mouse(): data = request.json renderer.set_mouse(data['x'], data['y']) return 'OK' @app.route('/click', methods=['POST']) def click(): data = request.json renderer.handle_click(data['x'], data['y']) return 'OK' @app.route('/sound/source', methods=['POST']) def sound_source(): data = request.json renderer.set_sound_source(data['source']) return 'OK' @app.route('/sound/amp') def sound_amp(): return {'amp': renderer.sound_amp} @app.route('/static/sound.mp3') def serve_sound(): if os.path.exists('sound.mp3'): with open('sound.mp3', 'rb') as f: return Response(f.read(), mimetype='audio/mpeg') return 'Sound not found', 404 # ============ HTML PAGE ============ @app.route('/') def index(): return render_template_string(f''' 🎮 Pygame 720p Streaming

🎮 Pygame 720p Streaming

X: 320, Y: 240
Format: MJPEG
Sound: None
Resolution: {VIDEO_WIDTH}x{VIDEO_HEIGHT}
''') if __name__ == '__main__': print("\n" + "="*70) print("🎮 Pygame 720p Streaming App") print("="*70) print(f"📡 Resolution: {VIDEO_WIDTH}x{VIDEO_HEIGHT} @ {VIDEO_FPS}fps") print("📡 Streaming endpoints:") print(" • /video/mjpeg - MJPEG stream") print(" • /video/webm - WebM stream") print(" • /video/mp4 - MP4 stream") print("🖱️ Interactive: mouse + clickable button") print(f"\n🌐 Main page: /") print("="*70 + "\n") port = int(os.environ.get('PORT', STREAM_PORT)) app.run(host='0.0.0.0', port=port, debug=False, threaded=True)