self._q_to_main.eput('EXCEPTION', e)
+class LogBuffer:
+ 'Collects line-shaped messages, scrolls and wraps them for display.'
+ _display_size: YX
+ _y_pgscroll: int
+
+ def __init__(self, wrap_func: Callable) -> None:
+ self._wrap = wrap_func
+ self._history: list[str] = []
+ self._wrapped: list[tuple[int, str]] = []
+ self._upscroll_history: int = 0
+ self._upscroll_wrapped: int = 0
+
+ def apply_geometry(self, display_size: YX) -> None:
+ 'Calcs .wrapped conditions based on new display_size, stored scroll.'
+ self._display_size = display_size
+ self._y_pgscroll = self._display_size.y // 2
+ self._wrapped.clear()
+ self._wrapped += [(-1, '')] * self._display_size.y
+ self._upscroll_wrapped = 0
+ if not self._history:
+ return
+ for idx_history, line in enumerate(self._history):
+ self._add_wrapped(idx_history, line)
+ last_by_upscroll_history = [
+ t for t in self._wrapped
+ if t[0] == len(self._history) - (self._upscroll_history + 1)]
+ idx_last = self._wrapped.index(last_by_upscroll_history[-1])
+ self._upscroll_wrapped = len(self._wrapped) - (idx_last + 1)
+
+ def _add_wrapped(self, idx_original, line) -> int:
+ wrapped_lines = self._wrap(line)
+ self._wrapped += [(idx_original, line) for line in wrapped_lines]
+ return len(wrapped_lines)
+
+ def append(self, line) -> None:
+ 'Adds line to history, and wrapped to .wrap; preserves scroll.'
+ self._history += [line]
+ n_wrapped = self._add_wrapped(len(self._history), line)
+ if self._upscroll_wrapped > 0:
+ self._upscroll_history += 1
+ self._upscroll_wrapped += n_wrapped
+
+ def _calc_upscroll_history(self) -> None:
+ self._upscroll_history = 0
+ if self._upscroll_wrapped:
+ idx_lowest = self._wrapped[-(self._upscroll_wrapped + 1)][0]
+ self._upscroll_history = len(self._history) - (idx_lowest + 1)
+
+ def scroll_up(self) -> None:
+ 'Scrolls view up by half of display size.'
+ self._upscroll_wrapped = min(
+ len(self._wrapped[self._display_size.y:]) - 2,
+ self._upscroll_wrapped + self._y_pgscroll)
+ self._calc_upscroll_history()
+
+ def scroll_down(self) -> None:
+ 'Scrolls view down by half of display size.'
+ self._upscroll_wrapped = max(0,
+ self._upscroll_wrapped - self._y_pgscroll)
+ self._calc_upscroll_history()
+
+ @property
+ def wrapped(self) -> list[str]:
+ 'Returns display_size/scroll-appropriately wrapped selection of lines.'
+ start_idx = len(self._wrapped) - (self._display_size.y
+ + self._upscroll_wrapped)
+ to_return = [t[1] for t in
+ self._wrapped[start_idx:-(self._upscroll_wrapped + 1)]]
+ if self._upscroll_wrapped:
+ scroll_info = f'vvv [{self._upscroll_wrapped}] '
+ scroll_info += 'v' * (self._display_size.x - len(scroll_info))
+ return to_return + [scroll_info]
+ return to_return + [self._wrapped[-1][1]]
+
+
class TuiLoop(Loop):
'Loop for drawing/updating TUI.'
def __init__(self, term: Terminal, *args, **kwargs) -> None:
self._term = term
self._prompt = ''
- self._log_buffer: list[str] = []
+ self._log_buffer = LogBuffer(self._term.wrap)
self._upscroll = 0
self._calc_and_draw_all()
self._term.flush()
if not super().process_main(event):
return False
if event.type_ in {'ALERT', 'RECV', 'SEND'}:
- self._log_buffer += [f'{event.type_} {event.payload}']
+ self._log_buffer.append(f'{event.type_} {event.payload}')
self._draw_log()
elif event.type_ == 'KEYBINDING':
getattr(self, f'_kb__{event.payload}')()
self._draw_prompt()
def _kb__scroll_up(self) -> None:
- self._upscroll = min(len(self._log_buffer) - 1,
- self._upscroll + self._y_pgscroll)
+ self._log_buffer.scroll_up()
self._draw_log()
def _kb__scroll_down(self) -> None:
- self._upscroll = max(0, self._upscroll - self._y_pgscroll)
+ self._log_buffer.scroll_down()
self._draw_log()
def _calc_and_draw_all(self) -> None:
self._term.calc_geometry()
self._y_prompt = self._term.size.y - 1
self._y_separator = self._term.size.y - 2
- self._y_pgscroll = self._y_separator // 2
+ self._log_buffer.apply_geometry(YX(self._y_separator,
+ self._term.size.x))
self._draw_frame()
self._draw_log()
self._draw_prompt()
self._term.write_yx(YX(self._y_prompt, 0), INPUT_PROMPT)
def _draw_log(self) -> None:
- temp_buffer = [''] * self._term.size.y
- if self._upscroll > 1:
- for line in self._log_buffer[:-(1 + self._upscroll)]:
- temp_buffer += self._term.wrap(line)
- scroll_info = f'vvv [{self._upscroll}] '
- scroll_info += 'v' * (self._term.size.x - len(scroll_info))
- temp_buffer += [scroll_info]
- else:
- for line in self._log_buffer:
- temp_buffer += self._term.wrap(line)
- for i, line in enumerate(temp_buffer[-self._y_separator:]):
+ for i, line in enumerate(self._log_buffer.wrapped):
self._term.write_yx(YX(i, 0), line)
def _draw_prompt(self) -> None: