handleNote basics

scorometer/chroma_case/Key.py

@@ -0,0 +1,10 @@
+class Key:
+    def __init__(self, key: int, start: int, duration: int):
+        self.key = key
+        self.start = start
+        self.duration = duration
+
+    def __str__(self):
+        return f"{self.key} ({self.start} - {self.duration})"
+
+

scorometer/chroma_case/Note.py

@@ -1,5 +1,3 @@
-
-
 class Note:
     def __init__(self, start_time, data) -> None:
 
@@ -10,4 +8,4 @@ class Note:
         return self.__start_time
     
     def get_data(self):
-        return self.__data
+        return self.__data

scorometer/chroma_case/Partition.py

@@ -1,40 +1,15 @@
-import asyncio, datetime
-from typing import Callable
-
-from .Note import Note
-
-async def wait_until(dt):
-	# sleep until the specified datetime
-	now = datetime.datetime.now()
-	await asyncio.sleep((dt - now).total_seconds())
-
-async def run_at(dt, coro):
-	await wait_until(dt)
-	return await coro
+from .Key import Key
 
 class Partition:
 
-	def __init__(self, name:str, notes:list[Note]) -> None:
+	def __init__(self, name:str, notes:list[Key]) -> None:
 
 		self.__name = name
 		self.__notes = notes
 
-	async def play(self, output_lambda:Callable[[object], None]):
-		now = datetime.datetime.now()
-		tasks_to_wait = []
-		for note in self.__notes:
-			tasks_to_wait.append(
-					asyncio.create_task(
-						run_at(
-							now + datetime.timedelta(milliseconds= note.get_start_time()),
-							output_lambda(note.get_data())
-							)
-						)
-					)
-		await asyncio.wait(tasks_to_wait)
 	def __repr__(self):
 		r = f"{self.__name}\n"
 		for i in self.__notes:
-			r += f"{i.get_data()}\n"
+			r += f"{i.__repr__()}\n"
 		return r
 

scorometer/main.py

@@ -1,7 +1,7 @@
 #!/usr/bin/python3
 from time import sleep
 from chroma_case.Partition import Partition
-from chroma_case.Note import Note
+from chroma_case.Key import Key
 import sys
 import select
 import json
@@ -10,21 +10,19 @@ from mido import MidiFile
 RATIO = float(sys.argv[2] if len(sys.argv) > 2 else 1)
 OCTAVE = 5
 OCTAVE_AMOUNT_KEYS = 12
-TRANSPOSE_AMOUNT = OCTAVE_AMOUNT_KEYS * OCTAVE
 
 class Scorometer():
 	def __init__(self, midiFile) -> None:
 		self.partition = self.getPartition(midiFile)
+		self.keys_down = []
 		pass
 	def getPartition(self, midiFile):
 		notes = []
-		# notes will start to play at 3500 ms 
 		s = 3500
 		notes_on = {}
 		prev_note_on = {}
 		for msg in MidiFile(midiFile):
 			d = msg.dict()
-#		print(msg, s)
 			s += d['time'] * 1000 * RATIO
 
 			if d["type"] == "note_on":
@@ -38,15 +36,49 @@ class Scorometer():
 				duration = s - notes_on[d["note"]]
 				note_start = notes_on[d["note"]]
 				# time value is only used during debug
-				notes.append(Note(note_start, {
-					"time": note_start,
-					"duration": duration - 10,
-					#		"color": default_color if note_color[d["note"]] else (255, 100, 0),
-					"key": d["note"]
-					}))
+				notes.append(Key(d["note"], note_start, duration - 10))
 				notes_on[d["note"]] = s # 500
 		return Partition(midiFile, notes)
 
+	def handleNote(self, obj):
+		_key = obj["note"]
+		status = obj["type"]
+		timestamp = obj["time"]
+		is_down = any(x[0] == _key for x in self.keys_down)
+		key = None
+		if status == "note_on" and not is_down:
+			self.keys_down.append((_key, timestamp))
+			# print(f"Midi: {status} - {key} - {intensity} - {data3} at {timestamp}")
+		elif status == "note_off" or is_down:
+			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, (timestamp - down_since))
+		if key is None:
+			return
+		to_play = next((i for i in self.partition.__notes if i.key == key.key and self.is_timing_close(key, i)), None)
+		if to_play == None:
+			pass
+			## TODO handle invalid key 
+			#points -= 50
+			#print(f"Invalid key.")
+		else:
+			tempo_percent = abs((key.duration / to_play.duration) - 1)
+			#points += tempo_percent * 50
+			if tempo_percent < .3 : 
+				timingScore = "good"
+			elif tempo_percent < .5:
+				timingScore = f"great"
+			else:
+				timingScore = "perfect"
+			
+			timingInformation = "fast" if key.start < to_play.start else "late"
+			if timingScore == "perfect": timingInformation = "perfect"
+			self.sendScore(obj["id"], timingScore, timingInformation)
+	
+
+
+	def is_timing_close(self, key: Key, i):
+		return abs(i.start - key.start) < 500
 
 	def handleMessage(self, message: str):
 		obj = json.loads(message)
@@ -54,17 +86,18 @@ class Scorometer():
 			self.sendError(message)
 			return
 		if obj["type"] == "note_on" or obj["type"] == "note_off":
-			pass
+			self.handleNote(obj)
 		if obj["type"] == "pause":
 			pass
 
 	def sendEnd(self, overall, difficulties):
 		print(json.dumps({"overallScore": overall, "score": difficulties}))
-		pass
 
 	def sendError(self, message):
 		print(json.dumps({"error": f"Could not handle message {message}"}))
-		pass
+
+	def sendScore(self, id, timingScore, timingInformation):
+		print(json.dumps({"id": id, "timingScore": timingScore, "timingInformation": timingInformation}))
 
 	def gameLoop(self):
 		while True: