A modern terminal UI framework powered by hypermedia served over HTTP.
from .application import HttpApplication

celx/__init__.py
shade40/celx
from argparse import ArgumentParser

from . import HttpApplication


def run(endpoint: str):
    with HttpApplication(endpoint, title="celx") as app:
        ...


def main() -> None:
    parser = ArgumentParser("A prototype browser for celx applications.")

    subs = parser.add_subparsers(required=True)

    run_command = subs.add_parser("run")
    run_command.set_defaults(func=run)
    run_command.add_argument("endpoint", help="The endpoint to connect to.")

    args = parser.parse_args()
    command = args.func

    opts = vars(args)
    del opts["func"]

    command(**vars(args))


if __name__ == "__main__":
    main()

celx/__main__.py
shade40/celx
from functools import wraps
from typing import Any, Callable
from threading import Thread
from urllib.parse import urlparse

from xml.etree.ElementTree import fromstring as ElementTree, Element

from celadon import Application, Page, Widget, Container
from requests import Session, Response

from .parsing import HTTPMethod, Instruction, Verb, TreeMethod, parse_widget, parse_page
from .lua import lua, init_runtime


__all__ = ["HttpApplication"]


def threaded(func: Callable[..., None]) -> Callable[..., None]:
    @wraps(func)
    def _inner(*args, **kwargs) -> None:
        Thread(target=func, args=args, kwargs=kwargs).start()

    return _inner


