diff --git a/base.py b/base.py
index 2501880..8c409b6 100644
--- a/base.py
+++ b/base.py
@@ -25,8 +25,6 @@ app = FastAPI(title="codey.lol API",
constants = importlib.import_module("constants").Constants()
util = importlib.import_module("util").Utilities(app, constants)
-glob_state = importlib.import_module("state").State(app, util, constants)
-
origins = [
"https://codey.lol",
@@ -83,23 +81,23 @@ End Blacklisted Routes
Actionable Routes
"""
-randmsg_endpoint = importlib.import_module("endpoints.rand_msg").RandMsg(app, util, constants, glob_state)
-transcription_endpoints = importlib.import_module("endpoints.transcriptions").Transcriptions(app, util, constants, glob_state)
-ai_endpoints = importlib.import_module("endpoints.ai").AI(app, util, constants, glob_state)
+randmsg_endpoint = importlib.import_module("endpoints.rand_msg").RandMsg(app, util, constants)
+transcription_endpoints = importlib.import_module("endpoints.transcriptions").Transcriptions(app, util, constants)
+ai_endpoints = importlib.import_module("endpoints.ai").AI(app, util, constants)
# Below also provides: /lyric_cache_list/ (in addition to /lyric_search/)
-lyric_search_endpoint = importlib.import_module("endpoints.lyric_search").LyricSearch(app, util, constants, glob_state)
+lyric_search_endpoint = importlib.import_module("endpoints.lyric_search").LyricSearch(app, util, constants)
# Below provides numerous LastFM-fed endpoints
-lastfm_endpoints = importlib.import_module("endpoints.lastfm").LastFM(app, util, constants, glob_state)
+lastfm_endpoints = importlib.import_module("endpoints.lastfm").LastFM(app, util, constants)
# Below: YT endpoint(s)
-yt_endpoints = importlib.import_module("endpoints.yt").YT(app, util, constants, glob_state)
+yt_endpoints = importlib.import_module("endpoints.yt").YT(app, util, constants)
# Below: XC endpoint(s)
-xc_endpoints = importlib.import_module("endpoints.xc").XC(app, util, constants, glob_state)
+xc_endpoints = importlib.import_module("endpoints.xc").XC(app, util, constants)
# Below: Karma endpoint(s)
-karma_endpoints = importlib.import_module("endpoints.karma").Karma(app, util, constants, glob_state)
+karma_endpoints = importlib.import_module("endpoints.karma").Karma(app, util, constants)
# Below: Radio endpoint(s) - in development, sporadically loaded as needed
-radio_endpoints = importlib.import_module("endpoints.radio").Radio(app, util, constants, glob_state)
+radio_endpoints = importlib.import_module("endpoints.radio").Radio(app, util, constants)
# Below: Misc endpoints
-misc_endpoints = importlib.import_module("endpoints.misc").Misc(app, util, constants, glob_state, radio_endpoints)
+misc_endpoints = importlib.import_module("endpoints.misc").Misc(app, util, constants, radio_endpoints)
"""
End Actionable Routes
@@ -109,6 +107,6 @@ End Actionable Routes
"""
Startup
"""
-redis_cache = redis_cache.RedisCache()
+redis = redis_cache.RedisCache()
asyncio.get_event_loop().create_task(
- redis_cache.create_index())
\ No newline at end of file
+ redis.create_index())
\ No newline at end of file
diff --git a/endpoints/ai.py b/endpoints/ai.py
index 967c9d1..ccb30d1 100644
--- a/endpoints/ai.py
+++ b/endpoints/ai.py
@@ -4,19 +4,20 @@
import logging
import traceback
import regex
+from regex import Pattern
+from typing import Union
from aiohttp import ClientSession, ClientTimeout
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from .constructors import ValidHookSongRequest, ValidAISongRequest
class AI(FastAPI):
"""AI Endpoints"""
- def __init__(self, app: FastAPI, my_util, constants, glob_state): # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, my_util, constants): # pylint: disable=super-init-not-called
self.app = app
self.util = my_util
self.constants = constants
- self.glob_state = glob_state
- self.url_clean_regex = regex.compile(r'^\/ai\/(openai|base)\/')
- self.endpoints = {
+ self.url_clean_regex: Pattern = regex.compile(r'^\/ai\/(openai|base)\/')
+ self.endpoints: dict = {
"ai/openai": self.ai_openai_handler,
"ai/base": self.ai_handler,
"ai/song": self.ai_song_handler,
@@ -29,16 +30,20 @@ class AI(FastAPI):
include_in_schema=False)
- async def respond_via_webhook(self, data: ValidHookSongRequest, originalRequest: Request):
+ async def respond_via_webhook(self, data: ValidHookSongRequest, originalRequest: Request) -> bool:
"""Respond via Webhook"""
try:
logging.debug("Request received: %s", data)
data2 = data.copy()
del data2.hook
+
+ if not data.hook:
+ return False
+
response = await self.ai_song_handler(data2, originalRequest)
if not response.get('resp'):
logging.critical("NO RESP!")
- return
+ return False
response = response.get('resp')
hook_data = {
'username': 'Claude',
@@ -82,7 +87,6 @@ class AI(FastAPI):
json=await request.json(),
headers=local_llm_headers,
timeout=ClientTimeout(connect=15, sock_read=30)) as out_request:
- await self.glob_state.increment_counter('ai_requests')
response = await out_request.json()
return response
except Exception as e: # pylint: disable=broad-exception-caught
@@ -117,7 +121,6 @@ class AI(FastAPI):
json=await request.json(),
headers=local_llm_headers,
timeout=ClientTimeout(connect=15, sock_read=30)) as out_request:
- await self.glob_state.increment_counter('ai_requests')
response = await out_request.json()
return response
except Exception as e: # pylint: disable=broad-exception-caught
@@ -134,7 +137,7 @@ class AI(FastAPI):
'success': True,
}
- async def ai_song_handler(self, data: ValidAISongRequest, request: Request):
+ async def ai_song_handler(self, data: Union[ValidAISongRequest, ValidHookSongRequest], request: Request):
"""
/ai/song
AI (Song Info) Request [Public]
@@ -165,9 +168,8 @@ class AI(FastAPI):
async with await session.post('https://api.anthropic.com/v1/messages',
json=request_data,
headers=local_llm_headers,
- timeout=ClientTimeout(connect=15, sock_read=30)) as request:
- await self.glob_state.increment_counter('claude_ai_requests')
- response = await request.json()
+ timeout=ClientTimeout(connect=15, sock_read=30)) as aiohttp_request:
+ response = await aiohttp_request.json()
logging.debug("Response: %s",
response)
if response.get('type') == 'error':
diff --git a/endpoints/karma.py b/endpoints/karma.py
index 5a15645..1511b7b 100644
--- a/endpoints/karma.py
+++ b/endpoints/karma.py
@@ -14,7 +14,7 @@ from .constructors import ValidTopKarmaRequest, ValidKarmaRetrievalRequest,\
class KarmaDB:
"""Karma DB Util"""
- def __init__(self):
+ def __init__(self) -> None:
self.db_path: LiteralString = os.path.join("/", "usr", "local", "share",
"sqlite_dbs", "karma.db")
@@ -36,7 +36,7 @@ class KarmaDB:
'errorText': f'No records for {keyword}',
}
- async def get_top(self, n: Optional[int] = 10) -> list[tuple]:
+ async def get_top(self, n: Optional[int] = 10) -> Optional[list[tuple]]:
"""Get Top n=10 Karma Entries
Args:
n (Optional[int]) = 10: The number of top results to return
@@ -49,7 +49,7 @@ class KarmaDB:
return await db_cursor.fetchall()
except:
traceback.print_exc()
- return
+ return None
async def update_karma(self, granter: str, keyword: str, flag: int) -> Optional[bool]:
"""Update Karma for Keyword
@@ -62,7 +62,7 @@ class KarmaDB:
"""
if not flag in [0, 1]:
- return
+ return None
modifier: str = "score + 1" if not flag else "score - 1"
query: str = f"UPDATE karma SET score = {modifier}, last_change = ? WHERE keyword LIKE ?"
@@ -90,13 +90,14 @@ class KarmaDB:
return True
else:
return False
+ return False
+
class Karma(FastAPI):
"""Karma Endpoints"""
- def __init__(self, app: FastAPI, util, constants, glob_state): # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, util, constants): # pylint: disable=super-init-not-called
self.app = app
self.util = util
self.constants = constants
- self.glob_state = glob_state
self.db = KarmaDB()
self.endpoints: dict = {
@@ -117,12 +118,17 @@ class Karma(FastAPI):
raise HTTPException(status_code=403, detail="Unauthorized")
n: int = 10
- if data:
- n: int = int(data.n)
+ if data and data.n:
+ n = int(data.n)
try:
- top10: list[tuple] = await self.db.get_top(n=n)
+ top10: Optional[list[tuple]] = await self.db.get_top(n=n)
+ if not top10:
+ return {
+ 'err': True,
+ 'errorText': 'General failure',
+ }
return top10
except:
traceback.print_exc()
diff --git a/endpoints/lastfm.py b/endpoints/lastfm.py
index 65c4a46..fc8af9a 100644
--- a/endpoints/lastfm.py
+++ b/endpoints/lastfm.py
@@ -10,14 +10,13 @@ from .constructors import ValidArtistSearchRequest, ValidAlbumDetailRequest,\
class LastFM(FastAPI):
"""Last.FM Endpoints"""
- def __init__(self, app: FastAPI, util, constants, glob_state) -> None: # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, util, constants) -> None: # pylint: disable=super-init-not-called
self.app = app
self.util = util
self.constants = constants
- self.glob_state = glob_state
self.lastfm = importlib.import_module("lastfm_wrapper").LastFM()
- self.endpoints = {
+ self.endpoints: dict = {
"lastfm/get_artist_by_name": self.artist_by_name_handler,
"lastfm/get_artist_albums": self.artist_album_handler,
"lastfm/get_release": self.release_detail_handler,
@@ -71,7 +70,7 @@ class LastFM(FastAPI):
seen_release_titles: list = []
for release in album_result:
- release_title: str = release.get('title')
+ release_title: str = release.get('title', 'Unknown')
if release_title.lower() in seen_release_titles:
continue
seen_release_titles.append(release_title.lower())
diff --git a/endpoints/lyric_search.py b/endpoints/lyric_search.py
index 13cd38d..6d95660 100644
--- a/endpoints/lyric_search.py
+++ b/endpoints/lyric_search.py
@@ -7,7 +7,8 @@ import urllib.parse
import regex
import aiosqlite as sqlite3
from fastapi import FastAPI, HTTPException
-from typing import LiteralString, Optional, Pattern
+from typing import LiteralString, Optional, Callable
+from regex import Pattern
from .constructors import ValidTypeAheadRequest, ValidLyricRequest
from lyric_search.constructors import LyricsResult
from lyric_search.sources import aggregate
@@ -15,32 +16,31 @@ from lyric_search import notifier
class CacheUtils:
"""Lyrics Cache DB Utils"""
- def __init__(self):
- self.lyrics_db_path: LiteralString = os.path.join("/", "usr", "local", "share",
+ def __init__(self) -> None:
+ self.lyrics_db_path: LiteralString = os.path.join("/usr/local/share",
"sqlite_dbs", "cached_lyrics.db")
- async def check_typeahead(self, s: str, pre_query: str | None = None) -> Optional[list[dict]]:
+ async def check_typeahead(self, s: str, pre_query: str | None = None) -> list[dict]:
"""Check s against artists stored - for typeahead"""
async with sqlite3.connect(self.lyrics_db_path,
timeout=2) as db_conn:
- db_conn.row_factory = lambda c, r: dict([(col[0], r[idx]) for idx, col in enumerate(c.description)])
+ db_conn.row_factory = sqlite3.Row
if not pre_query:
query: str = "SELECT distinct(artist) FROM lyrics WHERE artist LIKE ? LIMIT 15"
query_params: tuple = (f"%{s}%",)
else:
- query: str = "SELECT distinct(song) FROM lyrics WHERE artist LIKE ? AND song LIKE ? LIMIT 15"
- query_params: tuple = (f"%{pre_query}%", f"%{s}%",)
+ query = "SELECT distinct(song) FROM lyrics WHERE artist LIKE ? AND song LIKE ? LIMIT 15"
+ query_params = (f"%{pre_query}%", f"%{s}%",)
async with await db_conn.execute(query, query_params) as db_cursor:
return await db_cursor.fetchall()
class LyricSearch(FastAPI):
"""Lyric Search Endpoint"""
- def __init__(self, app: FastAPI, util, constants, glob_state): # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, util, constants): # pylint: disable=super-init-not-called
self.app = app
self.util = util
self.constants = constants
- self.glob_state = glob_state
self.cache_utils = CacheUtils()
self.notifier = notifier.DiscordNotifier()
@@ -78,7 +78,7 @@ class LyricSearch(FastAPI):
'errorText': 'Invalid request',
}
query: str = data.query
- typeahead_result: Optional[list[dict]] = await self.cache_utils.check_typeahead(query)
+ typeahead_result: list[dict] = await self.cache_utils.check_typeahead(query)
typeahead_list: list[str] = [str(r.get('artist')) for r in typeahead_result]
return typeahead_list
@@ -92,7 +92,7 @@ class LyricSearch(FastAPI):
}
pre_query: str = data.pre_query
query: str = data.query
- typeahead_result: Optional[list[dict]] = await self.cache_utils.check_typeahead(query, pre_query)
+ typeahead_result: list[dict] = await self.cache_utils.check_typeahead(query, pre_query)
typeahead_list: list[str] = [str(r.get('song')) for r in typeahead_result]
return typeahead_list
@@ -122,24 +122,30 @@ class LyricSearch(FastAPI):
}
if not data.t:
- search_artist: str = data.a
- search_song: str = data.s
+ search_artist: Optional[str] = data.a
+ search_song: Optional[str] = data.s
else:
t_split = data.t.split(" - ", maxsplit=1)
- search_artist: str = t_split[0]
- search_song: str = t_split[1]
+ search_artist = t_split[0]
+ search_song = t_split[1]
if search_artist and search_song:
- search_artist: str = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_artist.strip())
- search_song: str = self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_song.strip())
- search_artist: str = urllib.parse.unquote(search_artist)
- search_song: str = urllib.parse.unquote(search_song)
+ search_artist = str(self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_artist.strip()))
+ search_song = str(self.constants.DOUBLE_SPACE_REGEX.sub(" ", search_song.strip()))
+ search_artist = urllib.parse.unquote(search_artist)
+ search_song = urllib.parse.unquote(search_song)
- excluded_sources: list = data.excluded_sources
+ if not isinstance(search_artist, str) or not isinstance(search_song, str):
+ return {
+ 'err': True,
+ 'errorText': 'Invalid request',
+ }
+
+ excluded_sources: Optional[list] = data.excluded_sources
aggregate_search = aggregate.Aggregate(exclude_methods=excluded_sources)
plain_lyrics: bool = not data.lrc
- result: Optional[LyricsResult] = await aggregate_search.search(search_artist, search_song, plain_lyrics)
+ result: Optional[LyricsResult|dict] = await aggregate_search.search(search_artist, search_song, plain_lyrics)
if not result:
return {
@@ -147,15 +153,15 @@ class LyricSearch(FastAPI):
'errorText': 'Sources exhausted, lyrics not located.',
}
- result: dict = result.todict()
-
+ result = vars(result)
+
if data.sub and not data.lrc:
seeked_found_line: Optional[int] = None
lyric_lines: list[str] = result['lyrics'].strip().split(" / ")
for i, line in enumerate(lyric_lines):
- line: str = regex.sub(r'\u2064', '', line.strip())
+ line = regex.sub(r'\u2064', '', line.strip())
if data.sub.strip().lower() in line.strip().lower():
- seeked_found_line: int = i
+ seeked_found_line = i
logging.debug("Found %s at %s, match for %s!",
line, seeked_found_line, data.sub) # REMOVEME: DEBUG
break
diff --git a/endpoints/misc.py b/endpoints/misc.py
index 65e0bbf..c5a4f4b 100644
--- a/endpoints/misc.py
+++ b/endpoints/misc.py
@@ -9,11 +9,10 @@ from lyric_search.sources import private, cache as LyricsCache, redis_cache
class Misc(FastAPI):
"""Misc Endpoints"""
- def __init__(self, app: FastAPI, my_util, constants, glob_state, radio): # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, my_util, constants, radio): # pylint: disable=super-init-not-called
self.app = app
self.util = my_util
self.constants = constants
- self.glob_state = glob_state
self.radio_pubkey: str = "XC-AJCJS89-AOLOFKZ92921AK-AKASKZJAN178-3D1"
self.lyr_cache = LyricsCache.Cache()
self.redis_cache = redis_cache.RedisCache()
diff --git a/endpoints/radio.py b/endpoints/radio.py
index 0eb7fee..70d6fef 100644
--- a/endpoints/radio.py
+++ b/endpoints/radio.py
@@ -28,12 +28,11 @@ TODO:
class Radio(FastAPI):
"""Radio Endpoints"""
- def __init__(self, app: FastAPI, my_util, constants, glob_state) -> None: # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, my_util, constants) -> None: # pylint: disable=super-init-not-called
self.app = app
self.util = my_util
self.constants = constants
self.radio_util = radio_util.RadioUtil(self.constants)
- self.glob_state = glob_state
self.endpoints: dict = {
"radio/np": self.radio_now_playing,
@@ -64,8 +63,10 @@ class Radio(FastAPI):
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
if data.skipTo:
- (x, _) = self.radio_util.get_queue_item_by_uuid(data.skipTo)
- self.radio_util.active_playlist = self.radio_util.active_playlist[x:]
+ queue_item = self.radio_util.get_queue_item_by_uuid(data.skipTo)
+ if not queue_item:
+ return False
+ self.radio_util.active_playlist = self.radio_util.active_playlist[queue_item[0]:]
if not self.radio_util.active_playlist:
await self.radio_util.load_playlist()
return await self.radio_util._ls_skip()
@@ -112,7 +113,13 @@ class Radio(FastAPI):
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
- (x, item) = self.radio_util.get_queue_item_by_uuid(data.uuid)
+ queue_item = self.radio_util.get_queue_item_by_uuid(data.uuid)
+ if not queue_item:
+ return {
+ 'err': True,
+ 'errorText': 'Queue item not found.',
+ }
+ (x, item) = queue_item
self.radio_util.active_playlist.pop(x)
self.radio_util.active_playlist.insert(0, item)
if not data.next:
@@ -126,18 +133,18 @@ class Radio(FastAPI):
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
- (x, found_item) = self.radio_util.get_queue_item_by_uuid(data.uuid)
- if not found_item:
+ queue_item = self.radio_util.get_queue_item_by_uuid(data.uuid)
+ if not queue_item:
return {
- 'ok': False,
- 'err': 'UUID not found in play queue',
+ 'err': True,
+ 'errorText': 'Queue item not found.',
}
- self.radio_util.active_playlist.pop(x)
+ self.radio_util.active_playlist.pop(queue_item[0])
return {
'ok': True,
}
- async def album_art_handler(self, request: Request, track_id: Optional[int] = None) -> bytes:
+ async def album_art_handler(self, request: Request, track_id: Optional[int] = None) -> Response:
"""
Get album art, optional parameter track_id may be specified.
Otherwise, current track album art will be pulled.
@@ -182,13 +189,13 @@ class Radio(FastAPI):
if not isinstance(self.radio_util.active_playlist, list) or not self.radio_util.active_playlist:
await self.radio_util.load_playlist()
await self.radio_util._ls_skip()
- return
+ return None
next = self.radio_util.active_playlist.pop(0)
if not isinstance(next, dict):
logging.critical("next is of type: %s, reloading playlist...", type(next))
await self.radio_util.load_playlist()
await self.radio_util._ls_skip()
- return
+ return None
duration: int = next['duration']
time_started: int = int(time.time())
@@ -209,19 +216,21 @@ class Radio(FastAPI):
try:
if not await self.radio_util.get_album_art(file_path=next['file_path']):
album_art = await self.radio_util.get_album_art(file_path=next['file_path'])
+ if not album_art:
+ return None
await self.radio_util.cache_album_art(next['id'], album_art)
except:
traceback.print_exc()
return next
- async def radio_request(self, data: ValidRadioSongRequest, request: Request) -> Response:
+ async def radio_request(self, data: ValidRadioSongRequest, request: Request) -> dict:
"""Song request handler"""
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
- artistsong: str = data.artistsong
- artist: str = data.artist
- song: str = data.song
+ artistsong: Optional[str] = data.artistsong
+ artist: Optional[str] = data.artist
+ song: Optional[str] = data.song
if artistsong and (artist or song):
return {
'err': True,
diff --git a/endpoints/radio_util.py b/endpoints/radio_util.py
index 3b235b4..336d3fd 100644
--- a/endpoints/radio_util.py
+++ b/endpoints/radio_util.py
@@ -13,7 +13,7 @@ import os
import gpt
from aiohttp import ClientSession, ClientTimeout
import aiosqlite as sqlite3
-from typing import Optional, LiteralString
+from typing import Union, Optional, LiteralString
from uuid import uuid4 as uuid
from .constructors import RadioException
@@ -28,7 +28,7 @@ class RadioUtil:
self.active_playlist_path: str|LiteralString = os.path.join("/usr/local/share",
"sqlite_dbs", "track_file_map.db")
self.active_playlist_name = "default" # not used
- self.active_playlist: list = []
+ self.active_playlist: list[dict] = []
self.now_playing: dict = {
'artist': 'N/A',
'song': 'N/A',
@@ -75,8 +75,6 @@ class RadioUtil:
if not artistsong and (not artist or not song):
raise RadioException("No query provided")
try:
- search_artist: Optional[str] = None
- search_song: Optional[str] = None
search_query: str = 'SELECT id, artist, song, (artist || " - " || song) AS artistsong, genre, file_path, duration FROM tracks\
WHERE editdist3((lower(artist) || " " || lower(song)), (? || " " || ?))\
<= 410 ORDER BY editdist3((lower(artist) || " " || lower(song)), ?) ASC LIMIT 1'
@@ -84,10 +82,10 @@ class RadioUtil:
artistsong_split: list = artistsong.split(" - ", maxsplit=1)
(search_artist, search_song) = tuple(artistsong_split)
else:
- search_artist: str = artist
- search_song: str = song
+ search_artist = artist
+ search_song = song
if not artistsong:
- artistsong: str = f"{search_artist} - {search_song}"
+ artistsong = f"{search_artist} - {search_song}"
search_params = (search_artist.lower(), search_song.lower(), artistsong.lower(),)
async with sqlite3.connect(self.active_playlist_path,
timeout=2) as db_conn:
@@ -97,7 +95,7 @@ class RadioUtil:
db_conn.row_factory = sqlite3.Row
async with await db_conn.execute(search_query, search_params) as db_cursor:
result: Optional[sqlite3.Row|bool] = await db_cursor.fetchone()
- if not result:
+ if not result or not isinstance(result, sqlite3.Row):
return False
pushObj: dict = {
'id': result['id'],
@@ -116,7 +114,7 @@ class RadioUtil:
traceback.print_exc()
return False
- async def load_playlist(self):
+ async def load_playlist(self) -> None:
"""Load Playlist"""
try:
logging.info(f"Loading playlist...")
@@ -135,12 +133,19 @@ class RadioUtil:
"grunge", "house", "dubstep", "hardcore", "hair metal", "horror punk", "folk punk", "breakcore",\
"post-rock", "deathcore", "hardcore punk", "synthwave", "trap") GROUP BY artistdashsong ORDER BY RANDOM()'
+ """
+ LIMITED TO ONE ARTIST...
+ """
+
+ # db_query: str = 'SELECT distinct(artist || " - " || song) AS artistdashsong, id, artist, song, genre, file_path, duration FROM tracks\
+ # WHERE artist LIKE "%bad omens%" GROUP BY artistdashsong ORDER BY RANDOM()'
+
async with sqlite3.connect(self.active_playlist_path,
timeout=2) as db_conn:
db_conn.row_factory = sqlite3.Row
async with await db_conn.execute(db_query) as db_cursor:
- results: Optional[list[sqlite3.Row]] = await db_cursor.fetchall()
- self.active_playlist: list[dict] = [{
+ results: list[sqlite3.Row] = await db_cursor.fetchall()
+ self.active_playlist = [{
'uuid': str(uuid().hex),
'id': r['id'],
'artist': double_space.sub(' ', r['artist']).strip(),
@@ -174,7 +179,7 @@ class RadioUtil:
traceback.print_exc()
async def get_album_art(self, track_id: Optional[int] = None,
- file_path: Optional[str] = None) -> bytes:
+ file_path: Optional[str] = None) -> Optional[bytes]:
"""
Get Album Art
Args:
@@ -191,18 +196,18 @@ class RadioUtil:
query_params: tuple = (track_id,)
if file_path and not track_id:
- query: str = "SELECT album_art FROM tracks WHERE file_path = ?"
- query_params: tuple = (file_path,)
+ query = "SELECT album_art FROM tracks WHERE file_path = ?"
+ query_params = (file_path,)
async with await db_conn.execute(query,
query_params) as db_cursor:
result: Optional[sqlite3.Row|bool] = await db_cursor.fetchone()
- if not result:
- return
+ if not result or not isinstance(result, sqlite3.Row):
+ return None
return result['album_art']
except:
traceback.print_exc()
- return
+ return None
def get_queue_item_by_uuid(self, uuid: str) -> Optional[tuple[int, dict]]:
"""
@@ -249,7 +254,7 @@ class RadioUtil:
response: Optional[str] = await self.gpt.get_completion(prompt=f"I am going to listen to {song} by {artist}.")
if not response:
logging.critical("No response received from GPT?")
- return
+ return None
return response
async def webhook_song_change(self, track: dict) -> None:
@@ -314,7 +319,7 @@ class RadioUtil:
if not ai_response:
return
- hook_data: dict = {
+ hook_data = {
'username': 'GPT',
"embeds": [{
"title": "AI Feedback",
diff --git a/endpoints/rand_msg.py b/endpoints/rand_msg.py
index 3718b41..2bca2c6 100644
--- a/endpoints/rand_msg.py
+++ b/endpoints/rand_msg.py
@@ -2,29 +2,28 @@
import os
import random
-from typing import LiteralString
+from typing import LiteralString, Optional
import aiosqlite as sqlite3
from fastapi import FastAPI
from .constructors import RandMsgRequest
class RandMsg(FastAPI):
"""Random Message Endpoint"""
- def __init__(self, app: FastAPI, util, constants, glob_state): # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, util, constants): # pylint: disable=super-init-not-called
self.app = app
self.util = util
self.constants = constants
- self.glob_state = glob_state
-
self.endpoint_name = "randmsg"
app.add_api_route(f"/{self.endpoint_name}", self.randmsg_handler, methods=["POST"])
- async def randmsg_handler(self, data: RandMsgRequest = None):
+ async def randmsg_handler(self, data: RandMsgRequest):
"""
Get a randomly generated message
"""
random.seed()
- short: bool = data.short if data else False
+ short: bool = data.short if data.short else False
+
if not short:
db_rand_selected = random.choice([0, 1, 3])
else:
@@ -33,15 +32,13 @@ class RandMsg(FastAPI):
match db_rand_selected:
case 0:
- randmsg_db_path = os.path.join("/",
- "usr", "local", "share",
+ randmsg_db_path = os.path.join("/usr/local/share",
"sqlite_dbs", "qajoke.db") # For qajoke db
db_query = "SELECT id, ('Q: ' || question || '
A: ' \
|| answer) FROM jokes ORDER BY RANDOM() LIMIT 1" # For qajoke db
title_attr = "QA Joke DB"
case 1 | 9:
- randmsg_db_path = os.path.join("/",
- "usr", "local", "share",
+ randmsg_db_path = os.path.join("/usr/local/share",
"sqlite_dbs",
"randmsg.db") # For randmsg db
db_query = "SELECT id, msg FROM msgs WHERE \
@@ -50,46 +47,41 @@ class RandMsg(FastAPI):
db_query = db_query.replace("<= 180", "<= 126")
title_attr = "Random Msg DB"
case 2:
- randmsg_db_path = os.path.join("/",
- "usr", "local", "share",
+ randmsg_db_path = os.path.join("/usr/local/share",
"sqlite_dbs",
"trump.db") # For Trump Tweet DB
db_query = "SELECT id, content FROM tweets \
ORDER BY RANDOM() LIMIT 1" # For Trump Tweet DB
title_attr = "Trump Tweet DB"
case 3:
- randmsg_db_path: str|LiteralString = os.path.join("/",
- "usr", "local", "share",
+ randmsg_db_path = os.path.join("/usr/local/share",
"sqlite_dbs",
"philo.db") # For Philo DB
- db_query: str = "SELECT id, (content || '
- ' || speaker) FROM quotes \
+ db_query = "SELECT id, (content || '
- ' || speaker) FROM quotes \
ORDER BY RANDOM() LIMIT 1" # For Philo DB
- title_attr: str = "Philosophical Quotes DB"
+ title_attr = "Philosophical Quotes DB"
case 4:
- randmsg_db_path: str|LiteralString = os.path.join("/",
- "usr", "local", "share",
+ randmsg_db_path = os.path.join("/usr/local/share",
"sqlite_dbs",
"hate.db") # For Hate DB
- db_query: str = """SELECT id, ("" || comment) FROM hate_speech \
+ db_query = """SELECT id, ("" || comment) FROM hate_speech \
WHERE length(comment) <= 180 ORDER BY RANDOM() LIMIT 1"""
- title_attr: str = "Hate Speech DB"
+ title_attr = "Hate Speech DB"
case 5:
- randmsg_db_path: str|LiteralString = os.path.join("/",
- "usr", "local", "share",
+ randmsg_db_path = os.path.join("/usr/local/share",
"sqlite_dbs",
"rjokes.db") # r/jokes DB
- db_query: str = """SELECT id, (title || "
" || body) FROM jokes \
+ db_query = """SELECT id, (title || "
" || body) FROM jokes \
WHERE score >= 10000 ORDER BY RANDOM() LIMIT 1"""
- title_attr: str = "r/jokes DB"
+ title_attr = "r/jokes DB"
case 6:
- randmsg_db_path: str|LiteralString = os.path.join("/",
- "usr", "local", "share",
+ randmsg_db_path = os.path.join("/usr/local/share",
"sqlite_dbs",
"donnies.db") # Donnies DB
random.seed()
twilight_or_mice: str = random.choice(["twilight", "mice"])
- db_query: str = f"SELECT id, text FROM {twilight_or_mice} ORDER BY RANDOM() LIMIT 1"
- title_attr: str = "Donnies DB"
+ db_query = f"SELECT id, text FROM {twilight_or_mice} ORDER BY RANDOM() LIMIT 1"
+ title_attr = "Donnies DB"
async with sqlite3.connect(database=randmsg_db_path, timeout=1) as _db:
async with await _db.execute(db_query) as _cursor:
diff --git a/endpoints/transcriptions.py b/endpoints/transcriptions.py
index edcf615..63f2715 100644
--- a/endpoints/transcriptions.py
+++ b/endpoints/transcriptions.py
@@ -8,11 +8,10 @@ from .constructors import ValidShowEpisodeLineRequest, ValidShowEpisodeListReque
class Transcriptions(FastAPI):
"""Transcription Endpoints"""
- def __init__(self, app: FastAPI, util, constants, glob_state) -> None: # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, util, constants) -> None: # pylint: disable=super-init-not-called
self.app = app
self.util = util
self.constants = constants
- self.glob_state = glob_state
self.endpoints: dict = {
"transcriptions/get_episodes": self.get_episodes_handler,
@@ -34,39 +33,39 @@ class Transcriptions(FastAPI):
if show_id is None:
return {
'err': True,
- 'errorText': 'Invalid request'
+ 'errorText': 'Invalid request',
}
- show_id: int = int(show_id)
+ show_id = int(show_id)
if not(str(show_id).isnumeric()) or show_id not in [0, 1, 2]:
return {
'err': True,
- 'errorText': 'Show not found.'
+ 'errorText': 'Show not found.',
}
match show_id:
case 0:
- db_path: str|LiteralString = os.path.join("/", "usr", "local", "share",
+ db_path = os.path.join("/usr/local/share",
"sqlite_dbs", "sp.db")
- db_query: str = """SELECT DISTINCT(("S" || Season || "E" || Episode || " " || Title)), ID FROM SP_DAT ORDER BY Season, Episode"""
- show_title: str = "South Park"
+ db_query = """SELECT DISTINCT(("S" || Season || "E" || Episode || " " || Title)), ID FROM SP_DAT ORDER BY Season, Episode"""
+ show_title = "South Park"
case 1:
- db_path: str|LiteralString = os.path.join("/", "usr", "local", "share",
+ db_path = os.path.join("/usr/local/share",
"sqlite_dbs", "futur.db")
- db_query: str = """SELECT DISTINCT(("S" || EP_S || "E" || EP_EP || " " || EP_TITLE)), EP_ID FROM clean_dialog ORDER BY EP_S, EP_EP"""
- show_title: str = "Futurama"
+ db_query = """SELECT DISTINCT(("S" || EP_S || "E" || EP_EP || " " || EP_TITLE)), EP_ID FROM clean_dialog ORDER BY EP_S, EP_EP"""
+ show_title = "Futurama"
case 2:
- db_path: str|LiteralString = os.path.join("/", "usr", "local", "share",
+ db_path = os.path.join("/usr/local/share",
"sqlite_dbs", "parks.db")
- db_query: str = """SELECT DISTINCT(("S" || EP_S || "E" || EP_EP || " " || EP_TITLE)), EP_ID FROM clean_dialog ORDER BY EP_S, EP_EP"""
- show_title: str = "Parks And Rec"
+ db_query = """SELECT DISTINCT(("S" || EP_S || "E" || EP_EP || " " || EP_TITLE)), EP_ID FROM clean_dialog ORDER BY EP_S, EP_EP"""
+ show_title = "Parks And Rec"
case _:
return {
'err': True,
'errorText': 'Unknown error.'
}
- await self.glob_state.increment_counter('transcript_list_requests')
+
async with sqlite3.connect(database=db_path, timeout=1) as _db:
async with await _db.execute(db_query) as _cursor:
result: list[tuple] = await _cursor.fetchall()
@@ -86,17 +85,17 @@ class Transcriptions(FastAPI):
# pylint: disable=line-too-long
match show_id:
case 0:
- db_path: str|LiteralString = os.path.join("/", "usr", "local", "share",
+ db_path = os.path.join("/usr/local/share",
"sqlite_dbs", "sp.db")
- db_query: str = """SELECT ("S" || Season || "E" || Episode || " " || Title), Character, Line FROM SP_DAT WHERE ID = ?"""
+ db_query = """SELECT ("S" || Season || "E" || Episode || " " || Title), Character, Line FROM SP_DAT WHERE ID = ?"""
case 1:
- db_path: str|LiteralString = os.path.join("/", "usr", "local", "share",
+ db_path = os.path.join("/usr/local/share",
"sqlite_dbs", "futur.db")
- db_query: str = """SELECT ("S" || EP_S || "E" || EP_EP || " " || EP_TITLE || "
Opener: " || EP_OPENER || ""), EP_LINE_SPEAKER, EP_LINE FROM clean_dialog WHERE EP_ID = ? ORDER BY LINE_ID ASC"""
+ db_query = """SELECT ("S" || EP_S || "E" || EP_EP || " " || EP_TITLE || "
Opener: " || EP_OPENER || ""), EP_LINE_SPEAKER, EP_LINE FROM clean_dialog WHERE EP_ID = ? ORDER BY LINE_ID ASC"""
case 2:
- db_path: str|LiteralString = os.path.join("/", "usr", "local", "share",
+ db_path = os.path.join("/usr/local/share",
"sqlite_dbs", "parks.db")
- db_query: str = """SELECT ("S" || EP_S || "E" || EP_EP || " " || EP_TITLE), EP_LINE_SPEAKER, EP_LINE FROM clean_dialog WHERE EP_ID = ? ORDER BY id ASC"""
+ db_query = """SELECT ("S" || EP_S || "E" || EP_EP || " " || EP_TITLE), EP_LINE_SPEAKER, EP_LINE FROM clean_dialog WHERE EP_ID = ? ORDER BY id ASC"""
case _:
return {
@@ -104,7 +103,6 @@ class Transcriptions(FastAPI):
'errorText': 'Unknown error'
}
- await self.glob_state.increment_counter('transcript_requests')
async with sqlite3.connect(database=db_path, timeout=1) as _db:
params: tuple = (episode_id,)
async with await _db.execute(db_query, params) as _cursor:
diff --git a/endpoints/xc.py b/endpoints/xc.py
index 5f4498a..d1b75db 100644
--- a/endpoints/xc.py
+++ b/endpoints/xc.py
@@ -1,6 +1,7 @@
#!/usr/bin/env python3.12
import logging
+from typing import Optional
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
from aiohttp import ClientSession, ClientTimeout
@@ -9,12 +10,10 @@ from .constructors import ValidXCRequest
class XC(FastAPI):
"""XC (CrossComm) Endpoints"""
- def __init__(self, app: FastAPI, util, constants, glob_state) -> None: # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, util, constants) -> None: # pylint: disable=super-init-not-called
self.app = app
self.util = util
self.constants = constants
- self.glob_state = glob_state
-
self.endpoints: dict = {
"xc": self.xc_handler,
@@ -31,7 +30,7 @@ class XC(FastAPI):
key: str = data.key
bid: int = data.bid
cmd: str = data.cmd
- cmd_data: dict = data.data
+ cmd_data: Optional[dict] = data.data
if not self.util.check_key(path=request.url.path, req_type=0, key=key):
raise HTTPException(status_code=403, detail="Unauthorized")
@@ -50,11 +49,15 @@ class XC(FastAPI):
async with ClientSession() as session:
async with await session.post(f"{bot_api_url}{cmd}", json=cmd_data, headers={
'Content-Type': 'application/json; charset=utf-8'
- }, timeout=ClientTimeout(connect=5, sock_read=5)) as request:
- response: dict = await request.json()
+ }, timeout=ClientTimeout(connect=5, sock_read=5)) as aiohttp_request:
+ response: dict = await aiohttp_request.json()
return {
'success': True,
'response': response
}
except Exception as e:
- logging.debug("Error: %s", str(e))
\ No newline at end of file
+ logging.debug("Error: %s", str(e))
+ return {
+ 'err': True,
+ 'errorText': 'General error.',
+ }
\ No newline at end of file
diff --git a/endpoints/yt.py b/endpoints/yt.py
index c509b31..49ad8d2 100644
--- a/endpoints/yt.py
+++ b/endpoints/yt.py
@@ -8,11 +8,10 @@ from .constructors import ValidYTSearchRequest
class YT(FastAPI):
"""YT Endpoints"""
- def __init__(self, app: FastAPI, util, constants, glob_state) -> None: # pylint: disable=super-init-not-called
+ def __init__(self, app: FastAPI, util, constants) -> None: # pylint: disable=super-init-not-called
self.app = app
self.util = util
self.constants = constants
- self.glob_state = glob_state
self.ytsearch = importlib.import_module("youtube_search_async").YoutubeSearch()
self.endpoints: dict = {
diff --git a/gpt/__init__.py b/gpt/__init__.py
index 40cf305..08ad92a 100644
--- a/gpt/__init__.py
+++ b/gpt/__init__.py
@@ -12,7 +12,7 @@ class GPT:
)
self.default_system_prompt = "You are a helpful assistant who will provide only totally accurate tidbits of info on the specific songs the user may listen to."
- async def get_completion(self, prompt: str, system_prompt: Optional[str] = None) -> None:
+ async def get_completion(self, prompt: str, system_prompt: Optional[str] = None) -> str:
if not system_prompt:
system_prompt = self.default_system_prompt
chat_completion = await self.client.chat.completions.create(
diff --git a/lastfm_wrapper.py b/lastfm_wrapper.py
index 1ff1577..22260b9 100644
--- a/lastfm_wrapper.py
+++ b/lastfm_wrapper.py
@@ -3,7 +3,7 @@
import traceback
import logging
-from typing import Optional
+from typing import Optional, Union
import regex
from aiohttp import ClientSession, ClientTimeout
from constants import Constants
@@ -30,13 +30,13 @@ class LastFM:
timeout=ClientTimeout(connect=3, sock_read=8)) as request:
request.raise_for_status()
data: dict = await request.json()
- data = data.get('artist')
+ data = data.get('artist', 'N/A')
ret_obj: dict = {
'id': data.get('mbid'),
'touring': data.get('ontour'),
'name': data.get('name'),
- 'bio': data.get('bio').get('summary').strip()\
+ 'bio': data.get('bio', None).get('summary').strip()\
.split("= 50
@@ -143,7 +143,7 @@ class LastFM:
'err': 'Failed',
}
- async def get_artist_id(self, artist: Optional[str] = None) -> int|dict:
+ async def get_artist_id(self, artist: Optional[str] = None) -> int:
"""
Get Artist ID from LastFM
Args:
@@ -153,22 +153,16 @@ class LastFM:
"""
try:
if artist is None:
- return {
- 'err': 'No artist specified.',
- }
+ return -1
artist_search: dict = await self.search_artist(artist=artist)
if not artist_search:
logging.debug("[get_artist_id] Throwing no result error")
- return {
- 'err': 'No results.',
- }
+ return -1
artist_id: int = int(artist_search[0].get('id', 0))
return artist_id
except:
traceback.print_exc()
- return {
- 'err': 'Failed',
- }
+ return -1
async def get_artist_info_by_id(self, artist_id: Optional[int] = None) -> dict:
"""
@@ -219,12 +213,12 @@ class LastFM:
return {
'err': 'No artist specified.',
}
- artist_id: int = await self.get_artist_id(artist=artist)
+ artist_id: Optional[int] = await self.get_artist_id(artist=artist)
if not artist_id:
return {
'err': 'Failed',
}
- artist_info: dict = await self.get_artist_info_by_id(artist_id=artist_id)
+ artist_info: Optional[dict] = await self.get_artist_info_by_id(artist_id=artist_id)
if not artist_info:
return {
'err': 'Failed',
@@ -258,21 +252,21 @@ class LastFM:
async with await session.get(req_url,
timeout=ClientTimeout(connect=3, sock_read=8)) as request:
request.raise_for_status()
- data: dict = await request.json()
- data: dict = data.get('album')
+ json_data: dict = await request.json()
+ data: dict = json_data.get('album', None)
ret_obj: dict = {
'id': data.get('mbid'),
'artists': data.get('artist'),
'tags': data.get('tags'),
'title': data.get('name'),
- 'summary': data.get('wiki').get('summary').split(" dict:
- """Return as dict"""
- return asdict(self)
\ No newline at end of file
+ time: float = 0.00
\ No newline at end of file
diff --git a/lyric_search/sources/common.py b/lyric_search/sources/common.py
index 4b8be2b..87868f5 100644
--- a/lyric_search/sources/common.py
+++ b/lyric_search/sources/common.py
@@ -1,5 +1,5 @@
#!/usr/bin/env python3.12
-SCRAPE_HEADERS = {
+SCRAPE_HEADERS: dict[str, str] = {
'accept': '*/*',
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:130.0) Gecko/20100101 Firefox/130.0',
}
\ No newline at end of file
diff --git a/lyric_search/sources/genius.py b/lyric_search/sources/genius.py
index 58d65cf..d3698e1 100644
--- a/lyric_search/sources/genius.py
+++ b/lyric_search/sources/genius.py
@@ -14,7 +14,6 @@ from . import private, common, cache, redis_cache
from lyric_search import utils
from lyric_search.constructors import LyricsResult
-
logger = logging.getLogger()
log_level = logging.getLevelName(logger.level)
diff --git a/pyproject.toml b/pyproject.toml
index 41fd14b..0da987d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -23,3 +23,5 @@ dependencies = [
"typing-inspect>=0.9.0",
"http3>=0.6.7",
]
+[tool.mypy]
+disable_error_code = ["import-untyped"]
\ No newline at end of file
diff --git a/state.py b/state.py
deleted file mode 100644
index bb186ad..0000000
--- a/state.py
+++ /dev/null
@@ -1,111 +0,0 @@
-#!/usr/bin/env python3.12
-# pylint: disable=bare-except, broad-exception-raised
-
-"""Global State Storage/Counters"""
-
-import logging
-import os
-import aiosqlite as sqlite3
-
-from fastapi import FastAPI
-from fastapi_utils.tasks import repeat_every
-
-
-class State(FastAPI):
- """Global State for API"""
- def __init__(self, app: FastAPI, util, constants): # pylint: disable=unused-argument
- super().__init__()
- self.counter_db_path = os.path.join("/", "usr", "local", "share",
- "sqlite_dbs", "stats.db")
- self.counters = {
- str(counter): 0 for counter in constants.AVAILABLE_COUNTERS
- }
- self.counters_initialized = False
- logging.debug("[State] Counters: %s", self.counters)
-
- @app.on_event("startup")
- async def get_counters():
- logging.info("[State] Initializing counters...")
- async with sqlite3.connect(self.counter_db_path, timeout=2) as _db:
- _query = "SELECT ai_requests, lyric_requests, transcript_list_requests, transcript_requests, lyrichistory_requests, \
- failedlyric_requests, misc_failures, claude_ai_requests FROM counters LIMIT 1"
- await _db.executescript("pragma journal_mode = WAL; pragma synchronous = normal; pragma temp_store = memory; pragma mmap_size = 30000000000;")
- async with _db.execute(_query) as _cursor:
- _result = await _cursor.fetchone()
- (ai_requests,
- lyric_requests,
- transcript_list_requests,
- transcript_requests,
- lyrichistory_requests,
- failedlyric_requests,
- misc_failures,
- claude_ai_requests) = _result
- self.counters = {
- 'ai_requests': ai_requests,
- 'lyric_requests': lyric_requests,
- 'transcript_list_requests': transcript_list_requests,
- 'transcript_requests': transcript_requests,
- 'lyrichistory_requests': lyrichistory_requests,
- 'failedlyric_requests': failedlyric_requests,
- 'misc_failures': misc_failures,
- 'claude_ai_requests': claude_ai_requests
- }
- self.counters_initialized = True
- logging.info("Counters loaded from db: %s", self.counters)
-
-
- @app.on_event("startup")
- @repeat_every(seconds=10)
- async def update_db():
- if not self.counters_initialized:
- logging.debug("[State] TICK: Counters not yet initialized")
- return
-
- ai_requests = self.counters.get('ai_requests')
- lyric_requests = self.counters.get('lyric_requests')
- transcript_list_requests = self.counters.get('transcript_list_requests')
- transcript_requests = self.counters.get('transcript_requests')
- lyrichistory_requests = self.counters.get('lyrichistory_requests')
- failedlyric_requests = self.counters.get('failedlyric_requests')
- claude_ai_requests = self.counters.get('claude_ai_requests')
-
- async with sqlite3.connect(self.counter_db_path, timeout=2) as _db:
- _query = "UPDATE counters SET ai_requests = ?, lyric_requests = ?, transcript_list_requests = ?, \
- transcript_requests = ?, lyrichistory_requests = ?, failedlyric_requests = ?, \
- claude_ai_requests = ?"
-
- _params = (ai_requests,
- lyric_requests,
- transcript_list_requests,
- transcript_requests,
- lyrichistory_requests,
- failedlyric_requests,
- claude_ai_requests)
-
- await _db.executescript("pragma journal_mode = WAL; pragma synchronous = normal; pragma temp_store = memory; pragma mmap_size = 30000000000;")
- async with _db.execute(_query, _params) as _cursor:
- if _cursor.rowcount != 1:
- logging.error("Failed to update DB")
- return
- await _db.commit()
- logging.debug("[State] Updated DB")
-
-
- async def increment_counter(self, counter: str):
- """Increment Counter"""
- if not counter in self.counters.keys():
- raise BaseException(f"[State] Counter {counter} does not exist")
-
- self.counters[counter] += 1
- return True
-
- async def get_counter(self, counter: str):
- """Get Counter"""
- if not counter in self.counters.keys():
- raise BaseException(f"[State] Counter {counter} does not exist")
-
- return self.counters[counter]
-
- async def get_all_counters(self):
- """Get All Counters"""
- return self.counters
\ No newline at end of file