###
# Modern Albufeira Prolog Interpreter
#
# Warranty & Liability
# To the extent permitted by applicable law and unless explicitly
# otherwise agreed upon, XLOG Technologies AG makes no warranties
# regarding the provided information. XLOG Technologies AG assumes
# no liability that any problems might be solved with the information
# provided by XLOG Technologies AG.
#
# Rights & License
# All industrial property rights regarding the information - copyright
# and patent rights in particular - are the sole property of XLOG
# Technologies AG. If the company was not the originator of some
# excerpts, XLOG Technologies AG has at least obtained the right to
# reproduce, change and translate the information.
#
# Reproduction is restricted to the whole unaltered document. Reproduction
# of the information is only allowed for non-commercial uses. Selling,
# giving away or letting of the execution of the library is prohibited.
# The library can be distributed as part of your applications and libraries
# for execution provided this comment remains unchanged.
#
# Restrictions
# Only to be distributed with programs that add significant and primary
# functionality to the library. Not to be distributed with additional
# software intended to replace any components of the library.
#
# Trademarks
# Jekejeke is a registered trademark of XLOG Technologies AG.
##

import nova.store as store
from nova.store import (MASK_PRED_DYNAMIC, set_stage, pred_link,
     Clause, is_provable, remove_clause, is_logical, is_cache,
     set_partition, pred_destroy, clear, add_clause, Cache,
     make_defined, add, pred_touch, snapshot_data, is_goal,
     Variable, Compound, is_structure, Goal, is_clause,
     Place, Quote, Skeleton, unquote_objects, Frozen,
     deref, copy_term, MASK_PRED_BITS)
import nova.machine as machine
from nova.machine import (cont, gc_major,
     make_indicator, lookup_pred, exec_body, exec_head,
     defined_clauses, make_indicator_term, is_atom,
     Choice, more, unbind, exec_unify, exec_build,
     check_nonvar, VOID_ARGS, make_error, unify, register_signal,
     melt_directive, check_callable)
from nova.special import (check_atom, make_check,
     check_integer, make_special, list_objects, solve_seq)
import os
import platform
import asyncio
import stat
import errno

MAX_BUF = 4096

bootbase = ""


###
# Set the boot base.
#
# @param url The boot base.
##
def set_bootbase(url):
    global bootbase
    bootbase = url


###################################################################
# current_output/1 and current_error/1                            #
###################################################################

MASK_DST_COLR = 0x00000004
MASK_DST_CYCL = 0x00000008
MASK_DST_FEAT = MASK_DST_COLR | MASK_DST_CYCL


###
# Create a text output.
##
class Sink:
    def __init__(self):
        self.buf = ""
        self.send = lambda fd, buf: fd
        self.last = -1
        self.flags = 0
        self.notify = lambda fd: None
        self.release = lambda fd: None
        self.data = NotImplemented
        self.indent = 0


###
# current_output(S): [ISO 8.11.2]
# The built-in succeeds in S with the current output.
##
def test_current_output(args):
    alpha = store.engine.text_output
    return exec_unify(args[0], alpha)


###
# current_error(S):
# The built-in succeeds in S with the current error.
##
def test_current_error(args):
    alpha = store.engine.text_error
    return exec_unify(args[0], alpha)


###################################################################
# set_output/1 and set_error/1                                    #
###################################################################

###
# set_output(S): [ISO 8.11.4]
# The built-in succeeds. As a side effect the current output is set to S.
##
def test_set_output(args):
    obj = exec_build(args[0])
    check_sink(obj)
    store.engine.text_output = obj
    return True


###
# set_error(S):
# The built-in succeeds. As a side effect the current error is set to S.
##
def test_set_error(args):
    obj = exec_build(args[0])
    check_sink(obj)
    store.engine.text_error = obj
    return True


###
# Assure that the object is a sink.
#
# @param beta The object.
##
def check_sink(beta):
    if not isinstance(beta, Sink):
        check_nonvar(beta)
        beta = copy_term(beta)
        raise make_error(Compound("type_error", ["writer", beta]))


################################################################
# put_atom/2                                                   #
################################################################

###
# put_atom(S, A):
# The built-in succeeds. As a side effect, it adds
# the atom to the stream S.
##
def test_put_atom(args):
    stream = exec_build(args[0])
    check_sink(stream)
    text = exec_build(args[1])
    check_atom(text)
    put_atom(stream, text)
    return True


def put_atom(stream, text):
    if len(text) > 0:
        stream.last = last_code(text)
        if stream.buf is not None:
            stream.buf += text
            if len(stream.buf) >= MAX_BUF:
                flush_buffer(stream)
        else:
            stream.data = stream.send(stream.data, text)