class HttpApplication(Application):
    def __init__(self, domain: str, **app_args: Any) -> None:
        super().__init__(**app_args)

        init_runtime(lua, self)

        self._page = Page()
        self._url = urlparse(domain)
        self._session = Session()
        self._current_instructions = []

        def _clear_instructions():
            for instructions in self._current_instructions:
                instructions.clear()

        self.on_page_changed += _clear_instructions

        self.route(self._url.geturl())

    @property
    def session(self) -> Session:
        return self._session

    def _error(self, error: Exception) -> None:
        self.stop()

        self._raised = error

    def _prefix_endpoint(self, endpoint: str) -> str:
        """Prefixes hierarchy-only endpoints with the current url and its scheme."""

        if endpoint.startswith("/"):
            return self._url.scheme + "://" + self._url.netloc + endpoint

        return endpoint

    def _http(
        self,
        method: HTTPMethod,
        endpoint: str,
        data: dict[str, Any],
        handler: Callable[[str], None],
    ) -> None:
        endpoint = self._prefix_endpoint(endpoint)

        if method is HTTPMethod.GET:
            request_data = {"params": data}
        else:
            request_data = {"json": data}

        if not isinstance(method, HTTPMethod):
            self._error(TypeError(f"Invalid method {method!r}."))

        request = getattr(self._session, method.value.lower())

        def _execute() -> None:
            resp = request(endpoint, **request_data)

            if not 200 <= resp.status_code < 300:
                self.stop()
                resp.raise_for_status()

            url = urlparse(endpoint)

            if url.netloc != self._url.netloc:
                self._url = url

            tree = ElementTree(resp.text)

            for sourceable in ["styles", "lua"]:
                for node in tree.findall(f".//{sourceable}[@src]"):
                    resp = self._session.get(self._prefix_endpoint(node.attrib["src"]))

                    if not 200 <= resp.status_code < 300:
                        self.stop()
                        resp.raise_for_status()

                    node.text = resp.text
                    del node.attrib["src"]

            return handler(tree)

        thread = Thread(target=_execute)
        thread.start()

        return thread

    def _xml_page_route(self, node: Element) -> None:
        """Routes to a page loaded from the given XML."""

        try:
            page = parse_page(node)
        except Exception as exc:
            self._error(exc)
            return

        page.route_name = self._url.path

        # TODO: We don't have to request the page every time we go to it
        self.append(page)

        self._page = page
        self._mouse_target = self._page[0]
        self.on_page_changed()

        if page.route_name == "/":
            self._terminal.set_title(self.title)

        else:
            self._terminal.set_title(page.title)

        self.apply_rules()

    @threaded
    def run_instructions(self, instructions: list[Instruction], caller: Widget) -> None:
        """Runs through a list of instructions."""

        result = None

        def _set_result(xml: ElementTree) -> None:
            nonlocal result

            # Drill down to find the first widget, pass that on instead.
            if xml.tag == "celx":
                for node in xml.findall("./page//"):
                    if node.tag not in ["styles", "lua"]:
                        xml = node
                        break

                else:
                    self._error(ValueError("no widget in response"))
                    return

            result = parse_widget(xml)[0]

        self._current_instructions.append(instructions)

        for instr in instructions:
            if instr.verb.value in HTTPMethod.__members__:
                endpoint, container = instr.args

                body = caller.parent

                if container is not None:
                    try:
                        body = self.find(container)
                    except Exception as exc:
                        self._error(exc)
                        return

                    if body is None:
                        self._error(
                            ValueError(f"nothing matched selector {container!r}")
                        )
                        return

                content = body.serialize()

                self._http(
                    HTTPMethod(instr.verb.value), endpoint, content, _set_result
                ).join()

                continue

            if instr.verb.value in TreeMethod.__members__:
                if result is None:
                    self._error(ValueError("no result to update tree with"))
                    return

                selector, modifier = instr.args

                try:
                    target = self.find(selector)
                except Exception as exc:
                    self._error(exc)

                if target is None:
                    self._error(ValueError(f"nothing matched selector {selector!r}"))
                    return

                if instr.verb is Verb.SWAP:
                    offsets = {"before": -1, None: 0, "after": 1}

                    if modifier == "in":
                        target.update_children([result])

                    elif modifier in offsets:
                        target.parent.replace(target, result, offset=offsets[modifier])

                    else:
                        self._error(
                            ValueError(
                                f"unknown modifier {modifier!r} for verb {instr.verb!r}"
                            )
                        )
                        return

                elif instr.verb is Verb.INSERT:
                    if modifier == "in":
                        target.insert(0, result)
                    else:
                        index = target.parent.children.index(target)
                        offsets = {"before": index, "after": index + 1}

                        if modifier not in offsets:
                            self._error(
                                ValueError(
                                    f"unknown modifier {modifier!r}"
                                    + f" for verb {instr.verb!r}"
                                )
                            )
                            return

                        target.parent.insert(offsets[modifier], result)

                elif instr.verb is Verb.APPEND:
                    if modifier != "in":
                        self._error(
                            ValueError(
                                f"unknown modifier {modifier!r} for verb {instr.verb!r}"
                            )
                        )
                        return

                    target.append(result)

                # TODO: This is hacky as hell, but we need it for widgets to load in
                #       .styles
                parent = result.parent
                self._init_widget(result)
                result.parent = parent

                continue

            if instr.verb is Verb.SELECT:
                if result is None:
                    self._error(ValueError("no result to select from"))
                    return

                if not isinstance(result, Container):
                    self._error(
                        ValueError(f"cannot select from non container ({result!r})")
                    )
                    return

                try:
                    result = self.find(instr.args[0], scope=result)
                except Exception as exc:
                    self._error(exc)
                    return

                continue

        self._current_instructions.remove(instructions)

    def route(self, destination: str) -> None:
        """Routes to the given URL."""

        destination = self._prefix_endpoint(destination)

        self._http(HTTPMethod.GET, destination, {}, self._xml_page_route)

celx/application.py
shade40/celx
from __future__ import annotations

from typing import TYPE_CHECKING, Any
from lupa import LuaRuntime

from celadon import Widget
from celadon.style_map import StyleMap
from zenith import zml_alias, zml_macro

if TYPE_CHECKING:
    from .application import HttpApplication


def _attr_filter(obj, attr, is_setting):
    if attr.startswith("_"):
        raise AttributeError("access denied")

    return attr


class LuaStyleWrapper:
    def __init__(self, widget: Widget) -> None:
        self._widget = widget

    def __setattr__(self, attr: str, value: Any) -> None:
        if attr == "_widget":
            super().__setattr__(attr, value)
            return

        self._widget.app.rule(
            self._widget.as_query(), score=9999, **{f"{attr}_style": value}
        )

    def __call__(self, widget: Widget) -> LuaStyleWrapper:
        return self.__class__(widget)


