codon/stdlib/internal/python.codon

# (c) 2022 Exaloop Inc. All rights reserved.

import os
from internal.dlopen import *

PyUnicode_AsEncodedString = Function[[cobj, cobj, cobj], cobj](cobj())
PyBytes_AsString = Function[[cobj], cobj](cobj())
PyErr_Fetch = Function[[Ptr[cobj], Ptr[cobj], Ptr[cobj]], void](cobj())
PyObject_GetAttrString = Function[[cobj, cobj], cobj](cobj())
PyObject_GetAttr = Function[[cobj, cobj], cobj](cobj())
PyObject_Str = Function[[cobj], cobj](cobj())
PyRun_SimpleString = Function[[cobj], void](cobj())
Py_IncRef = Function[[cobj], void](cobj())
Py_DecRef = Function[[cobj], void](cobj())
PyObject_Call = Function[[cobj, cobj, cobj], cobj](cobj())
PyObject_SetAttrString = Function[[cobj, cobj, cobj], cobj](cobj())
PyObject_Length = Function[[cobj], int](cobj())
Py_Initialize = Function[[], void](cobj())
PyImport_ImportModule = Function[[cobj], cobj](cobj())
PyLong_FromLong = Function[[int], cobj](cobj())
PyLong_AsLong = Function[[cobj], int](cobj())
PyFloat_FromDouble = Function[[float], cobj](cobj())
PyFloat_AsDouble = Function[[cobj], float](cobj())
PyBool_FromLong = Function[[int], cobj](cobj())
PyObject_IsTrue = Function[[cobj], int](cobj())
PyUnicode_DecodeFSDefaultAndSize = Function[[cobj, int], cobj](cobj())
PyTuple_New = Function[[int], cobj](cobj())
PyTuple_SetItem = Function[[cobj, int, cobj], void](cobj())
PyTuple_GetItem = Function[[cobj, int], cobj](cobj())
PyList_New = Function[[int], cobj](cobj())
PyList_SetItem = Function[[cobj, int, cobj], cobj](cobj())
PyList_GetItem = Function[[cobj, int], cobj](cobj())
PySet_New = Function[[cobj], cobj](cobj())
PySet_Add = Function[[cobj, cobj], cobj](cobj())
PyDict_New = Function[[], cobj](cobj())
PyDict_SetItem = Function[[cobj, cobj, cobj], cobj](cobj())
PyDict_Next = Function[[cobj, Ptr[int], Ptr[cobj], Ptr[cobj]], int](cobj())
PyObject_GetIter = Function[[cobj], cobj](cobj())
PyIter_Next = Function[[cobj], cobj](cobj())
PyObject_HasAttrString = Function[[cobj, cobj], int](cobj())
PyImport_AddModule = Function[[cobj], cobj](cobj())

_PY_MODULE_CACHE = Dict[str, pyobj]()

_PY_INIT = """
import io

clsf = None
clsa = None
plt = None
try:
    import matplotlib.figure
    import matplotlib.pyplot
    plt = matplotlib.pyplot
    clsf = matplotlib.figure.Figure
    clsa = matplotlib.artist.Artist
except ModuleNotFoundError:
    pass

def __codon_repr__(fig):
    if clsf and isinstance(fig, clsf):
        stream = io.StringIO()
        fig.savefig(stream, format="svg")
        return 'image/svg+xml', stream.getvalue()
    elif clsa and isinstance(fig, list) and all(
        isinstance(i, clsa) for i in fig
    ):
        stream = io.StringIO()
        plt.gcf().savefig(stream, format="svg")
        return 'image/svg+xml', stream.getvalue()
    elif hasattr(fig, "_repr_html_"):
        return 'text/html', fig._repr_html_()
    else:
        return 'text/plain', fig.__repr__()
"""

_PY_INITIALIZED = False