def last_code(text):
    return ord(text[len(text)-1])


def flush_buffer(stream):
    if stream.buf is not None and len(stream.buf) > 0:
        text = stream.buf
        stream.buf = ""
        stream.data = stream.send(stream.data, text)


################################################################
# current_input/1 and set_input/1                              #
################################################################

MASK_SRC_SKIP = 0x00000001
MASK_SRC_AREAD = 0x00000002


###
# Create a text input.
##
class Source:
    def __init__(self):
        self.buf = ""
        self.pos = 0
        self.receive = lambda fd: ""
        self.flags = 0
        self.lineno = 0
        self.release = lambda fd: None
        self.data = NotImplemented


def illegal_state():
    raise make_error(Compound("resource_error", ["illegal_state"]))


###
# current_input(S): [ISO 8.11.1]
# The built-in succeeds in S with the current input.
##
def test_current_input(args):
    alpha = store.engine.text_input
    return exec_unify(args[0], alpha)


###
# set_input(S): [ISO 8.11.3]
# The built-in succeeds. As a side effect it sets the current input to S.
##
def test_set_input(args):
    obj = exec_build(args[0])
    check_source(obj)
    store.engine.text_input = obj
    return True


###
# Assure that the object is a source.
#
# @param beta The object.
##
def check_source(beta):
    if not isinstance(beta, Source):
        check_nonvar(beta)
        beta = copy_term(beta)
        raise make_error(Compound("type_error", ["reader", beta]))


################################################################
# os_get_code/2 and os_peek_code/2                             #
################################################################

###
# os_get_code(S, C):
# The predicate succeeds in C with the Unicode point from the stream buffer S.
# As a side effect the stream position is advanced.
##
def test_os_get_code(args):
    stream = exec_build(args[0])
    pos = stream.pos
    buf = stream.buf
    if pos < len(buf):
        ch = ord(buf[pos])
        pos += 1
        if ch == 13 or (ch == 10 and (stream.flags & MASK_SRC_SKIP) == 0):
            stream.lineno += 1
        if ch == 13:
            stream.flags |= MASK_SRC_SKIP
        else:
            stream.flags &= ~MASK_SRC_SKIP
        stream.pos = pos
        return exec_unify(args[1], ch)
    else:
        return False


###
# os_peek_code(S, C):
# The built-in succeeds in C with the Unicode point from the stream buffer S.
##
def test_os_peek_code(args):
    stream = exec_build(args[0])
    pos = stream.pos
    buf = stream.buf
    if pos < len(buf):
        ch = ord(buf[pos])
        return exec_unify(args[1], ch)
    else:
        return False


################################################################
# os_open_promise_opts/4                                       #
################################################################

###
# os_open_promise_opts(P, L, S, Q):
# The predicate succeeds in Q with a promise for open input S
# on path P and option list L.
##
def test_os_open_promise_opts(args):
    url = exec_build(args[0])
    check_atom(url)
    opts = exec_build(args[1])
    stream = Source()
    if not exec_unify(args[2], stream):
        return False
    buf = machine.ctx
    prom = open_file_promise_opts(buf, stream, url, opts)
    return exec_unify(args[3], lambda: prom)


def get_encoding(opts):
    if opts is not None and "encoding" in opts:
        return opts["encoding"]
    else:
        return "utf8"


async def open_file_promise_opts(buf, stream, url, opts):
    enc = get_encoding(opts)
    try:
        file = await asyncio.to_thread(open, url, encoding=enc, newline='')
    except OSError as err:
        register_signal(buf, map_file_error(err, url))
        return
    stream.data = file
    stream.receive = file_read_promise
    stream.release = file_close_promise
    stream.flags |= MASK_SRC_AREAD


def map_file_error(err, url):
    if hasattr(err, 'reason'):
        code = err.reason.errno
        if code == 11001:
            return Compound("resource_error", ["unknown_host"])
        else:
            return Compound("resource_error", ["connect_failed"])
    else:
        code = err.errno
        if code == errno.ENOENT or errno.EACCES:
            return Compound("existence_error", ["source_sink", url])
        else:
            return Compound("resource_error", ["remote_error"])


###################################################################
# os_read_promise/2 and os_close_promise/2                        #
###################################################################


###
# os_read_promise(S, P):
# The predicate suceeds in P with a read promise for a input S.
##
def test_os_read_promise(args):
    stream = exec_build(args[0])
    buf = machine.ctx
    return exec_unify(args[1], lambda: stream.receive(buf, stream))


