plain2code_state.py

"""Contains all state and context information we need for the rendering process."""

import time
import uuid
from typing import Optional


class RunState:
    """Contains information about the identifiable state of the rendering process."""

    def __init__(self, spec_filename: str, replay_with: Optional[str] = None, job_tag: Optional[str] = None):
        self.replay: bool = replay_with is not None
        self.render_succeeded: bool = False
        self.render_cancelled: bool = False
        # Why the render failed, as the TUI showed it; printed in the summary after the TUI closes.
        self.render_error_message: Optional[str] = None
        self.render_generated_code_path: Optional[str] = None
        self.rendered_functionalities: int = 0
        if replay_with:
            self.render_id: str = replay_with
        else:
            self.render_id: str = str(uuid.uuid4())
        self.spec_filename: str = spec_filename
        self.job_tag: Optional[str] = job_tag
        self.call_count: int = 0
        self.unittest_batch_id: int = 0
        self.render_time_accumulated: int = 0
        self.last_render_start_timestamp: float = time.monotonic()
        # Mirrored from the render state machine for crash reporting; not authoritative.
        self.current_module: Optional[str] = None
        self.current_frid: Optional[str] = None
        self.current_render_state: Optional[str] = None
        self.user_email: Optional[str] = None

    def increment_call_count(self):
        self.call_count += 1

    def increment_unittest_batch_id(self):
        self.unittest_batch_id += 1

    def set_render_succeeded(self, succeeded: bool):
        self.render_succeeded = succeeded

    def set_render_error_message(self, error_message: str):
        self.render_error_message = error_message

    def set_render_cancelled(self):
        self.render_cancelled = True

    def set_render_generated_code_path(self, generated_code_path: str):
        self.render_generated_code_path = generated_code_path

    def increment_rendered_functionalities(self):
        self.rendered_functionalities += 1

    def add_to_render_time(self):
        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()

    def to_dict(self):
        state = {
            "render_id": self.render_id,
            "call_count": self.call_count,
            "replay": self.replay,
            "spec_filename": self.spec_filename,
        }
        if self.job_tag is not None:
            state["job_tag"] = self.job_tag
        return state