def init_runtime(runtime: LuaRuntime, application: "HttpApplication") -> None:
    def _multi_find(selector: str, multiple: bool = False) -> Widget | list[Widget]:
        if multiple:
            return lua.table_from([*application.find_all(selector)])

        return application.find(selector)

    def _zml_define(name: str, macro: MacroType) -> None:
        def _inner(*args) -> str:
            return macro(*args)

        _inner.__name__ = name.replace(" ", "_")
        zml_macro(_inner)

    lua.execute(
        """
        sandbox = {
          ipairs = ipairs,
          next = next,
          pairs = pairs,
          pcall = pcall,
          tonumber = tonumber,
          tostring = tostring,
          type = type,
          unpack = unpack,
          coroutine = { create = coroutine.create, resume = coroutine.resume,
              running = coroutine.running, status = coroutine.status,
              wrap = coroutine.wrap },
          string = { byte = string.byte, char = string.char, find = string.find,
              format = string.format, gmatch = string.gmatch, gsub = string.gsub,
              len = string.len, lower = string.lower, match = string.match,
              rep = string.rep, reverse = string.reverse, sub = string.sub,
              upper = string.upper },
          table = { insert = table.insert, maxn = table.maxn, remove = table.remove,
              sort = table.sort },
          math = { abs = math.abs, acos = math.acos, asin = math.asin,
              atan = math.atan, atan2 = math.atan2, ceil = math.ceil, cos = math.cos,
              cosh = math.cosh, deg = math.deg, exp = math.exp, floor = math.floor,
              fmod = math.fmod, frexp = math.frexp, huge = math.huge,
              ldexp = math.ldexp, log = math.log, log10 = math.log10, max = math.max,
              min = math.min, modf = math.modf, pi = math.pi, pow = math.pow,
              rad = math.rad, random = math.random, sin = math.sin, sinh = math.sinh,
              sqrt = math.sqrt, tan = math.tan, tanh = math.tanh },
          os = { clock = os.clock, difftime = os.difftime, time = os.time },
          scopes = {},
        }
        """
    )

    inject = lua.eval("function(obj, name) sandbox[name] = obj end")

    inject({"alias": zml_alias, "define": _zml_define}, "zml")
    inject(_multi_find, "find")
    inject(LuaStyleWrapper, "styles")


lua = LuaRuntime(
    register_eval=False,
    register_builtins=False,
    attribute_filter=_attr_filter,
)

celx/lua.py
shade40/celx
import re
from xml.etree.ElementTree import Element

from dataclasses import dataclass
from enum import Enum
from typing import Any, Callable
from textwrap import indent, dedent

from celadon import Widget, widgets, load_rules, Page

from .lua import lua

WIDGET_TYPES = {
    key.lower(): value
    for key, value in vars(widgets).items()
    if isinstance(value, type) and issubclass(value, Widget)
}


STYLE_TEMPLATE = """\
{query}:
{indented_content}\
"""

LUA_SCRIPT_TEMPLATE = """\
function(widget)
    local env = {{}}

    for k, v in pairs(sandbox) do
        env[k] = v
    end

    local _ENV = env

    self = widget
    styles = styles(widget)

{indented_content}

    return _ENV
end\
"""


class HTTPMethod(Enum):
    GET = "GET"
    POST = "POST"
    DELETE = "DELETE"
    PUT = "PUT"
    PATCH = "PATCH"


class TreeMethod(Enum):
    INSERT = "INSERT"
    SWAP = "SWAP"
    APPEND = "APPEND"


class Verb(Enum):
    GET = HTTPMethod.GET.value
    POST = HTTPMethod.POST.value
    DELETE = HTTPMethod.DELETE.value
    PUT = HTTPMethod.PUT.value
    PATCH = HTTPMethod.PATCH.value

    INSERT = TreeMethod.INSERT.value
    SWAP = TreeMethod.SWAP.value
    APPEND = TreeMethod.APPEND.value

    SELECT = "SELECT"


@dataclass
class Instruction:
    verb: Verb
    args: tuple[str, ...]


def lua_formatted_get_content(scope: dict[str, Any]) -> Callable[[Widget], list[str]]:
    def _get_content(self) -> None:
        lines = []

        for line in self.__class__.get_content(self):
            in_var = False
            parts = []

            for word in re.split(r"([^a-zA-Z0-9_\.])", line):
                if word == "$":
                    in_var = True
                    continue

                if in_var:
                    in_var = False

                    if "." in word:
                        *scopeids, word = word.split(".")

                        inner = scope

                        for scopeid in scopeids:
                            if scopeid not in inner:
                                raise ValueError(f"unregistered scopeid {scopeid!r}")

                            inner = inner[scopeid]

                        parts.append(str(inner[word]))
                        continue

                    word = str(scope[word])

                parts.append(word)

            lines.append("".join(parts))

        return lines

    return _get_content