async def file_read_promise(buf, stream):
    try:
        res = await asyncio.to_thread(blocking_read, stream.data)
        stream.buf = res
        stream.pos = 0
    except IOError as err:
        register_signal(buf, map_stream_error(err))


def blocking_read(data):
    if data.fileno() == 0:
        return data.readline(8192)
    else:
        return data.read(8192)


###
# Map a stream error when in transit.
#
# @param err The offending error.
# @return The Prolog error term.
##
def map_stream_error(err):
    code = err.errno
    if code == errno.ETIMEDOUT:
        return Compound("resource_error", ["socket_timeout"])
    elif code == errno.EADDRINUSE:
        return Compound("resource_error", ["port_error"])
    elif code == errno.ECONNRESET:
        return Compound("resource_error", ["remote_error"])
    else:
        return Compound("resource_error", ["io_exception"])


###
# os_close_promise(S, P):
# The predicate suceeds in P with a read promise for a input S.
##
def test_os_close_promise(args):
    stream = exec_build(args[0])
    buf = machine.ctx
    return exec_unify(args[1], lambda: stream.release(buf, stream))


async def file_close_promise(buf, stream):
    try:
        await asyncio.to_thread(blocking_close, stream.data)
    except IOError as err:
        register_signal(buf, map_stream_error(err))


def blocking_close(data):
    return data.close()


################################################################
# os_prop_promise/3                                            #
################################################################

###
# os_prop_promise(F, M, Q):
# The predicate succeeds in Q with with a promise for the
# properties M of the file F. Barks if path F doesn't exist
# or io exception while resolving.
##
def test_os_prop_promise(args):
    url = exec_build(args[0])
    check_atom(url)
    res = {}
    if not exec_unify(args[1], res):
        return False
    buf = machine.ctx
    prom = prop_file_promise(buf, url, res)
    return exec_unify(args[2], lambda: prom)


async def prop_file_promise(buf, url, res):
    try:
        stats = await asyncio.to_thread(os.stat, url)
        ftype = "regular" if stat.S_ISREG(stats.st_mode) else \
            ("directory" if stat.S_ISDIR(stats.st_mode) else "other")

        res["last_modified"] = stats.st_mtime_ns // 1000000
        res["absolute_path"] = os.path.abspath(url)
        res["type"] = ftype
        res["can_write"] = "true" if (stats.st_mode & 0x80) != 0 else "false"
    except OSError as err:
        register_signal(buf, map_file_error(err, url))
        return


################################################################
# os_stream_flags/2 and os_open_sync_opts/4                    #
################################################################

###
# os_stream_flags(S, F):
# The predicate succeeds in F with the flags of the stream S.
##
def test_os_stream_flags(args):
    obj = exec_build(args[0])
    if isinstance(obj, Source):
        flags = obj.flags
    else:
        check_sink(obj)
        flags = obj.flags
    return exec_unify(args[1], flags)


###
# os_open_sync_opts(P, M, L, S):
# The predicate succeeds. As a side effect the stream S is
# opened on the path P with the mode M and the option list L.
##
def test_os_open_sync_opts(args):
    url = exec_build(args[0])
    check_atom(url)
    mode = exec_build(args[1])
    check_atom(mode)
    opts = exec_build(args[2])
    if "read" == mode:
        raise make_error(Compound("resource_error",
                                  ["not_implemented"]))
    elif "write" == mode:
        stream = open_write(url, "w", opts)
    elif "append" == mode:
        stream = open_write(url, "a", opts)
    else:
        raise make_error(Compound("domain_error",
                                  ["io_mode", mode]))
    return exec_unify(args[3], stream)


def open_write(url, mode, opts):
    enc = get_encoding(opts)
    try:
        file = open(url, mode, encoding=enc, newline='')
    except OSError as err:
        raise make_error(map_file_error(err, url))
    dst = Sink()
    dst.data = file
    dst.send = file_write
    dst.release = file_close
    return dst


def file_write(data, buf):
    try:
        data.write(buf)
        return data
    except IOError as err:
        raise make_error(map_stream_error(err))


def file_close(data):
    try:
        data.close()
    except IOError as err:
        raise make_error(map_stream_error(err))


################################################################
# flush_output/1 and os_close_sync/1                           #
################################################################

###
# flush_output(S): [ISO 8.11.7]
# The built-in succeeds. As a side effect, it flushes the stream buffer.
##
def test_flush_output(args):
    stream = exec_build(args[0])
    check_sink(stream)
    stream_flush(stream)
    return True


def stream_flush(stream):
    flush_buffer(stream)
    stream.notify(stream.data)


