Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 2 additions & 2 deletions cli_output/render_summary.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

from plain2code_console import console
from plain2code_state import RunState
from plain2code_utils import format_duration_hms
from usage_summary import format_usage_summary


def print_exit_summary(
Expand All @@ -24,7 +24,7 @@ def print_exit_summary(
msg += f" [#8E8F91]render id:\t\t\t[#FFFFFF]{run_state.render_id}\n"
msg += f" [#8E8F91]input file:\t\t\t[#FFFFFF]{spec_filename}\n"
msg += f" [#8E8F91]generated code folder:\t[#FFFFFF]{run_state.render_generated_code_path or '-'}\n\n"
msg += f"[#8E8F91]functionalities [#FFFFFF]{run_state.rendered_functionalities} [#8E8F91]used credits [#FFFFFF]{run_state.rendered_functionalities} [#8E8F91]render time [#FFFFFF]{format_duration_hms(run_state.render_time_accumulated)}\n"
msg += format_usage_summary(run_state.rendered_functionalities, run_state.render_time_accumulated) + "\n"
console.print(msg)

if not run_state.render_succeeded and error_message:
Expand Down
1 change: 1 addition & 0 deletions plain2code.py
Original file line number Diff line number Diff line change
Expand Up @@ -286,6 +286,7 @@ def run_render():
render_thread = threading.Thread(target=run_render, daemon=True)
app = Plain2CodeTUI(
event_bus=event_bus,
run_state=run_state,
on_ready=render_thread.start,
render_id=run_state.render_id,
unittests_script=args.unittests_script,
Expand Down
9 changes: 2 additions & 7 deletions plain2code_logger.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import logging
import time

from event_bus import EventBus
from plain2code_events import LogMessageEmitted
Expand Down Expand Up @@ -42,9 +41,7 @@ def __init__(self, run_state: RunState, fmt: str = "%(elapsed_time)s %(levelname
def format(self, record):
# Calculate elapsed time the same way as LoggingHandler does for the TUI
try:
offset_seconds = self.run_state.render_time_accumulated + int(
time.monotonic() - self.run_state.last_render_start_timestamp
)
offset_seconds = self.run_state.get_live_render_time()
except Exception:
# If RunState is not available or there's any error, default to 00:00:00
offset_seconds = 0
Expand Down Expand Up @@ -74,9 +71,7 @@ def __init__(self, event_bus: EventBus, run_state: RunState):

def emit(self, record):
try:
offset_seconds = self.run_state.render_time_accumulated + int(
time.monotonic() - self.run_state.last_render_start_timestamp
)
offset_seconds = self.run_state.get_live_render_time()

hours = offset_seconds // 3600
minutes = (offset_seconds % 3600) // 60
Expand Down
18 changes: 17 additions & 1 deletion plain2code_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,23 @@ def increment_rendered_functionalities(self):
self.rendered_functionalities += 1

def add_to_render_time(self):
self.render_time_accumulated += int(time.monotonic() - self.last_render_start_timestamp)
now = time.monotonic()
self.render_time_accumulated += int(now - self.last_render_start_timestamp)
# Reset the segment start so get_live_render_time() does not re-count the
# span that was just banked (e.g. after a render completes).
self.last_render_start_timestamp = now

def get_live_render_time(self) -> int:
"""Render time so far in whole seconds, including the in-progress segment.

This is the single source of truth for elapsed render time, shared by the
log timestamps, the TUI usage line, and the post-render summary. Pause time
is excluded because ``add_to_render_time`` banks the elapsed span at the
start of a pause and ``set_last_render_start_timestamp`` restarts the
segment on resume. The value is only transiently inflated while the code
sits inside the pause loop itself, which no consumer samples.
"""
return self.render_time_accumulated + int(time.monotonic() - self.last_render_start_timestamp)

def set_last_render_start_timestamp(self):
self.last_render_start_timestamp = time.monotonic()
Expand Down
30 changes: 17 additions & 13 deletions plain2code_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,19 +16,23 @@ def find_large_base64_blob(text: str) -> Optional[str]:
return match.group(0) if match else None


def format_duration_hms(total_seconds: int) -> str:
"""Format a duration in seconds as hours, minutes, and seconds (e.g. ``1h 2m 3.45s``, ``45.67s``)."""
if total_seconds < 0:
total_seconds = 0
h = int(total_seconds // 3600)
m = int((total_seconds % 3600) // 60)
s = total_seconds % 60
if h:
return f"{h}h {m}m {s}s"
if m:
return f"{m}m {s}s"
text = f"{s}".rstrip("0").rstrip(".")
return f"{text}s" if text else "0s"
def format_duration_hms(total_seconds: float) -> str:
"""Format a whole-second duration compactly (e.g. ``10s``, ``5m 49s``, ``1h 2m``).

Fractional seconds are truncated. Durations under a minute render as ``{s}s``,
under an hour as ``{m}m {s}s``, and beyond as ``{h}h {m}m``.
"""
elapsed = int(total_seconds)
if elapsed < 0:
elapsed = 0
if elapsed < 60:
return f"{elapsed}s"
minutes = elapsed // 60
seconds = elapsed % 60
if minutes < 60:
return f"{minutes}m {seconds}s"
hours = minutes // 60
return f"{hours}h {minutes % 60}m"


AMBIGUITY_CAUSES = {
Expand Down
36 changes: 36 additions & 0 deletions tests/test_plain2code_state.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
"""Tests for RunState render-time accounting."""

import time

from plain2code_state import RunState


def test_get_live_render_time_includes_in_progress_segment():
run_state = RunState(spec_filename="x.plain")
run_state.render_time_accumulated = 100
run_state.last_render_start_timestamp = time.monotonic() - 5
# 100 banked + ~5 in the current segment.
assert run_state.get_live_render_time() == 105


def test_add_to_render_time_banks_and_resets_segment():
run_state = RunState(spec_filename="x.plain")
run_state.last_render_start_timestamp = time.monotonic() - 10

run_state.add_to_render_time()

# The 10s segment is banked...
assert run_state.render_time_accumulated == 10
# ...and the segment start is reset, so the value is not re-counted afterwards.
assert run_state.get_live_render_time() == 10


def test_add_to_render_time_is_cumulative_across_segments():
run_state = RunState(spec_filename="x.plain")

run_state.last_render_start_timestamp = time.monotonic() - 3
run_state.add_to_render_time()
run_state.last_render_start_timestamp = time.monotonic() - 4
run_state.add_to_render_time()

assert run_state.render_time_accumulated == 7
36 changes: 36 additions & 0 deletions tests/test_plain2code_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
"""Tests for pure helpers in plain2code_utils."""

from plain2code_utils import format_duration_hms


class TestFormatDurationHms:
def test_zero_seconds(self):
assert format_duration_hms(0) == "0s"

def test_sub_minute_seconds(self):
assert format_duration_hms(5) == "5s"
assert format_duration_hms(49) == "49s"

def test_sub_minute_multiple_of_ten(self):
# Regression: trailing-zero seconds must not be stripped (10s, not 1s).
assert format_duration_hms(10) == "10s"
assert format_duration_hms(20) == "20s"
assert format_duration_hms(30) == "30s"

def test_exact_minute(self):
assert format_duration_hms(60) == "1m 0s"

def test_minutes_and_seconds(self):
assert format_duration_hms(90) == "1m 30s"
assert format_duration_hms(349) == "5m 49s"

def test_hours_drop_seconds(self):
assert format_duration_hms(3600) == "1h 0m"
assert format_duration_hms(3661) == "1h 1m"

def test_float_input_is_truncated(self):
assert format_duration_hms(10.9) == "10s"
assert format_duration_hms(349.4) == "5m 49s"

def test_negative_input_clamped_to_zero(self):
assert format_duration_hms(-5) == "0s"
209 changes: 209 additions & 0 deletions tests/test_tui_usage.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,209 @@
"""Headless TUI tests for the live credit-usage line and terminal summaries."""

import asyncio
import time

from textual.widgets import Static

from event_bus import EventBus
from plain2code_events import RenderCompleted, RenderFailed
from plain2code_state import RunState
from tui.components import TUIComponents
from tui.plain2code_tui import Plain2CodeTUI


def _make_app(run_state: RunState, event_bus: EventBus) -> Plain2CodeTUI:
return Plain2CodeTUI(
event_bus=event_bus,
run_state=run_state,
on_ready=lambda: None,
render_id="test-render-id",
unittests_script=None,
conformance_tests_script=None,
prepare_environment_script=None,
state_machine_version="0.0.0",
css_path="styles.css",
)


def _set_live_render_time(run_state: RunState, seconds: int) -> None:
"""Make run_state.get_live_render_time() return ``seconds`` deterministically.

Models an in-progress segment (nothing banked yet) that started ``seconds`` ago,
which is the state on the exception/cancel paths the render machine never
finalizes.
"""
run_state.render_time_accumulated = 0
run_state.last_render_start_timestamp = time.monotonic() - seconds


def _usage_text(app: Plain2CodeTUI) -> str:
widget = app.query_one(f"#{TUIComponents.RENDER_USAGE_WIDGET.value}", Static)
return str(widget.content)


def _status_text(app: Plain2CodeTUI) -> str:
widget = app.query_one(f"#{TUIComponents.RENDER_STATUS_WIDGET.value}", Static)
return str(widget.content)


def test_usage_line_reflects_live_progress():
async def scenario():
event_bus = EventBus()
run_state = RunState(spec_filename="x.plain")
app = _make_app(run_state, event_bus)
async with app.run_test() as pilot:
run_state.rendered_functionalities = 3
_set_live_render_time(run_state, 349)
app._refresh_usage_summary()
await pilot.pause()

text = _usage_text(app)
assert "functionalities [#FFFFFF]3" in text
assert "used credits [#FFFFFF]3" in text
assert "render time [#FFFFFF]5m 49s" in text

asyncio.run(scenario())


def test_usage_line_frozen_while_paused():
async def scenario():
event_bus = EventBus()
run_state = RunState(spec_filename="x.plain")
app = _make_app(run_state, event_bus)
async with app.run_test():
_set_live_render_time(run_state, 10)
app._refresh_usage_summary()
assert "render time [#FFFFFF]10s" in _usage_text(app)

# While paused, a refresh must not sample the render time again.
app._usage_paused = True
_set_live_render_time(run_state, 999) # 16m 39s
app._refresh_usage_summary()
assert "render time [#FFFFFF]10s" in _usage_text(app)
assert "16m 39s" not in _usage_text(app)

# Resuming samples again.
app._usage_paused = False
app._refresh_usage_summary()
assert "render time [#FFFFFF]16m 39s" in _usage_text(app)

asyncio.run(scenario())


def test_usage_line_finalized_on_success():
async def scenario():
event_bus = EventBus()
run_state = RunState(spec_filename="x.plain")
app = _make_app(run_state, event_bus)
async with app.run_test() as pilot:
run_state.rendered_functionalities = 6
_set_live_render_time(run_state, 349)
event_bus.publish(RenderCompleted(rendered_code_path="plain_modules/hello_world_python/"))
await pilot.pause()

status = _status_text(app)
assert "rendering completed!" in status
assert "generated code folder: plain_modules/hello_world_python/" in status

usage = _usage_text(app)
assert "functionalities [#FFFFFF]6" in usage
assert "used credits [#FFFFFF]6" in usage
assert "render time [#FFFFFF]5m 49s" in usage
assert app._render_finished is True
# The live value is captured onto the run state so the console summary matches.
assert run_state.render_time_accumulated == 349

asyncio.run(scenario())


def test_usage_line_below_error_on_failure():
async def scenario():
event_bus = EventBus()
run_state = RunState(spec_filename="x.plain")
app = _make_app(run_state, event_bus)
async with app.run_test() as pilot:
# Failure after some functionalities and elapsed time. The render machine
# did NOT finalize run_state.render_time_accumulated (exception failure
# path), so the shared live value must supply the real elapsed time.
run_state.rendered_functionalities = 2
_set_live_render_time(run_state, 349)

error = "Conformance tests failed for functionality 3"
event_bus.publish(RenderFailed(error_message=error))
await pilot.pause()

status = _status_text(app)
assert error in status

usage = _usage_text(app)
assert "functionalities [#FFFFFF]2" in usage
assert "used credits [#FFFFFF]2" in usage
# Regression: this was "0s" because the summary trusted the unfinalized state.
assert "render time [#FFFFFF]5m 49s" in usage
assert app._render_finished is True
assert run_state.render_time_accumulated == 349

asyncio.run(scenario())


def test_usage_line_zero_when_failure_before_any_progress():
async def scenario():
event_bus = EventBus()
run_state = RunState(spec_filename="x.plain")
app = _make_app(run_state, event_bus)
async with app.run_test() as pilot:
# Failure before any functionality/time (e.g. a syntax error at parse time).
_set_live_render_time(run_state, 0)
error = "Syntax error at line 1: Invalid specification heading (`implementation req`)"
event_bus.publish(RenderFailed(error_message=error))
await pilot.pause()

usage = _usage_text(app)
assert "functionalities [#FFFFFF]0" in usage
assert "used credits [#FFFFFF]0" in usage
assert "render time [#FFFFFF]0s" in usage

asyncio.run(scenario())


def test_usage_line_stops_updating_after_completion():
async def scenario():
event_bus = EventBus()
run_state = RunState(spec_filename="x.plain")
app = _make_app(run_state, event_bus)
async with app.run_test() as pilot:
run_state.rendered_functionalities = 2
_set_live_render_time(run_state, 5)
event_bus.publish(RenderCompleted(rendered_code_path="plain_modules/x/"))
await pilot.pause()

# Late mutation must not change the frozen final usage line.
run_state.rendered_functionalities = 99
app._refresh_usage_summary()
await pilot.pause()

usage = _usage_text(app)
assert "functionalities [#FFFFFF]2" in usage
assert "functionalities [#FFFFFF]99" not in usage

asyncio.run(scenario())


def test_cancel_records_render_time_for_summary():
async def scenario():
event_bus = EventBus()
run_state = RunState(spec_filename="x.plain")
app = _make_app(run_state, event_bus)
app._on_cancel = run_state.set_render_cancelled
async with app.run_test():
# The render machine never finalizes render_time_accumulated on cancel,
# so the TUI must capture the live value for the summary.
run_state.rendered_functionalities = 2
_set_live_render_time(run_state, 42)
app.action_quit()

assert run_state.render_cancelled is True
assert run_state.render_time_accumulated == 42

asyncio.run(scenario())
Loading
Loading