Skip to content

Player Parser Refactor Pt1 #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Jan 8, 2022
Merged

Player Parser Refactor Pt1 #9

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
606c67f
Factor out a function
Jan 6, 2022
e9c697b
Fix comment for LoopResult enum
Jan 7, 2022
26c8761
Factor out the prompt generating function
Jan 7, 2022
c07a6a7
renamed next_row function
flipcoder Jan 8, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
159 changes: 89 additions & 70 deletions textbeat/player.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,18 @@

from collections import deque
from prompt_toolkit import PromptSession
from enum import Enum

STACK_SIZE = 64

class LoopResult(Enum):
"""This enum types help us decide whether we should continue in the loop, break, or proceed.
This type is needed so we can choose how to deal interact with a loop outside a functions scope.
"""
PROCEED = 0 # Continue forward in the current loop
CONTINUE = 1 # Skip to the next iteration
BREAK = 2 # Break from the loop

class StackFrame(object):
"""
Stack frames are items on the Player's call stack.
Expand Down Expand Up @@ -261,6 +270,7 @@ def write_midi_tempo(self):
'set_tempo', tempo=mido.bpm2tempo(self.tempo)
))


async def run(self):
for ch in self.tracks:
ch.refresh()
Expand All @@ -277,76 +287,10 @@ async def run(self):

try:
self.line = '.'
try:
self.line = self.buf[self.row]
if self.row == self.startrow:
self.startrow = -1
if self.stoprow!=-1 and self.row == self.stoprow:
self.buf = []
raise IndexError
except IndexError:
if self.has_flags(Player.Flag.LOOP):
self.row = 0
continue

self.row = len(self.buf)
# done with file, finish playing some stuff

arps_remaining = 0
if self.interactive or self.cmdmode in ['c','l']: # finish arps in shell mode
for ch in self.tracks[:self.tracks_active]:
if ch.arp_enabled:
if ch.arp_cycle_limit or not ch.arp_once:
arps_remaining += 1
self.line = '.'
if not arps_remaining and not self.shell and self.cmdmode not in ['c','l']:
break
self.line = '.'

if not arps_remaining and not self.schedule.pending():
if self.interactive:
for ch in self.tracks[:self.tracks_active]:
ch.release_all()

if self.shell:
# self.shell PROMPT
# log(orr(self.tracks[0].scale,self.scale).mode_name(orr(self.tracks[0].mode,self.mode)))
# cur_oct = self.tracks[0].octave
# cline = FG.GREEN + 'txbt> '+FG.BLUE+ '('+str(int(self.tempo))+'bpm x'+str(int(self.grid))+' '+\
# note_name(self.tracks[0].transpose) + ' ' +\
# orr(self.tracks[0].scale,self.scale).mode_name(orr(self.tracks[0].mode,self.mode,-1))+\
# ')> '
modename = orr(self.tracks[0].scale,self.scale).mode_name(orr(self.tracks[0].mode,self.mode,-1))

keynote = note_name(self.transpose + self.tracks[0].transpose)
keynote = keynote if keynote!='C' else ''
parts = [
str(int(self.tempo))+'bpm', # tempo
'x'+str(int(self.grid)), # subdiv
keynote,
('' if modename=='ionian' else modename)
]
cline = 'txbt> ('+ \
' '.join(filter(lambda x: x, parts))+ \
')> '
# if bufline.endswith('.txbt'):
# play file?
# bufline = raw_input(cline)
bufline = await prompt_session.prompt_async(cline)
bufline = list(filter(None, bufline.split(' ')))
bufline = list(map(lambda b: b.replace(';',' '), bufline))
# elif self.remote:
# pass # not yet implemented
else:
assert False

self.buf += bufline

continue

else:
break

loop_result = await self.next_row(prompt_session)
if loop_result == LoopResult.CONTINUE: continue
elif loop_result == LoopResult.BREAK: break

log(FG.MAGENTA + self.line)

# cells = line.split(' '*2)
Expand Down Expand Up @@ -1930,3 +1874,78 @@ async def run(self):

self.row += 1

async def next_row(self, prompt_session:PromptSession) -> LoopResult:
try:
self.line = self.buf[self.row]
if self.row == self.startrow:
self.startrow = -1
if self.stoprow!=-1 and self.row == self.stoprow:
self.buf = []
raise IndexError
return LoopResult.PROCEED
except IndexError:
if self.has_flags(Player.Flag.LOOP):
self.row = 0
return LoopResult.CONTINUE

self.row = len(self.buf)
# done with file, finish playing some stuff

arps_remaining = 0
if self.interactive or self.cmdmode in ['c','l']: # finish arps in shell mode
for ch in self.tracks[:self.tracks_active]:
if ch.arp_enabled:
if ch.arp_cycle_limit or not ch.arp_once:
arps_remaining += 1
self.line = '.'
if not arps_remaining and not self.shell and self.cmdmode not in ['c','l']:
return LoopResult.BREAK
self.line = '.'

if not arps_remaining and not self.schedule.pending():
if self.interactive:
for ch in self.tracks[:self.tracks_active]:
ch.release_all()

if self.shell:
self.buf += await self.mk_prompt(prompt_session)
# elif self.remote:
# pass # not yet implemented
else:
assert False
return LoopResult.CONTINUE

else:
return LoopResult.BREAK
return LoopResult.PROCEED

async def mk_prompt(self, prompt_session:PromptSession):
"""Creates a new prompt for the current line"""
# self.shell PROMPT
# log(orr(self.tracks[0].scale,self.scale).mode_name(orr(self.tracks[0].mode,self.mode)))
# cur_oct = self.tracks[0].octave
# cline = FG.GREEN + 'txbt> '+FG.BLUE+ '('+str(int(self.tempo))+'bpm x'+str(int(self.grid))+' '+\
# note_name(self.tracks[0].transpose) + ' ' +\
# orr(self.tracks[0].scale,self.scale).mode_name(orr(self.tracks[0].mode,self.mode,-1))+\
# ')> '
modename = orr(self.tracks[0].scale,self.scale).mode_name(orr(self.tracks[0].mode,self.mode,-1))

keynote = note_name(self.transpose + self.tracks[0].transpose)
keynote = keynote if keynote!='C' else ''
parts = [
str(int(self.tempo))+'bpm', # tempo
'x'+str(int(self.grid)), # subdiv
keynote,
('' if modename=='ionian' else modename)
]
cline = 'txbt> ('+ \
' '.join(filter(lambda x: x, parts))+ \
')> '
# if bufline.endswith('.txbt'):
# play file?
# bufline = raw_input(cline)
bufline = await prompt_session.prompt_async(cline)
bufline = list(filter(None, bufline.split(' ')))
bufline = list(map(lambda b: b.replace(';',' '), bufline))

return bufline