kennethreitz/replit-py
1
0
Fork 0
mirror of https://github.com/kennethreitz/replit-py.git synced 2026-06-05 23:10:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make types import relative and format with black

Browse Source
This commit is contained in:
Scoder12
2020-07-27 14:05:30 -07:00
parent 736683f28e
commit 71b547f8b7
1 changed files with 107 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/replit/audio/replit/__init__.py
+107 -106
View File
@@ -1,14 +1,22 @@
import json
import time
from replit.types import ReaderType, RequestArgs, RequestData, SourceData, AudioStatus, WaveType, file_types
from .types import (
ReaderType,
RequestArgs,
RequestData,
SourceData,
AudioStatus,
WaveType,
file_types,
)
from typing import List
from datetime import datetime, timedelta
from os import path
def clear():
'Clear is used to clear the terminal.'
print('\033[H\033[2J', end='', flush=True)
"Clear is used to clear the terminal."
print("\033[H\033[2J", end="", flush=True)
class InvalidFileType(Exception):
@@ -22,7 +30,7 @@ class NoSuchSourceException(Exception):
class Source:
'''A Source is used to get audio that is sent to the user.
"""A Source is used to get audio that is sent to the user.
Parameters
----------
@@ -31,7 +39,8 @@ class Source:
loops : int
How many times the source should loop.
'''
"""
__payload: SourceData
_loops: bool
_name: str
@@ -39,14 +48,14 @@ class Source:
def __init__(self, payload: SourceData, loops: bool):
self.__payload = payload
self._loops = loops
self._name = payload['Name']
self._name = payload["Name"]
def __get_source(self) -> SourceData or None:
source = None
with open('/tmp/audioStatus.json', 'r') as f:
with open("/tmp/audioStatus.json", "r") as f:
data = json.loads(f.read())
for s in data['Sources']:
if s['ID'] == self.id:
for s in data["Sources"]:
if s["ID"] == self.id:
source = s
break
if source:
@@ -57,38 +66,38 @@ class Source:
s = self.__get_source()
if not s:
raise NoSuchSourceException(
f'No player with id "{id}" found! It might be done playing.')
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:
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'
"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")
"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 wrapper for :py:meth:`~replit.Source.get_start_time`"
@property
def path(self) -> str or None:
'The path to the source, if available.'
"The path to the source, if available."
data = self.__payload
if ReaderType(data['Type']) in file_types:
return self.__payload['Request']['Args']['Path']
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']
"The ID of the source."
return self.__payload["ID"]
def get_remaining(self) -> timedelta:
"The estimated time remaining in the source's current loop."
@@ -96,44 +105,43 @@ class Source:
if not data:
return timedelta(millaseconds=0)
return timedelta(milliseconds=data['Remaining'])
return timedelta(milliseconds=data["Remaining"])
remaining: int = property(get_remaining)
'Property wrapper for :py:meth:`~replit.Source.get_remaining`'
"Property wrapper for :py:meth:`~replit.Source.get_remaining`"
def get_end_time(self) -> datetime or None:
'''The estimated time when the sourcce will be done playing.
"""The estimated time when the sourcce will be done playing.
Returns None if the source has finished playing.
Note: this is the estimation for the end of the current loop.'''
Note: this is the estimation for the end of the current loop."""
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")
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 wrapper for :py:meth:`~replit.Source.get_end_time`"
@property
def does_loop(self) -> bool:
'Wether the source repeats itself or not.'
"Wether the source repeats itself or not."
return self._loops
@property
def duration(self) -> timedelta:
'The duration of the source.'
return timedelta(millaseconds=self.__payload['Duration'])
"The duration of the source."
return timedelta(millaseconds=self.__payload["Duration"])
def get_volume(self) -> float:
'The volume the source is set to.'
"The volume the source is set to."
self.__get_source()
return self.__payload['Volume']
return self.__payload["Volume"]
def set_volume(self, volume: float):
'''
"""
Parameters
----------
volume: float
@@ -143,19 +151,19 @@ class Source:
------
NoSuchSourceException
If the source is no longer known to the audio manager.
'''
"""
self.__update_source(volume=volume)
volume: float = property(get_volume, set_volume)
'Property wrapper for :py:meth:`~replit.Source.get_volume` and :py:meth:`~replit.Source.set_volume`'
"Property wrapper for :py:meth:`~replit.Source.get_volume` and :py:meth:`~replit.Source.set_volume`"
def get_paused(self) -> bool:
'Wether the source is paused or not.'
"Wether the source is paused or not."
self.__get_source()
return self.__payload['Paused']
return self.__payload["Paused"]
def set_paused(self, paused: bool):
'''
"""
Parameters
----------
paused: bool
@@ -165,14 +173,14 @@ class Source:
------
NoSuchSourceException
If the source is no longer known to the audio manager.
'''
"""
self.__update_source(paused=paused)
paused = property(get_paused, set_paused)
'Property wrapper for :py:meth:`~replit.Source.get_paused` and :py:meth:`~replit.Source.set_paused`'
"Property wrapper for :py:meth:`~replit.Source.get_paused` and :py:meth:`~replit.Source.set_paused`"
def get_loops_remaining(self) -> int or None:
'''The remaining amount of times the file will restart. Returns none if the source is done playing.
"""The remaining amount of times the file will restart. Returns none if the source is done playing.
Returns
-------
@@ -181,7 +189,7 @@ class Source:
None
The source can't be found, either because it has finished playing or an error occured.
'''
"""
if not self._loops:
return 0
@@ -189,13 +197,13 @@ class Source:
if not s:
return None
if s['ID'] == self.id:
loops = s['Loop']
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.
"""Set the remaining amount of loops for the source.
Set loop_count to a negative value to repeat forever.
Parameters
@@ -209,34 +217,35 @@ class Source:
------
NoSuchSourceException
If the source is no longer known to the audio manager.
'''
"""
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`'
"Property wrapper for :py:meth:`~replit.Source.get_loops_remaining`"
def toggle_playing(self) -> None:
'''Play/pause the source.'''
"""Play/pause the source."""
self.set_paused(not self.paused)
class Audio():
'''The basic audio manager.
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() -> str:
return f'Source {time.time()}'
return f"Source {time.time()}"
def __get_new_source(self, name: str, does_loop: bool) -> Source:
new_source = None
@@ -244,17 +253,15 @@ class Audio():
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])
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(f'Source was not created within 2 seconds.')
raise TimeoutError(f"Source was not created within 2 seconds.")
return Source(new_source, does_loop)
@@ -264,9 +271,9 @@ class Audio():
volume: float = 1,
does_loop: bool = False,
loop_count: int = 0,
name: str = __gen_name()
name: str = __gen_name(),
) -> Source:
'''Sends a request to play a file, assuming the file is valid.
"""Sends a request to play a file, assuming the file is valid.
Parameters
----------
@@ -295,14 +302,14 @@ class Audio():
If the file type is not valid.
ValueError
If the type is not a valid type for a source.
'''
"""
if not path.exists(file_path):
raise FileNotFoundError(f'File "{file_path}" not found.')
file_type = file_path.split('.')[-1]
file_type = file_path.split(".")[-1]
if ReaderType(file_type) not in file_types:
raise InvalidFileType(f'Type {file_type} is not supported.')
raise InvalidFileType(f"Type {file_type} is not supported.")
data = RequestData(
Type=file_type,
@@ -310,12 +317,10 @@ class Audio():
DoesLoop=does_loop,
LoopCount=loop_count,
Name=name,
Args=RequestArgs(
Path=file_path
)
Args=RequestArgs(Path=file_path),
)
with open('/tmp/audio', 'w') as p:
with open("/tmp/audio", "w") as p:
p.write(json.dumps(dict(data)))
return self.__get_new_source(name, does_loop)
@@ -330,7 +335,7 @@ class Audio():
volume: float = 1,
name: str = __gen_name(),
) -> Source:
'''Play a tone from a frequency and wave type.
"""Play a tone from a frequency and wave type.
Parameters
----------
@@ -356,7 +361,7 @@ class Audio():
If the source isn't found after 2 seconds.
ValueError
If the wave type isn't valid.
'''
"""
# ensure the wave type is valid. This will throw an error if it isn't.
WaveType(wave_type)
@@ -367,20 +372,16 @@ class Audio():
LoopCount=loop_count,
Volume=volume,
Type=str(ReaderType.tone),
Args=RequestArgs(
WaveType=wave_type,
Pitch=pitch,
Seconds=duration,
)
Args=RequestArgs(WaveType=wave_type, Pitch=pitch, Seconds=duration,),
)
with open('/tmp/audio', 'w') as f:
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
"""Get a source by it's ID
Parameters
----------
@@ -396,78 +397,78 @@ class Audio():
------
:py:exc:`~replit.NoSourceFoundException`
If the source isnt found or there isn't any sources known to the audio manager.
'''
"""
source = None
with open('/tmp/audioStatus.json', 'r') as f:
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 not data["Sources"]:
raise NoSuchSourceException("No sources exist yet.")
for s in data["Sources"]:
if s['ID'] == int(source_id):
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'])
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
"""Get the raw data for what's playing. This is an api call, and shouldn't be needed
for general usage.
Returns
-------
AudioStaus
The contents of /tmp/audioStatus.json
'''
with open('/tmp/audioStatus.json', 'r') as f:
"""
with open("/tmp/audioStatus.json", "r") as f:
data = AudioStatus(json.loads(f.read()))
if data['Sources'] == None:
data['Sources']: List[SourceData] = []
if data["Sources"] == None:
data["Sources"]: List[SourceData] = []
return data
def get_playing(self) -> List[Source]:
'''Get a list of playing sources.
"""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']]
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.
"""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']]
sources = data["Sources"]
return [Source(s, s["Loop"]) for s in sources if s["Paused"]]
def get_sources(self) -> List[Source]:
'''Gets all sources.
"""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]
sources = data["Sources"]
return [Source(s, s["Loop"]) for s in sources]
audio = Audio()
'''The interface used for all things audio.
"""The interface used for all things audio.
Can be used to play and fetch audio sources.
'''
"""