Tap Trainer

import ctypes
import datetime
import json
import pyaudio
from pynput import keyboard
import math
import matplotlib.pyplot as pyplot
from matplotlib.colors import LinearSegmentedColormap
import numpy
import queue
import struct
import time
import uuid

# Enable support for ANSI escape codes to move the terminal cursor
ctypes.windll.kernel32.SetConsoleMode(ctypes.windll.kernel32.GetStdHandle(-11), 7)

def get_focused_window_handle():
  return ctypes.windll.user32.GetForegroundWindow()

log_filename = "tap_trainer_log.txt"

this_window_handle = get_focused_window_handle()
keyboard_queue = queue.SimpleQueue()
PRESS = 0
RELEASE = 1

pressed_keys = set()

# Mimics fret order from low to high frets, used to detect if a "lower" fret is pressed before a "higher" fret
key_rows = [
  "poiuytrewq",
  "lkjhgfdsa",
  "mnbvcxz",
]

valid_characters = "".join(key_rows)

def get_next_keyboard_event(block = True, timeout = None):
  try:
    # Loop to avoid returning held keys
    while True:
      (event_type, key, seconds) = keyboard_queue.get(block, timeout)
      if event_type == PRESS and key not in pressed_keys:
        pressed_keys.add(key)
        return (event_type, key, seconds)
      elif event_type == RELEASE and key in pressed_keys:
        pressed_keys.remove(key)
        return (event_type, key, seconds)
  except queue.Empty:
    return None

def on_press(key):
  if this_window_handle == get_focused_window_handle():
    seconds = time.perf_counter()
    keyboard_queue.put((PRESS, key, seconds))

def on_release(key):
  if this_window_handle == get_focused_window_handle():
    seconds = time.perf_counter()
    keyboard_queue.put((RELEASE, key, seconds))

def drain_keyboard_queue():
  try:
    get_next_keyboard_event(False)
  except queue.Empty:
    return

def wait_for_keypress(expected_keys):
  while True:
    (event_type, key, seconds) = get_next_keyboard_event()
    if event_type == PRESS and (expected_keys is None or key in expected_keys):
      return key

def is_valid_sequence_character(key):
  return isinstance(key, keyboard.KeyCode) and key.char is not None and key.char in valid_characters

def input_sequence():
  sequence = []

  while True:
    (event_type, key, seconds) = get_next_keyboard_event()
    if event_type == PRESS:
      if key is keyboard.Key.esc:
        print("")
        return None

      if key is keyboard.Key.space:
        print("")
        if len(sequence) > 0 and all(v == sequence[0] for v in sequence):
          print("A sequence cannot consist of only a single key")
          return None
        return sequence

      if key is keyboard.Key.backspace or key is keyboard.Key.delete:
        if len(sequence) > 0:
          sequence.pop()
          print("\b \b", end = "", flush = True)
      elif is_valid_sequence_character(key):
        sequence.append(key)
        print(key.char, end = "", flush = True)

def detect_sequence(match_sequence):
  # Try to determine what the sequence is - requires three perfect reps
  required_repetitions_to_match = 3
  min_sequence_length = 3
  max_sequence_length = 8
  for sequence_length in range(min_sequence_length, max_sequence_length + 1):
    if len(match_sequence) >= sequence_length * required_repetitions_to_match:
      match = True
      for i in range(sequence_length, len(match_sequence)):
        if match_sequence[i] != match_sequence[i % sequence_length]:
          match = False
          break

      if match:
        return match_sequence[:sequence_length]

  if len(match_sequence) < max_sequence_length * required_repetitions_to_match:
    return None # No match found yet, keep adding input
  else:
    return [] # Could not find a match

class MultiLineMessage:
  def __init__(self):
    self.message = ""

  def add(self, text):
    self.message += text

  def add_line(self, line):
    self.message += line + "\n"

  def clear_current_line(self):
    self.message += "\033[K"

  def goto_previous_line(self, count = 1):
    for i in range(count):
      self.message += "\033[F"

  def print(self):
    print(self.message, end = "", flush = True)
    self.message = ""