###
# close(S): [ISO 8.11.6]
# The built-in succeeds. As a side effect, the stream S is closed.
##
def test_os_close_sync(args):
    stream = exec_build(args[0])
    if isinstance(stream, Sink):
        pass
    else:
        check_source(stream)
    stream_close(stream)
    return True


def stream_close(stream):
    if isinstance(stream, Sink):
        flush_buffer(stream)
        stream.release(stream.data)
    else:
        stream.release(stream.data)


################################################################
# os_cntrl_sync/2 and os_read_sync/1                           #
################################################################

###
# os_cntrl_sync(F, P):
# The predicate assigns the property P to the file F.
##
def test_os_cntrl_sync(args):
    url = exec_build(args[0])
    check_atom(url)
    prop = exec_build(args[1])
    if (is_structure(prop) and
            prop.functor == "last_modified"
            and len(prop.args) == 1):
        val = deref(prop.args[0])
        check_integer(val)
        if val < 0:
            raise make_error(Compound("domain_error",
                     ["not_less_than_zero", val]))
        try:
            os.utime(url, (val/1000, val/1000))
        except OSError as err:
            raise make_error(map_file_error(err, url))
    elif (is_structure(prop) and
            prop.functor == "can_write"
            and len(prop.args) == 1):
        val = deref(prop.args[0])
        check_atom(val)
        if "true" != val and "false" != val:
            raise make_error(Compound("domain_error",
                     ["boolean", val]))
        try:
            stats = os.stat(url)
            if "true" == val:
                os.chmod(url, (stats.st_mode | 0x80))
            else:
                os.chmod(url, (stats.st_mode & ~0x80))
        except OSError as err:
            raise make_error(map_file_error(err, url))
    else:
        check_nonvar(prop)
        prop = copy_term(prop)
        raise make_error(Compound("domain_error", ["prolog_property", prop]))
    return True


###
# os_read_sync(S):
# The predicate succeeds. As a side effect the stream buffer is read.
##
def test_os_read_sync(args):
    stream = exec_build(args[0])
    stream.buf = stream.receive(stream.data)
    stream.pos = 0
    return True


################################################################
# ir_anon_new/1, ir_place_new/2 and ir_skeleton_new/3          #
################################################################

###
# ir_anon_new(I, S):
# The predicate succeeds in S with a new annonymous variable.
##
def test_ir_anon_new(args):
    alpha = NotImplemented
    return exec_unify(args[0], Quote(alpha))


###
# ir_place_new(I, S):
# The predicate succeeds in S with a new place for index I.
##
def test_ir_place_new(args):
    alpha = exec_build(args[0])
    check_integer(alpha)
    alpha = Place(alpha)
    return exec_unify(args[1], Quote(alpha))


###
# ir_skeleton_new(F, L, S):
# The predicate succeeds in S with a new
# skeleton for functor F and list L.
##
def test_ir_skeleton_new(args):
    alpha = exec_build(args[0])
    beta = exec_build(args[1])
    beta = list_objects(beta)
    unquote_objects(beta)
    alpha = Skeleton(alpha, beta)
    return exec_unify(args[2], Quote(alpha))


################################################################
# ir_make_frozen/2                                             #
################################################################

###
# ir_make_frozen(F, L, S):
# The predicate succeeds in S with a new frozen
# compound for functor F and list L.
##
def test_ir_make_frozen(args):
    alpha = exec_build(args[0])
    beta = exec_build(args[1])
    beta = list_objects(beta)
    alpha = Frozen(alpha, beta)
    return exec_unify(args[2], alpha)


################################################################
# ir_pred_site/2, ir_is_site/1 and ir_site_name/2              #
################################################################

###
# ir_pred_site(X, Y): internal only
# The built-in succeeds in Y with a cacheable X.
##
def test_ir_pred_site(args):
    alpha = exec_build(args[0])
    if is_structure(alpha):
        if is_atom(alpha.functor):
            alpha = Compound(Cache(alpha.functor), alpha.args)
    elif is_atom(alpha):
        alpha = Cache(alpha)
    return exec_unify(args[1], alpha)


###
# ir_is_site(X): internal only
# The built-in succeeds if X is cacheable.
##
def test_ir_is_site(args):
    alpha = exec_build(args[0])
    if is_structure(alpha):
        return is_cache(alpha.functor)
    return is_cache(alpha)


###
# ir_site_name(X, Y): internal only
# The built-in succeeds in Y with a cachefree X.
##
def test_ir_site_name(args):
    alpha = exec_build(args[0])
    if is_structure(alpha):
        if is_cache(alpha.functor):
            alpha = Compound(alpha.functor.name, alpha.args)
    elif is_cache(alpha):
        alpha = alpha.name
    return exec_unify(args[1], alpha)


