work on music player
This commit is contained in:
Lucca Pirovano
+112
-63
@@ -10,7 +10,8 @@ import queue
|
||||
import cv2
|
||||
import numpy as np
|
||||
import pygame
|
||||
import subprocess
|
||||
import yt_dlp
|
||||
import os
|
||||
|
||||
# --- Audio for mic ---
|
||||
try:
|
||||
@@ -18,10 +19,9 @@ try:
|
||||
HAS_AUDIO = True
|
||||
except ImportError:
|
||||
HAS_AUDIO = False
|
||||
print("⚠️ PyAudio not installed.")
|
||||
|
||||
# ==========================================
|
||||
# NAO MIC STREAM TO LAPTOP
|
||||
# NAO MIC STREAM
|
||||
# ==========================================
|
||||
class SoundReceiver:
|
||||
PREBUFFER_CHUNKS = 3
|
||||
@@ -47,8 +47,7 @@ class SoundReceiver:
|
||||
chosen = (self.p.get_device_info_by_index(output_device_index) if output_device_index is not None
|
||||
else self.p.get_default_output_device_info())
|
||||
print(f"🔈 Using: [{chosen['index']}] {chosen['name']}")
|
||||
except:
|
||||
pass
|
||||
except: pass
|
||||
|
||||
self.stream = self.p.open(format=pyaudio.paInt16, channels=1, rate=16000,
|
||||
output=True, output_device_index=output_device_index,
|
||||
@@ -138,73 +137,81 @@ class NaoMusicPlayer:
|
||||
self.music_dir = "/home/nao/music"
|
||||
try:
|
||||
self.session.service("ALFileManager").createFolder(self.music_dir)
|
||||
except:
|
||||
pass
|
||||
except: pass
|
||||
self.current_file = None
|
||||
self.is_playing = False
|
||||
self.current_title = ""
|
||||
|
||||
def search_and_play(self, query):
|
||||
print(f"🔍 Searching: {query}")
|
||||
def search(self, query):
|
||||
try:
|
||||
result = subprocess.run(['ytfzf', '-m', '-L', query], capture_output=True, text=True, timeout=15)
|
||||
url = result.stdout.strip().split('\n')[0] if result.stdout.strip() else None
|
||||
if not url:
|
||||
import yt_dlp
|
||||
with yt_dlp.YoutubeDL({'quiet': True, 'extract_flat': True}) as ydl:
|
||||
info = ydl.extract_info(f"ytsearch1:{query}", download=False)
|
||||
url = info['entries'][0]['url']
|
||||
except Exception as e:
|
||||
print(f"Search failed: {e}")
|
||||
return False
|
||||
|
||||
self.play_on_nao(url, query)
|
||||
return True
|
||||
|
||||
def play_on_nao(self, url, title=""):
|
||||
self.stop()
|
||||
try:
|
||||
filename = f"{int(time.time())}.mp3"
|
||||
local_path = f"/tmp/{filename}"
|
||||
remote_path = f"{self.music_dir}/{filename}"
|
||||
|
||||
import yt_dlp
|
||||
ydl_opts = {'format': 'bestaudio/best', 'outtmpl': local_path, 'quiet': True,
|
||||
'postprocessors': [{'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3'}]}
|
||||
ydl_opts = {
|
||||
'format': 'bestaudio/best',
|
||||
'quiet': True,
|
||||
'noplaylist': True,
|
||||
'extract_flat': True,
|
||||
}
|
||||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||||
ydl.download([url])
|
||||
info = ydl.extract_info(f"ytsearch5:{query}", download=False)
|
||||
return info.get('entries', [])
|
||||
except:
|
||||
return []
|
||||
|
||||
def play(self, url, title):
|
||||
self.stop()
|
||||
self.current_title = title
|
||||
try:
|
||||
ydl_opts = {
|
||||
'format': 'bestaudio/best',
|
||||
'outtmpl': '/tmp/%(id)s.%(ext)s',
|
||||
'quiet': True,
|
||||
'noplaylist': True,
|
||||
'extractaudio': True,
|
||||
'postprocessors': [{'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3'}],
|
||||
'ignoreerrors': True,
|
||||
'no_warnings': True,
|
||||
}
|
||||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||||
info = ydl.extract_info(url, download=True)
|
||||
if not info:
|
||||
return False
|
||||
downloaded = ydl.prepare_filename(info)
|
||||
|
||||
remote_path = f"{self.music_dir}/{int(time.time())}.mp3"
|
||||
file_manager = self.session.service("ALFileManager")
|
||||
file_manager.put(local_path, remote_path)
|
||||
# Use uploadFile instead of put
|
||||
file_manager.uploadFile(downloaded, remote_path)
|
||||
|
||||
self.current_file = remote_path
|
||||
self.player.playFile(remote_path)
|
||||
self.is_playing = True
|
||||
print(f"♪ Playing on NAO: {title}")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Playback error: {e}")
|
||||
return False
|
||||
|
||||
def pause(self):
|
||||
if not self.current_file: return
|
||||
if self.is_playing:
|
||||
self.player.pause()
|
||||
self.is_playing = False
|
||||
print("⏸ Paused")
|
||||
else:
|
||||
self.player.play(self.current_file)
|
||||
self.is_playing = True
|
||||
print("▶ Resumed")
|
||||
|
||||
def stop(self):
|
||||
self.player.stopAll()
|
||||
try:
|
||||
self.player.stopAll()
|
||||
except:
|
||||
pass
|
||||
self.is_playing = False
|
||||
print("⏹ Stopped")
|
||||
self.current_file = None
|
||||
|
||||
def seek(self, seconds):
|
||||
try:
|
||||
pos = self.player.getCurrentPosition() / 1000.0
|
||||
self.player.goTo(pos + seconds)
|
||||
print(f"Seeked {seconds:+.0f}s")
|
||||
except:
|
||||
print("Seek unavailable")
|
||||
except: pass
|
||||
|
||||
|
||||
# ==========================================
|
||||
@@ -221,7 +228,9 @@ class NaoTeleop:
|
||||
except:
|
||||
self.battery = None
|
||||
|
||||
self.music = NaoMusicPlayer(session) # Music on NAO
|
||||
self.music = NaoMusicPlayer(session)
|
||||
self.music_results = []
|
||||
self.selected_result = 0
|
||||
|
||||
self.audio_device = None
|
||||
self.audio_service_name = "SoundReceiver"
|
||||
@@ -238,6 +247,8 @@ class NaoTeleop:
|
||||
|
||||
self.typing_mode = False
|
||||
self.chat_message = ""
|
||||
self.music_search_mode = False
|
||||
self.music_search_text = ""
|
||||
|
||||
self.video_scale = video_scale
|
||||
self.display_width = int(320 * video_scale)
|
||||
@@ -312,12 +323,11 @@ class NaoTeleop:
|
||||
neutral = [1.5, -0.15, 1.2, 0.5, 0.0, 0.0]
|
||||
self.motion.angleInterpolationWithSpeed(names, neutral, 0.3)
|
||||
self.motion.setStiffnesses("RArm", 0.6)
|
||||
print("✅ Wave completed")
|
||||
except Exception as e:
|
||||
print(f"Wave error: {e}")
|
||||
|
||||
def handle_head_movement(self):
|
||||
if self.typing_mode: return
|
||||
if self.typing_mode or self.music_search_mode: return
|
||||
keys = pygame.key.get_pressed()
|
||||
speed = 0.3
|
||||
changed = False
|
||||
@@ -332,7 +342,6 @@ class NaoTeleop:
|
||||
self.head_yaw = 0.0
|
||||
self.head_pitch = 0.0
|
||||
self.motion.setAngles(["HeadYaw", "HeadPitch"], [0.0, 0.0], 0.4)
|
||||
print("Head reset to neutral")
|
||||
|
||||
def update_title(self):
|
||||
mode = "[TYPING] " if self.typing_mode else ""
|
||||
@@ -365,17 +374,7 @@ class NaoTeleop:
|
||||
|
||||
carpet_config = [["StepHeight", 0.028], ["MaxStepFrequency", 0.6], ["TorsoWy", 0.05]]
|
||||
|
||||
print("\n🎮 Controls:")
|
||||
print(" WASD/Arrows = Walk")
|
||||
print(" I J K L = Head")
|
||||
print(" SPACE = Reset head")
|
||||
print(" 1 = Wave")
|
||||
print(" M = Music search (plays on NAO)")
|
||||
print(" P = Pause/Resume music")
|
||||
print(" S = Stop music")
|
||||
print(" ← → = Seek ±10s")
|
||||
print(" T = TTS")
|
||||
print(" ESC = Quit")
|
||||
print("\n🎮 Controls: WASD=Walk | IJKL=Head | SPACE=Reset head | 1=Wave | M=Music search | P=Pause | X=Stop | ←→=Seek | T=TTS | ESC=Quit")
|
||||
|
||||
while self.running:
|
||||
self.check_battery()
|
||||
@@ -384,7 +383,30 @@ class NaoTeleop:
|
||||
if event.type == pygame.QUIT:
|
||||
self.running = False
|
||||
elif event.type == pygame.KEYDOWN:
|
||||
if self.typing_mode:
|
||||
if self.music_search_mode:
|
||||
if event.key == pygame.K_RETURN:
|
||||
if self.music_search_text.strip():
|
||||
self.music_results = self.music.search(self.music_search_text)
|
||||
if self.music_results:
|
||||
self.selected_result = 0
|
||||
else:
|
||||
self.music_search_mode = False
|
||||
else:
|
||||
self.music_search_mode = False
|
||||
elif event.key == pygame.K_ESCAPE:
|
||||
self.music_search_mode = False
|
||||
self.music_search_text = ""
|
||||
elif event.key == pygame.K_BACKSPACE:
|
||||
self.music_search_text = self.music_search_text[:-1]
|
||||
elif event.key in (pygame.K_1, pygame.K_2, pygame.K_3, pygame.K_4, pygame.K_5):
|
||||
idx = int(event.unicode) - 1
|
||||
if idx < len(self.music_results):
|
||||
entry = self.music_results[idx]
|
||||
self.music.play(entry['url'], entry.get('title', 'Unknown'))
|
||||
self.music_search_mode = False
|
||||
else:
|
||||
self.music_search_text += event.unicode
|
||||
elif self.typing_mode:
|
||||
if event.key == pygame.K_RETURN:
|
||||
if self.chat_message.strip():
|
||||
threading.Thread(target=self.tts.say, args=(self.chat_message,)).start()
|
||||
@@ -405,12 +427,12 @@ class NaoTeleop:
|
||||
elif event.key == pygame.K_SPACE:
|
||||
self.reset_head()
|
||||
elif event.key == pygame.K_m:
|
||||
query = input("Search music to play on NAO: ").strip()
|
||||
if query:
|
||||
self.music.search_and_play(query)
|
||||
self.music_search_mode = True
|
||||
self.music_search_text = ""
|
||||
self.music_results = []
|
||||
elif event.key == pygame.K_p:
|
||||
self.music.pause()
|
||||
elif event.key == pygame.K_s:
|
||||
elif event.key == pygame.K_x: # Stop music
|
||||
self.music.stop()
|
||||
elif event.key == pygame.K_LEFT:
|
||||
self.music.seek(-10)
|
||||
@@ -420,7 +442,7 @@ class NaoTeleop:
|
||||
self.typing_mode = True
|
||||
self.motion.stopMove()
|
||||
|
||||
if not self.typing_mode:
|
||||
if not self.typing_mode and not self.music_search_mode:
|
||||
keys = pygame.key.get_pressed()
|
||||
x = y = theta = 0.0
|
||||
if keys[pygame.K_w] or keys[pygame.K_UP]: x = 0.6
|
||||
@@ -443,6 +465,33 @@ class NaoTeleop:
|
||||
else:
|
||||
self.screen.fill((10, 10, 30))
|
||||
|
||||
# Music status
|
||||
if self.music.current_title:
|
||||
status = f"♪ {self.music.current_title[:45]} {'(Paused)' if not self.music.is_playing else ''}"
|
||||
text = self.font.render(status, True, (0, 255, 100))
|
||||
self.screen.blit(text, (10, 10))
|
||||
|
||||
# Music search interface
|
||||
if self.music_search_mode:
|
||||
s = pygame.Surface((self.display_width, self.display_height//2))
|
||||
s.set_alpha(220)
|
||||
s.fill((0, 0, 40))
|
||||
self.screen.blit(s, (0, 80))
|
||||
|
||||
text = self.font.render(f"Search: {self.music_search_text}", True, (255, 255, 255))
|
||||
self.screen.blit(text, (20, 100))
|
||||
|
||||
if self.music_results:
|
||||
for i, entry in enumerate(self.music_results[:5]):
|
||||
color = (255, 255, 0) if i == self.selected_result else (200, 200, 200)
|
||||
title = entry.get('title', 'No title')[:60]
|
||||
line = self.font.render(f"{i+1}. {title}", True, color)
|
||||
self.screen.blit(line, (20, 140 + i*30))
|
||||
else:
|
||||
text = self.font.render("Searching...", True, (200, 200, 200))
|
||||
self.screen.blit(text, (20, 140))
|
||||
|
||||
# TTS box
|
||||
if self.typing_mode:
|
||||
s = pygame.Surface((self.display_width, 40))
|
||||
s.set_alpha(180)
|
||||
|
||||
Reference in New Issue
Block a user