Admin User, created Apr 26. 2025
###
# 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 os
import signal
import sys
import asyncio
from nova.core import (Compound, make_error, set_bootbase,
file_read_promise, Sink, Source, map_stream_error,
MASK_SRC_AREAD, post, perform_async, perform, set_stage,
set_partition, get_engine)
###
# Initializes the Prolog interpreter.
##
def init():
set_stage(0)
set_partition("user")
################################################################
# Plain Console #
################################################################
def plain_out(data, buf):
try:
data.write(buf)
return data
except IOError as err:
raise make_error(map_stream_error(err))
def console_flush(data):
try:
data.flush()
except IOError as err:
raise make_error(map_stream_error(err))
################################################################
# Color Console #
################################################################
ANSI_RESET = "\x1B[0m"
ANSI_INPUT = "\x1B[32m" # Green
ANSI_ERROR = "\x1B[31m" # Red
def ansi_out(data, buf):
try:
data.write(ANSI_RESET)
data.write(buf)
data.write(ANSI_INPUT)
return data
except IOError as err:
raise make_error(map_stream_error(err))
def ansi_err(data, buf):
try:
data.write(ANSI_ERROR)
data.write(buf)
data.write(ANSI_INPUT)
return data
except IOError as err:
raise make_error(map_stream_error(err))
################################################################
# When Main #
################################################################
def ctrlc_abort(signum, frame):
post(Compound("system_error", ["user_abort"]))
###
# Simply colored async process standard input/output.
##
async def dogelog_async():
back = signal.signal(signal.SIGINT, ctrlc_abort)
try:
init()
os.system("")
perform("sys_baseline")
if len(sys.argv) > 1:
await perform_async("sys_launch")
else:
dst = Sink()
dst.data = sys.stdout
dst.send = ansi_out
dst.notify = console_flush
get_engine().text_output = dst
dst = Sink()
dst.data = sys.stderr
dst.send = ansi_err
dst.notify = console_flush
get_engine().text_error = dst
await perform_async("sys_launch")
sys.stdout.write(ANSI_RESET)
finally:
signal.signal(signal.SIGINT, back)
set_bootbase(__file__)
dst = Sink()
dst.data = sys.stdout
dst.send = plain_out
dst.notify = console_flush
get_engine().text_output = dst
dst = Sink()
dst.data = sys.stderr
dst.send = plain_out
dst.notify = console_flush
get_engine().text_error = dst
src = Source()
src.data = sys.stdin
src.receive = file_read_promise
src.flags |= MASK_SRC_AREAD
get_engine().text_input = src
if __name__ == "__main__":
try:
asyncio.run(dogelog_async())
except Exception as err:
perform(Compound("sys_print_error", [err.args[0]]))
sys.exit(1)