################################################################
# ir_clause_new/5 and ir_clause_add/4                          #
################################################################

###
# ir_clause_new(S, H, B, R, D, C):
# The built-in succeeds in C with a Java object representing
# a clause with variable count S, head instructions H, body instructions B,
# cut variable index R and head index option D.
##
def test_ir_clause_new(args):
    alpha = exec_build(args[0])
    check_integer(alpha)
    beta = exec_build(args[1])
    beta = list_objects(beta)
    gamma = exec_build(args[2])
    gamma = list_objects(gamma)
    delta = exec_build(args[3])
    check_integer(delta)
    mue = exec_build(args[4])
    if is_structure(mue) and mue.functor == "just" and len(mue.args) == 1:
        mue = deref(mue.args[0])
    else:
        mue = NotImplemented
    unquote_objects(beta)
    unquote_objects(gamma)
    return unify(args[5], Clause(alpha, beta, gamma, delta, mue))


###
# ir_clause_add(F, A, C, O): internal only
# The built-in succeeds. As a side effect the Python object clause C
# is added according to options O to the knowledge base for the predicate
# indicator F/A. If C is not a clause, but directly a Python object
# provable, the built-in definition of the indicator F/A is updated.
##
def test_ir_clause_add(args):
    functor = exec_build(args[0])
    check_atom(functor)
    arity = exec_build(args[1])
    check_integer(arity)
    gamma = exec_build(args[2])
    flags = exec_build(args[3])
    check_integer(flags)
    add_clause(functor, arity, gamma, flags)
    return True


################################################################
# ir_goal_new/4 and ir_goal_call/1                             #
################################################################

###
# ir_goal_new(S, B, R, G):
# The built-in succeeds in G with a Python object representing
# a goal with variable count S, body instructions B and the
# cut variable index R.
##
def test_ir_goal_new(args):
    alpha = exec_build(args[0])
    check_integer(alpha)
    beta = exec_build(args[1])
    beta = list_objects(beta)
    gamma = exec_build(args[2])
    check_integer(gamma)
    unquote_objects(beta)
    return exec_unify(args[3], Goal(alpha, beta, gamma))


###
# ir_goal_call(A):
# The predicate succeeds whenever the compiled goal A succeeds.
##
def special_ir_goal_call(args):
    goal = deref(args[0])
    check_goal(goal)
    goal = melt_directive(goal)
    solve_seq(goal)
    return True


def check_goal(beta):
    if not is_goal(beta):
        check_nonvar(beta)
        beta = copy_term(beta)
        raise make_error(Compound("type_error", ["goal", beta]))


################################################################
# kb_clause_ref/3 and kb_pred_touch/3                          #
################################################################

MASK_FIND_MODIFY = 0x00000001
MASK_FIND_DYNAMIC = 0x00000002
MASK_FIND_REVERSE = 0x00000004


###
# kb_clause_ref(H, F, C): internal only
# The built-in succeeds in C with the clause references
# for the head H and the flags F.
##
def special_kb_clause_ref(args):
    head = deref(args[0])
    check_callable(head)
    flags = deref(args[1])
    check_integer(flags)
    peek = lookup_pred(head)
    if peek is NotImplemented:
        return False
    if (flags & MASK_FIND_DYNAMIC) != 0:
        if (peek.flags & MASK_PRED_DYNAMIC) == 0:
            make_error_find(head, flags)
    if not is_logical(peek.rope):
        return False
    if is_structure(head):
        head = head.args
    else:
        head = VOID_ARGS
    peek = defined_clauses(peek, head)
    peek = snapshot_data(peek)
    if (flags & MASK_FIND_REVERSE) == 0:
        return solve2_ref(args, peek, 0, None)
    else:
        return solve2_ref_reverse(args, peek, len(peek), None)


def make_error_find(head, flags):
    if (flags & MASK_FIND_MODIFY) != 0:
        raise make_error(Compound("permission_error",
            ["modify", "static_procedure", make_indicator_term(head)]))
    else:
        raise make_error(Compound("permission_error",
            ["access", "private_procedure", make_indicator_term(head)]))


def solve_ref(rope, at, choice):
    goal = deref(machine.call.args[0])
    return solve2_ref(goal.args, rope, at, choice)


def solve_ref_reverse(rope, at, choice):
    goal = deref(machine.call.args[0])
    return solve2_ref_reverse(goal.args, rope, at, choice)