def init() -> void:
    global _PY_INITIALIZED
    if _PY_INITIALIZED:
        return

    LD = os.getenv("CODON_PYTHON", default="libpython." + dlext())
    hnd = dlopen(LD, RTLD_LOCAL | RTLD_NOW)

    global PyUnicode_AsEncodedString
    PyUnicode_AsEncodedString = dlsym(hnd, "PyUnicode_AsEncodedString")
    global PyBytes_AsString
    PyBytes_AsString = dlsym(hnd, "PyBytes_AsString")
    global PyErr_Fetch
    PyErr_Fetch = dlsym(hnd, "PyErr_Fetch")
    global PyObject_GetAttrString
    PyObject_GetAttrString = dlsym(hnd, "PyObject_GetAttrString")
    global PyObject_GetAttr
    PyObject_GetAttr = dlsym(hnd, "PyObject_GetAttr")
    global PyObject_Str
    PyObject_Str = dlsym(hnd, "PyObject_Str")
    global PyRun_SimpleString
    PyRun_SimpleString = dlsym(hnd, "PyRun_SimpleString")
    global Py_IncRef
    Py_IncRef = dlsym(hnd, "Py_IncRef")
    global Py_DecRef
    Py_DecRef = dlsym(hnd, "Py_DecRef")
    global PyObject_Call
    PyObject_Call = dlsym(hnd, "PyObject_Call")
    global PyObject_SetAttrString
    PyObject_SetAttrString = dlsym(hnd, "PyObject_SetAttrString")
    global PyObject_Length
    PyObject_Length = dlsym(hnd, "PyObject_Length")
    global Py_Initialize
    Py_Initialize = dlsym(hnd, "Py_Initialize")
    global PyImport_ImportModule
    PyImport_ImportModule = dlsym(hnd, "PyImport_ImportModule")
    global PyLong_FromLong
    PyLong_FromLong = dlsym(hnd, "PyLong_FromLong")
    global PyLong_AsLong
    PyLong_AsLong = dlsym(hnd, "PyLong_AsLong")
    global PyFloat_FromDouble
    PyFloat_FromDouble = dlsym(hnd, "PyFloat_FromDouble")
    global PyFloat_AsDouble
    PyFloat_AsDouble = dlsym(hnd, "PyFloat_AsDouble")
    global PyBool_FromLong
    PyBool_FromLong = dlsym(hnd, "PyBool_FromLong")
    global PyObject_IsTrue
    PyObject_IsTrue = dlsym(hnd, "PyObject_IsTrue")
    global PyUnicode_DecodeFSDefaultAndSize
    PyUnicode_DecodeFSDefaultAndSize = dlsym(hnd, "PyUnicode_DecodeFSDefaultAndSize")
    global PyTuple_New
    PyTuple_New = dlsym(hnd, "PyTuple_New")
    global PyTuple_SetItem
    PyTuple_SetItem = dlsym(hnd, "PyTuple_SetItem")
    global PyTuple_GetItem
    PyTuple_GetItem = dlsym(hnd, "PyTuple_GetItem")
    global PyList_New
    PyList_New = dlsym(hnd, "PyList_New")
    global PyList_SetItem
    PyList_SetItem = dlsym(hnd, "PyList_SetItem")
    global PyList_GetItem
    PyList_GetItem = dlsym(hnd, "PyList_GetItem")
    global PySet_New
    PySet_New = dlsym(hnd, "PySet_New")
    global PySet_Add
    PySet_Add = dlsym(hnd, "PySet_Add")
    global PyDict_New
    PyDict_New = dlsym(hnd, "PyDict_New")
    global PyDict_SetItem
    PyDict_SetItem = dlsym(hnd, "PyDict_SetItem")
    global PyDict_Next
    PyDict_Next = dlsym(hnd, "PyDict_Next")
    global PyObject_GetIter
    PyObject_GetIter = dlsym(hnd, "PyObject_GetIter")
    global PyIter_Next
    PyIter_Next = dlsym(hnd, "PyIter_Next")
    global PyObject_HasAttrString
    PyObject_HasAttrString = dlsym(hnd, "PyObject_HasAttrString")
    global PyImport_AddModule
    PyImport_AddModule = dlsym(hnd, "PyImport_AddModule")

    Py_Initialize()
    PyRun_SimpleString(_PY_INIT.c_str())
    _PY_INITIALIZED = True


