# Copyright 2025 Bytedance Ltd. and/or its affiliates # Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import asyncio import re import traceback from pathlib import Path from typing import Annotated, Optional import aiofiles try: import ujson as json except ImportError: import json import typer from rich.highlighter import ReprHighlighter from rich.markdown import Markdown from rich.table import Table from rich.text import Text from textual import on from textual.app import App, ComposeResult from textual.containers import Horizontal, Vertical, VerticalScroll from textual.widgets import Input, ProgressBar, Select, SelectionList, Static INDEX_KEY = "__IDX" FILE_SUFFIX = ".jsonl" def check_textual_version(): # check if textual version is equal to 0.52.1 import textual from packaging.version import Version if Version(textual.__version__) != Version("0.52.1"): raise ImportError(f"Textual version {textual.__version__} is not supported, please pip install textual==0.52.1") check_textual_version() async def load_path(p: Path, data: dict, mask_strs: str, idx: int, pbar): samples = [] async with aiofiles.open(p, encoding="utf-8") as f: async for line in f: d = json.loads(line) for k in d: if isinstance(d[k], str): if mask_strs: d[k] = re.sub(rf"{mask_strs}", "*", d[k]) else: d[k] = json.dumps(d[k], ensure_ascii=False, indent=4) d[INDEX_KEY] = len(samples) samples.append(d) data[idx] = {"samples": samples} print(f"path {p} loaded") pbar.advance(1) async def load_dir(path: Path, data: dict[int, dict], pbar, mask_strs: str = ""): paths = list(path.glob(f"*{FILE_SUFFIX}")) paths = sorted(paths, key=lambda x: int(x.stem)) tasks = [load_path(p, data, mask_strs, i, pbar) for i, p in enumerate(paths)] await asyncio.gather(*tasks) class Highlighter(ReprHighlighter): highlights = ReprHighlighter.highlights + [ r"(?P[][\<\>{}()\|()【】\[\]=`])", r"\<\|(?P[\w\W]*?)\|\>", ] def center_word_with_equals_exactly(word: str, total_length: int, char: str = "=") -> str: if len(word) > total_length: return word padding = total_length - len(word) left_pad = (padding) // 2 right_pad = (padding + 1) // 2 return char * left_pad + " " + word + " " + char * right_pad def highlight_keyword(content: str, keyword: Optional[str]): if not keyword: return Text(content) text = Text() parts = content.split(keyword) for i, part in enumerate(parts): text.append(part, style=None) if i < len(parts) - 1: # text.append(keyword, style=Style(color="#d154d1", bgcolor="yellow", bold=True)) text.append(keyword, style="on #8f51b5") return text help_doc = """ ⌨️ keybinds: - `f/esc`: find/cancel - `tab/←/→`: change focus - `j/k`: page down/up - `g/G`: scroll home/end - `n/N`: next sample/step - `p/P`: previous sample/step - `s`: switch display mode - plain text - rich table """ class JsonLineViewer(App): BINDINGS = [ ("left", "focus_previous", "Focus Previous"), ("right", "focus_next", "Focus Next"), ("s", "swith_render", "switch render"), # control ("n", "next_sample", "Next Sample"), ("N", "next_step", "Next Step"), ("p", "previous_sample", "Previous Sample"), ("P", "previous_step", "Previous Step"), # search ("f", "toggle_search", "find"), ("enter", "next_search", "find next"), ("escape", "cancel_search", "cancel find"), # scroll ("j", "page_down", "page down"), ("k", "page_up", "page up"), ("g", "page_home", "page home"), ("G", "page_end", "page end"), ] CSS = """ Select:focus > SelectCurrent { border: tall #8f51b5; } Select.-expanded > SelectCurrent { border: tall #8f51b5; } #select-container { width: 15%; height: 100%; align: center top; } #search-container { height: 10%; align: center top; } #search-box { width: 50%; } #reqid-box { width: 50%; } """ def __init__(self, step_num: int, data: dict[int, dict], pbar): super().__init__() self.step_num = step_num self.data = data self.render_table = False self.selected_step_index = 0 self.selected_sample_index = 0 self.pbar = pbar self.matches = [] self.current_match_index = 0 self.highlighter = Highlighter() first_samples = data[list(data.keys())[0]]["samples"] # Prepare the initial field filter list (all keys from the first sample) self.filter_fields = [(f, f, True) for f in first_samples[0].keys()] # Internal set used for fast membership checks when we add new fields on the fly. # We keep it here so that when new columns appear in later steps (e.g. `request_id`), # they can be added to the UI automatically without restarting the viewer. self._field_set: set[str] = set(first_samples[0].keys()) self.sample_num = len(first_samples) def compose(self) -> ComposeResult: with Horizontal(id="search-container"): yield Input(placeholder="find something...", id="search-box") yield Input(placeholder="request id...", id="reqid-box") with Vertical(id="search-container2"): yield self.pbar yield Static("", id="search-status") with Horizontal(): with Vertical(id="select-container"): yield Static("\n") yield Static( renderable=Markdown( help_doc, ), markup=False, ) yield Static("\n") yield Select( id="step-select", value=0, prompt="select step", options=[("step: 1", 0)], allow_blank=False, ) yield Select( id="sample-select", value=0, prompt="select sample", options=[("sample: 1", 0)], allow_blank=False, ) yield Select( id="sample-sort", value=0, prompt="排序", options=[ ("sort", 0), ("score asc", 1), ("score desc", 2), ], allow_blank=False, ) yield SelectionList[int](("Select ALL", 1, True), id="fields-select-all") with VerticalScroll(id="scroll-view2"): yield SelectionList[str](*self.filter_fields, id="fields-select") with VerticalScroll(id="scroll-view"): yield Static(id="content", markup=False) async def on_mount(self) -> None: self.step_select = self.query_one("#step-select", Select) self.sample_select = self.query_one("#sample-select", Select) self.sample_sort = self.query_one("#sample-sort", Select) self.content_display = self.query_one("#content", Static) self.search_box = self.query_one("#search-box", Input) self.reqid_box = self.query_one("#reqid-box", Input) self.scroll_view = self.query_one("#scroll-view", VerticalScroll) self.search_status = self.query_one("#search-status", Static) self.fields_select = self.query_one("#fields-select", SelectionList) self.fields_select.border_title = "field filter" if self.data: self.step_select.set_options([(f"step: {i + 1}", i) for i in range(self.step_num)]) self.sample_select.set_options([(f"sample: {i + 1}", i) for i in range(self.sample_num)]) self.step_select.focus() await self.update_content() def update_result_options(self, offset: int = 0, sort_desc: Optional[bool] = None): options = [] if isinstance(self.selected_step_index, int) and self.selected_step_index < len(self.data): if self.sample_num is None or sort_desc is not None: samples = self.data[self.selected_step_index].get("samples", []) if not samples: self.selected_sample_index = offset return if sort_desc is not None: samples = sorted( samples, key=lambda x: x.get("score", x.get("score_1", 0)), reverse=sort_desc, ) options = [(f"sample: {r[INDEX_KEY] + 1}", r[INDEX_KEY]) for r in samples] self.sample_select.set_options(options) self.sample_num = len(samples) if sort_desc is not None and options: self.selected_sample_index = options[0][1] else: self.selected_sample_index = offset async def update_content(self, search_keyword: Optional[str] = None): content = "" try: samples = self.data[self.selected_step_index].get("samples", []) content_dict_full = samples[self.selected_sample_index] # Dynamically track any NEW keys that appear and add them to the field filter. self._update_fields_select(content_dict_full.keys()) # Apply field selection filter (only show selected fields) content_dict = {k: v for k, v in content_dict_full.items() if k in self.fields_select.selected} if self.render_table: content = Table("key", "value", show_lines=True) for k in content_dict: v = content_dict[k] v = f"{v}" content.add_row( k, self.highlighter(highlight_keyword(v, search_keyword)), ) else: text = Text() for k in content_dict: v = content_dict[k] s = center_word_with_equals_exactly(k, 64) + f"\n{v}\n" text.append(highlight_keyword(s, search_keyword)) content = self.highlighter(text) except KeyError: content = f"Loading data asynchronously, progress: {len(self.data)}/{self.step_num} step" except Exception: content = self.highlighter(traceback.format_exc()) self.content_display.update(content) # --------------------------------------------------------------------- # Request-ID jump logic # --------------------------------------------------------------------- @on(Input.Submitted, "#reqid-box") async def on_reqid_submitted(self, event: Input.Submitted) -> None: """Jump to the sample that has a matching `request_id`.""" req_id_raw = event.value.strip() # Remove hyphens so search is tolerant to different id formats req_id = req_id_raw.replace("-", "") if not req_id: return found = False for step_idx, step_data in self.data.items(): for sample in step_data.get("samples", []): sample_id = str(sample.get("request_id", "")) if sample_id.replace("-", "") == req_id: # Update selected indices self.selected_step_index = step_idx self.step_select.value = step_idx # Ensure sample list is updated and select sample self.update_result_options(offset=sample[INDEX_KEY]) self.selected_sample_index = sample[INDEX_KEY] self.sample_select.value = sample[INDEX_KEY] await self._clear_search() await self.update_content() found = True break if found: break if not found: self.search_status.update(Text(f"request_id '{req_id_raw}' not found", style="bold red")) else: # Keep the typed id in the input box so users see what was searched. pass # --------------------------------------------------------------------- # Helper: add new fields to SelectionList on-the-fly # --------------------------------------------------------------------- def _update_fields_select(self, keys): """Add any unseen *keys* to the field-selection widget so they can be toggled. The viewer is often launched with only the first step loaded. Later steps may introduce new columns (e.g. `request_id`). This helper ensures those fields become visible without requiring a restart. """ # Ensure we have the widget (only after on_mount) if not hasattr(self, "fields_select"): return for k in keys: if k not in self._field_set: self._field_set.add(k) try: # By default, new fields are selected so they appear immediately. self.fields_select.add_option(k, k, selected=True) except Exception: # Fallback for older textual versions where signature is different. self.fields_select.add_option((k, k, True)) @on(Select.Changed, "#step-select") async def step_changed(self, event): self.selected_step_index = event.value self.update_result_options() await self.update_content() @on(Select.Changed, "#sample-select") async def sample_changed(self, event): self.selected_sample_index = event.value await self._clear_search() await self.update_content() @on(Select.Changed, "#sample-sort") async def sort_changed(self, event): v = event.value self.update_result_options(sort_desc=None if v == 0 else False if v == 1 else True) await self.update_content() @on(SelectionList.SelectedChanged, "#fields-select") async def fields_changed(self, event): await self.update_content() @on(SelectionList.SelectedChanged, "#fields-select-all") async def fields_all_changed(self, event): s = self.query_one("#fields-select-all", SelectionList) if s.selected: self.fields_select.select_all() else: self.fields_select.deselect_all() def action_focus_previous(self): self.screen.focus_previous() def action_focus_next(self): self.screen.focus_next() async def action_next_step(self) -> None: self.selected_step_index += 1 if self.selected_step_index >= self.step_num: self.selected_step_index = 0 self.step_select.value = self.selected_step_index self.update_result_options() await self.update_content() async def action_next_sample(self) -> None: self.selected_sample_index += 1 if not self.sample_num or self.selected_sample_index >= self.sample_num: self.selected_sample_index = 0 self.sample_select.value = self.selected_sample_index await self._clear_search() await self.update_content() async def action_previous_step(self) -> None: self.selected_step_index -= 1 if self.selected_step_index < 0: self.selected_step_index = self.step_num - 1 self.step_select.value = self.selected_step_index self.update_result_options() await self.update_content() async def action_previous_sample(self) -> None: self.selected_sample_index -= 1 if self.selected_sample_index < 0: self.selected_sample_index = self.sample_num - 1 self.sample_select.value = self.selected_sample_index await self._clear_search() await self.update_content() async def action_swith_render(self): self.render_table = not self.render_table await self.update_content() def action_toggle_search(self) -> None: self.search_box.focus() async def action_cancel_search(self) -> None: self.search_box.value = "" await self._clear_search() await self.update_content() async def _clear_search(self): self.matches = [] self.search_status.update("") self.current_match_index = 0 @on(Input.Submitted, "#search-box") async def on_search_submitted(self, event: Input.Submitted) -> None: self.matches = [] self.current_match_index = 0 if event.value: await self.update_content(event.value) renderable = self.content_display.render() if isinstance(renderable, Table): return assert isinstance(renderable, Text) console = self.content_display._console lines = renderable.wrap(console, self.scroll_view.container_size.width) line_idx_recorded = set() for line_idx, line in enumerate(lines): if line_idx in line_idx_recorded: continue if event.value in line: self.matches.append( { "line": line_idx, "word": event.value, } ) line_idx_recorded.add(line_idx) self.scroll_view.focus() await self.action_next_search() async def action_next_search(self) -> None: if not self.matches or self.current_match_index >= len(self.matches): return target_line = self.matches[self.current_match_index]["line"] self.scroll_view.scroll_to(x=0, y=target_line * 1, animate=False) self.current_match_index = (self.current_match_index + 1) % len(self.matches) self.search_status.update( Text( f"Find :{self.current_match_index + 1}/{len(self.matches)}", style="bold on #8f51b5", ) ) def action_page_up(self): self.scroll_view.scroll_page_up(animate=False) def action_page_down(self): self.scroll_view.scroll_page_down(animate=False) def action_page_home(self): self.scroll_view.scroll_home(animate=False) def action_page_end(self): self.scroll_view.scroll_end(animate=False) async def _run(path: Path, mask_str: str): assert path.exists(), f"{path} not exist" paths = list(path.glob(f"*{FILE_SUFFIX}")) paths = sorted(paths, key=lambda x: int(x.stem)) if not paths: raise ValueError(f"no available reward dump files under f{path}") print(f"get jsonl file nums: {len(paths)}") pbar = ProgressBar(total=len(paths), name="data load progress") data = {} await load_path(paths[0], data, mask_str, 0, pbar) app = JsonLineViewer(step_num=len(paths), data=data, pbar=pbar) await asyncio.gather(load_dir(path, data, pbar, mask_str), app.run_async()) app = typer.Typer() @app.command(help="launch TUI APP") def run( rollout_data_dir: Path, mask_str: Annotated[str, typer.Option(help="string that will be masked to *")] = "<\|image_pad\|>|<\|imgpad\|>", ): loop = asyncio.get_event_loop() loop.run_until_complete(_run(rollout_data_dir, mask_str)) if __name__ == "__main__": app()