###
# Search a Prolog clause and return it.
#
# @param args The current arguments.
# @param rope The clause list.
# @param at The clause index.
# @param choice The choice point for reuse or null.
# @return True if search succeeds, otherwise false.
##
def solve2_ref(args, rope, at, choice):
    mark = machine.trail
    while at < len(rope):
        clause = rope[at]
        at += 1
        if unify(args[2], clause):
            if at < len(rope):
                if choice is None:
                    choice = Choice(solve_ref, rope, at, mark)
                else:
                    choice.at = at
                more(choice)
            cont(machine.call.args[1])
            return True
        unbind(mark)
    return False


###
# Search a Prolog clause backwards and return it.
#
# @param args The current arguments.
# @param rope The clause list.
# @param at The clause index.
# @param choice The choice point for reuse or null.
# @return True if search succeeds, otherwise false.
##
def solve2_ref_reverse(args, rope, at, choice):
    mark = machine.trail
    while at > 0:
        at -= 1
        clause = rope[at]
        if unify(args[2], clause):
            if at > 0:
                if choice is None:
                    choice = Choice(solve_ref_reverse, rope, at, mark)
                else:
                    choice.at = at
                more(choice)
            cont(machine.call.args[1])
            return True
        unbind(mark)
    return False


###
# kb_pred_touch(F, N, O): internal only
# The built-in succeeds. As a side effect the predicate
# indicator F/N with options O is touched.
##
def test_kb_pred_touch(args):
    functor = exec_build(args[0])
    check_atom(functor)
    arity = exec_build(args[1])
    check_integer(arity)
    flags = exec_build(args[2])
    check_integer(flags)
    pred_touch(functor, arity, flags)
    return True


################################################################
# kb_clause_remove/4 and kb_pred_destroy/2                     #
################################################################

###
# kb_clause_remove(F, N, G, R): internal only
# The built-in succeeds if the clause or predicate R could
# be removed from the predicate indicator F/N and the flags G.
##
def test_kb_clause_remove(args):
    functor = exec_build(args[0])
    check_atom(functor)
    arity = exec_build(args[1])
    check_integer(arity)
    flags = exec_build(args[2])
    check_integer(flags)
    clause = exec_build(args[3])
    check_clause(clause)
    return remove_clause(functor, arity, clause, flags)


def check_clause(beta):
    if not is_clause(beta):
        check_nonvar(beta)
        beta = copy_term(beta)
        raise make_error(Compound("type_error", ["clause", beta]))


###
# kb_pred_destroy(F, N): internal only
# The built-in succeeds. As a side effect the
# predicate indicator F/N is destroyed.
##
def test_kb_pred_destroy(args):
    functor = exec_build(args[0])
    check_atom(functor)
    arity = exec_build(args[1])
    check_integer(arity)
    pred_destroy(functor, arity)
    return True


################################################################
# kb_make_defined/2 and kb_pred_link/3                         #
################################################################

###
# kb_make_defined(L, P): internal only
# The built-in succeeds in P with an annonymous predicate for the clauses L.
##
def test_kb_make_defined(args):
    alpha = exec_build(args[0])
    alpha = list_objects(alpha)
    offset = exec_build(args[1])
    check_integer(offset)
    return exec_unify(args[2], make_defined(alpha, offset))


###
# kb_pred_link(F, N, Q): internal only
# The built-in succeeds in Q with the provable of the
# predicate indicator F/A. Otherwise if no such provable
# exists the built-in fails.
##
def test_kb_pred_link(args):
    functor = exec_build(args[0])
    check_atom(functor)
    arity = exec_build(args[1])
    check_integer(arity)
    peek = pred_link(functor, arity)
    if peek is NotImplemented:
        return False
    return exec_unify(args[2], peek)


################################################################
# kb_is_link/1, kb_link_flags/2 and kb_pred_list/1             #
################################################################

###
# kb_is_link(Q): internal only
# The built-in succeeds if Q is a provable.
##
def test_kb_is_link(args):
    alpha = exec_build(args[0])
    if is_structure(alpha):
        return is_provable(alpha.functor)
    return is_provable(alpha)


###
# kb_link_flags(Q, F): internal only
# The built-in succeeds in F with the flags of the provable Q.
##
def test_kb_link_flags(args):
    peek = exec_build(args[0])
    check_provable(peek)
    res = peek.flags >> MASK_PRED_BITS
    return exec_unify(args[1], res)


###
# Assure that the object is a provable.
#
# @param beta The object.
##
def check_provable(beta):
    if not is_provable(beta):
        check_nonvar(beta)
        beta = copy_term(beta)
        raise make_error(Compound("type_error",
            ["provable", beta]))