HIGHLIGHT_BACKGROUND = 0
HIGHLIGHT_TEXT = 1

class CaptureResult:
  def __init__(self):
    self.sequence = []
    self.captured_length = 0
    self.captured_duration = 0
    self.notes_per_second = 0
    self.timing_variance = 0
    self.error_count = 0
    self.captured_notes = []

class CaptureData:
  def __init__(self):
    # Array of (index in sequence, time)
    self.notes = []

  def add_sample(self, sequence_index, note_time):
    self.notes.append((sequence_index, note_time))

  def valid_note_count(self):
    return len([v for v in self.notes if v[0] >= 0])

  def get_nps_and_timing_variance(self, recent_sample_count = None):
    valid_note_times = [v[1] for v in self.notes if v[0] >= 0]
    total_sample_count = len(valid_note_times)
    if recent_sample_count is None:
      recent_sample_count = total_sample_count
    sample_count = min(recent_sample_count, total_sample_count)
    start_index = total_sample_count - sample_count

    if sample_count < 2:
        return None

    # Use running least squares to determine BPM, i.e. we will solve
    # y = A + Bx + err
    # where y is note time and x is note index (0, 1, 2, 3, ...)
    # B = (N * (sum xy) - (sum x) * (sum y)) / (N * (sum x^2) - (sum x)^2)
    # A = (sum y - B * sum x) / N
    sum_x = 0.0
    sum_x2 = 0.0
    sum_y = 0.0
    sum_xy = 0.0
    for i in range(start_index, total_sample_count):
      sum_x += i
      sum_x2 += i * i
      sum_y += valid_note_times[i]
      sum_xy += i * valid_note_times[i]

    slope = (sample_count * sum_xy - sum_x * sum_y) / (sample_count * sum_x2 - sum_x * sum_x)
    # intercept = (sum_y - intercept * sum_x) / sample_count # Not needed

    # Calculate average timing error
    expected_seconds_per_note = slope
    mean_abs_error_sum = 0
    for i in range(start_index + 1, total_sample_count):
      seconds_per_note = valid_note_times[i] - valid_note_times[i - 1]
      mean_abs_error_sum += abs(seconds_per_note - expected_seconds_per_note)
    mean_abs_error = mean_abs_error_sum / (sample_count - 1)

    notes_per_second = 1.0 / slope
    timing_variance = mean_abs_error / expected_seconds_per_note

    return (notes_per_second, timing_variance)

  def analyze_timing(self, sequence_length, recent_sample_count, most_recent_only, highlight_sequence_index = None, highlight_style = HIGHLIGHT_BACKGROUND):
    total_sample_count = len(self.notes)
    if recent_sample_count is None:
      recent_sample_count = total_sample_count
    sample_count = min(recent_sample_count, total_sample_count)
    start_index = total_sample_count - sample_count

    note_times = [0] * sequence_length
    note_counts = [0] * sequence_length
    # Iterate in reverse so most_recent_only works. Note that start_index is explicitly not included in this loop.
    for i in range(start_index + sample_count - 1, start_index, -1):
      sequence_index = self.notes[i - 1][0]
      if sequence_index < 0:
        continue
      if not most_recent_only or note_counts[sequence_index] == 0:
        note_time = self.notes[i][1] - self.notes[i - 1][1]
        note_times[sequence_index] += note_time
        note_counts[sequence_index] += 1

    if any(v == 0 for v in note_counts):
      return "" # Not all notes have been captured

    for i in range(sequence_length):
      note_times[i] /= note_counts[i]

    chart_width = 128
    chart = ["|"] + ["_"] * chart_width + [" "] * 8 # Add padding to be safe against overflows

    total_note_times = sum(note_times)
    current_note_offset = 0
    previous_chart_position = 0
    highlight_position = None
    for i in range(sequence_length):
      current_note_offset += note_times[i]
      chart_position = round(current_note_offset * chart_width / total_note_times)
      chart[chart_position] = "|"
      note_time_percent = f"{100 * note_times[i] * sequence_length / total_note_times:.0f}%"
      percent_chart_position = round((previous_chart_position + chart_position - len(note_time_percent)) / 2)
      for t, c in enumerate(note_time_percent):
        chart[percent_chart_position + t] = c

      if i == highlight_sequence_index:
        highlight_position = (previous_chart_position, chart_position)

      previous_chart_position = chart_position

    if highlight_position is not None:
      start_index, end_index = highlight_position
      start_color_code = "\u001b[41m" if highlight_style == HIGHLIGHT_BACKGROUND else "\u001b[38;5;255m" # Red background
      end_color_code = "\u001b[0m" if highlight_style == HIGHLIGHT_BACKGROUND else "\u001b[38;5;242m" # Reset
      chart = chart[:end_index + 1] + list(end_color_code) + chart[end_index + 1:]
      chart = chart[:start_index] + list(start_color_code) + chart[start_index:]
      if highlight_style == HIGHLIGHT_TEXT:
        chart = list("\u001b[38;5;242m") + chart + list("\u001b[0m")

    return "".join(chart)