def ensure_initialized() -> void:
    if not _PY_INITIALIZED:
        init()
        # raise ValueError("Python not initialized; make sure to 'import python'")


@extend
class pyobj:
    def __new__(p: Ptr[byte]) -> pyobj:
        return (p,)

    def _getattr(self, name: str) -> pyobj:
        return pyobj.exc_wrap(pyobj(PyObject_GetAttrString(self.p, name.c_str())))

    def __setitem__(self, name: str, val: pyobj) -> pyobj:
        return pyobj.exc_wrap(
            pyobj(PyObject_SetAttrString(self.p, name.c_str(), val.p))
        )

    def __len__(self) -> int:
        return pyobj.exc_wrap(PyObject_Length(self.p))

    def __to_py__(self) -> pyobj:
        return self

    def __from_py__(self) -> pyobj:
        return self

    def __str__(self) -> str:
        return str.__from_py__(self._getattr("__str__").__call__())

    def __repr__(self) -> str:
        return str.__from_py__(self._getattr("__repr__").__call__())

    def __iter__(self) -> Generator[pyobj]:
        it = PyObject_GetIter(self.p)
        if not it:
            raise ValueError("Python object is not iterable")
        while i := PyIter_Next(it):
            yield pyobj(pyobj.exc_wrap(i))
            pyobj(i).decref()
        pyobj(it).decref()
        pyobj.exc_check()

    def to_str(self, errors: str, empty: str = "") -> str:
        obj = PyUnicode_AsEncodedString(self.p, "utf-8".c_str(), errors.c_str())
        if obj == cobj():
            return empty
        bts = PyBytes_AsString(obj)
        pyobj(obj).decref()
        return str.from_ptr(bts)

    def exc_check() -> void:
        ptype, pvalue, ptraceback = cobj(), cobj(), cobj()
        PyErr_Fetch(__ptr__(ptype), __ptr__(pvalue), __ptr__(ptraceback))
        if ptype != cobj():
            py_msg = PyObject_Str(pvalue) if pvalue != cobj() else pvalue
            msg = pyobj(py_msg).to_str("ignore", "<empty Python message>")
            typ = pyobj.to_str(
                pyobj(PyObject_GetAttrString(ptype, "__name__".c_str())), "ignore"
            )

            pyobj(ptype).decref()
            pyobj(pvalue).decref()
            pyobj(ptraceback).decref()
            pyobj(py_msg).decref()

            raise PyError(msg, typ)

    def exc_wrap(_retval: T, T: type) -> T:
        pyobj.exc_check()
        return _retval

    def incref(self) -> void:
        Py_IncRef(self.p)

    def decref(self) -> void:
        Py_DecRef(self.p)

    def __call__(self, *args, **kwargs) -> pyobj:
        names = iter(kwargs.__dict__())
        kws = Dict[str, pyobj]()
        if staticlen(kwargs) > 0:
            kws = {next(names): i.__to_py__() for i in kwargs}
        return pyobj.exc_wrap(
            pyobj(PyObject_Call(self.p, args.__to_py__().p, kws.__to_py__().p))
        )

    def _tuple_new(length: int) -> pyobj:
        t = PyTuple_New(length)
        pyobj.exc_check()
        return pyobj(t)

    def _tuple_set(self, idx: int, val: pyobj) -> void:
        PyTuple_SetItem(self.p, idx, val.p)
        pyobj.exc_check()

    def _tuple_get(self, idx: int) -> pyobj:
        t = PyTuple_GetItem(self.p, idx)
        pyobj.exc_check()
        return pyobj(t)

    def _import(name: str) -> pyobj:
        ensure_initialized()
        if name in _PY_MODULE_CACHE:
            return _PY_MODULE_CACHE[name]
        m = pyobj.exc_wrap(pyobj(PyImport_ImportModule(name.c_str())))
        _PY_MODULE_CACHE[name] = m
        return m

    def _exec(code: str) -> void:
        ensure_initialized()
        PyRun_SimpleString(code.c_str())

    def get(self, T: type) -> T:
        return T.__from_py__(self)

    def _main_module() -> pyobj:
        m = PyImport_AddModule("__main__".c_str())
        return pyobj(m)

    def _repr_mimebundle_(self, bundle=Set[str]()) -> Dict[str, str]:
        fn = pyobj._main_module()._getattr("__codon_repr__")
        assert fn.p != cobj(), "cannot find python.__codon_repr__"
        mime, txt = fn.__call__(self).get(Tuple[str, str])
        return {mime: txt}