###
# kb_pred_list(L): internal only
# The built-in succeeds in L with the current predicate indicators.
##
def test_kb_pred_list(args):
    res = kb_pred_list()
    return exec_unify(args[0], res)


def kb_pred_list():
    back = None
    res = None
    for functor in store.kb:
        temp = store.kb[functor]
        i = 0
        while i < len(temp):
            peek = temp[i]
            if peek is NotImplemented or peek.remover is not NotImplemented:
                i += 1
                continue
            peek = Compound(".", [make_indicator(functor, i), NotImplemented])
            if back is None:
                res = peek
            else:
                back.args[1] = peek
            back = peek
            i += 1
    if back is None:
        res = "[]"
    else:
        back.args[1] = "[]"
    return res


################################################################
# kb_clause_shard/2 and kb_clause_arity/2                      #
################################################################

###
# kb_clause_shard(C, S): internal only
# The built-in succeeds in S with the shard of the clause C.
##
def test_kb_clause_shard(args):
    clause = exec_build(args[0])
    check_clause(clause)
    return exec_unify(args[1], clause.shard)


###
# kb_clause_arity(C, A): internal only
# The built-in succeeds in A with the arity of the clause C.
##
def test_kb_clause_arity(args):
    clause = exec_build(args[0])
    check_clause(clause)
    res = len(clause.head)
    return exec_unify(args[1], res)


################################################################
# kb_clause_head/2 and kb_clause_data/4                        #
################################################################

###
# kb_clause_head(C, H): internal only
# The built-in succeeds in H with the head of the clause C.
##
def special_kb_clause_head(args):
    clause = deref(args[0])
    check_clause(clause)
    head = deref(args[1])
    if is_structure(head):
        paras = head.args
    else:
        paras = VOID_ARGS
    if clause.size != 0:
        display = [NotImplemented] * clause.size
    else:
        display = None
    if not exec_head(clause.head, display, paras):
        return False
    cont(machine.call.args[1])
    return True


###
# kb_clause_data(C, H, O, L): internal only
# The built-in succeeds in H, O and L with the head,
# the cut var and the translated body of the clause C.
##
def special_kb_clause_data(args):
    clause = deref(args[0])
    check_clause(clause)
    head = deref(args[1])
    if is_structure(head):
        paras = head.args
    else:
        paras = VOID_ARGS
    if clause.size != 0:
        display = [NotImplemented] * clause.size
    else:
        display = None
    if not exec_head(clause.head, display, paras):
        return False
    peek = clause.cutvar
    if peek != -1:
        temp = Variable()
        display[peek] = temp
        temp = Compound("just", [temp])
    else:
        temp = "nothing"
    if not unify(args[2], temp):
        return False
    temp = exec_body(clause.body, display)
    if not unify(args[3], temp):
        return False
    cont(machine.call.args[1])
    return True


################################################################
# dg_get_partition/1, dg_set_partition/1 and sys_flags_get/1   #
################################################################

###
# dg_get_partition(D): internal only
# The built-in succeeds in D with the current partition.
##
def test_dg_get_partition(args):
    return exec_unify(args[0], store.engine.partition)


###
# dg_set_partition(D): internal only
# The built-in succeeds. As a side effect it changes the current partition.
##
def test_dg_set_partition(args):
    value = exec_build(args[0])
    check_atom(value)
    set_partition(value)
    return True


###
# sys_flags_get(W): internal only
# The built-in succeeds in W with the garbage collector flags.
##
def test_sys_flags_get(args):
    return exec_unify(args[0], store.engine.flags)


####################################################################
# sys_flag_map/1, os_get_workdir/1 and os_set_workdir/1            #
####################################################################

###
# sys_flag_map(M): internal only
# The built-in succeeds in W with a Prolog flag map.
##
def test_sys_flag_map(args):
    res = {}
    res["system_url"] = bootbase
    res["foreign_ext"] = ".py"
    res["host_info"] = platform.python_implementation()+", Python "+platform.python_version()
    res["mach_info"] = platform.processor()+", "+platform.platform()
    return exec_unify(args[0], res)


###
# os_get_workdir(D): internal only
# The built-in succeeds in D with the working directory.
##
def test_os_get_workdir(args):
    url = os.getcwd()+os.path.sep
    return exec_unify(args[0], url)


###
# os_set_workdir(D): internal only
# The built-in succeeds. As a side effect it changes the working directory to D.
##
def test_os_set_workdir(args):
    url = exec_build(args[0])
    check_atom(url)
    try:
        os.chdir(url)
    except OSError as err:
        raise make_error(map_file_error(err, url))
    return True


