renaming files
This commit is contained in:
@@ -1,34 +0,0 @@
|
||||
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
|
||||
|
||||
|
||||
class Partition:
|
||||
|
||||
def __init__(self, name:str, notes:list[Note]) -> None:
|
||||
|
||||
self.__name = name
|
||||
self.__notes = notes
|
||||
|
||||
|
||||
async def play(self, output_lambda:Callable([str, tuple[int, int, int], int], None)):
|
||||
now = datetime.datetime.now()
|
||||
for note in self.__notes:
|
||||
asyncio.create_task(run_at(
|
||||
now + datetime.time(milliseconds = note.get_start_time()),
|
||||
output_lambda(
|
||||
note.get_key(),
|
||||
note.get_color(),
|
||||
note.get_duration()
|
||||
)
|
||||
))
|
||||
@@ -11,6 +11,9 @@ class Note:
|
||||
def get_key(self):
|
||||
return self.__key
|
||||
|
||||
def get_color(self):
|
||||
return self.__color
|
||||
|
||||
def get_start_time(self):
|
||||
return self.__start_time
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
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
|
||||
|
||||
class Partition:
|
||||
|
||||
def __init__(self, name:str, notes:list[Note]) -> None:
|
||||
|
||||
self.__name = name
|
||||
self.__notes = notes
|
||||
|
||||
|
||||
def play(self, output_lambda:Callable[[str, tuple[int, int, int], int], None]):
|
||||
now = datetime.datetime.now()
|
||||
tasks_to_wait = []
|
||||
for note in self.__notes:
|
||||
tasks_to_wait.append(
|
||||
asyncio.create_task(
|
||||
lambda: asyncio.wait(
|
||||
run_at(
|
||||
now + datetime.timedelta(milliseconds= note.get_start_time()),
|
||||
output_lambda(
|
||||
note.get_key(),
|
||||
note.get_color(),
|
||||
note.get_duration()
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
asyncio.wait(tasks_to_wait)
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -27,7 +27,6 @@ def playNote(color, secondsToStay, pixelsToFill):
|
||||
pixels[pixelIndex] = color
|
||||
time.sleep(secondsToStay)
|
||||
|
||||
def incomingNote(color, time, pixelsToFill):
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
from chroma_case.Partition import Partition
|
||||
from chroma_case.Note import Note
|
||||
import asyncio
|
||||
import sys
|
||||
|
||||
def printing(key, color, duration):
|
||||
print(f"key: {key}, c:{color} for {duration / 1000}s")
|
||||
|
||||
|
||||
def main():
|
||||
n = Note("fa", (0, 255, 0), 1000, 1000)
|
||||
|
||||
p = Partition("test", [n])
|
||||
|
||||
p.play(printing)
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(asyncio.run(main()))
|
||||
@@ -0,0 +1,15 @@
|
||||
import asyncio
|
||||
|
||||
async def nested():
|
||||
return 42
|
||||
|
||||
async def main():
|
||||
# Schedule nested() to run soon concurrently
|
||||
# with "main()".
|
||||
task = asyncio.create_task(nested())
|
||||
|
||||
# "task" can now be used to cancel "nested()", or
|
||||
# can simply be awaited to wait until it is complete:
|
||||
await task
|
||||
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user