alterator-python-functions/alterator_python_functions.py

# -*- coding: utf-8 -*-
"""Bindings to write alterator backends in Python

Alterator backends use a very simple communication protocol and can be written in any language.
This library contains a set of functions to simplify development of backend on `Python`.
Main event loop is provided by `message_loop` function. You must provide your own `on_message` callback
(that accepts contents of Alterator's message in `message` argument). 
All incoming parameters are provided as-is
(unlike `alterator-sh-functions`, i.e. without `in_` prefix, exactly like in `alterator-perl-functions`). 
For example, incoming parameter `'foo'` can be found as a key `"foo"` in `message` `dict`.

You should write output parameters using functions with prefix `write_`.
Some parameter names have special meaning. There are: `action`, `_objects`, `language`, `expert_mode`.

See also Project's README.md for more thorough description.

Attributes:
    ALTERATOR_DEBUG (bool): Indicates if additional debug output 
        (to stderr) is needed (messages from alterator, etc.).
        Also with this flag `write_debug` won't be silent
        (will print messages to stderr)

    TEXTDOMAIN (str): Sets dictionary used for translations. 
        If not set manually, warning will be printed and default value
        (running backend file name) will be set
"""

import re
import sys
import os
from io import StringIO

ALTERATOR_DEBUG = False  # May be set from backend for debug-messages
TEXTDOMAIN = None  # Should be set by backend


_LANGUAGE = "en_US"  # Will be set from language parameter
_OUT_BUF = None  # Module-local variable


### Internal function
def _validate_symbol(str: str):
    if not re.match(r"^\w+$", str):
        sys.exit("wrong attribute name: {}".format(str))


### Quote
def string_quote(str: str):
    """Escapes \" and \\"""
    escaped = str.translate(str.maketrans({"\\": r"\\", '"': r"\""}))
    return escaped


### Input/Output functions


def write_string(str: str):
    """Writes string with escaping \" and \\"""
    print(string_quote(str), end="", file=_OUT_BUF)


def get_bool_string(value):
    """Inteprets value as Scheme's boolean (value can be Python's bool or str)"""
    if (
        value == "yes"
        or value == "true"
        or value == "on"
        or value == "y"
        or value == "1"
        or value == "#t"
        or value == True
    ):
        return "#t"
    return "#f"


def test_bool(input: str):
    """Simple wrapper to cast Scheme's boolean to Python's"""
    return input == "#t"


### High-Level Output


def write_string_param(name: str, value: str):
    """Writes attribute's value as string"""
    _validate_symbol(name)
    print('{} "{}"'.format(name, string_quote(value)), file=_OUT_BUF)


def write_bool_param(name: str, value):
    """Writes attribute's value as Scheme's bool"""
    _validate_symbol(name)
    print(name, get_bool_string(value), file=_OUT_BUF)


def write_enum_item(name: str, label: str = None):
    """Writes enum entry (sets label value to name if not given)"""
    _validate_symbol(name)

    if label is None:
        label = name

    print('(name "{}" label "{}")'.format(name, string_quote(label)), file=_OUT_BUF)


def write_table_item(row: dict):
    """Writes table row (dictionary with column-value pairs)"""
    print("(", end="", file=_OUT_BUF)
    for name, value in row.items():
        _validate_symbol(name)

        if value is None:
            value = name

        if value == "#t" or value == "#f":
            print(" {} {}".format(name, string_quote(value)), end="", file=_OUT_BUF)
        else:
            print(' {} "{}"'.format(name, string_quote(value)), end="", file=_OUT_BUF)

    print(")", end="", file=_OUT_BUF)


def write_type_item(name: str, type: str):
    """Declares attribute's type for Alterator"""
    write_string_param(name, type)


def write_error(message: str):
    """Writes an error to Alterator"""
    global _OUT_BUF
    _OUT_BUF = StringIO()
    write_string_param("error", message)


def write_debug(message: str):
    """Writes debug message in ALTERATOR_DEBUG mode"""
    if ALTERATOR_DEBUG:
        print(message, file=sys.stderr)


### Localization Support


def _(text: str, domain=None):
    """Provides text translation from given domain (TEXTDOMAIN by default)"""
    if domain is None:
        domain = TEXTDOMAIN

    lang_list = _LANGUAGE.split(":")
    if len(lang_list) <= 0:
        return text

    return 'LANGUAGE="{}" LANG="{}.UTF8" gettext {} "{}"'.format(
        _LANGUAGE, lang_list[0], domain, text
    )


### Message Loop


# Unimplemented for now
def _check_response(response: str):
    return True


def _redirect_stdout():
    saved_stdout_fd = os.dup(sys.stdout.fileno())
    backendout = os.fdopen(saved_stdout_fd, "w")
    os.dup2(sys.stderr.fileno(), sys.stdout.fileno())
    return backendout


def _ensure_textdomain():
    global TEXTDOMAIN
    if TEXTDOMAIN is None:
        backend_name = os.path.splitext(os.path.basename(sys.argv[0]))[0]
        TEXTDOMAIN = "alterator-" + backend_name
        print(
            "alterator_python_functions: TEXTDOMAIN variable is undefined! Setting to {}".format(
                TEXTDOMAIN
            ),
            file=sys.stderr,
        )


def _have_language(message: dict):
    if "language" in message and message["language"]:
        global _LANGUAGE
        _LANGUAGE = message["language"].replace(";", ":")


def message_loop(handler):
    """Main function to communicate with Alterator"""
    _ensure_textdomain()

    # Redirecting streams (according to Alterator API)
    backendout = _redirect_stdout()

    message = {}
    reading = False
    for line in sys.stdin:
        write_debug(">>>{}".format(line))

        if line == "_message:begin\n":
            message = {}
            reading = True

        elif reading and line == "_message:end\n":
            _have_language(message)

            reading = False
            global _OUT_BUF
            _OUT_BUF = StringIO()

            handler(message)
            response = _OUT_BUF.getvalue()

            # TODO: use _check_response here
            write_debug("response >>>({})<<<".format(response))
            print("({})".format(response), end="", file=backendout, flush=True)

        elif reading:
            name, value = line.split(":", 1)
            value = value.rstrip("\n")

            value = re.sub(r"([^\\])\\n", r"\1\n", value)
            value = re.sub(r"\\\\", r"\\", value)

            message[name] = value