################################################################
# dg_clear_stage/0, dg_get_stage/1 and dg_set_stage/1          #
################################################################

###
# dg_clear_stage: internal only
# The built-in succeeds. As a side effect it clears the current stage.
##
def test_dg_clear_stage(args):
    clear()
    return True


###
# dg_get_stage(D): internal only
# The built-in succeeds in D with the current stage.
##
def test_dg_get_stage(args):
    return exec_unify(args[0], store.stage)


###
# dg_set_stage(D): internal only
# The built-in succeeds. As a side effect it changes the current stage.
##
def test_dg_set_stage(args):
    value = exec_build(args[0])
    check_integer(value)
    set_stage(value)
    return True


################################################################
# Runtime Init                                                 #
################################################################

# stream specials, output
add("current_output", 1, make_check(test_current_output))
add("current_error", 1, make_check(test_current_error))
add("set_output", 1, make_check(test_set_output))
add("set_error", 1, make_check(test_set_error))
add("put_atom", 2, make_check(test_put_atom))

# stream specials, async
add("current_input", 1, make_check(test_current_input))
add("set_input", 1, make_check(test_set_input))
add("os_get_code", 2, make_check(test_os_get_code))
add("os_peek_code", 2, make_check(test_os_peek_code))
add("os_open_promise_opts", 4, make_check(test_os_open_promise_opts))
add("os_read_promise", 2, make_check(test_os_read_promise))
add("os_close_promise", 2, make_check(test_os_close_promise))
add("os_prop_promise", 3, make_check(test_os_prop_promise))

# stream specials, sync
add("os_stream_flags", 2, make_check(test_os_stream_flags))
add("os_open_sync_opts", 4, make_check(test_os_open_sync_opts))
add("flush_output", 1, make_check(test_flush_output))
add("os_close_sync", 1, make_check(test_os_close_sync))
add("os_cntrl_sync", 2, make_check(test_os_cntrl_sync))
add("os_read_sync", 1, make_check(test_os_read_sync))

# intermediate representation, Albufeira code
add("ir_anon_new", 1, make_check(test_ir_anon_new))
add("ir_place_new", 2, make_check(test_ir_place_new))
add("ir_skeleton_new", 3, make_check(test_ir_skeleton_new))
add("ir_make_frozen", 3, make_check(test_ir_make_frozen))
add("ir_pred_site", 2, make_check(test_ir_pred_site))
add("ir_is_site", 1, make_check(test_ir_is_site))
add("ir_site_name", 2, make_check(test_ir_site_name))

# intermediate representation specials, consult text, internal only
add("ir_clause_new", 6, make_check(test_ir_clause_new))
add("ir_clause_add", 4, make_check(test_ir_clause_add))
add("ir_goal_new", 4, make_check(test_ir_goal_new))
add("ir_goal_call", 1, make_special(special_ir_goal_call))

# knowledge base specials, dynamic database, internal only
add("kb_clause_ref", 3, make_special(special_kb_clause_ref))
add("kb_pred_touch", 3, make_check(test_kb_pred_touch))
add("kb_clause_remove", 4, make_check(test_kb_clause_remove))
add("kb_pred_destroy", 2, make_check(test_kb_pred_destroy))

# knowledge base specials, linked provables, internal only
add("kb_make_defined", 3, make_check(test_kb_make_defined))
add("kb_pred_link", 3, make_check(test_kb_pred_link))
add("kb_is_link", 1, make_check(test_kb_is_link))
add("kb_link_flags", 2, make_check(test_kb_link_flags))
add("kb_pred_list", 1, make_check(test_kb_pred_list))

# knowledge base specials, meta data, internal only
add("kb_clause_shard", 2, make_check(test_kb_clause_shard))
add("kb_clause_arity", 2, make_check(test_kb_clause_arity))
add("kb_clause_head", 2, make_special(special_kb_clause_head))
add("kb_clause_data", 4, make_special(special_kb_clause_data))
add("dg_get_partition", 1, make_check(test_dg_get_partition))
add("dg_set_partition", 1, make_check(test_dg_set_partition))

# system specials, flags, internal only
add("sys_flags_get", 1, make_check(test_sys_flags_get))
add("sys_flag_map", 1, make_check(test_sys_flag_map))
add("os_get_workdir", 1, make_check(test_os_get_workdir))
add("os_set_workdir", 1, make_check(test_os_set_workdir))

# system specials, staging, internal only
add("dg_clear_stage", 0, make_check(test_dg_clear_stage))
add("dg_get_stage", 1, make_check(test_dg_get_stage))
add("dg_set_stage", 1, make_check(test_dg_set_stage))

