Front: Pull Scorometer
This commit is contained in:
@@ -7,72 +7,72 @@ from validated_dc import ValidatedDC, get_errors, is_valid
|
||||
|
||||
@dataclass
|
||||
class InvalidMessage:
|
||||
message: str
|
||||
message: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class StartMessage(ValidatedDC):
|
||||
id: int
|
||||
bearer: str
|
||||
mode: Literal["normal", "practice"]
|
||||
type: Literal["start"] = "start"
|
||||
id: int
|
||||
bearer: str
|
||||
mode: Literal["normal", "practice"]
|
||||
type: Literal["start"] = "start"
|
||||
|
||||
|
||||
@dataclass
|
||||
class EndMessage(ValidatedDC):
|
||||
type: Literal["end"] = "end"
|
||||
type: Literal["end"] = "end"
|
||||
|
||||
|
||||
@dataclass
|
||||
class NoteOnMessage(ValidatedDC):
|
||||
time: int
|
||||
note: int
|
||||
id: int
|
||||
type: Literal["note_on"] = "note_on"
|
||||
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"
|
||||
time: int
|
||||
note: int
|
||||
id: int
|
||||
type: Literal["note_off"] = "note_off"
|
||||
|
||||
|
||||
@dataclass
|
||||
class PauseMessage(ValidatedDC):
|
||||
paused: bool
|
||||
time: int
|
||||
type: Literal["pause"] = "pause"
|
||||
paused: bool
|
||||
time: int
|
||||
type: Literal["pause"] = "pause"
|
||||
|
||||
|
||||
message_map = {
|
||||
"start": StartMessage,
|
||||
"end": EndMessage,
|
||||
"note_on": NoteOnMessage,
|
||||
"note_off": NoteOffMessage,
|
||||
"pause": PauseMessage,
|
||||
"start": StartMessage,
|
||||
"end": EndMessage,
|
||||
"note_on": NoteOnMessage,
|
||||
"note_off": NoteOffMessage,
|
||||
"pause": PauseMessage,
|
||||
}
|
||||
|
||||
|
||||
def getMessage() -> (
|
||||
Tuple[
|
||||
StartMessage
|
||||
| EndMessage
|
||||
| NoteOnMessage
|
||||
| NoteOffMessage
|
||||
| PauseMessage
|
||||
| InvalidMessage,
|
||||
str,
|
||||
]
|
||||
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)), ""
|
||||
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)), ""
|
||||
@@ -4,8 +4,8 @@ import json
|
||||
import logging
|
||||
import operator
|
||||
import os
|
||||
import select
|
||||
import sys
|
||||
from typing import TypedDict
|
||||
|
||||
import requests
|
||||
from chroma_case.Key import Key
|
||||
@@ -33,6 +33,15 @@ NORMAL = 0
|
||||
PRACTICE = 1
|
||||
|
||||
|
||||
class ScoroInfo(TypedDict):
|
||||
max_score: int
|
||||
score: int
|
||||
missed: int
|
||||
perfect: int
|
||||
great: int
|
||||
good: int
|
||||
|
||||
|
||||
def send(o):
|
||||
print(json.dumps(o), flush=True)
|
||||
|
||||
@@ -45,13 +54,20 @@ class Scorometer:
|
||||
self.mode: int = mode
|
||||
self.song_id: int = song_id
|
||||
self.user_id: int = user_id
|
||||
self.score: int = 0
|
||||
self.missed: int = 0
|
||||
self.perfect: int = 0
|
||||
self.great: int = 0
|
||||
self.good: int = 0
|
||||
self.wrong_ids = []
|
||||
self.difficulties = {}
|
||||
self.info: ScoroInfo = {
|
||||
"max_score": len(self.partition.notes) * 100,
|
||||
"score": 0,
|
||||
"missed": 0,
|
||||
"perfect": 0,
|
||||
"great": 0,
|
||||
"good": 0,
|
||||
}
|
||||
|
||||
def send(self, obj):
|
||||
obj["info"] = self.info
|
||||
send(obj)
|
||||
|
||||
def getPartition(self, midiFile: str):
|
||||
notes = []
|
||||
@@ -106,12 +122,12 @@ class Scorometer:
|
||||
if to_play:
|
||||
perf = self.getTimingScore(key, to_play)
|
||||
logging.debug({"note_on": f"{perf} on {message.note}"})
|
||||
send({"type": "timing", "id": message.id, "timing": perf})
|
||||
self.send({"type": "timing", "id": message.id, "timing": perf})
|
||||
else:
|
||||
self.score -= 50
|
||||
self.info["score"] -= 50
|
||||
self.wrong_ids += [message.id]
|
||||
logging.debug({"note_on": f"wrong key {message.note}"})
|
||||
send({"type": "timing", "id": message.id, "timing": "wrong"})
|
||||
self.send({"type": "timing", "id": message.id, "timing": "wrong"})
|
||||
|
||||
def handleNoteOff(self, message: NoteOffMessage):
|
||||
logging.debug({"note_off": message.note})
|
||||
@@ -121,7 +137,7 @@ class Scorometer:
|
||||
self.keys_down.remove((message.note, down_since))
|
||||
if message.id in self.wrong_ids:
|
||||
logging.debug({"note_off": f"wrong key {message.note}"})
|
||||
send({"type": "duration", "id": message.id, "duration": "wrong"})
|
||||
self.send({"type": "duration", "id": message.id, "duration": "wrong"})
|
||||
return
|
||||
key = Key(
|
||||
key=message.note, start=down_since, duration=(message.time - down_since)
|
||||
@@ -136,7 +152,7 @@ class Scorometer:
|
||||
)
|
||||
if to_play:
|
||||
perf = self.getDurationScore(key, to_play)
|
||||
self.score += (
|
||||
self.info["score"] += (
|
||||
100
|
||||
if perf == "perfect"
|
||||
else 75
|
||||
@@ -145,7 +161,7 @@ class Scorometer:
|
||||
)
|
||||
to_play.done = True
|
||||
logging.debug({"note_off": f"{perf} on {message.note}"})
|
||||
send({"type": "duration", "id": message.id, "duration": perf})
|
||||
self.send({"type": "duration", "id": message.id, "duration": perf})
|
||||
else:
|
||||
logging.warning("note_off: no key to play but it was not a wrong note_on")
|
||||
|
||||
@@ -161,7 +177,7 @@ class Scorometer:
|
||||
None,
|
||||
)
|
||||
if keys_to_play is None:
|
||||
send({"type": "error", "error": "no keys should be played"})
|
||||
self.send({"type": "error", "error": "no keys should be played"})
|
||||
return
|
||||
to_play = next(
|
||||
(i for i in keys_to_play if i.key == key.key and i.done is not True), None
|
||||
@@ -169,11 +185,11 @@ class Scorometer:
|
||||
if to_play:
|
||||
perf = "practice"
|
||||
logging.debug({"note_on": f"{perf} on {message.note}"})
|
||||
send({"type": "timing", "id": message.id, "timing": perf})
|
||||
self.send({"type": "timing", "id": message.id, "timing": perf})
|
||||
else:
|
||||
self.wrong_ids += [message.id]
|
||||
logging.debug({"note_on": f"wrong key {message.note}"})
|
||||
send({"type": "timing", "id": message.id, "timing": "wrong"})
|
||||
self.send({"type": "timing", "id": message.id, "timing": "wrong"})
|
||||
|
||||
def handleNoteOffPractice(self, message: NoteOffMessage):
|
||||
logging.debug({"note_off": message.note})
|
||||
@@ -183,7 +199,7 @@ class Scorometer:
|
||||
self.keys_down.remove((message.note, down_since))
|
||||
if message.id in self.wrong_ids:
|
||||
logging.debug({"note_off": f"wrong key {message.note}"})
|
||||
send({"type": "duration", "id": message.id, "duration": "wrong"})
|
||||
self.send({"type": "duration", "id": message.id, "duration": "wrong"})
|
||||
return
|
||||
key = Key(
|
||||
key=message.note, start=down_since, duration=(message.time - down_since)
|
||||
@@ -194,8 +210,9 @@ class Scorometer:
|
||||
)
|
||||
if keys_to_play is None:
|
||||
logging.info("Invalid key.")
|
||||
self.score -= 50
|
||||
self.sendScore(message.id, "wrong key", "wrong key")
|
||||
self.info["score"] -= 50
|
||||
# TODO: I dont think this if is right
|
||||
# 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
|
||||
@@ -204,9 +221,9 @@ class Scorometer:
|
||||
perf = "practice"
|
||||
to_play.done = True
|
||||
logging.debug({"note_off": f"{perf} on {message.note}"})
|
||||
send({"type": "duration", "id": message.id, "duration": perf})
|
||||
self.send({"type": "duration", "id": message.id, "duration": perf})
|
||||
else:
|
||||
send({"type": "duration", "id": message.id, "duration": "wrong"})
|
||||
self.send({"type": "duration", "id": message.id, "duration": "wrong"})
|
||||
|
||||
def getDurationScore(self, key: Key, to_play: Key):
|
||||
tempo_percent = abs((key.duration / to_play.duration) - 1)
|
||||
@@ -244,7 +261,7 @@ class Scorometer:
|
||||
match message:
|
||||
case InvalidMessage(error):
|
||||
logging.warning(f"Invalid message {line} with error: {error}")
|
||||
send({"error": f"Invalid message {line} with error: {error}"})
|
||||
self.send({"error": f"Invalid message {line} with error: {error}"})
|
||||
case NoteOnMessage():
|
||||
if self.mode == NORMAL:
|
||||
self.handleNoteOn(message)
|
||||
@@ -264,15 +281,6 @@ class Scorometer:
|
||||
f"Expected note_on note_off or pause message but got {message.type} instead"
|
||||
)
|
||||
|
||||
def sendScore(self, id, timingScore, timingInformation):
|
||||
send(
|
||||
{
|
||||
"id": id,
|
||||
"timingScore": timingScore,
|
||||
"timingInformation": timingInformation,
|
||||
}
|
||||
)
|
||||
|
||||
def gameLoop(self):
|
||||
while True:
|
||||
message, line = getMessage()
|
||||
@@ -282,16 +290,16 @@ class Scorometer:
|
||||
def endGame(self):
|
||||
for i in self.partition.notes:
|
||||
if i.done is False:
|
||||
self.score -= 50
|
||||
self.missed += 1
|
||||
self.info["score"] -= 50
|
||||
self.info["missed"] += 1
|
||||
send(
|
||||
{
|
||||
"overallScore": self.score,
|
||||
"overallScore": self.info["score"],
|
||||
"score": {
|
||||
"missed": self.missed,
|
||||
"good": self.good,
|
||||
"great": self.great,
|
||||
"perfect": self.perfect,
|
||||
"missed": self.info["missed"],
|
||||
"good": self.info["good"],
|
||||
"great": self.info["great"],
|
||||
"perfect": self.info["perfect"],
|
||||
"maxScore": len(self.partition.notes) * 100,
|
||||
},
|
||||
}
|
||||
@@ -302,7 +310,7 @@ class Scorometer:
|
||||
json={
|
||||
"songID": self.song_id,
|
||||
"userID": self.user_id,
|
||||
"score": self.score,
|
||||
"score": self.info["score"],
|
||||
"difficulties": self.difficulties,
|
||||
},
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user