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

format

Browse Source
This commit is contained in:
GitBluub
2023-04-02 23:12:32 +09:00
parent 437d5c7b5c
commit 445b949fa8
4 changed files with 326 additions and 329 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scorometer/chroma_case/Key.py
+7 -9
View File
@@ -1,11 +1,9 @@
class Key: class Key:
def __init__(self, key: int, start: int, duration: int): def __init__(self, key: int, start: int, duration: int):
self.key = key self.key = key
self.start = start self.start = start
self.duration = duration self.duration = duration
self.done = False self.done = False
def __repr__(self):
return f"{self.key} ({self.start} - {self.duration})"
def __repr__(self):
return f"{self.key} ({self.start} - {self.duration})"
scorometer/chroma_case/Note.py
+7 -7
View File
@@ -1,11 +1,11 @@
class Note: class Note:
def __init__(self, start_time, data) -> None: def __init__(self, start_time, data) -> None:
self.__start_time = start_time self.__start_time = start_time
self.__data = data self.__data = data
def get_start_time(self): def get_start_time(self):
return self.__start_time return self.__start_time
def get_data(self): def get_data(self):
return self.__data return self.__data
scorometer/chroma_case/Partition.py
+2 -3
View File
@@ -1,8 +1,8 @@
from .Key import Key from .Key import Key
class Partition:
def __init__(self, name:str, notes:list[Key]) -> None: class Partition:
def __init__(self, name: str, notes: list[Key]) -> None:
self.__name = name self.__name = name
self.notes = notes self.notes = notes
@@ -12,4 +12,3 @@ class Partition:
for i in self.notes: for i in self.notes:
r += f"{i.__repr__()}\n" r += f"{i.__repr__()}\n"
return r return r
scorometer/main.py
+309 -309
View File
@@ -29,374 +29,374 @@ PRACTICE = 1
@dataclass @dataclass
class InvalidMessage: class InvalidMessage:
message: str message: str
@dataclass @dataclass
class StartMessage(ValidatedDC): class StartMessage(ValidatedDC):
id: int id: int
bearer: str bearer: str
mode: Literal["normal", "practice"] mode: Literal["normal", "practice"]
type: Literal["start"] = "start" type: Literal["start"] = "start"
@dataclass @dataclass
class EndMessage(ValidatedDC): class EndMessage(ValidatedDC):
type: Literal["end"] = "end" type: Literal["end"] = "end"
@dataclass @dataclass
class NoteOnMessage(ValidatedDC): class NoteOnMessage(ValidatedDC):
time: int time: int
note: int note: int
id: int id: int
type: Literal["note_on"] = "note_on" type: Literal["note_on"] = "note_on"
@dataclass @dataclass
class NoteOffMessage(ValidatedDC): class NoteOffMessage(ValidatedDC):
time: int time: int
note: int note: int
id: int id: int
type: Literal["note_off"] = "note_off" type: Literal["note_off"] = "note_off"
@dataclass @dataclass
class PauseMessage(ValidatedDC): class PauseMessage(ValidatedDC):
paused: bool paused: bool
time: int time: int
type: Literal["pause"] = "pause" type: Literal["pause"] = "pause"
message_map = { message_map = {
"start": StartMessage, "start": StartMessage,
"end": EndMessage, "end": EndMessage,
"note_on": NoteOnMessage, "note_on": NoteOnMessage,
"note_off": NoteOffMessage, "note_off": NoteOffMessage,
"pause": PauseMessage, "pause": PauseMessage,
} }
def getMessage() -> ( def getMessage() -> (
Tuple[ Tuple[
StartMessage StartMessage
| EndMessage | EndMessage
| NoteOnMessage | NoteOnMessage
| NoteOffMessage | NoteOffMessage
| PauseMessage | PauseMessage
| InvalidMessage, | InvalidMessage,
str, str,
] ]
): ):
try: try:
msg = input() msg = input()
obj = json.loads(msg) obj = json.loads(msg)
res = message_map[obj["type"]](**obj) res = message_map[obj["type"]](**obj)
if is_valid(res): if is_valid(res):
return res, msg return res, msg
else: else:
return InvalidMessage(str(get_errors(res))), msg return InvalidMessage(str(get_errors(res))), msg
except Exception as e: except Exception as e:
return InvalidMessage(str(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, midiFile, song_id, user_id) -> None:
self.partition = self.getPartition(midiFile) self.partition = self.getPartition(midiFile)
self.keys_down = [] self.keys_down = []
self.mode = mode self.mode = mode
self.song_id = song_id self.song_id = song_id
self.user_id = user_id self.user_id = user_id
self.score = 0 self.score = 0
self.missed = 0 self.missed = 0
self.perfect = 0 self.perfect = 0
self.great = 0 self.great = 0
self.good = 0 self.good = 0
self.difficulties = {} self.difficulties = {}
if mode == PRACTICE: if mode == PRACTICE:
get_start = operator.attrgetter("start") get_start = operator.attrgetter("start")
self.practice_partition = [ self.practice_partition = [
list(g) list(g)
for _, g in itertools.groupby( for _, g in itertools.groupby(
sorted(self.partition.notes, key=get_start), get_start sorted(self.partition.notes, key=get_start), get_start
) )
] ]
else: else:
self.practice_partition: list[list[Key]] = [] self.practice_partition: list[list[Key]] = []
def getPartition(self, midiFile): def getPartition(self, midiFile):
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 handleNoteOn(self, message: NoteOnMessage):
_key = message.note _key = message.note
timestamp = message.time 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: if not is_down:
self.keys_down.append((_key, timestamp)) self.keys_down.append((_key, timestamp))
logging.debug({"note": _key}) logging.debug({"note": _key})
def handleNoteOff(self, message: NoteOffMessage): def handleNoteOff(self, message: NoteOffMessage):
_key = message.note _key = message.note
timestamp = message.time timestamp = message.time
down_since = next(since for (h_key, since) in self.keys_down if h_key == _key) down_since = next(since for (h_key, since) in self.keys_down if h_key == _key)
self.keys_down.remove((_key, down_since)) self.keys_down.remove((_key, down_since))
key = Key(_key, down_since, (timestamp - down_since)) key = Key(_key, down_since, (timestamp - down_since))
# debug({key: key}) # debug({key: key})
to_play = next( to_play = next(
( (
i i
for i in self.partition.notes for i in self.partition.notes
if i.key == key.key and self.is_timing_close(key, i) and i.done is False if i.key == key.key and self.is_timing_close(key, i) and i.done is False
), ),
None, None,
) )
if to_play is None: if to_play is None:
self.score -= 50 self.score -= 50
logging.info("Invalid key.") logging.info("Invalid key.")
else: else:
timingScore, timingInformation = self.getTiming(key, to_play) timingScore, timingInformation = self.getTiming(key, to_play)
self.score += ( self.score += (
100 100
if timingScore == "perfect" if timingScore == "perfect"
else 75 else 75
if timingScore == "great" if timingScore == "great"
else 50 else 50
) )
to_play.done = True to_play.done = True
self.sendScore(message.id, timingScore, timingInformation) self.sendScore(message.id, timingScore, timingInformation)
def handleNoteOnPractice(self, message: NoteOnMessage): def handleNoteOnPractice(self, message: NoteOnMessage):
_key = message.note _key = message.note
timestamp = message.time 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: if not is_down:
self.keys_down.append((_key, timestamp)) self.keys_down.append((_key, timestamp))
logging.debug({"note": _key}) logging.debug({"note": _key})
def handleNoteOffPractice(self, message: NoteOffMessage): def handleNoteOffPractice(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) # 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) down_since = next(since for (h_key, since) in self.keys_down if h_key == _key)
self.keys_down.remove((_key, down_since)) self.keys_down.remove((_key, down_since))
key = Key(_key, down_since, (timestamp - down_since)) key = Key(_key, down_since, (timestamp - down_since))
keys_to_play = next( keys_to_play = next(
(i for i in self.practice_partition if any(x.done is not True for x in i)), (i for i in self.practice_partition if any(x.done is not True for x in i)),
None, None,
) )
if keys_to_play is None: if keys_to_play is None:
logging.info("Key sent but there is no keys to play") logging.info("Key sent but there is no keys to play")
self.score -= 50 self.score -= 50
return return
to_play = next( to_play = next(
(i for i in keys_to_play if i.key == key.key and i.done is not True), None (i for i in keys_to_play if i.key == key.key and i.done is not True), None
) )
if to_play is None: if to_play is None:
self.score -= 50 self.score -= 50
logging.info("Invalid key.") logging.info("Invalid key.")
else: else:
timingScore, _ = self.getTiming(key, to_play) timingScore, _ = self.getTiming(key, to_play)
self.score += ( self.score += (
100 100
if timingScore == "perfect" if timingScore == "perfect"
else 75 else 75
if timingScore == "great" if timingScore == "great"
else 50 else 50
) )
to_play.done = True to_play.done = True
self.sendScore(message.id, timingScore, "practice") self.sendScore(message.id, timingScore, "practice")
def getTiming(self, key: Key, to_play: Key): def getTiming(self, key: Key, to_play: Key):
return self.getTimingScore(key, to_play), self.getTimingInfo(key, to_play) return self.getTimingScore(key, to_play), self.getTimingInfo(key, to_play)
def getTimingScore(self, key: Key, to_play: Key): def getTimingScore(self, key: Key, to_play: Key):
tempo_percent = abs((key.duration / to_play.duration) - 1) tempo_percent = abs((key.duration / to_play.duration) - 1)
if tempo_percent < 0.3: if tempo_percent < 0.3:
timingScore = "perfect" timingScore = "perfect"
elif tempo_percent < 0.5: elif tempo_percent < 0.5:
timingScore = "great" timingScore = "great"
else: else:
timingScore = "good" timingScore = "good"
return timingScore return timingScore
def getTimingInfo(self, key: Key, to_play: Key): def getTimingInfo(self, key: Key, to_play: Key):
return ( return (
"perfect" "perfect"
if abs(key.start - to_play.start) < 200 if abs(key.start - to_play.start) < 200
else "fast" else "fast"
if key.start < to_play.start if key.start < to_play.start
else "late" else "late"
) )
# is it in the 500 ms range # is it in the 500 ms range
def is_timing_close(self, key: Key, i: Key): def is_timing_close(self, key: Key, i: Key):
return abs(i.start - key.start) < 500 return abs(i.start - key.start) < 500
def handleMessage( def handleMessage(
self, self,
message: StartMessage message: StartMessage
| EndMessage | EndMessage
| NoteOnMessage | NoteOnMessage
| NoteOffMessage | NoteOffMessage
| PauseMessage | PauseMessage
| InvalidMessage, | InvalidMessage,
line: str, line: str,
): ):
match message: match message:
case InvalidMessage(error): case InvalidMessage(error):
logging.warning(f"Invalid message {line} with error: {error}") logging.warning(f"Invalid message {line} with error: {error}")
send({"error": f"Invalid message {line} with error: {error}"}) send({"error": f"Invalid message {line} with error: {error}"})
case NoteOnMessage(): case NoteOnMessage():
if self.mode == NORMAL: if self.mode == NORMAL:
self.handleNoteOn(message) self.handleNoteOn(message)
elif self.mode == PRACTICE: elif self.mode == PRACTICE:
self.handleNoteOnPractice(message) self.handleNoteOnPractice(message)
case NoteOffMessage(): case NoteOffMessage():
if self.mode == NORMAL: if self.mode == NORMAL:
self.handleNoteOff(message) self.handleNoteOff(message)
elif self.mode == PRACTICE: elif self.mode == PRACTICE:
self.handleNoteOffPractice(message) self.handleNoteOffPractice(message)
case PauseMessage(): case PauseMessage():
pass pass
case EndMessage(): case EndMessage():
self.endGame() self.endGame()
case _: case _:
logging.warning( logging.warning(
f"Expected note_on note_off or pause message but got {message.type} instead" f"Expected note_on note_off or pause message but got {message.type} instead"
) )
def sendScore(self, id, timingScore, timingInformation): def sendScore(self, id, timingScore, timingInformation):
send( send(
{ {
"id": id, "id": id,
"timingScore": timingScore, "timingScore": timingScore,
"timingInformation": timingInformation, "timingInformation": timingInformation,
} }
) )
def gameLoop(self): def gameLoop(self):
while True: while True:
if select.select( if select.select(
[ [
sys.stdin, sys.stdin,
], ],
[], [],
[], [],
0.0, 0.0,
)[0]: )[0]:
message, line = getMessage() message, line = getMessage()
logging.info(f"handling message {line}") logging.info(f"handling message {line}")
self.handleMessage(message, line) self.handleMessage(message, line)
else: else:
pass pass
def endGame(self): def endGame(self):
for i in self.partition.notes: for i in self.partition.notes:
if i.done is False: if i.done is False:
self.score -= 50 self.score -= 50
send( send(
{ {
"overallScore": self.score, "overallScore": self.score,
"score": { "score": {
"missed": self.missed, "missed": self.missed,
"good": self.good, "good": self.good,
"great": self.great, "great": self.great,
"perfect": self.perfect, "perfect": self.perfect,
"maxScore": len(self.partition.notes) * 100, "maxScore": len(self.partition.notes) * 100,
}, },
} }
) )
if self.user_id != -1: if self.user_id != -1:
requests.post( requests.post(
f"{BACK_URL}/history", f"{BACK_URL}/history",
json={ json={
"songID": self.song_id, "songID": self.song_id,
"userID": self.user_id, "userID": self.user_id,
"score": self.score, "score": self.score,
"difficulties": self.difficulties, "difficulties": self.difficulties,
}, },
) )
exit() 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()