zoriya/Chromacase
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: messages in separate file

Browse Source
This commit is contained in:
GitBluub
2023-04-03 01:57:48 +09:00
parent 445b949fa8
commit f1f7500b44
2 changed files with 350 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scorometer/chroma_case/Message.py
+77
View File
@@ -0,0 +1,77 @@
from dataclasses import dataclass
from typing import Literal, Tuple
from validated_dc import ValidatedDC, get_errors, is_valid
@dataclass
class InvalidMessage:
message: str
@dataclass
class StartMessage(ValidatedDC):
id: int
bearer: str
mode: Literal["normal", "practice"]
type: Literal["start"] = "start"
@dataclass
class EndMessage(ValidatedDC):
type: Literal["end"] = "end"
@dataclass
class NoteOnMessage(ValidatedDC):
time: int
note: int
id: int
type: Literal["note_on"] = "note_on"
@dataclass
class NoteOffMessage(ValidatedDC):
time: int
note: int
id: int
type: Literal["note_off"] = "note_off"
@dataclass
class PauseMessage(ValidatedDC):
paused: bool
time: int
type: Literal["pause"] = "pause"
message_map = {
"start": StartMessage,
"end": EndMessage,
"note_on": NoteOnMessage,
"note_off": NoteOffMessage,
"pause": PauseMessage,
}
def getMessage() -> (
Tuple[
StartMessage
| EndMessage
| NoteOnMessage
| NoteOffMessage
| PauseMessage
| InvalidMessage,
str,
]
):
try:
msg = input()
obj = json.loads(msg)
res = message_map[obj["type"]](**obj)
if is_valid(res):
return res, msg
else:
return InvalidMessage(str(get_errors(res))), msg
except Exception as e:
return InvalidMessage(str(e)), ""
scorometer/main.py
+273 -344
View File
@@ -6,14 +6,20 @@ import operator
import os import os
import select import select
import sys import sys
from dataclasses import dataclass
from typing import Literal, Tuple
import requests import requests
from chroma_case.Key import Key from chroma_case.Key import Key
from chroma_case.Message import (
EndMessage,
InvalidMessage,
NoteOffMessage,
NoteOnMessage,
PauseMessage,
StartMessage,
getMessage,
)
from chroma_case.Partition import Partition from chroma_case.Partition import Partition
from mido import MidiFile from mido import MidiFile
from validated_dc import ValidatedDC, get_errors, is_valid
BACK_URL = os.environ.get("BACK_URL") or "http://back:3000" BACK_URL = os.environ.get("BACK_URL") or "http://back:3000"
MUSICS_FOLDER = os.environ.get("MUSICS_FOLDER") or "/musics/" MUSICS_FOLDER = os.environ.get("MUSICS_FOLDER") or "/musics/"
@@ -27,376 +33,299 @@ NORMAL = 0
PRACTICE = 1 PRACTICE = 1
@dataclass
class InvalidMessage:
message: str
@dataclass
class StartMessage(ValidatedDC):
id: int
bearer: str
mode: Literal["normal", "practice"]
type: Literal["start"] = "start"
@dataclass
class EndMessage(ValidatedDC):
type: Literal["end"] = "end"
@dataclass
class NoteOnMessage(ValidatedDC):
time: int
note: int
id: int
type: Literal["note_on"] = "note_on"
@dataclass
class NoteOffMessage(ValidatedDC):
time: int
note: int
id: int
type: Literal["note_off"] = "note_off"
@dataclass
class PauseMessage(ValidatedDC):
paused: bool
time: int
type: Literal["pause"] = "pause"
message_map = {
"start": StartMessage,
"end": EndMessage,
"note_on": NoteOnMessage,
"note_off": NoteOffMessage,
"pause": PauseMessage,
}
def getMessage() -> (
Tuple[
StartMessage
| EndMessage
| NoteOnMessage
| NoteOffMessage
| PauseMessage
| InvalidMessage,
str,
]
):
try:
msg = input()
obj = json.loads(msg)
res = message_map[obj["type"]](**obj)
if is_valid(res):
return res, msg
else:
return InvalidMessage(str(get_errors(res))), msg
except Exception as e:
return InvalidMessage(str(e)), ""
def send(o): def send(o):
print(json.dumps(o), flush=True) print(json.dumps(o), flush=True)
class Scorometer: class Scorometer:
def __init__(self, mode, midiFile, song_id, user_id) -> None: def __init__(self, mode: int, midiFile: str, song_id: int, user_id: int) -> None:
self.partition = self.getPartition(midiFile) self.partition: Partition = self.getPartition(midiFile)
self.keys_down = [] self.practice_partition: list[list[Key]] = self.getPracticePartition(mode)
self.mode = mode self.keys_down = []
self.song_id = song_id self.mode: int = mode
self.user_id = user_id self.song_id: int = song_id
self.score = 0 self.user_id: int = user_id
self.missed = 0 self.score: int = 0
self.perfect = 0 self.missed: int = 0
self.great = 0 self.perfect: int = 0
self.good = 0 self.great: int = 0
self.difficulties = {} self.good: int = 0
if mode == PRACTICE: self.difficulties = {}
get_start = operator.attrgetter("start")
self.practice_partition = [
list(g)
for _, g in itertools.groupby(
sorted(self.partition.notes, key=get_start), get_start
)
]
else:
self.practice_partition: list[list[Key]] = []
def getPartition(self, midiFile): def getPartition(self, midiFile: str):
notes = [] notes = []
s = 3500 s = 3500
notes_on = {} notes_on = {}
prev_note_on = {} prev_note_on = {}
for msg in MidiFile(midiFile): for msg in MidiFile(midiFile):
d = msg.dict() d = msg.dict()
s += d["time"] * 1000 * RATIO s += d["time"] * 1000 * RATIO
if d["type"] == "note_on": if d["type"] == "note_on":
prev_note_on[d["note"]] = 0 prev_note_on[d["note"]] = 0
if d["note"] in notes_on: if d["note"] in notes_on:
prev_note_on[d["note"]] = notes_on[d["note"]] # 500 prev_note_on[d["note"]] = notes_on[d["note"]] # 500
notes_on[d["note"]] = s # 0 notes_on[d["note"]] = s # 0
if d["type"] == "note_off": if d["type"] == "note_off":
duration = s - notes_on[d["note"]] duration = s - notes_on[d["note"]]
note_start = notes_on[d["note"]] note_start = notes_on[d["note"]]
notes.append(Key(d["note"], note_start, duration - 10)) notes.append(Key(d["note"], note_start, duration - 10))
notes_on[d["note"]] = s # 500 notes_on[d["note"]] = s # 500
return Partition(midiFile, notes) return Partition(midiFile, notes)
def handleNoteOn(self, message: NoteOnMessage): def getPracticePartition(self, mode: int) -> list[list[Key]]:
_key = message.note get_start = operator.attrgetter("start")
timestamp = message.time return (
is_down = any(x[0] == _key for x in self.keys_down) [
if not is_down: list(g)
self.keys_down.append((_key, timestamp)) for _, g in itertools.groupby(
logging.debug({"note": _key}) sorted(self.partition.notes, key=get_start), get_start
)
]
if mode == PRACTICE
else []
)
def handleNoteOff(self, message: NoteOffMessage): def handleNoteOn(self, message: NoteOnMessage):
_key = message.note key = message.note
timestamp = message.time is_down = any(x[0] == key for x in self.keys_down)
down_since = next(since for (h_key, since) in self.keys_down if h_key == _key) if not is_down:
self.keys_down.remove((_key, down_since)) self.keys_down.append((key, message.time))
key = Key(_key, down_since, (timestamp - down_since)) logging.debug({"note": key})
# debug({key: key})
to_play = next(
(
i
for i in self.partition.notes
if i.key == key.key and self.is_timing_close(key, i) and i.done is False
),
None,
)
if to_play is None:
self.score -= 50
logging.info("Invalid key.")
else:
timingScore, timingInformation = self.getTiming(key, to_play)
self.score += (
100
if timingScore == "perfect"
else 75
if timingScore == "great"
else 50
)
to_play.done = True
self.sendScore(message.id, timingScore, timingInformation)
def handleNoteOnPractice(self, message: NoteOnMessage): def handleNoteOff(self, message: NoteOffMessage):
_key = message.note _key = message.note
timestamp = message.time timestamp = message.time
is_down = any(x[0] == _key for x in self.keys_down) down_since = next(since for (h_key, since) in self.keys_down if h_key == _key)
if not is_down: self.keys_down.remove((_key, down_since))
self.keys_down.append((_key, timestamp)) key = Key(_key, down_since, (timestamp - down_since))
logging.debug({"note": _key}) to_play = next(
(
i
for i in self.partition.notes
if i.key == key.key and self.is_timing_close(key, i) and i.done is False
),
None,
)
if to_play is None:
self.score -= 50
logging.info("Invalid key.")
self.sendScore(message.id, "wrong key", "wrong key")
else:
timingScore, timingInformation = self.getTiming(key, to_play)
self.score += (
100
if timingScore == "perfect"
else 75
if timingScore == "great"
else 50
)
to_play.done = True
self.sendScore(message.id, timingScore, timingInformation)
def handleNoteOffPractice(self, message: NoteOffMessage): def handleNoteOnPractice(self, message: NoteOnMessage):
_key = message.note key = message.note
timestamp = message.time is_down = any(x[0] == key for x in self.keys_down)
# is_down = any(x[0] == _key for x in self.keys_down) if not is_down:
down_since = next(since for (h_key, since) in self.keys_down if h_key == _key) self.keys_down.append((key, message.time))
self.keys_down.remove((_key, down_since)) logging.debug({"note": key})
key = Key(_key, down_since, (timestamp - down_since))
keys_to_play = next(
(i for i in self.practice_partition if any(x.done is not True for x in i)),
None,
)
if keys_to_play is None:
logging.info("Key sent but there is no keys to play")
self.score -= 50
return
to_play = next(
(i for i in keys_to_play if i.key == key.key and i.done is not True), None
)
if to_play is None:
self.score -= 50
logging.info("Invalid key.")
else:
timingScore, _ = self.getTiming(key, to_play)
self.score += (
100
if timingScore == "perfect"
else 75
if timingScore == "great"
else 50
)
to_play.done = True
self.sendScore(message.id, timingScore, "practice")
def getTiming(self, key: Key, to_play: Key): def handleNoteOffPractice(self, message: NoteOffMessage):
return self.getTimingScore(key, to_play), self.getTimingInfo(key, to_play) _key = message.note
down_since = next(since for (h_key, since) in self.keys_down if h_key == _key)
self.keys_down.remove((_key, down_since))
key = Key(_key, down_since, (message.time - down_since))
keys_to_play = next(
(i for i in self.practice_partition if any(x.done is not True for x in i)),
None,
)
if keys_to_play is None:
logging.info("Invalid key.")
self.score -= 50
self.sendScore(message.id, "wrong key", "wrong key")
return
to_play = next(
(i for i in keys_to_play if i.key == key.key and i.done is not True), None
)
if to_play is None:
logging.info(f"Wrong key sent: {message.note} with id {message.id}")
self.score -= 50
self.sendScore(message.id, "wrong key", "wrong key")
else:
timingScore, _ = self.getTiming(key, to_play)
self.score += (
100
if timingScore == "perfect"
else 75
if timingScore == "great"
else 50
)
to_play.done = True
self.sendScore(message.id, timingScore, "practice")
def getTimingScore(self, key: Key, to_play: Key): def getTiming(self, key: Key, to_play: Key):
tempo_percent = abs((key.duration / to_play.duration) - 1) return self.getTimingScore(key, to_play), self.getTimingInfo(key, to_play)
if tempo_percent < 0.3:
timingScore = "perfect"
elif tempo_percent < 0.5:
timingScore = "great"
else:
timingScore = "good"
return timingScore
def getTimingInfo(self, key: Key, to_play: Key): def getTimingScore(self, key: Key, to_play: Key):
return ( tempo_percent = abs((key.duration / to_play.duration) - 1)
"perfect" if tempo_percent < 0.3:
if abs(key.start - to_play.start) < 200 self.perfect += 1
else "fast" timingScore = "perfect"
if key.start < to_play.start elif tempo_percent < 0.5:
else "late" self.great += 1
) timingScore = "great"
else:
self.good += 1
timingScore = "good"
return timingScore
# is it in the 500 ms range def getTimingInfo(self, key: Key, to_play: Key):
def is_timing_close(self, key: Key, i: Key): return (
return abs(i.start - key.start) < 500 "perfect"
if abs(key.start - to_play.start) < 200
else "fast"
if key.start < to_play.start
else "late"
)
def handleMessage( # is it in the 500 ms range
self, def is_timing_close(self, key: Key, i: Key):
message: StartMessage return abs(i.start - key.start) < 500
| EndMessage
| NoteOnMessage
| NoteOffMessage
| PauseMessage
| InvalidMessage,
line: str,
):
match message:
case InvalidMessage(error):
logging.warning(f"Invalid message {line} with error: {error}")
send({"error": f"Invalid message {line} with error: {error}"})
case NoteOnMessage():
if self.mode == NORMAL:
self.handleNoteOn(message)
elif self.mode == PRACTICE:
self.handleNoteOnPractice(message)
case NoteOffMessage():
if self.mode == NORMAL:
self.handleNoteOff(message)
elif self.mode == PRACTICE:
self.handleNoteOffPractice(message)
case PauseMessage():
pass
case EndMessage():
self.endGame()
case _:
logging.warning(
f"Expected note_on note_off or pause message but got {message.type} instead"
)
def sendScore(self, id, timingScore, timingInformation): def handleMessage(
send( self,
{ message: StartMessage
"id": id, | EndMessage
"timingScore": timingScore, | NoteOnMessage
"timingInformation": timingInformation, | NoteOffMessage
} | PauseMessage
) | InvalidMessage,
line: str,
):
match message:
case InvalidMessage(error):
logging.warning(f"Invalid message {line} with error: {error}")
send({"error": f"Invalid message {line} with error: {error}"})
case NoteOnMessage():
if self.mode == NORMAL:
self.handleNoteOn(message)
elif self.mode == PRACTICE:
self.handleNoteOnPractice(message)
case NoteOffMessage():
if self.mode == NORMAL:
self.handleNoteOff(message)
elif self.mode == PRACTICE:
self.handleNoteOffPractice(message)
case PauseMessage():
pass
case EndMessage():
self.endGame()
case _:
logging.warning(
f"Expected note_on note_off or pause message but got {message.type} instead"
)
def gameLoop(self): def sendScore(self, id, timingScore, timingInformation):
while True: send(
if select.select( {
[ "id": id,
sys.stdin, "timingScore": timingScore,
], "timingInformation": timingInformation,
[], }
[], )
0.0,
)[0]:
message, line = getMessage()
logging.info(f"handling message {line}")
self.handleMessage(message, line)
else:
pass
def endGame(self): def gameLoop(self):
for i in self.partition.notes: while True:
if i.done is False: message, line = getMessage()
self.score -= 50 logging.info(f"handling message {line}")
send( self.handleMessage(message, line)
{
"overallScore": self.score, def endGame(self):
"score": { for i in self.partition.notes:
"missed": self.missed, if i.done is False:
"good": self.good, self.score -= 50
"great": self.great, self.missed += 1
"perfect": self.perfect, send(
"maxScore": len(self.partition.notes) * 100, {
}, "overallScore": self.score,
} "score": {
) "missed": self.missed,
if self.user_id != -1: "good": self.good,
requests.post( "great": self.great,
f"{BACK_URL}/history", "perfect": self.perfect,
json={ "maxScore": len(self.partition.notes) * 100,
"songID": self.song_id, },
"userID": self.user_id, }
"score": self.score, )
"difficulties": self.difficulties, if self.user_id != -1:
}, requests.post(
) f"{BACK_URL}/history",
exit() json={
"songID": self.song_id,
"userID": self.user_id,
"score": self.score,
"difficulties": self.difficulties,
},
)
exit()
def handleStartMessage(start_message: StartMessage): def handleStartMessage(start_message: StartMessage):
mode = PRACTICE if start_message.mode == "practice" else NORMAL mode = PRACTICE if start_message.mode == "practice" else NORMAL
song_id = start_message.id song_id = start_message.id
user_id = -1 user_id = -1
try: try:
if start_message.bearer != "": if start_message.bearer != "":
r = requests.get( r = requests.get(
f"{BACK_URL}/auth/me", f"{BACK_URL}/auth/me",
headers={"Authorization": f"Bearer {start_message.bearer}"}, headers={"Authorization": f"Bearer {start_message.bearer}"},
) )
r.raise_for_status() r.raise_for_status()
user_id = r.json()["id"] user_id = r.json()["id"]
except Exception as e: except Exception as e:
logging.fatal("Could not get user id with given bearer", exc_info=e) logging.fatal("Could not get user id with given bearer", exc_info=e)
send({"error": "Could not get user id with given bearer"}) send({"error": "Could not get user id with given bearer"})
exit() exit()
try: try:
r = requests.get(f"{BACK_URL}/song/{song_id}") r = requests.get(f"{BACK_URL}/song/{song_id}")
r.raise_for_status() r.raise_for_status()
song_path = r.json()["midiPath"] song_path = r.json()["midiPath"]
song_path = song_path.replace("/musics/", MUSICS_FOLDER) song_path = song_path.replace("/musics/", MUSICS_FOLDER)
except Exception as e: except Exception as e:
logging.fatal("Invalid song id", exc_info=e) logging.fatal("Invalid song id", exc_info=e)
send({"error": "Invalid song id, song does not exist"}) send({"error": "Invalid song id, song does not exist"})
exit() exit()
return mode, song_path, song_id, user_id return mode, song_path, song_id, user_id
def startGame(start_message: StartMessage): def startGame(start_message: StartMessage):
mode, song_path, song_id, user_id = handleStartMessage(start_message) mode, song_path, song_id, user_id = handleStartMessage(start_message)
sc = Scorometer(mode, song_path, song_id, user_id) sc = Scorometer(mode, song_path, song_id, user_id)
sc.gameLoop() sc.gameLoop()
def main(): def main():
try: try:
msg, _ = getMessage() msg, _ = getMessage()
match msg: match msg:
case StartMessage(): case StartMessage():
startGame(msg) startGame(msg)
case EndMessage(): case EndMessage():
logging.info("scorometer ended before a start message") logging.info("scorometer ended before a start message")
send({"error": "Did not receive a start message"}) send({"error": "Did not receive a start message"})
exit() exit()
case InvalidMessage(error): case InvalidMessage(error):
logging.warning(f"invalid message with error: {error}") logging.warning(f"invalid message with error: {error}")
send({"error": "Invalid input, expected a start message"}) send({"error": "Invalid input, expected a start message"})
case _: case _:
logging.warning(f"invalid message with type: {msg.type}") logging.warning(f"invalid message with type: {msg.type}")
send({"error": "Invalid input, expected a start message"}) send({"error": "Invalid input, expected a start message"})
except Exception as e: except Exception as e:
logging.fatal("error", exc_info=e) logging.fatal("error", exc_info=e)
send({"error": "a fatal error occured"}) send({"error": "a fatal error occured"})
if __name__ == "__main__": if __name__ == "__main__":
main() main()