def capture_sequence(sequence):
  if sequence is None:
    sequence_repeat_counts = None
  else:
    # Determine how many times any keys repeat. Note that a sequence should never all be the same key.
    sequence_repeat_counts = [1] * len(sequence)
    for i in range(len(sequence)):
      for j in range(1, len(sequence) - 1):
        if sequence[i] == sequence[(i + j) % len(sequence)]:
          sequence_repeat_counts[i] += 1
        else:
          break

  match_sequence = []

  capture_pressed_keys = set()
  next_sequence_index = 0
  prev_sequence_index = None
  start_time = None
  capture_data = CaptureData()
  error_count = 0
  running_error_count = 0
  last_key = None

  message = MultiLineMessage()
  message.add_line("")
  message.add_line("")
  message.add_line("Capture active (...):")
  message.print()

  previous_timing_chart_immediate_no_color = None
  while True:
    return_result = False

    # Grab events which all occur very close in time and only process the most recent one
    data_to_process = None
    pending_data = get_next_keyboard_event()
    while True:
      (event_type, key, seconds) = pending_data

      if event_type == PRESS:
        if key not in capture_pressed_keys:
          capture_pressed_keys.add(key)

        if key is keyboard.Key.esc:
          # Discard this capture
          print("")
          print("Discarding capture")
          return None

        if key is keyboard.Key.space:
          if sequence is None or capture_data.valid_note_count() < 16:
            print("")
            print("Not enough samples, discarding capture")
            return None
          return_result = True
          break

        if is_valid_sequence_character(key):
          commit_pending_data = True

          # Make sure there is not a note on a "higher fret" that is already pressed
          row = next(r for r in key_rows if key.char in r)
          for index in range(row.index(key.char) + 1, len(row)):
            key_to_check = keyboard.KeyCode.from_char(row[index])
            if key_to_check in pressed_keys:
              # This key is pressed
              commit_pending_data = False
              break

          if commit_pending_data:
            data_to_process = (key, seconds)
      else:
        assert event_type == RELEASE

        # $TODO add release timing variance?
        if key in capture_pressed_keys:
          capture_pressed_keys.remove(key)

        if is_valid_sequence_character(key):
          # Make sure there is not a note on a "higher fret" that is already pressed
          blocked = False
          row = next(r for r in key_rows if key.char in r)
          for index in range(row.index(key.char) + 1, len(row)):
            key_to_check = keyboard.KeyCode.from_char(row[index])
            if key_to_check in pressed_keys:
              # This key is pressed
              blocked = True
              break

          if not blocked:
            # Determine if there is a note on a "lower fret" that is now pressed
            for index in range(row.index(key.char) - 1, -1, -1):
              key_to_check = keyboard.KeyCode.from_char(row[index])
              if key_to_check in pressed_keys:
                # This key is pressed
                data_to_process = (key_to_check, seconds)
                break

      pending_data = get_next_keyboard_event(timeout = 0.005)
      if pending_data is None:
        break

    did_add_sample = False
    if data_to_process is not None:
      (key, seconds) = data_to_process
      last_key = key

      if sequence is None:
        if not is_valid_sequence_character(key):
          print("")
          print("Invalid sequence specified, discarding capture")
          return None

        match_sequence.append(key)
        sequence = detect_sequence(match_sequence)
        if sequence is not None:
          if len(sequence) == 0:
            print("")
            print("Unable to determine the sequence, discarding capture")
            return None
          else:
            sequence_repeat_counts = [1] * len(sequence)
            capture_data.notes = [(i % len(sequence), v[1]) for i, v in enumerate(capture_data.notes)]

      if sequence is None:
        sequence_index = -1
        repeat_count = 1
      else:
        repeat_count = 1 if prev_sequence_index is None else sequence_repeat_counts[prev_sequence_index]
        expected_key = sequence[next_sequence_index]
        if running_error_count > 0:
          # An error was previously made. This could be one of three types of errors:
          # (1) The wrong key was simply pressed
          # (2) An extra key was pressed
          # (3) A key was skipped
          # We will try to handle all of these cases but the results will depend on the particular sequence.
          if key == expected_key:
            # The previous key was wrong but this one is right - we hit the wrong key once
            sequence_index = next_sequence_index
            running_error_count = 0
          else:
            next_expected_key = sequence[(next_sequence_index + 1) % len(sequence)]
            previous_expected_key = sequence[(next_sequence_index - 1) % len(sequence)] # Python's % operator works for this

            # $TODO figure out repeat count recovery here?
            if key == previous_expected_key:
              # We pressed an extra key so we're 1 behind
              sequence_index = (next_sequence_index - 1) % len(sequence)
              prev_sequence_index = next_sequence_index
              next_sequence_index = (next_sequence_index - 1) % len(sequence)
              running_error_count = 0
            elif key == next_expected_key:
              # We skipped a key so we're 1 ahead
              sequence_index = (next_sequence_index + 1) % len(sequence)
              prev_sequence_index = next_sequence_index
              next_sequence_index = (next_sequence_index + 1) % len(sequence)
              running_error_count = 0
            else:
              sequence_index = -1
              # We made another error that doesn't fall into one of the 3 cases that we handle
              if running_error_count < 2:
                # If we make consecutive errors, don't keep penalizing - record up to 2 and then wait to get back in sync
                error_count += 1
              running_error_count += 1
        elif key == expected_key:
          # Correct entry
          sequence_index = next_sequence_index
        else:
          sequence_index = -1
          error_count += 1
          running_error_count += 1

        # Advance in the sequence unless we've made too many consecutive errors - then just wait to get back in sync
        if running_error_count < 2:
          prev_sequence_index = next_sequence_index
          next_sequence_index = (next_sequence_index + sequence_repeat_counts[next_sequence_index]) % len(sequence)

      if start_time is None:
        start_time = seconds
      note_time = seconds - start_time

      # Record timing info for the key event (and any repeated previous events)
      prev_note_time = capture_data.notes[-1][1] if len(capture_data.notes) > 0 else 0.0
      for repeat_index in range(1, repeat_count + 1):
        # Lerp across repeated notes, evenly spreading the total note duration
        u = repeat_index / repeat_count
        lerped_note_time = prev_note_time * (1 - u) + note_time * u
        lerped_sequence_index = -1 if sequence is None else (sequence_index - repeat_count + repeat_index) % len(sequence)
        capture_data.add_sample(lerped_sequence_index, lerped_note_time)
        did_add_sample = True

    message.goto_previous_line(3)
    message.clear_current_line()

    timing_chart_immediate = None
    timing_chart_immediate_no_color = None
    if sequence is not None and len(capture_data.notes) > 0:
      last_note_sequence_index = capture_data.notes[-1][0]
      highlight_sequence_index = None if return_result else last_note_sequence_index
      timing_chart = capture_data.analyze_timing(len(sequence), recent_sample_count, False, highlight_sequence_index, HIGHLIGHT_BACKGROUND)
      if did_add_sample:
        timing_chart_immediate = capture_data.analyze_timing(len(sequence), recent_sample_count, True, highlight_sequence_index, HIGHLIGHT_BACKGROUND)
        timing_chart_immediate_no_color = capture_data.analyze_timing(len(sequence), recent_sample_count, True, highlight_sequence_index, HIGHLIGHT_TEXT)
    else:
      timing_chart = ""

    # Add the latest timing data whenever there is an update so we get a continuous "stream" of timing charts
    if timing_chart_immediate is not None:
      # Overwrite the previous timing chart so that only the latest one has coloring
      if previous_timing_chart_immediate_no_color is not None:
        message.goto_previous_line()
        message.add_line(f"  {previous_timing_chart_immediate_no_color}")

      previous_timing_chart_immediate_no_color = timing_chart_immediate_no_color
      message.add_line(f"  {timing_chart_immediate}")
    elif return_result and previous_timing_chart_immediate_no_color is not None:
      # Rewrite the last line without background color
      message.goto_previous_line()
      message.add_line(f"  {previous_timing_chart_immediate_no_color}")

    message.clear_current_line()
    message.add_line("")
    message.clear_current_line()
    message.add_line(f"  {timing_chart}")

    message.clear_current_line()
    message.add("Capture active (...): " if sequence is None else f"Capture active ({''.join(x.char for x in sequence)}): ")

    recent_sample_count = None if return_result or sequence is None else len(sequence) * 8
    note_count = len(capture_data.notes)
    nps_and_timing_variance = capture_data.get_nps_and_timing_variance(recent_sample_count)
    if nps_and_timing_variance is not None:
      notes_per_second, timing_variance = nps_and_timing_variance
      bpm_16ths = notes_per_second * (60 / 4)

      message.add(f"{notes_per_second:.1f}nps ({bpm_16ths:.0f}bpm 16ths), ")
      message.add(f"{100 * timing_variance:.0f}% timing variance, ")

    if note_count >= 1:
      message.add(f"{error_count} error(s) ({100 * error_count / note_count:.0f}% error rate) ")

    if last_key is not None:
      message.add(f"{last_key.char} ")

    message.add_line(f"{'' if running_error_count == 0 else '(!)'}")

    # Backtrack so we can overwrite the previous message next time
    message.print()

    if return_result:
      assert note_count >= 2

      result = CaptureResult()
      result.sequence = sequence
      result.captured_length = len(capture_data.notes)
      result.captured_duration = capture_data.notes[-1][1] - capture_data.notes[0][1]
      result.notes_per_second = notes_per_second
      result.timing_variance = timing_variance
      result.error_count = error_count
      result.captured_notes = capture_data.notes
      return result

