import os import json import time import threading import librosa import numpy as np import random import base import rest from config import cfg from actuator import Actuator from audio import AudioController DANCE_MUSIC_FOLDER = cfg.get("dance", "musicfd") DANCE_LIGHT_LOUDNESS_LINE = cfg.getfloat("dance", "lightloudline") DANCE_MIN_LOUDNESS_LINE = cfg.getfloat("dance", "minloudline") DANCE_SILENT_TIME = cfg.getfloat("dance", "silenttime") DANCE_MUSIC_MAX_BEAT = 80 DANCE_MUSIC_MIN_BEAT = 30 RECORD_MUSICSEGMENT_DIR = "/mdata/musicsegment.wav" RECORD_MUSICSEGMENT_TIME = 10 # s ACCOMPAY_DANCE_TIME = 120 # s _dancethreadsinglelock = False class _dancethread(threading.Thread): def __init__(self, token, music): self.running = False self.acting = False self.token = token self.music = music threading.Thread.__init__(self) def run(self): global _dancethreadsinglelock if _dancethreadsinglelock : return _dancethreadsinglelock = True self.running = True Dance.SetStatus(DanceState.ANALYZE) # if music is empty, record an music file if self.music == "" : musicdir = RECORD_MUSICSEGMENT_DIR if not self._recordsegment(musicdir): Dance.SetStatus(DanceState.STOP) self.running = False # for break _dancethreadsinglelock = False return else : musicdir = os.path.join(DANCE_MUSIC_FOLDER , self.music+".wav") # analyze music file (beat, duration) = self._analyzeaudio(musicdir) if beat < DANCE_MUSIC_MIN_BEAT or beat > DANCE_MUSIC_MAX_BEAT: Dance.SetStatus(DanceState.STOP) self.running = False # for break _dancethreadsinglelock = False return # if music is not empty, play music if self.music != "" : Dance.SetStatus(DanceState.PLAY) if not AudioController.AsynPlay(self.token, musicdir) : Dance.SetStatus(DanceState.STOP) self.running = False # for break _dancethreadsinglelock = False return time.sleep(DANCE_SILENT_TIME) # dance Dance.SetStatus(DanceState.DANCE) interval = round(60 / beat, 2) if self.music == "" : duration = ACCOMPAY_DANCE_TIME else: duration = duration - DANCE_SILENT_TIME * 2 if duration > 0 : curduration = 0 while curduration < duration and self.running : actiongroup = Dance.GetRandomAction() agduration = self._doactgroup(actiongroup, interval) curduration = curduration + agduration # if music is not empty, stop play music if self.music != "" : AudioController.ForceStop(self.token) Dance.SetStatus(DanceState.STOP) self.running = False # for break _dancethreadsinglelock = False def _recordsegment(self, dir): if AudioController.OpenMic(self.token): if AudioController.Record(self.token, False, dir, RECORD_MUSICSEGMENT_TIME): if AudioController.CloseMic(self.token): return True return False def _analyzeaudio(self, audiodir): try: y, sr = librosa.load(audiodir) loudnesses = librosa.amplitude_to_db(librosa.feature.rms(y=y)) avgloudness = np.mean(loudnesses[0]) if avgloudness < DANCE_MIN_LOUDNESS_LINE: # if average loudness less than min loudness, return failed return (-1, -1) beat, _ = librosa.beat.beat_track(y=y, sr=sr) # get available beat between max and min while beat > DANCE_MUSIC_MAX_BEAT: beat = beat / 2 if beat < DANCE_MUSIC_MIN_BEAT: return (-1, -1) if avgloudness < DANCE_LIGHT_LOUDNESS_LINE: # if average loudness less than loudness line, use half beat to make dance slow halfbeat = beat / 2 if halfbeat > DANCE_MUSIC_MAX_BEAT: beat = halfbeat duration = librosa.get_duration(y=y) return (beat, duration) except Exception as e: return (-1, -1) def _doactgroup(self, actiongroup, interval): agduration = 0 for a in actiongroup : starttime = time.time() aduration = interval * a["count"] if aduration <= 0: continue match a["action"]: case "f": Actuator.ForwardByTime(self.token, aduration) case "b": Actuator.BackwardByTime(self.token, aduration) case "l": Actuator.LeftByTime(self.token, aduration) case "r": Actuator.RightByTime(self.token, aduration) case "lf": Actuator.LeftForwardByTime(self.token, aduration) case "lb": Actuator.LeftBackwardByTime(self.token, aduration) case "rf": Actuator.RightForwardByTime(self.token, aduration) case "rb": Actuator.RightBackwardByTime(self.token, aduration) stoptime = time.time() sleepduration = aduration + interval - (stoptime - starttime) while sleepduration < 0: sleepduration = sleepduration + interval time.sleep(sleepduration) agduration = agduration + aduration + interval return agduration def IsRunning(self): return _dancethreadsinglelock def StopThread(self): self.running = False while(self.is_alive()): time.sleep(1) class DanceState(): STOP = 0 ANALYZE = 1 PLAY = 2 DANCE = 3 class _dance: def __init__(self): self.actionf = cfg.get("dance", "actionf") self.actions = self._initactions() self.actioncount = len(self.actions) self.dancethread = None self.status = DanceState.STOP def _initactions(self): with open(self.actionf, "r") as jsonf: return json.load(jsonf) return [] def GetRandomAction(self): ri = random.randint(0, self.actioncount-1) return self.actions[ri] def GetMusics(self): musics = [] for root, dirs, files in os.walk(DANCE_MUSIC_FOLDER): for f in files: if f.endswith(".wav"): musics.append(f[:(len(f)-4)]) return musics def SetStatus(self, status): self.status = status def GetDanceInfo(self): return {"status": self.status} def StartDance(self, token, music): if self.dancethread is not None: if self.dancethread.IsRunning() : return False if self.actioncount < 1 : return False self.dancethread = _dancethread(token, music) self.dancethread.start() return True def StopDance(self, token): if self.dancethread is not None: if self.dancethread.IsRunning(): self.dancethread.StopThread() return True Dance = _dance()