Tu's in die Cloud, höhöhö....
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

campatmo.py 4.9KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165
  1. #!/usr/bin/env python3
  2. from glob import glob
  3. import os
  4. import random
  5. import sys
  6. from threading import Thread
  7. import time
  8. from typing import Union, List, Mapping
  9. os.environ["PYGAME_HIDE_SUPPORT_PROMPT"] = "hide"
  10. from pygame import mixer
  11. from collection import SampleCollection
  12. from atmorepl import AtmoReplRunner
  13. VERBOSITY = 0
  14. channel_index = 0
  15. class CampAtmo:
  16. tracks_per_type = 2
  17. single_track_types = ["lines", "reactions"]
  18. background_track_types = ["background", "buzzwords"]
  19. type_mix = {
  20. "lines": 0.8,
  21. "reactions": 0.7,
  22. "buzzwords": 0.6,
  23. "background": 0.3,
  24. "other": 0.5,
  25. }
  26. lines_delta_sec = (5, 15)
  27. line_reaction_wait_sec = 0.8
  28. def __init__(self):
  29. self.stopped = False
  30. mixer.init(frequency=22050 * 2)
  31. type_dirs = glob("samples/*/")
  32. self.types = {}
  33. for type_dir in type_dirs:
  34. type_name = type_dir.split("/")[1]
  35. self.types[type_name] = SampleCollection(
  36. type_dir, volume=self.type_mix[type_name]
  37. )
  38. i = 0
  39. mixer.set_num_channels(
  40. sum(
  41. [
  42. 1 if name in self.single_track_types else self.tracks_per_type
  43. for name in self.types
  44. ]
  45. )
  46. )
  47. v(f"{mixer.get_num_channels()} channels set")
  48. self.type_channels = {}
  49. for type_name in self.types:
  50. self.type_channels[type_name] = []
  51. for k in range(
  52. 1 if type_name in self.single_track_types else self.tracks_per_type
  53. ):
  54. self.type_channels[type_name].append(mixer.Channel(i))
  55. i += 1
  56. def start(self):
  57. self.background_thread = Thread(
  58. None, self.run_forever, "atmo-background", daemon=True
  59. )
  60. self.background_thread.start()
  61. self.lines_thread = Thread(
  62. None, self.run_lines_forever, "atmo-lines", daemon=True
  63. )
  64. self.lines_thread.start()
  65. def stop(self):
  66. self.stopped = True
  67. self.background_thread.join(1.2)
  68. self.lines_thread.join(2)
  69. def load_sounds(self):
  70. for _, collection in self.types.items():
  71. collection.load()
  72. def run_forever(self):
  73. while not self.stopped:
  74. for type_name, channels in self.type_channels.items():
  75. if (
  76. type_name not in self.background_track_types
  77. or len(self.types[type_name]) == 0
  78. ):
  79. continue
  80. self.play_type_sounds(type_name)
  81. time.sleep(1)
  82. def run_lines_forever(self):
  83. while not self.stopped:
  84. time.sleep(random.randrange(*self.lines_delta_sec))
  85. length_sec = self.play_type_sounds("lines")
  86. time.sleep(length_sec + self.line_reaction_wait_sec)
  87. length_sec = self.play_type_sounds("reactions")
  88. time.sleep(length_sec + self.line_reaction_wait_sec)
  89. self.play_type_sounds("reactions")
  90. def play_type_sounds(self, type_name) -> int:
  91. if type_name in self.types and len(self.types[type_name]) == 0:
  92. return
  93. max_sound_length = 0
  94. for channel in self.type_channels[type_name]:
  95. if channel.get_queue() is not None and len(self.types[type_name]) > 1:
  96. continue
  97. sample = self.types[type_name].next()
  98. max_sound_length = max(max_sound_length, sample.sound.get_length())
  99. channel.queue(sample.sound)
  100. if (
  101. type_name not in self.single_track_types
  102. and channel.get_queue() is None
  103. and len(self.types[type_name]) > 1
  104. ):
  105. channel.queue(self.types[type_name].next_sound())
  106. return max_sound_length
  107. def get_status(self):
  108. ret_strs = []
  109. for type_name, channels in self.type_channels.items():
  110. samples = []
  111. for chan in channels:
  112. sample = self.types[type_name].get_sample_from_channel(chan)
  113. if sample is not None:
  114. samples.append(sample)
  115. if samples:
  116. sample_infos = [
  117. f"{s.path} @{s.sound.get_volume():.0%}vol" for s in samples
  118. ]
  119. ret_strs.append(f"{type_name}: {sample_infos}")
  120. return "\n".join(ret_strs)
  121. def pause(self):
  122. if mixer.get_busy():
  123. mixer.pause()
  124. return False
  125. else:
  126. mixer.unpause()
  127. return True
  128. def main():
  129. atmo = CampAtmo()
  130. atmoRepl = AtmoReplRunner(atmo, port=7723)
  131. atmoRepl.start()
  132. atmo.start()
  133. try:
  134. while True:
  135. time.sleep(10)
  136. except KeyboardInterrupt:
  137. print("exiting...")
  138. atmo.stop()
  139. sys.exit(0)
  140. def v(*msg):
  141. if VERBOSITY >= 1:
  142. print(*msg)
  143. if __name__ == "__main__":
  144. main()