def format_duration(total_seconds):
  rounded_seconds = math.floor(total_seconds)
  seconds = rounded_seconds % 60
  total_minutes = math.floor(rounded_seconds / 60)
  minutes = total_minutes % 60
  total_hours = math.floor(total_minutes / 60)
  return f"{total_hours}:{minutes:02d}:{seconds:02d}"

def list_logged_sequences():
  sequences = {}
  with open(log_filename, "r") as log_file:
    lines = log_file.readlines()
    for line in lines:
      entry = json.loads(line)
      sequence = entry["sequence"]
      value = sequences[sequence] if sequence in sequences else (0, 0.0)
      count, duration = value
      count += 1
      if "captured_duration" in entry:
        duration += entry["captured_duration"]
      else:
        duration += entry["captured_length"] / entry["notes_per_second"] # Handle old log entry
      sequences[sequence] = (count, duration)

  sorted_sequences = sorted(sequences, key = lambda k: sequences[k][0], reverse = True)
  total_duration = sum(v[1] for v in sequences.values())
  print(f"  Total duration: {format_duration(total_duration)}")
  for sequence in sorted_sequences:
    count, duration = sequences[sequence]
    print(f"  {sequence} - {count}, {format_duration(duration)}")
  print("")

PLOT_TYPE_INDEX_BPM_TIMING_VARIANCE = 1
PLOT_TYPE_INDEX_BPM_TIMING_VARIANCE_MS = 2
PLOT_TYPE_INDEX_BPM_ERROR_RATE = 3
PLOT_TYPE_DATE_BPM_TIMING_VARIANCE = 4
PLOT_TYPE_DATE_BPM_TIMING_VARIANCE_MS = 5
PLOT_TYPE_DATE_BPM_ERROR_RATE = 6
PLOT_TYPE_BPM_TIMING_VARIANCE = 7
PLOT_TYPE_BPM_TIMING_VARIANCE_MS = 8
PLOT_TYPE_BPM_ERROR_RATE = 9