def none():
    raise NotImplementedError()


# Type conversions


def py(x) -> pyobj:
    return x.__to_py__()


def get(x: pyobj, T: type) -> T:
    return T.__from_py__(x)


@extend
class int:
    def __to_py__(self) -> pyobj:
        return pyobj.exc_wrap(pyobj(PyLong_FromLong(self)))

    def __from_py__(i: pyobj) -> int:
        return pyobj.exc_wrap(PyLong_AsLong(i.p))


@extend
class float:
    def __to_py__(self) -> pyobj:
        return pyobj.exc_wrap(pyobj(PyFloat_FromDouble(self)))

    def __from_py__(d: pyobj) -> float:
        return pyobj.exc_wrap(PyFloat_AsDouble(d.p))


@extend
class bool:
    def __to_py__(self) -> pyobj:
        return pyobj.exc_wrap(pyobj(PyBool_FromLong(int(self))))

    def __from_py__(b: pyobj) -> bool:
        return pyobj.exc_wrap(PyObject_IsTrue(b.p)) != 0


@extend
class byte:
    def __to_py__(self) -> pyobj:
        return str.__to_py__(str(__ptr__(self), 1))

    def __from_py__(c: pyobj) -> byte:
        return str.__from_py__(c).p[0]


@extend
class str:
    def __to_py__(self) -> pyobj:
        return pyobj.exc_wrap(
            pyobj(PyUnicode_DecodeFSDefaultAndSize(self.ptr, self.len))
        )

    def __from_py__(s: pyobj) -> str:
        r = s.to_str("strict")
        pyobj.exc_check()
        return r


@extend
class List:
    def __to_py__(self) -> pyobj:
        pylist = PyList_New(len(self))
        pyobj.exc_check()
        idx = 0
        for a in self:
            PyList_SetItem(pylist, idx, py(a).p)
            pyobj.exc_check()
            idx += 1
        return pyobj(pylist)

    def __from_py__(v: pyobj) -> List[T]:
        n = v.__len__()
        t = List[T](n)
        for i in range(n):
            elem = pyobj(PyList_GetItem(v.p, i))
            pyobj.exc_check()
            t.append(get(elem, T))
        return t


@extend
class Dict:
    def __to_py__(self) -> pyobj:
        pydict = PyDict_New()
        pyobj.exc_check()
        for k, v in self.items():
            PyDict_SetItem(pydict, py(k).p, py(v).p)
            pyobj.exc_check()
        return pyobj(pydict)

    def __from_py__(d: pyobj) -> Dict[K, V]:
        b = Dict[K, V]()
        pos = 0
        k_ptr = cobj()
        v_ptr = cobj()
        while PyDict_Next(d.p, __ptr__(pos), __ptr__(k_ptr), __ptr__(v_ptr)):
            pyobj.exc_check()
            k = get(pyobj(k_ptr), K)
            v = get(pyobj(v_ptr), V)
            b[k] = v
        return b


@extend
class Set:
    def __to_py__(self) -> pyobj:
        pyset = PySet_New(cobj())
        pyobj.exc_check()
        for a in self:
            PySet_Add(pyset, py(a).p)
            pyobj.exc_check()
        return pyobj(pyset)