From e51ae13459504ccfcda9bf62d99fac2870e51b7f Mon Sep 17 00:00:00 2001 From: Ryan Mulligan Date: Fri, 17 May 2024 17:48:15 +0000 Subject: [PATCH] remove audio Why === * The audio service is not frequently used. * The audio service is not core to Replit. * Replit still supports audio playing via VNC. * We are planning to remove the audio service. What changed === * Remove all audio code Test plan === * CI passes --- README.md | 1 - docs/api.rst | 13 -- docs/index.rst | 1 - main.py | 0 src/replit/__init__.py | 4 - src/replit/audio/Makefile | 18 -- src/replit/audio/__init__.py | 415 ----------------------------------- src/replit/audio/test.py | 67 ------ src/replit/audio/types.py | 117 ---------- 9 files changed, 636 deletions(-) create mode 100644 main.py delete mode 100644 src/replit/audio/Makefile delete mode 100644 src/replit/audio/__init__.py delete mode 100644 src/replit/audio/test.py delete mode 100644 src/replit/audio/types.py diff --git a/README.md b/README.md index aff3d751..96a294ac 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,6 @@ This repository is the home for the `replit` Python package, which provides: - A fully-featured database client for [Replit DB](https://docs.replit.com/category/databases). - Tools and utilities for Flask Web Development, including an interface to Replit's User Authetication service - Replit user profile metadata retrieval (more coming here!). -- A simple audio library that can play tones and audio files! ### Open Source License diff --git a/docs/api.rst b/docs/api.rst index 8cd5a37d..ea03f313 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -56,16 +56,3 @@ replit.info module :members: :undoc-members: :show-inheritance: - -replit.audio module -------------------- - -.. automodule:: replit.audio - :members: - :undoc-members: - :show-inheritance: - -.. automodule:: replit.audio.types - :members: - :undoc-members: - :show-inheritance: diff --git a/docs/index.rst b/docs/index.rst index 5440758c..e01b8820 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -14,7 +14,6 @@ provides: - A fully-featured `Replit DB `_ client and CLI. - A Flask–based application framework for accellerating development on the platform. -- A simple audio library that can play tones and audio files! - A library to authenticate between Repls. Table of Contents diff --git a/main.py b/main.py new file mode 100644 index 00000000..e69de29b diff --git a/src/replit/__init__.py b/src/replit/__init__.py index 66f5ef4f..ea1e7009 100644 --- a/src/replit/__init__.py +++ b/src/replit/__init__.py @@ -5,7 +5,6 @@ from typing import Any from . import database, web -from .audio import Audio from .database import ( Database, AsyncDatabase, @@ -23,9 +22,6 @@ def clear() -> None: print("\033[H\033[2J", end="", flush=True) -audio = Audio() - - # Previous versions of this library would just have side-effects and always set # up a database unconditionally. That is very undesirable, so instead of doing # that, we are using this egregious hack to get the database / database URL diff --git a/src/replit/audio/Makefile b/src/replit/audio/Makefile deleted file mode 100644 index d730584d..00000000 --- a/src/replit/audio/Makefile +++ /dev/null @@ -1,18 +0,0 @@ - - - - - -.PHONY: test, docs - - -test: - @python3 test.py - -docs: - @cd ./docs && make - -docs-%: - - @echo $(shell echo $@ | cut -c6-) - @cd ./docs && make $(shell echo $@ | cut -c6-) diff --git a/src/replit/audio/__init__.py b/src/replit/audio/__init__.py deleted file mode 100644 index ec736e21..00000000 --- a/src/replit/audio/__init__.py +++ /dev/null @@ -1,415 +0,0 @@ -# flake8: noqa -"""A library to play audio in a repl.""" -from datetime import datetime, timedelta -import json -from os import path -import time -from typing import Any, List, Optional - -from .types import ( - AudioStatus, - file_types, - ReaderType, - RequestArgs, - RequestData, - SourceData, - WaveType, -) - - -class InvalidFileType(Exception): - """Exception for when a requested file's type isnt valid.""" - - pass - - -class NoSuchSourceException(Exception): - """Exception used when a source doesn't exist.""" - - pass - - -class Source: - """A Source is used to get audio that is sent to the user.""" - - __payload: SourceData - _loops: bool - _name: str - - def __init__(self, payload: SourceData, loops: bool) -> None: - """Initialize the class. - - Args: - payload (SourceData): The payload for the source. - loops (bool): How many times the source should loop. - """ - self.__payload = payload - self._loops = loops - self._name = payload["Name"] - - def __get_source(self) -> SourceData or None: - source = None - with open("/tmp/audioStatus.json", "r") as f: - data = json.loads(f.read()) - for s in data["Sources"]: - if s["ID"] == self.id: - source = s - break - if source: - self.__payload = source - return source - - def __update_source(self, **changes: Any) -> None: - s = self.__get_source() - if not s: - raise NoSuchSourceException( - f'No player with id "{id}" found! It might be done playing.' - ) - s.update({key.title(): changes[key] for key in changes}) - with open("/tmp/audio", "w") as f: - f.write(json.dumps(s)) - self.__get_source() - - @property - def name(self) -> str: - """The name of the source.""" - return self._name - - def get_start_time(self) -> datetime: - """When the source started plaing.""" - timestamp_str = self.__payload["StartTime"] - timestamp = datetime.strptime(timestamp_str[:-4], "%Y-%m-%dT%H:%M:%S.%f") - return timestamp - - start_time: datetime = property(get_start_time) - "Property wrapper for :py:meth:`~replit.Source.get_start_time`" - - @property - def path(self) -> str or None: - """The path to the source, if available.""" - data = self.__payload - if ReaderType(data["Type"]) in file_types: - return self.__payload["Request"]["Args"]["Path"] - - @property - def id(self) -> int: - """The ID of the source.""" - return self.__payload["ID"] - - def get_remaining(self) -> timedelta: - """The estimated time remaining in the source's current loop.""" - data = self.__get_source() - if not data: - return timedelta(milliseconds=0) - - return timedelta(milliseconds=data["Remaining"]) - - remaining: int = property(get_remaining) - "Property wrapper for :py:meth:`~replit.Source.get_remaining`" - - def get_end_time(self) -> Optional[datetime]: - """The estimated time when the source will be done playing. - - Returns: - Optional[datetime]: The estimated time when the source will be done playing - or None if it is already finished. - """ - s = self.__get_source() - if not s: - return None - - timestamp_str = s["EndTime"] - timestamp = datetime.strptime(timestamp_str[:-4], "%Y-%m-%dT%H:%M:%S.%f") - return timestamp - - end_time: datetime or None = property(get_end_time) - "Property wrapper for :py:meth:`~replit.Source.get_end_time`" - - @property - def does_loop(self) -> bool: - """Whether the source repeats itself or not.""" - return self._loops - - @property - def duration(self) -> timedelta: - """The duration of the source.""" - return timedelta(milliseconds=self.__payload["Duration"]) - - def get_volume(self) -> float: - """The volume the source is set to.""" - self.__get_source() - return self.__payload["Volume"] - - def set_volume(self, volume: float) -> None: - """Set the volume. - - Args: - volume (float): The volume the source should be set to. - """ - self.__update_source(volume=volume) - - volume: float = property(get_volume, set_volume) - "Property wrapper for `replit.Source.get_volume` and `replit.Source.set_volume`" - - def get_paused(self) -> bool: - """Whether the source is paused.""" - self.__get_source() - return self.__payload["Paused"] - - def set_paused(self, paused: bool) -> None: - """Change if the source is paused. - - Args: - paused (bool): Whether the source should be paused. - """ - self.__update_source(paused=paused) - - paused = property(get_paused, set_paused) - "Property wrapper for `replit.Source.get_paused` and `replit.Source.set_paused`" - - def get_loops_remaining(self) -> Optional[int]: - """The remaining amount of times the file will restart. - - Returns: - Optional[int]: The number of loops remaining or None if the source can't be - found, either because it has finished playing or an error occured. - """ - if not self._loops: - return 0 - - s = self.__get_source() - if not s: - return None - - if s["ID"] == self.id: - loops = s["Loop"] - - return loops - - def set_loop(self, loop_count: int) -> None: - """Set the remaining amount of loops for the source. - - Args: - loop_count (int): How many times the source should repeat itself. Set to a - negative value for infinite. - """ - does_loop = loop_count != 0 - self._loops = does_loop - self.__update_source(doesLoop=does_loop, loopCount=loop_count) - - loops_remaining: int or None = property(get_loops_remaining) - "Property wrapper for :py:meth:`~replit.Source.get_loops_remaining`" - - def toggle_playing(self) -> None: - """Play/pause the source.""" - self.set_paused(not self.paused) - - -class Audio: - """The basic audio manager. - - Notes - ----- - This is not intended to be called directly, instead use :py:const:`audio`. - - Using this in addition to `audio` can cause **major** issues. - """ - - __known_ids = [] - __names_created = 0 - - def __gen_name(self) -> str: - return f"Source {time.time()}" - - def __get_new_source(self, name: str, does_loop: bool) -> Source: - new_source = None - timeOut = datetime.now() + timedelta(seconds=2) - - while not new_source and datetime.now() < timeOut: - try: - sources = AudioStatus(self.read_status())["Sources"] - new_source = SourceData([s for s in sources if s["Name"] == name][0]) - except IndexError: - pass - except json.JSONDecodeError: - pass - - if not new_source: - raise TimeoutError("Source was not created within 2 seconds.") - - return Source(new_source, does_loop) - - def play_file( - self, - file_path: str, - volume: float = 1, - does_loop: bool = False, - loop_count: int = 0, - name: Optional[str] = None, - ) -> Source: - """Sends a request to play a file, assuming the file is valid. - - Args: - file_path (str): The path to the file that should be played. Can be - absolute or relative. - volume (float): The volume the source should be played at. (1 being - 100%) - does_loop (bool): Wether the source should repeat itself or not. Note, if - you set this you should also set loop_count. - loop_count (int): How many times the source should repeat itself. Set to 0 - to have the source play only once, or set to a negative value for the - source to repeat forever. - name (str): The name of the file. Default value is a unique name for the - source. - - Returns: - Source: The source created with the provided data. - - Raises: - FileNotFoundError: If the file is not found. - InvalidFileType: If the file type is not valid. - """ - name = name or self.__gen_name() - - if not path.exists(file_path): - raise FileNotFoundError(f'File "{file_path}" not found.') - - file_type = file_path.split(".")[-1] - - if ReaderType(file_type) not in file_types: - raise InvalidFileType(f"Type {file_type} is not supported.") - - data = RequestData( - Type=file_type, - Volume=volume, - DoesLoop=does_loop, - LoopCount=loop_count, - Name=name, - Args=RequestArgs(Path=file_path), - ) - - with open("/tmp/audio", "w") as p: - p.write(json.dumps(dict(data))) - - return self.__get_new_source(name, does_loop) - - def play_tone( - self, - duration: float, - pitch: int, - wave_type: WaveType, - does_loop: bool = False, - loop_count: int = 0, - volume: float = 1, - name: Optional[str] = None, - ) -> Source: - """Play a tone from a frequency and wave type. - - Args: - duration (float): How long the tone should be played (in seconds). - pitch (int): The frequency the tone should be played at. - wave_type (WaveType): The wave shape used to generate the tone. - does_loop (bool): Wether the source should repeat itself or not. Note, if - you set this you should also set loop_count. - loop_count (int): How many times the source should repeat itself. Set to 0 - to have the source play only once, or set to a negative value for the - source to repeat forever. - volume (float): The volume the tone should be played at (1 being 100%). - name (str): The name of the file. Default value is a unique name for the - source. - - Returns: - Source: The source for the tone. - """ - name = name or self.__gen_name() - - # ensure the wave type is valid. This will throw an error if it isn't. - WaveType(wave_type) - - data = RequestData( - Name=name, - DoesLoop=does_loop, - LoopCount=loop_count, - Volume=volume, - Type=str(ReaderType.tone), - Args=RequestArgs( - WaveType=wave_type, - Pitch=pitch, - Seconds=duration, - ), - ) - - with open("/tmp/audio", "w") as f: - f.write(json.dumps(data)) - - return self.__get_new_source(name, does_loop) - - def get_source(self, source_id: int) -> Source or None: - """Get a source by it's ID. - - Args: - source_id (int): The ID for the source that should be found. - - Raises: - NoSuchSourceException: If the source isnt found or there isn't any sources - known to the audio manager. - - Returns: - Source: The source with the ID provided. - """ - source = None - with open("/tmp/audioStatus.json", "r") as f: - data = AudioStatus(json.loads(f.read())) - if not data["Sources"]: - raise NoSuchSourceException("No sources exist yet.") - for s in data["Sources"]: - if s["ID"] == int(source_id): - source = s - break - if not source: - raise NoSuchSourceException(f'Could not find source with ID "{source_id}"') - return Source(source, source["Loop"]) - - def read_status(self) -> AudioStatus: - """Get the raw data for what's playing. - - This is an api call, and shouldn't be needed for general usage. - - Returns: - AudioStatus: The contents of /tmp/audioStatus.json - """ - with open("/tmp/audioStatus.json", "r") as f: - data = AudioStatus(json.loads(f.read())) - if data["Sources"] is None: - data["Sources"]: List[SourceData] = [] - return data - - def get_playing(self) -> List[Source]: - """Get a list of playing sources. - - Returns: - List[Source]: A list of sources that aren't paused. - """ - data = self.read_status() - sources = data["Sources"] - return [Source(s, s["Loop"]) for s in sources if not s["Paused"]] - - def get_paused(self) -> List[Source]: - """Get a list of paused sources. - - Returns: - List[Source]: A list of sources that are paused. - """ - data = self.read_status() - sources = data["Sources"] - return [Source(s, s["Loop"]) for s in sources if s["Paused"]] - - def get_sources(self) -> List[Source]: - """Gets all sources. - - Returns: - List[Source]: Every source known to the audio manager, paused or playing. - """ - data = self.read_status() - sources = data["Sources"] - return [Source(s, s["Loop"]) for s in sources] diff --git a/src/replit/audio/test.py b/src/replit/audio/test.py deleted file mode 100644 index b5c55265..00000000 --- a/src/replit/audio/test.py +++ /dev/null @@ -1,67 +0,0 @@ -# flake8: noqa -import time -import unittest -import replit -from .. import audio -from . import types - -test_file = "../test.mp3" - - -class TestAudio(unittest.TestCase): - def test_creation(self): - source = audio.play_file(test_file) - self.assertEqual(source.path, test_file) - source.paused = True - time.sleep(1) - self.assertEqual(source.paused, True, "Pausing Source") - - def test_pause(self): - source = audio.play_file(test_file) - source.volume = 2 - time.sleep(1) - self.assertEqual(source.volume, 2, "Volume set to 2") - - source.paused = True - time.sleep(1) - self.assertEqual(source.paused, True, "Pausing Source") - - source.volume = 0.2 - time.sleep(1) - self.assertEqual(source.volume, 0.2, "Volume set to .2") - - source.paused = True - time.sleep(1) - self.assertEqual(source.paused, True, "Pausing Source") - - def test_loop_setting(self): - source = audio.play_file(test_file) - - self.assertEqual(source.loops_remaining, 0, "0 loops remaining") - source.set_loop(2) - time.sleep(1) - - self.assertEqual(source.loops_remaining, 2, "2 loops remaining") - source.paused = True - time.sleep(1) - self.assertEqual(source.paused, True, "Pausing Source") - - def test_other(self): - source = audio.play_file(test_file) - - self.assertIsNotNone(source.end_time) - self.assertIsNotNone(source.start_time) - self.assertIsNotNone(source.remaining) - source.paused = True - time.sleep(1) - self.assertEqual(source.paused, True, "Pausing Source") - - def test_tones(self): - try: - audio.play_tone(2, 400, 2) - except TimeoutError or ValueError as e: - self.fail(e) - - -if __name__ == "__main__": - unittest.main() diff --git a/src/replit/audio/types.py b/src/replit/audio/types.py deleted file mode 100644 index 0fc0ebf0..00000000 --- a/src/replit/audio/types.py +++ /dev/null @@ -1,117 +0,0 @@ -# flake8: noqa -from typing import List -from typing_extensions import TypedDict -from enum import Enum - - -class ReaderType(Enum): - "An Enum for the types of sources." - - def __str__(self) -> str: - return self._value_ - - def __repr__(self) -> str: - return f"ReaderType.{self._name_}" - - wav_file = "wav" - "ReaderType : The type for a .wav file." - aiff_file = "aiff" - "ReaderType : The type for a .aiff file." - mp3_file = "mp3" - "ReaderType : The type for a .mp3 file." - tone = "tone" - "ReaderType : The type for a generated tone." - - -class WaveType(Enum): - "The different wave shapes that can be used for tone generation." - - def __str__(self) -> str: - return self._value_ - - WaveSine = 0 - "WaveType : The WaveSine wave shape." - WaveTriangle = 1 - "WaveType : The Triangle wave shape." - WaveSaw = 2 - "WaveType : The Saw wave shape." - WaveSqr = 3 - "WaveType : The Square wave shape." - - -file_types: List[ReaderType] = [ - ReaderType.aiff_file, - ReaderType.wav_file, - ReaderType.mp3_file, -] -"The different file types for sources in a list." - - -class RequestArgs(TypedDict, total=False): - "The additional arguments for a request that are type-specific." - Pitch: float - "float : The pitch/frequency of the tone. Only used if the request type is tone." - Seconds: float - - "float : The duration for the tone to be played. Only used if the request type is tone." - WaveType: WaveType or int - "WaveType : The wave type of the tone. Only used if the request type is tone." - Path: str - "str : The path to the file to be read. Only used if the request is for a file type." - - -class RequestData(TypedDict): - "A request to pid1 for a source to be played." - ID: int - "int : The ID of the source. Only used for updating a pre-existing source." - Paused: bool or None - "bool or None : Wether the source with the provided ID should be paused or not. Can only be used when updating a source." - Volume: float - "float : The volume the source should be played at. (1 being 100%)" - DoesLoop: bool - "bool : Wether the source should loop / repeat or not. Defaults to false." - LoopCount: int - "int : How many times the source should loop / repeat. Defaults to 0." - Name: str - "str : The name of the source." - Type: ReaderType or str - "ReaderType : The type of the source." - Args: RequestArgs - "RequestArgs : The additional arguments for the source." - - -class SourceData(TypedDict): - """A source's raw data, as a payload.""" - - Name: str - "str : The name of the source." - Type: str - "str : The type of the source." - Volume: float - "float : The volume of the source." - Duration: int - "int : The duration of the source in milliseconds." - Remaining: int - "int : How many more milliseconds the source will be playing." - Paused: bool - "bool : Wether the source is paused or not." - Loop: int - "int : How many times the source will loop. If 0, the source will not repeat itself." - ID: int - "int : The ID of the source." - EndTime: str - "str : The estimated timestamp for when the source will finish playing." - StartTime: str - "str : When the source started playing." - Request: RequestData - "RequestData : The request used to create the source." - - -class AudioStatus(TypedDict): - "The raw data read from /tmp/audioStatus.json." - Sources: List[SourceData] or None - "List[SourceData] : The sources that are know to the audio manager." - Running: bool - "bool : Wether the audio manager knows any sources or not." - Disabled: bool - "bool : Wether the audio manager is disabled or not."