def plot_logged_sequence(sequence, plot_type):
  sequence = "".join(x.char for x in sequence)
  entries = []
  with open(log_filename, "r") as log_file:
    lines = log_file.readlines()
    for line in lines:
      entry = json.loads(line)
      if entry["sequence"] == sequence:
        entries.append(entry)

  if len(entries) == 0:
    print("No logged entries found")
    return

  entries.sort(key = lambda entry: datetime.datetime.strptime(entry["date"], "%Y-%m-%d %H:%M:%S.%f"))

  indices = []
  dates = []
  bpms_16ths = []
  timing_variances = []
  timing_variances_ms = []
  error_rates = []
  for i, entry in enumerate(entries):
    indices.append(i)
    date = datetime.datetime.strptime(entry["date"], "%Y-%m-%d %H:%M:%S.%f")
    dates.append(date)

    notes_per_second = entry["notes_per_second"]
    bpm_16ths = notes_per_second * (60 / 4)
    bpms_16ths.append(bpm_16ths)

    timing_variance = entry["timing_variance"]
    timing_variances.append(100 * timing_variance)

    note_length_ms = 1000.0 / notes_per_second
    timing_variances_ms.append(timing_variance * note_length_ms)

    error_rates.append(100 * entry["error_count"] / entry["captured_length"])

  rotate_x_labels = False
  fit = False
  if plot_type == PLOT_TYPE_INDEX_BPM_TIMING_VARIANCE:
    x = indices
    y = bpms_16ths
    colors = timing_variances
  elif plot_type == PLOT_TYPE_INDEX_BPM_TIMING_VARIANCE_MS:
    x = indices
    y = bpms_16ths
    colors = timing_variances_ms
  elif plot_type == PLOT_TYPE_INDEX_BPM_ERROR_RATE:
    x = indices
    y = bpms_16ths
    colors = error_rates
  elif plot_type == PLOT_TYPE_DATE_BPM_TIMING_VARIANCE:
    x = dates
    y = bpms_16ths
    colors = timing_variances
    rotate_x_labels = True
  elif plot_type == PLOT_TYPE_DATE_BPM_TIMING_VARIANCE_MS:
    x = dates
    y = bpms_16ths
    colors = timing_variances_ms
    rotate_x_labels = True
  elif plot_type == PLOT_TYPE_DATE_BPM_ERROR_RATE:
    x = dates
    y = bpms_16ths
    colors = error_rates
    rotate_x_labels = True
  elif plot_type == PLOT_TYPE_BPM_TIMING_VARIANCE:
    x = bpms_16ths
    y = timing_variances
    colors = timing_variances
    fit = True
  elif plot_type == PLOT_TYPE_BPM_TIMING_VARIANCE_MS:
    x = bpms_16ths
    y = timing_variances_ms
    colors = timing_variances_ms
    fit = True
  elif plot_type == PLOT_TYPE_BPM_ERROR_RATE:
    x = bpms_16ths
    y = error_rates
    colors = error_rates
    fit = True
  else:
    assert False

  if fit:
    trend = numpy.polyfit(x, y, deg = 1)

  cmap = LinearSegmentedColormap.from_list("gor", ["g", "#ff8000", "r"], N = 256)
  pyplot.scatter(x, y, c = colors, cmap = cmap)
  if fit:
    trendpoly = numpy.poly1d(trend)
    pyplot.plot(x, trendpoly(x))
  if rotate_x_labels:
    pyplot.xticks(rotation = 60)
  pyplot
  pyplot.colorbar()
  pyplot.tight_layout()
  pyplot.show()

