import pygame
import time
import threading
import base64
import os
from flask import Flask, render_template_string
from flask_socketio import SocketIO, emit
from io import BytesIO
os.environ['SDL_VIDEODRIVER'] = 'dummy'
try:
pygame.init()
print("โ
PyGame initialized successfully")
except Exception as e:
print(f"โ PyGame init failed: {e}")
# Create a fallback mode
app = Flask(__name__)
app.config['SECRET_KEY'] = 'simple_stream_key'
# Use threading mode - avoids eventlet issues
socketio = SocketIO(app, async_mode='threading', cors_allowed_origins="*")
WIDTH, HEIGHT = 400, 300
shared = {
"streaming": True,
"x": 200,
"frame_count": 0,
"clients": 0
}
def game_loop():
"""Simple PyGame loop"""
print("๐ฌ Starting PyGame loop")
# Test PyGame drawing
test_screen = pygame.Surface((100, 100))
test_screen.fill((255, 0, 0))
pygame.draw.circle(test_screen, (0, 255, 0), (50, 50), 30)
test_pixel = test_screen.get_at((50, 50))
print(f"๐งช PyGame test - Pixel at (50,50): {test_pixel}")
if test_pixel != (0, 255, 0, 255):
print("โ ๏ธ WARNING: PyGame drawing might not work correctly")
# Main game surface
screen = pygame.Surface((WIDTH, HEIGHT))
x, speed = 200, 5
frame_counter = 0
last_log_time = time.time()
while shared["streaming"]:
start_time = time.time()
# Update
x += speed
if x < 30 or x > WIDTH-30:
speed *= -1
# Draw
screen.fill((25, 25, 45))
pygame.draw.circle(screen, (255, 80, 80), (int(x), 150), 20)
# Convert to JPEG
buf = BytesIO()
pygame.image.save(screen, buf, format='JPEG', quality=85)
frame_b64 = base64.b64encode(buf.getvalue()).decode('utf-8')
# Broadcast frame
socketio.emit('new_frame', {
'frame': frame_b64,
'count': shared['frame_count'],
'x': x,
'timestamp': time.time()
})
# Update shared state
shared['x'] = x
shared['frame_count'] += 1
frame_counter += 1
# Log every second
current_time = time.time()
if current_time - last_log_time >= 1.0:
print(f"๐ FPS: {frame_counter} | Total: {shared['frame_count']} | Clients: {shared['clients']}")
frame_counter = 0
last_log_time = current_time
# Frame rate control
elapsed = time.time() - start_time
if elapsed < 1.0/30:
time.sleep(1.0/30 - elapsed)
print("๐ PyGame loop stopped")
@app.route('/')
def index():
return render_template_string('''
๐ Working PyGame Stream
![]()
Frame: 0
Position: X=200
Status: Connecting...
Clients: 0
''')
@socketio.on('connect')
def handle_connect():
shared['clients'] += 1
print(f"Client connected. Total: {shared['clients']}")
socketio.emit('client_count', shared['clients'])
@socketio.on('disconnect')
def handle_disconnect():
shared['clients'] -= 1
print(f"Client disconnected. Total: {shared['clients']}")
socketio.emit('client_count', shared['clients'])
if __name__ == '__main__':
print("๐ Starting server...")
# Start PyGame thread
game_thread = threading.Thread(target=game_loop, daemon=True)
game_thread.start()
# Give it a moment
time.sleep(1)
# Start server
socketio.run(
app,
host='0.0.0.0',
port=int(os.environ.get('PORT', 7860)),
debug=False,
use_reloader=False,
allow_unsafe_werkzeug=True
)