UNLOZE/projects-jenz
9
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

reverted the changes from metroid to torchlight

Browse Source
This commit is contained in:
jenz
2026-08-15 21:30:51 +02:00
parent e0e043d2ad
commit fd84509996
3 changed files with 1549 additions and 2734 deletions
Download Patch File Download Diff File Expand all files Collapse all files
torchlight_changes_unloze/torchlight3/Torchlight/AudioManager.py
+352 -443
View File
@@ -1,443 +1,352 @@
#!/usr/bin/python3 #!/usr/bin/python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import logging import logging
import sys import sys
import math import io
import time import math
from .FFmpegAudioPlayer import FFmpegAudioPlayerFactory from .FFmpegAudioPlayer import FFmpegAudioPlayerFactory
def _progress_delta(old_position, new_position, elapsed = None, jitter = 0.25): class AudioPlayerFactory():
try: AUDIOPLAYER_FFMPEG = 1
delta = abs(float(new_position) - float(old_position))
except (TypeError, ValueError): def __init__(self, master):
return 0.0 self.Logger = logging.getLogger(__class__.__name__)
if not math.isfinite(delta): self.Master = master
return 0.0 self.Torchlight = self.Master.Torchlight
if elapsed is not None:
try: self.FFmpegAudioPlayerFactory = FFmpegAudioPlayerFactory(self)
elapsed = float(elapsed)
except (TypeError, ValueError): def __del__(self):
return 0.0 self.Logger.info("~AudioPlayerFactory()")
if not math.isfinite(elapsed) or elapsed < 0.0:
return 0.0 def NewPlayer(self, _type):
delta = min(delta, elapsed + jitter) if _type == self.AUDIOPLAYER_FFMPEG:
return delta return self.FFmpegAudioPlayerFactory.NewPlayer()
class AudioPlayerFactory(): class AntiSpam():
AUDIOPLAYER_FFMPEG = 1 def __init__(self, master):
self.Logger = logging.getLogger(__class__.__name__)
def __init__(self, master): self.Master = master
self.Logger = logging.getLogger(__class__.__name__) self.Torchlight = self.Master.Torchlight
self.Master = master
self.Torchlight = self.Master.Torchlight self.LastClips = dict()
self.DisabledTime = None
self.FFmpegAudioPlayerFactory = FFmpegAudioPlayerFactory(self) self.SaidHint = False
def __del__(self): def CheckAntiSpam(self, player):
self.Logger.info("~AudioPlayerFactory()") if self.DisabledTime and self.DisabledTime > self.Torchlight().Master.Loop.time() and \
not (player.Access and player.Access["level"] >= self.Torchlight().Config["AntiSpam"]["ImmunityLevel"]):
def NewPlayer(self, _type):
if _type == self.AUDIOPLAYER_FFMPEG: self.Torchlight().SayPrivate(player, "Torchlight is currently on cooldown! ({0} seconds left)".format(
return self.FFmpegAudioPlayerFactory.NewPlayer() math.ceil(self.DisabledTime - self.Torchlight().Master.Loop.time())))
raise ValueError("Unsupported audio player type: {0}".format(_type)) return False
return True
class AntiSpam():
GLOBAL_PLAY_TIME_IMMUNITY_LEVEL = 10 def SpamCheck(self, Delta):
Now = self.Torchlight().Master.Loop.time()
def __init__(self, master): Duration = 0.0
self.Logger = logging.getLogger(__class__.__name__)
self.Master = master for Key, Clip in list(self.LastClips.items()):
self.Torchlight = self.Master.Torchlight if not Clip["timestamp"]:
continue
self.LastClips = dict()
self.DisabledTime = None if Clip["timestamp"] + Clip["duration"] + self.Torchlight().Config["AntiSpam"]["MaxUsageSpan"] < Now:
if not Clip["active"]:
def _counts_toward_global_play_time(self, level): del self.LastClips[Key]
return level < self.GLOBAL_PLAY_TIME_IMMUNITY_LEVEL continue
def CheckAntiSpam(self, player): Duration += Clip["duration"]
if self.DisabledTime and self.DisabledTime > self.Torchlight().Master.Loop.time() and \
not (player.Access and player.Access["level"] >= self.Torchlight().Config["AntiSpam"]["ImmunityLevel"]): if Duration > self.Torchlight().Config["AntiSpam"]["MaxUsageTime"]:
self.DisabledTime = self.Torchlight().Master.Loop.time() + self.Torchlight().Config["AntiSpam"]["PunishDelay"]
self.Torchlight().SayPrivate(player, "Torchlight is currently on cooldown! ({0} seconds left)".format( self.Torchlight().SayChat("Blocked voice commands for the next {0} seconds. Used {1} seconds within {2} seconds.".format(
math.ceil(self.DisabledTime - self.Torchlight().Master.Loop.time()))) self.Torchlight().Config["AntiSpam"]["PunishDelay"], self.Torchlight().Config["AntiSpam"]["MaxUsageTime"], self.Torchlight().Config["AntiSpam"]["MaxUsageSpan"]))
return False
# Make a copy of the list since AudioClip.Stop() will change the list
return True for AudioClip in self.Master.AudioClips[:]:
if AudioClip.Level < self.Torchlight().Config["AntiSpam"]["ImmunityLevel"]:
def SpamCheck(self): AudioClip.Stop()
Now = self.Torchlight().Master.Loop.time()
Duration = 0.0 self.LastClips.clear()
config = self.Torchlight().Config["AntiSpam"]
def OnPlay(self, clip):
for Key, Clip in list(self.LastClips.items()): Now = self.Torchlight().Master.Loop.time()
if not Clip["timestamp"]: self.LastClips[hash(clip)] = dict({"timestamp": Now, "duration": 0.0, "dominant": False, "active": True})
continue
HasDominant = False
if Clip["timestamp"] + Clip["duration"] + config["MaxUsageSpan"] < Now: for Key, Clip in self.LastClips.items():
if not Clip["active"]: if Clip["dominant"]:
del self.LastClips[Key] HasDominant = True
continue break
Duration += Clip["duration"] self.LastClips[hash(clip)]["dominant"] = not HasDominant
if Duration > config["MaxUsageTime"]: def OnStop(self, clip):
self.DisabledTime = self.Torchlight().Master.Loop.time() + config["PunishDelay"] if hash(clip) not in self.LastClips:
self.Torchlight().SayChat("Blocked voice commands for the next {0} seconds. Used {1} seconds within {2} seconds.".format( return
config["PunishDelay"], config["MaxUsageTime"], config["MaxUsageSpan"]))
self.LastClips[hash(clip)]["active"] = False
# Make a copy of the list since AudioClip.Stop() will change the list
for AudioClip in self.Master.AudioClips[:]: if self.LastClips[hash(clip)]["dominant"]:
if AudioClip.Level < config["ImmunityLevel"]: for Key, Clip in self.LastClips.items():
AudioClip.Stop() if Clip["active"]:
Clip["dominant"] = True
self.LastClips.clear() break
def OnPlay(self, clip): self.LastClips[hash(clip)]["dominant"] = False
if not self._counts_toward_global_play_time(clip.Level):
return def OnUpdate(self, clip, old_position, new_position):
Delta = new_position - old_position
Now = self.Torchlight().Master.Loop.time() Clip = self.LastClips[hash(clip)]
self.LastClips[clip] = dict({"timestamp": Now, "duration": 0.0, "dominant": False, "active": True, "last_update_realtime": time.time()})
if not Clip["dominant"]:
HasDominant = False return
for Key, Clip in self.LastClips.items():
if Clip["dominant"]: Clip["duration"] += Delta
HasDominant = True self.SpamCheck(Delta)
break
self.LastClips[clip]["dominant"] = not HasDominant class Advertiser():
def __init__(self, master):
def OnStop(self, clip): self.Logger = logging.getLogger(__class__.__name__)
if not self._counts_toward_global_play_time(clip.Level): self.Master = master
return self.Torchlight = self.Master.Torchlight
if clip not in self.LastClips: self.LastClips = dict()
return self.AdStop = 0
self.NextAdStop = 0
self.LastClips[clip]["active"] = False
def Think(self, Delta):
if self.LastClips[clip]["dominant"]: Now = self.Torchlight().Master.Loop.time()
for Key, Clip in self.LastClips.items(): Duration = 0.0
if Clip["active"]:
Clip["dominant"] = True for Key, Clip in list(self.LastClips.items()):
break if not Clip["timestamp"]:
continue
self.LastClips[clip]["dominant"] = False
if Clip["timestamp"] + Clip["duration"] + self.Torchlight().Config["Advertiser"]["MaxSpan"] < Now:
def OnUpdate(self, clip, old_position, new_position): if not Clip["active"]:
if not self._counts_toward_global_play_time(clip.Level): del self.LastClips[Key]
return continue
if clip not in self.LastClips: Duration += Clip["duration"]
if self.Logger.isEnabledFor(logging.DEBUG):
self.Logger.debug( self.NextAdStop -= Delta
"OnUpdate called for unknown clip key %r; %d known keys", CeilDur = math.ceil(Duration)
clip, if CeilDur > self.AdStop and self.NextAdStop <= 0 and CeilDur % self.Torchlight().Config["Advertiser"]["AdStop"] == 0:
len(self.LastClips), self.Torchlight().SayChat("Hint: Type \x07FF0000!stop(ze) !pls(mg)\x01 to stop all currently playing sounds.")
) self.AdStop = CeilDur
return self.NextAdStop = 0
Clip = self.LastClips[clip] elif CeilDur < self.AdStop:
now = time.time() self.AdStop = 0
elapsed = None self.NextAdStop = self.Torchlight().Config["Advertiser"]["AdStop"] / 2
if Clip.get("last_update_realtime") is not None:
elapsed = now - Clip["last_update_realtime"] def OnPlay(self, clip):
Clip["last_update_realtime"] = now Now = self.Torchlight().Master.Loop.time()
Delta = _progress_delta(old_position, new_position, elapsed = elapsed) self.LastClips[hash(clip)] = dict({"timestamp": Now, "duration": 0.0, "dominant": False, "active": True})
if Delta <= 0.0 or not Clip["dominant"]: HasDominant = False
return for Key, Clip in self.LastClips.items():
if Clip["dominant"]:
Clip["duration"] += Delta HasDominant = True
self.SpamCheck() break
self.LastClips[hash(clip)]["dominant"] = not HasDominant
class Advertiser():
def __init__(self, master): def OnStop(self, clip):
self.Logger = logging.getLogger(__class__.__name__) if hash(clip) not in self.LastClips:
self.Master = master return
self.Torchlight = self.Master.Torchlight
self.LastClips[hash(clip)]["active"] = False
self.LastClips = dict()
if self.LastClips[hash(clip)]["dominant"]:
def Think(self, clip): for Key, Clip in self.LastClips.items():
Now = self.Torchlight().Master.Loop.time() if Clip["active"]:
config = self.Torchlight().Config["Advertiser"] Clip["dominant"] = True
break
for Key, Clip in list(self.LastClips.items()):
if not Clip["timestamp"]: self.LastClips[hash(clip)]["dominant"] = False
continue
def OnUpdate(self, clip, old_position, new_position):
if Clip["timestamp"] + Clip["duration"] + config["MaxSpan"] < Now: Delta = new_position - old_position
if not Clip["active"]: Clip = self.LastClips[hash(clip)]
del self.LastClips[Key]
continue if not Clip["dominant"]:
return
Clip = self.LastClips.get(clip) Clip["duration"] += Delta
if not Clip: self.Think(Delta)
return
if not clip.StopHinted and not Clip.get("hinted") and Clip["duration"] >= config["AdStop"]: class AudioManager():
self.Torchlight().SayChat("Hint: Type \x07FF0000!stop(ze) !pls(mg)\x01 to stop all currently playing sounds.") def __init__(self, torchlight):
Clip["hinted"] = True self.Logger = logging.getLogger(__class__.__name__)
clip.StopHinted = True self.Torchlight = torchlight
self.AntiSpam = AntiSpam(self)
def OnPlay(self, clip): self.Advertiser = Advertiser(self)
Now = self.Torchlight().Master.Loop.time() self.AudioPlayerFactory = AudioPlayerFactory(self)
self.LastClips[clip] = dict({"timestamp": Now, "duration": 0.0, "dominant": False, "active": True, "hinted": False, "last_update_realtime": time.time()}) self.AudioClips = []
HasDominant = False def __del__(self):
for Key, Clip in self.LastClips.items(): self.Logger.info("~AudioManager()")
if Clip["dominant"]:
HasDominant = True def CheckLimits(self, player):
break Level = 0
if player.Access:
self.LastClips[clip]["dominant"] = not HasDominant Level = player.Access["level"]
def OnStop(self, clip): if str(Level) in self.Torchlight().Config["AudioLimits"]:
if clip not in self.LastClips: if self.Torchlight().Config["AudioLimits"][str(Level)]["Uses"] >= 0 and \
return player.Storage["Audio"]["Uses"] >= self.Torchlight().Config["AudioLimits"][str(Level)]["Uses"]:
self.LastClips[clip]["active"] = False self.Torchlight().SayPrivate(player, "You have used up all of your free uses! ({0} uses)".format(
self.Torchlight().Config["AudioLimits"][str(Level)]["Uses"]))
if self.LastClips[clip]["dominant"]: return False
for Key, Clip in self.LastClips.items():
if Clip["active"]: if player.Storage["Audio"]["TimeUsed"] >= self.Torchlight().Config["AudioLimits"][str(Level)]["TotalTime"]:
Clip["dominant"] = True self.Torchlight().SayPrivate(player, "You have used up all of your free time! ({0} seconds)".format(
break self.Torchlight().Config["AudioLimits"][str(Level)]["TotalTime"]))
return False
self.LastClips[clip]["dominant"] = False
TimeElapsed = self.Torchlight().Master.Loop.time() - player.Storage["Audio"]["LastUse"]
def OnUpdate(self, clip, old_position, new_position): UseDelay = player.Storage["Audio"]["LastUseLength"] * self.Torchlight().Config["AudioLimits"][str(Level)]["DelayFactor"]
if clip not in self.LastClips:
return if TimeElapsed < UseDelay:
Clip = self.LastClips[clip] self.Torchlight().SayPrivate(player, "You are currently on cooldown! ({0} seconds left)".format(
now = time.time() round(UseDelay - TimeElapsed)))
elapsed = None return False
if Clip.get("last_update_realtime") is not None:
elapsed = now - Clip["last_update_realtime"] return True
Clip["last_update_realtime"] = now
Delta = _progress_delta(old_position, new_position, elapsed = elapsed) def Stop(self, player, extra):
Level = 0
if Delta <= 0.0 or not Clip["dominant"]: if player.Access:
return Level = player.Access["level"]
Clip["duration"] += Delta for AudioClip in self.AudioClips[:]:
self.Think(clip) if extra and not extra.lower() in AudioClip.Player.Name.lower():
continue
class AudioManager(): if not Level or (Level < AudioClip.Level and Level < self.Torchlight().Config["AntiSpam"]["StopLevel"]):
def __init__(self, torchlight): AudioClip.Stops.add(player.UserID)
self.Logger = logging.getLogger(__class__.__name__)
self.Torchlight = torchlight if len(AudioClip.Stops) >= 3:
self.AntiSpam = AntiSpam(self) AudioClip.Stop()
self.Advertiser = Advertiser(self) self.Torchlight().SayPrivate(AudioClip.Player, "Your audio clip was stopped.")
self.AudioPlayerFactory = AudioPlayerFactory(self) if player != AudioClip.Player:
self.AudioClips = [] self.Torchlight().SayPrivate(player, "Stopped \"{0}\"({1}) audio clip.".format(AudioClip.Player.Name, AudioClip.Player.UserID))
else:
def __del__(self): self.Torchlight().SayPrivate(player, "This audio clip needs {0} more !stop's.".format(3 - len(AudioClip.Stops)))
self.Logger.info("~AudioManager()") else:
AudioClip.Stop()
def CheckLimits(self, player): self.Torchlight().SayPrivate(AudioClip.Player, "Your audio clip was stopped.")
Level = 0 if player != AudioClip.Player:
if player.Access: self.Torchlight().SayPrivate(player, "Stopped \"{0}\"({1}) audio clip.".format(AudioClip.Player.Name, AudioClip.Player.UserID))
Level = player.Access["level"]
def AudioClip(self, player, uri, _type = AudioPlayerFactory.AUDIOPLAYER_FFMPEG):
config = self.Torchlight().Config["AudioLimits"] Level = 0
if str(Level) in config: if player.Access:
level_config = config[str(Level)] Level = player.Access["level"]
if level_config["Uses"] >= 0 and \
player.Storage["Audio"]["Uses"] >= level_config["Uses"]: if self.Torchlight().Disabled and self.Torchlight().Disabled > Level:
self.Torchlight().SayPrivate(player, "Torchlight is currently disabled!")
self.Torchlight().SayPrivate(player, "You have used up all of your free uses! ({0} uses)".format( return None
level_config["Uses"]))
return False if not self.AntiSpam.CheckAntiSpam(player):
return None
if player.Storage["Audio"]["TimeUsed"] >= level_config["TotalTime"]:
self.Torchlight().SayPrivate(player, "You have used up all of your free time! ({0} seconds)".format( if not self.CheckLimits(player):
level_config["TotalTime"])) return None
return False
Clip = AudioClip(self, player, uri, _type)
TimeElapsed = self.Torchlight().Master.Loop.time() - player.Storage["Audio"]["LastUse"] self.AudioClips.append(Clip)
UseDelay = player.Storage["Audio"]["LastUseLength"] * level_config["DelayFactor"]
if not player.Access or player.Access["level"] < self.Torchlight().Config["AntiSpam"]["ImmunityLevel"]:
if TimeElapsed < UseDelay: Clip.AudioPlayer.AddCallback("Play", lambda *args: self.AntiSpam.OnPlay(Clip, *args))
self.Torchlight().SayPrivate(player, "You are currently on cooldown! ({0} seconds left)".format( Clip.AudioPlayer.AddCallback("Stop", lambda *args: self.AntiSpam.OnStop(Clip, *args))
round(UseDelay - TimeElapsed))) Clip.AudioPlayer.AddCallback("Update", lambda *args: self.AntiSpam.OnUpdate(Clip, *args))
return False
Clip.AudioPlayer.AddCallback("Play", lambda *args: self.Advertiser.OnPlay(Clip, *args))
return True Clip.AudioPlayer.AddCallback("Stop", lambda *args: self.Advertiser.OnStop(Clip, *args))
Clip.AudioPlayer.AddCallback("Update", lambda *args: self.Advertiser.OnUpdate(Clip, *args))
def Stop(self, player, extra):
Result = {"active": 0, "matched": 0, "stopped": 0, "pending": 0} return Clip
stopper_level = player.Access["level"] if player.Access else 0
def OnDisconnect(self, player):
for AudioClip in self.AudioClips[:]: for AudioClip in self.AudioClips[:]:
Result["active"] += 1 if AudioClip.Player == player:
target_level = AudioClip.Player.Access["level"] if AudioClip.Player.Access else 0 AudioClip.Stop()
same_player = False
if getattr(player, "UniqueID", None) is not None and getattr(AudioClip.Player, "UniqueID", None) is not None:
same_player = player.UniqueID == AudioClip.Player.UniqueID class AudioClip():
elif getattr(player, "UserID", None) is not None and getattr(AudioClip.Player, "UserID", None) is not None: def __init__(self, master, player, uri, _type):
same_player = player.UserID == AudioClip.Player.UserID self.Logger = logging.getLogger(__class__.__name__)
else: self.Master = master
same_player = player == AudioClip.Player self.Torchlight = self.Master.Torchlight
self.Player = player
if not same_player and target_level > stopper_level: self.Type = _type
Result["pending"] += 1 self.URI = uri
continue self.LastPosition = None
Result["matched"] += 1 self.Stops = set()
AudioClip.Stop() self.Level = 0
Result["stopped"] += 1 if self.Player.Access:
self.Torchlight().SayPrivate(AudioClip.Player, "Your audio clip was stopped.") self.Level = self.Player.Access["level"]
if not same_player:
self.Torchlight().SayPrivate(player, "Stopped \"{0}\"({1}) audio clip.".format(AudioClip.Player.Name, AudioClip.Player.UserID)) self.AudioPlayer = self.Master.AudioPlayerFactory.NewPlayer(self.Type)
self.AudioPlayer.AddCallback("Play", self.OnPlay)
return Result self.AudioPlayer.AddCallback("Stop", self.OnStop)
self.AudioPlayer.AddCallback("Update", self.OnUpdate)
def CreateAudioClip(self, player, uri, _type = AudioPlayerFactory.AUDIOPLAYER_FFMPEG):
Level = 0 def __del__(self):
if player.Access: self.Logger.info("~AudioClip()")
Level = player.Access["level"]
def Play(self, seconds = None, rubberband = None, dec_params = None, bitrate = None, backwards = None, *args):
if self.Torchlight().Disabled and self.Torchlight().Disabled > Level: return self.AudioPlayer.PlayURI(self.URI, position = seconds, rubberband = rubberband, dec_params = dec_params, bitrate = bitrate,
self.Torchlight().SayPrivate(player, "Torchlight is currently disabled!") backwards = backwards, *args)
return None
def Stop(self):
if not self.AntiSpam.CheckAntiSpam(player): return self.AudioPlayer.Stop()
return None
def OnPlay(self):
if not self.CheckLimits(player): self.Logger.debug(sys._getframe().f_code.co_name + ' ' + self.URI)
return None
self.Player.Storage["Audio"]["Uses"] += 1
try: self.Player.Storage["Audio"]["LastUse"] = self.Torchlight().Master.Loop.time()
Clip = AudioClip(self, player, uri, _type) self.Player.Storage["Audio"]["LastUseLength"] = 0.0
except ValueError as ex:
self.Logger.warning("Rejected unsupported audio player type %r: %s", _type, ex) def OnStop(self):
self.Torchlight().SayPrivate(player, "Unsupported audio player type.") self.Logger.debug(sys._getframe().f_code.co_name + ' ' + self.URI)
return None self.Master.AudioClips.remove(self)
self.AudioClips.append(Clip)
if self.AudioPlayer.Playing:
anti_spam_track_ceiling = min( Delta = self.AudioPlayer.Position - self.LastPosition
self.Torchlight().Config["AntiSpam"]["ImmunityLevel"], self.Player.Storage["Audio"]["TimeUsed"] += Delta
self.AntiSpam.GLOBAL_PLAY_TIME_IMMUNITY_LEVEL, self.Player.Storage["Audio"]["LastUseLength"] += Delta
)
if not player.Access or player.Access["level"] < anti_spam_track_ceiling: if str(self.Level) in self.Torchlight().Config["AudioLimits"]:
Clip.AudioPlayer.AddCallback("Play", lambda *args: self.AntiSpam.OnPlay(Clip, *args)) if self.Player.Storage:
Clip.AudioPlayer.AddCallback("Stop", lambda *args: self.AntiSpam.OnStop(Clip, *args)) if self.Player.Storage["Audio"]["TimeUsed"] >= self.Torchlight().Config["AudioLimits"][str(self.Level)]["TotalTime"]:
Clip.AudioPlayer.AddCallback("Update", lambda *args: self.AntiSpam.OnUpdate(Clip, *args)) self.Torchlight().SayPrivate(self.Player, "You have used up all of your free time! ({0} seconds)".format(
self.Torchlight().Config["AudioLimits"][str(self.Level)]["TotalTime"]))
Clip.AudioPlayer.AddCallback("Play", lambda *args: self.Advertiser.OnPlay(Clip, *args)) elif self.Player.Storage["Audio"]["LastUseLength"] >= self.Torchlight().Config["AudioLimits"][str(self.Level)]["MaxLength"]:
Clip.AudioPlayer.AddCallback("Stop", lambda *args: self.Advertiser.OnStop(Clip, *args)) self.Torchlight().SayPrivate(self.Player, "Your audio clip exceeded the maximum length! ({0} seconds)".format(
Clip.AudioPlayer.AddCallback("Update", lambda *args: self.Advertiser.OnUpdate(Clip, *args)) self.Torchlight().Config["AudioLimits"][str(self.Level)]["MaxLength"]))
return Clip del self.AudioPlayer
def AudioClip(self, player, uri, _type = AudioPlayerFactory.AUDIOPLAYER_FFMPEG): def OnUpdate(self, old_position, new_position):
return self.CreateAudioClip(player, uri, _type) Delta = new_position - old_position
self.LastPosition = new_position
def OnDisconnect(self, player):
for AudioClip in self.AudioClips[:]: self.Player.Storage["Audio"]["TimeUsed"] += Delta
if AudioClip.Player == player: self.Player.Storage["Audio"]["LastUseLength"] += Delta
AudioClip.Stop()
if not str(self.Level) in self.Torchlight().Config["AudioLimits"]:
return
class AudioClip():
def __init__(self, master, player, uri, _type): if (self.Player.Storage["Audio"]["TimeUsed"] >= self.Torchlight().Config["AudioLimits"][str(self.Level)]["TotalTime"] or
self.Logger = logging.getLogger(__class__.__name__) self.Player.Storage["Audio"]["LastUseLength"] >= self.Torchlight().Config["AudioLimits"][str(self.Level)]["MaxLength"]):
self.Master = master self.Stop()
self.Torchlight = self.Master.Torchlight
self.Player = player
self.Type = _type
self.URI = uri
self.LastPosition = None
self.LastUpdateRealtime = None
self.StartedAt = None
self.StopHinted = False
self.Stops = set()
self.Level = 0
if self.Player.Access:
self.Level = self.Player.Access["level"]
self.AudioPlayer = self.Master.AudioPlayerFactory.NewPlayer(self.Type)
self.AudioPlayer.AddCallback("Play", self.OnPlay)
self.AudioPlayer.AddCallback("Stop", self.OnStop)
self.AudioPlayer.AddCallback("Update", self.OnUpdate)
def __del__(self):
self.Logger.info("~AudioClip()")
def Play(self, seconds = None, rubberband = None, dec_params = None, bitrate = None, backwards = None, *args):
return self.AudioPlayer.PlayURI(self.URI, position = seconds, rubberband = rubberband, dec_params = dec_params, bitrate = bitrate,
backwards = backwards, *args)
def Stop(self):
return self.AudioPlayer.Stop()
def OnPlay(self):
self.Logger.debug(sys._getframe().f_code.co_name + ' ' + self.URI)
self.Player.Storage["Audio"]["Uses"] += 1
self.Player.Storage["Audio"]["LastUse"] = self.Torchlight().Master.Loop.time()
self.Player.Storage["Audio"]["LastUseLength"] = 0.0
self.StartedAt = self.Player.Storage["Audio"]["LastUse"]
self.LastUpdateRealtime = time.time()
self.StopHinted = False
def OnStop(self):
self.Logger.debug(sys._getframe().f_code.co_name + ' ' + self.URI)
if self in self.Master.AudioClips:
self.Master.AudioClips.remove(self)
played = None
if self.AudioPlayer and self.AudioPlayer.StartedPlaying is not None:
elapsed = time.time() - self.AudioPlayer.StartedPlaying
if self.AudioPlayer.Seconds:
elapsed = min(elapsed, self.AudioPlayer.Seconds)
played = max(0.0, elapsed)
if played is not None:
last_length = self.Player.Storage["Audio"]["LastUseLength"]
if played > last_length:
delta = played - last_length
self.Player.Storage["Audio"]["TimeUsed"] += delta
self.Player.Storage["Audio"]["LastUseLength"] += delta
if str(self.Level) in self.Torchlight().Config["AudioLimits"]:
if self.Player.Storage:
if self.Player.Storage["Audio"]["TimeUsed"] >= self.Torchlight().Config["AudioLimits"][str(self.Level)]["TotalTime"]:
self.Torchlight().SayPrivate(self.Player, "You have used up all of your free time! ({0} seconds)".format(
self.Torchlight().Config["AudioLimits"][str(self.Level)]["TotalTime"]))
elif self.Player.Storage["Audio"]["LastUseLength"] >= self.Torchlight().Config["AudioLimits"][str(self.Level)]["MaxLength"]:
self.Torchlight().SayPrivate(self.Player, "Your audio clip exceeded the maximum length! ({0} seconds)".format(
self.Torchlight().Config["AudioLimits"][str(self.Level)]["MaxLength"]))
del self.AudioPlayer
def OnUpdate(self, old_position, new_position):
now = time.time()
elapsed = None
if self.LastUpdateRealtime is not None:
elapsed = now - self.LastUpdateRealtime
self.LastUpdateRealtime = now
Delta = _progress_delta(old_position, new_position, elapsed = elapsed)
self.LastPosition = new_position
if Delta <= 0.0:
return
self.Player.Storage["Audio"]["TimeUsed"] += Delta
self.Player.Storage["Audio"]["LastUseLength"] += Delta
if not str(self.Level) in self.Torchlight().Config["AudioLimits"]:
return
if (self.Player.Storage["Audio"]["TimeUsed"] >= self.Torchlight().Config["AudioLimits"][str(self.Level)]["TotalTime"] or
self.Player.Storage["Audio"]["LastUseLength"] >= self.Torchlight().Config["AudioLimits"][str(self.Level)]["MaxLength"]):
self.Stop()
torchlight_changes_unloze/torchlight3/Torchlight/Commands.py
+990 -2023
View File
File diff suppressed because it is too large Load Diff
torchlight_changes_unloze/torchlight3/Torchlight/FFmpegAudioPlayer.py
+207 -268
View File
@@ -1,268 +1,207 @@
#!/usr/bin/python3 #!/usr/bin/python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import logging import logging
import traceback import traceback
import asyncio import asyncio
import datetime import datetime
import time import time
import socket import socket
import struct import struct
import sys import sys
SAMPLEBYTES = 2 SAMPLEBYTES = 2
class FFmpegAudioPlayerFactory(): class FFmpegAudioPlayerFactory():
VALID_CALLBACKS = ["Play", "Stop", "Update"] VALID_CALLBACKS = ["Play", "Stop", "Update"]
def __init__(self, master): def __init__(self, master):
self.Logger = logging.getLogger(__class__.__name__) self.Logger = logging.getLogger(__class__.__name__)
self.Master = master self.Master = master
self.Torchlight = self.Master.Torchlight self.Torchlight = self.Master.Torchlight
def __del__(self): def __del__(self):
self.Master.Logger.info("~FFmpegAudioPlayerFactory()") self.Master.Logger.info("~FFmpegAudioPlayerFactory()")
self.Quit() self.Quit()
def NewPlayer(self): def NewPlayer(self):
self.Logger.debug(sys._getframe().f_code.co_name) self.Logger.debug(sys._getframe().f_code.co_name)
Player = FFmpegAudioPlayer(self) Player = FFmpegAudioPlayer(self)
return Player return Player
def Quit(self): def Quit(self):
self.Master.Logger.info("FFmpegAudioPlayerFactory->Quit()") self.Master.Logger.info("FFmpegAudioPlayerFactory->Quit()")
class FFmpegAudioPlayer(): class FFmpegAudioPlayer():
def __init__(self, master): def __init__(self, master):
self.Master = master self.Master = master
self.Torchlight = self.Master.Torchlight self.Torchlight = self.Master.Torchlight
self.Playing = False self.Playing = False
self.Host = ( self.Host = (
self.Torchlight().Config["VoiceServer"]["Host"], self.Torchlight().Config["VoiceServer"]["Host"],
self.Torchlight().Config["VoiceServer"]["Port"] self.Torchlight().Config["VoiceServer"]["Port"]
) )
self.SampleRate = float(self.Torchlight().Config["VoiceServer"]["SampleRate"]) self.SampleRate = float(self.Torchlight().Config["VoiceServer"]["SampleRate"])
self.Channels = int(self.Torchlight().Config["VoiceServer"].get("Channels", 2))
self.StartedPlaying = None
self.StartedPlaying = None self.StoppedPlaying = None
self.StoppedPlaying = None self.Seconds = 0.0
self.Seconds = 0.0
self.Writer = None
self.Writer = None self.Process = None
self.Process = None
self.Callbacks = []
self.Callbacks = []
def __del__(self):
def __del__(self): self.Master.Logger.debug("~FFmpegAudioPlayer()")
self.Master.Logger.debug("~FFmpegAudioPlayer()") self.Stop()
self.Stop()
def PlayURI(self, uri, position, rubberband = None, dec_params = None, bitrate = None,
def PlayURI(self, uri, position, rubberband = None, dec_params = None, bitrate = None, backwards = None, *args):
backwards = None, *args): if position:
if self.Playing: PosStr = str(datetime.timedelta(seconds = position))
self.Stop() #Command = ["/usr/bin/ffmpeg", "-ss", PosStr, "-i", uri, "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn", *args]
Command = ["/usr/bin/ffmpeg", "-ss", PosStr, "-i", uri, "-af", "highpass=f=20", "-acodec", "pcm_s16le", "-ac", "2", "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn", *args]
proxy_url = "" else:
if uri.startswith("http://") or uri.startswith("https://"): #Command = ["/usr/bin/ffmpeg", "-i", uri, "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn", *args]
proxy_url = self.Torchlight().Config["Proxy"] Command = ["/usr/bin/ffmpeg", "-i", uri, "-af", "highpass=f=20", "-acodec", "pcm_s16le", "-ac", "2", "-ar", str(int(self.SampleRate)),"-f", "s16le", "-vn"]
if position: self.Playing = True
PosStr = str(datetime.timedelta(seconds=position)) if dec_params:
Command = ["/usr/bin/ffmpeg"] Command += dec_params
if proxy_url:
Command += ["-http_proxy", proxy_url] if rubberband and backwards:
Command += ["-ss", PosStr, "-i", uri, "-acodec", "pcm_s16le", "-ac", str(self.Channels), "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn"] Command += ["-filter:a"]
else: rubberCommand = ""
Command = ["/usr/bin/ffmpeg"] for rubber in rubberband:
if proxy_url: rubberCommand += rubber + ", "
Command += ["-http_proxy", proxy_url] rubberCommand = rubberCommand[:-2]
Command += ["-i", uri, "-acodec", "pcm_s16le", "-ac", str(self.Channels), "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn"] Command += [rubberCommand + "[reversed];[reversed]areverse"] #[reversed] is intermediate stream label so reverse knows what stream label to reverse
else:
if args: if rubberband:
Command += list(args) Command += ["-filter:a"]
rubberCommand = ""
# Reset per-play counters/state in case this instance is reused. for rubber in rubberband:
self.Seconds = 0.0 rubberCommand += rubber + ", "
self.StartedPlaying = None rubberCommand = rubberCommand[:-2]
self.StoppedPlaying = None Command += [rubberCommand]
self._byte_remainder = 0 if backwards:
self.Playing = True Command += ["-af"]
Command += ["areverse"]
if dec_params: if bitrate:
Command += dec_params Command += ["-ab ", str(bitrate), "k"]
self.Master.Logger.debug(f"command: {Command}")
filter_chain = ["highpass=f=20"] Command += ["-"]
if rubberband: #self.Master.Logger.debug(f"command: {Command}")
# Treat rubberband as either a single pre-built filtergraph string asyncio.ensure_future(self._stream_subprocess(Command))
# or an iterable of simple, comma-safe filters. Validate to avoid return True
# generating invalid ffmpeg filter syntax.
rubberband_filters = [] def Stop(self, force = True):
if isinstance(rubberband, str): if not self.Playing:
rubberband_filters = [rubberband] return False
else:
try: if self.Process:
iterator = iter(rubberband) try:
except TypeError: self.Process.terminate()
# Not iterable: coerce to string and treat as single element. self.Process.kill()
rubberband_filters = [str(rubberband)] self.Process = None
else: except ProcessLookupError:
for f in iterator: pass
if not isinstance(f, str):
f = str(f) if self.Writer:
# Reject potentially complex/unsafe filter fragments that if force:
# could break the filtergraph when joined with commas. Socket = self.Writer.transport.get_extra_info("socket")
if any(ch in f for ch in [",", ";", "[", "]"]): if Socket:
self.Master.Logger.error( Socket.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
"Skipping unsafe rubberband filter fragment %r: " struct.pack("ii", 1, 0))
"contains ',', ';', '[' or ']'", f
) self.Writer.transport.abort()
continue
rubberband_filters.append(f) self.Writer.close()
filter_chain.extend(rubberband_filters)
if backwards: self.Playing = False
filter_chain.append("areverse")
Command += ["-filter:a", ",".join(filter_chain)] self.Callback("Stop")
if bitrate: del self.Callbacks
Command += ["-b:a", f"{bitrate}k"]
self.Master.Logger.debug(f"command: {Command}") return True
Command += ["-"]
#self.Master.Logger.debug(f"command: {Command}") def AddCallback(self, cbtype, cbfunc):
asyncio.ensure_future(self._stream_subprocess(Command)) if not cbtype in FFmpegAudioPlayerFactory.VALID_CALLBACKS:
return True return False
def Stop(self, force = True): self.Callbacks.append((cbtype, cbfunc))
if not self.Playing: return True
return False
def Callback(self, cbtype, *args, **kwargs):
if self.Process: for Callback in self.Callbacks:
try: if Callback[0] == cbtype:
self.Process.terminate() try:
self.Process.kill() Callback[1](*args, **kwargs)
self.Process = None except Exception as e:
except ProcessLookupError: self.Master.Logger.error(traceback.format_exc())
pass
async def _updater(self):
if self.Writer: LastSecondsElapsed = 0.0
if force:
Socket = self.Writer.transport.get_extra_info("socket") while self.Playing:
if Socket: SecondsElapsed = time.time() - self.StartedPlaying
Socket.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
struct.pack("ii", 1, 0)) if SecondsElapsed > self.Seconds:
SecondsElapsed = self.Seconds
self.Writer.transport.abort()
self.Callback("Update", LastSecondsElapsed, SecondsElapsed)
self.Writer.close()
if SecondsElapsed >= self.Seconds:
self.Playing = False if not self.StoppedPlaying:
print("BUFFER UNDERRUN!")
self.Callback("Stop") self.Stop(False)
self.Callbacks = [] return
return True LastSecondsElapsed = SecondsElapsed
def AddCallback(self, cbtype, cbfunc): await asyncio.sleep(0.1)
if not cbtype in FFmpegAudioPlayerFactory.VALID_CALLBACKS:
return False async def _read_stream(self, stream, writer):
Started = False
self.Callbacks.append((cbtype, cbfunc))
return True while stream and self.Playing:
Data = await stream.read(65536)
def Callback(self, cbtype, *args, **kwargs):
for Callback in self.Callbacks: if Data:
if Callback[0] == cbtype: writer.write(Data)
try: await writer.drain()
Callback[1](*args, **kwargs)
except Exception as e: Bytes = len(Data)
self.Master.Logger.error(traceback.format_exc()) Samples = Bytes / SAMPLEBYTES
Seconds = Samples / self.SampleRate
async def _updater(self):
LastSecondsElapsed = 0.0 self.Seconds += Seconds
while self.Playing: if not Started:
SecondsElapsed = time.time() - self.StartedPlaying Started = True
self.Callback("Play")
if SecondsElapsed > self.Seconds: self.StartedPlaying = time.time()
SecondsElapsed = self.Seconds asyncio.ensure_future(self._updater())
else:
self.Callback("Update", LastSecondsElapsed, SecondsElapsed) self.Process = None
LastSecondsElapsed = SecondsElapsed break
if SecondsElapsed >= self.Seconds: self.StoppedPlaying = time.time()
if not self.StoppedPlaying:
self.Master.Logger.warning("BUFFER UNDERRUN") async def _stream_subprocess(self, cmd):
await asyncio.sleep(0.05) if not self.Playing:
continue return
self.Stop(False)
return _, self.Writer = await asyncio.open_connection(self.Host[0], self.Host[1])
await asyncio.sleep(0.1) Process = await asyncio.create_subprocess_exec(*cmd,
stdout = asyncio.subprocess.PIPE, stderr = asyncio.subprocess.DEVNULL)
async def _read_stream(self, stream, writer): self.Process = Process
Started = False
try: await self._read_stream(Process.stdout, self.Writer)
while stream and self.Playing: await Process.wait()
Data = await stream.read(65536)
if self.Seconds == 0.0:
if Data: self.Stop()
try:
writer.write(Data)
await writer.drain()
except (ConnectionError, BrokenPipeError, OSError) as ex:
self.Master.Logger.warning("Voice socket write failed: %s", ex)
break
Bytes = len(Data)
frame_size = SAMPLEBYTES * self.Channels
# Guard against invalid configuration and use integral frame counting
if frame_size <= 0 or self.SampleRate <= 0:
self.Master.Logger.error(
"Invalid audio configuration: Channels=%r, SampleRate=%r",
self.Channels,
self.SampleRate,
)
else:
# Accumulate remainder bytes to avoid systematic undercounting
total_bytes = getattr(self, "_byte_remainder", 0) + Bytes
Frames = total_bytes // frame_size
self._byte_remainder = total_bytes % frame_size
Seconds = Frames / self.SampleRate
self.Seconds += Seconds
if not Started:
Started = True
self.Callback("Play")
self.StartedPlaying = time.time()
asyncio.ensure_future(self._updater())
else:
self.Process = None
break
finally:
self.StoppedPlaying = time.time()
async def _stream_subprocess(self, cmd):
if not self.Playing:
return
try:
_, self.Writer = await asyncio.open_connection(self.Host[0], self.Host[1])
Process = await asyncio.create_subprocess_exec(*cmd,
stdout = asyncio.subprocess.PIPE, stderr = asyncio.subprocess.DEVNULL)
self.Process = Process
await self._read_stream(Process.stdout, self.Writer)
await Process.wait()
if self.Playing:
# Keep the clip active until the updater reaches self.Seconds.
# This allows !stop to interrupt short clips that are already
# buffered to the voice socket but still audible in-game.
if self.StartedPlaying is None:
self.Stop(False)
except asyncio.CancelledError:
raise
except Exception:
self.Master.Logger.error(traceback.format_exc())
if self.Playing:
self.Stop()