def _instruction_runner(instructions: list[Instruction]) -> Callable[[str], bool]:
    """Creates a function to runs the given instructions on the calller widget's app."""

    def _interpret(self: Widget) -> bool:
        self.app.run_instructions(instructions, self)

    return _interpret


def parse_callback(text: str) -> Callable[[str], bool]:
    """Parses a callback descriptor into a list of Instructions."""

    lines = re.split("[;\n]", text)

    instructions = []

    for line in lines:
        verb_str, *args = line.strip().split()
        verb = Verb(verb_str.upper())

        if verb is Verb.SELECT:
            if len(args) > 1:
                raise ValueError(f"too many arguments for verb {verb!r}")

            instructions.append(Instruction(verb, (args[0],)))

        else:
            if len(args) > 2:
                raise ValueError(f"too many arguments for verb {verb!r}")

            modifier = None
            arg = args[0]

            if len(args) == 2:
                modifier, arg = args

            instructions.append(Instruction(verb, (arg, modifier)))

    return _instruction_runner(instructions)


def parse_rules(text: str, query: str | None = None) -> dict[str, Any]:
    """Parses a block of YAML rules into a dictionary."""

    if query is None:
        style = dedent(text)
    else:
        style = STYLE_TEMPLATE.format(
            query=query, indented_content=indent(dedent(text), 4 * " ")
        )

    return load_rules(style)


def parse_widget(node: Element) -> Widget:
    init = {}

    for key, value in node.attrib.items():
        if key == "groups":
            init["groups"] = tuple(value.split(" "))
            continue

        if key.startswith("on-"):
            key = key.replace("-", "_")
            init[key] = [parse_callback(value)]
            continue

        init[key] = value

    text = node.text

    if text is None or text.strip() == "":
        text = ""
        skipped = 0
        total = 0

        for total, child in enumerate(node):
            if child.tail is None:
                skipped += 1
                continue

            text = text + child.tail

        if skipped == total + 1:
            text = None

    cls = WIDGET_TYPES[node.tag]

    if text is not None and text.strip() != "":
        widget = cls(text.strip(), **init)
    else:
        widget = cls(**init)

    query = widget.as_query()
    scope = None

    if (script_node := node.find("lua")) is not None:
        # Set up widget & styles globals
        setup_scope = lua.eval(
            LUA_SCRIPT_TEMPLATE.format(
                indented_content=indent(dedent(script_node.text), 4 * " ")
            )
        )
        scope = setup_scope(widget)

        lua.eval("function(widget, scope) sandbox.scopes[widget] = scope end")(
            widget, scope
        )

        for key, value in scope.items():
            if key.startswith(("pre_", "on_")):
                event = getattr(widget, key)

                if event is None:
                    raise ValueError(f"invalid event handler {key!r}")

                event += value

        node.remove(script_node)

    rules = {}

    # Save current outer scope
    old_outer = lua.eval("sandbox.outer")
    scope = scope or lua.table_from({"outer": lua.eval("sandbox.outer")})

    # Set outer scope to this widget's inner scope
    lua.eval("function(widget) sandbox.outer = sandbox.scopes[widget] end")(widget)

    for child in node:
        if child.tag == "styles":
            rules.update(**parse_rules(child.text, query))
            continue

        parsed, parsed_rules = parse_widget(child)

        widget += parsed
        rules.update(**parsed_rules)

    # Set formatted get content for the widget
    get_content = lua_formatted_get_content(scope)
    widget.get_content = get_content.__get__(widget, widget.__class__)

    # Reset outer scope to what it used to be
    lua.eval("function(scope) sandbox.outer = scope end")(old_outer)

    return widget, rules


def parse_page(node: Element) -> Page:
    page_node = node.find("page")

    page = Page(**page_node.attrib)

    for child in page_node:
        if child.tag in WIDGET_TYPES:
            widget, rules = parse_widget(child)
            page += widget

            for selector, rule in rules.items():
                page.rule(selector, **rule)

        elif child.tag == "styles":
            for selector, rule in parse_rules(child.text).items():
                page.rule(selector, **rule)

        else:
            raise ValueError(child.tag)

    return page

celx/parsing.py
shade40/celx