session_id = uuid.uuid4()
with keyboard.Listener(on_press = on_press, on_release = on_release) as listener:
  # Wait for a moment and drain the queue to clear out any initial events (e.g. the enter key release)
  time.sleep(0.1)
  drain_keyboard_queue()

  # Note: annoyingly, input into stdin builds up in the background. I'm not sure how to avoid this.
  while True:
    print('Press <space> to start, <L> to view logged sequences, and <P> to plot sequence history')
    key = wait_for_keypress([keyboard.Key.space, keyboard.KeyCode.from_char('l'), keyboard.KeyCode.from_char('p')])
    if key == keyboard.Key.space:
      print("Enter sequence and press <space> (empty sequence will auto-detect on capture): ", end = "", flush = True)
      sequence = input_sequence()
      if sequence is None:
        continue

      capture_result = capture_sequence(sequence if len(sequence) > 0 else None)
      print("")
      if capture_result is None:
        continue

      # Write each log file line as its own JSON string for parsing convenience
      result_json = json.dumps(
        {
          "session": str(session_id),
          "date": str(datetime.datetime.now()),
          "sequence": "".join(x.char for x in capture_result.sequence),
          "captured_length": capture_result.captured_length,
          "captured_duration": capture_result.captured_duration,
          "notes_per_second": capture_result.notes_per_second,
          "timing_variance": capture_result.timing_variance,
          "error_count": capture_result.error_count,
          "captured_notes": capture_result.captured_notes,
        })

      with open(log_filename, "a") as log_file:
        log_file.write(result_json + "\n")
        log_file.flush()
    elif key == keyboard.KeyCode.from_char("l"):
      list_logged_sequences()
    elif key == keyboard.KeyCode.from_char("p"):
      print("Enter sequence and press <space>: ", end = "", flush = True)
      sequence = input_sequence()
      if sequence is None:
        continue

      print("Enter plot type:")
      print("  <1>: entry index - BPM (16ths) - timing variance %")
      print("  <2>: entry index - BPM (16ths) - timing variance ms")
      print("  <3>: entry index - BPM (16ths) - error rate %")
      print("  <4>: entry date - BPM (16ths) - timing variance %")
      print("  <5>: entry date - BPM (16ths) - timing variance ms")
      print("  <6>: entry date - BPM (16ths) - error rate %")
      print("  <7>: BPM (16ths) - timing variance %")
      print("  <8>: BPM (16ths) - timing variance ms")
      print("  <9>: BPM (16ths) - error rate")
      plot_type_key = wait_for_keypress(None)
      if plot_type_key == keyboard.Key.esc:
        continue

      plot_types = {
        keyboard.KeyCode.from_char("1"): PLOT_TYPE_INDEX_BPM_TIMING_VARIANCE,
        keyboard.KeyCode.from_char("2"): PLOT_TYPE_INDEX_BPM_TIMING_VARIANCE_MS,
        keyboard.KeyCode.from_char("3"): PLOT_TYPE_INDEX_BPM_ERROR_RATE,
        keyboard.KeyCode.from_char("4"): PLOT_TYPE_DATE_BPM_TIMING_VARIANCE,
        keyboard.KeyCode.from_char("5"): PLOT_TYPE_DATE_BPM_TIMING_VARIANCE_MS,
        keyboard.KeyCode.from_char("6"): PLOT_TYPE_DATE_BPM_ERROR_RATE,
        keyboard.KeyCode.from_char("7"): PLOT_TYPE_BPM_TIMING_VARIANCE,
        keyboard.KeyCode.from_char("8"): PLOT_TYPE_BPM_TIMING_VARIANCE_MS,
        keyboard.KeyCode.from_char("9"): PLOT_TYPE_BPM_ERROR_RATE,
      }

      if plot_type_key not in plot_types:
        print("Invalid plot type")
        continue

      plot_logged_sequence(sequence, plot_types[plot_type_key])
      print("")