runner.py


Runner class

Display Footer

Runner source
class Runner:
    """Display Footer"""

    _status = "waiting"
    status = "starting"
    last_error = ""
    title = "runner"
    border = "green"

    _dirhash = ""
    time = time.time()

    def __init__(self, markata: "Markata") -> None:
        self.m = markata
        self._dirhash = self.m.content_dir_hash
        self._run()
        atexit.register(self.kill)

    def kill(self) -> None:
        self.proc.stdout.close()
        self.proc.stderr.close()
        self.proc.kill()
        self.proc.wait()

    def run(self) -> None:
        "Runs the build only if one is not already running."
        if self.proc.poll() is not None:
            self._run()

    def _run(self) -> None:
        "Runs the build and sets the proc"
        self.status = "running"
        self.time = time.time()
        self.proc = subprocess.Popen(
            ["markata", "build"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

    @property
    def status_message(self) -> str:
        "returns the status message to display"
        num_lines = self.m.console.height - 4
        last_error = "\n".join(self.last_error.split("\n")[-num_lines:])
        if self.status == "running":
            self.title = "runner running"
            self.border = "gold1"
        elif last_error == "":
            self.title = "runner succeded"
            self.border = "green"
        else:
            self.title = "runner failed"
            self.border = "red"
        self.title = f"{self.title} [blue]({round(time.time() - self.time)}s)[/]"

        return f"runner is {self.status} {round(time.time() - self.time)}\npid: {self.proc.pid}\nhash: {self.m.content_dir_hash}\n{last_error}"

    def __rich__(self) -> Panel:

        if self.proc:
            if self.proc.poll() is None:
                return Panel(
                    Text(self.status_message),
                    border_style=self.border,
                    title=self.title,
                    expand=True,
                )

        if self.status == "running":
            self.status = "waiting"
            self.time = time.time()
            if self.proc:
                self.last_error = self.proc.stderr.read().decode()

        if self._dirhash != self.m.content_dir_hash:
            self.run()
            self._dirhash = self.m.content_dir_hash

        return Panel(
            Text(self.status_message),
            border_style=self.border,
            title=self.title,
            expand=True,
        )

configure function

None

configure source
def configure(markata: "Markata") -> None:
    def get_runner(self):
        try:
            return self._runner
        except AttributeError:
            self._runner: Runner = Runner(self)
            return self._runner

    from markata import Markata

    Markata.runner = property(get_runner)

init method

None

init source
def __init__(self, markata: "Markata") -> None:
        self.m = markata
        self._dirhash = self.m.content_dir_hash
        self._run()
        atexit.register(self.kill)

kill method

None

kill source
def kill(self) -> None:
        self.proc.stdout.close()
        self.proc.stderr.close()
        self.proc.kill()
        self.proc.wait()

run method

Runs the build only if one is not already running.

run source
def run(self) -> None:
        "Runs the build only if one is not already running."
        if self.proc.poll() is not None:
            self._run()

_run method

Runs the build and sets the proc

_run source
def _run(self) -> None:
        "Runs the build and sets the proc"
        self.status = "running"
        self.time = time.time()
        self.proc = subprocess.Popen(
            ["markata", "build"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )

status_message method

returns the status message to display

status_message source
def status_message(self) -> str:
        "returns the status message to display"
        num_lines = self.m.console.height - 4
        last_error = "\n".join(self.last_error.split("\n")[-num_lines:])
        if self.status == "running":
            self.title = "runner running"
            self.border = "gold1"
        elif last_error == "":
            self.title = "runner succeded"
            self.border = "green"
        else:
            self.title = "runner failed"
            self.border = "red"
        self.title = f"{self.title} [blue]({round(time.time() - self.time)}s)[/]"

        return f"runner is {self.status} {round(time.time() - self.time)}\npid: {self.proc.pid}\nhash: {self.m.content_dir_hash}\n{last_error}"

rich method

None

rich source
def __rich__(self) -> Panel:

        if self.proc:
            if self.proc.poll() is None:
                return Panel(
                    Text(self.status_message),
                    border_style=self.border,
                    title=self.title,
                    expand=True,
                )

        if self.status == "running":
            self.status = "waiting"
            self.time = time.time()
            if self.proc:
                self.last_error = self.proc.stderr.read().decode()

        if self._dirhash != self.m.content_dir_hash:
            self.run()
            self._dirhash = self.m.content_dir_hash

        return Panel(
            Text(self.status_message),
            border_style=self.border,
            title=self.title,
            expand=True,
        )

get_runner function

None

get_runner source
def get_runner(self):
        try:
            return self._runner
        except AttributeError:
            self._runner: Runner = Runner(self)
            return self._runner