Home Page
Site Map
Login/out
Message Board
Download(c) 2005-2024 XLOG Technologies AG
###
# 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
from dogelog import (exec_build, add, make_check, deref,
exec_unify, check_atom, make_error, Compound)
#####################################################################
# File Access #
#####################################################################
###
# directory_files(F, L):
# The predicate succeeds in L with the entries of the directory F.
# Barks if path F doesn't exist or if path F doesn't point to directory.
##
def test_directory_files(args):
url = deref(exec_build(args[0]))
check_atom(url)
try:
entries = os.listdir(url if url != "" else os.getcwd())
except OSError as err:
if err.errno == 2:
raise make_error(Compound("existence_error", ["source_sink", url]))
else:
raise make_error(Compound("resource_error", ["io_exception"]))
res = "[]"
i = len(entries) - 1
while i >= 0:
val = entries[i]
res = Compound(".", [val, res])
i -= 1
return exec_unify(args[1], res)
###
# file_exists(F):
# The predicate succeeds if the file F exists, otherwise fails.
##
def test_file_exists(args):
url = deref(exec_build(args[0]))
check_atom(url)
return os.path.exists(url)
###
# make_directory(F):
# The predicate succeeds. As a side effect a directory F is created.
# Barks if the parent of F doesn't exist or if F already exists.
##
def test_make_directory(args):
url = deref(exec_build(args[0]))
check_atom(url)
try:
os.mkdir(url)
except OSError as err:
if err.errno == 2:
raise make_error(Compound("existence_error", ["source_sink", url]))
else:
raise make_error(Compound("resource_error", ["io_exception"]))
return True
###
# delete_file(F):
# The predicate succeeds. As a side effect the file F is deleted.
# Barks if the file F doesn't exist or if its a directory.
##
def test_delete_file(args):
url = deref(exec_build(args[0]))
check_atom(url)
try:
os.unlink(url)
except OSError as err:
if err.errno == 2:
raise make_error(Compound("existence_error", ["source_sink", url]))
else:
raise make_error(Compound("resource_error", ["io_exception"]))
return True
###
# copy_binary(F, G):
# The predicate succeeds. As a side effect the file F is copied
# to the file G, without rollback upon error. An already existing
# file G is silently overwritten.
##
def test_copy_binary(args):
src = deref(exec_build(args[0]))
check_atom(src)
dst = deref(exec_build(args[1]))
check_atom(dst)
try:
copy_binary(src, dst)
except OSError as err:
if err.errno == 2:
raise make_error(Compound("existence_error",
["source_sink", src]))
else:
raise make_error(Compound("resource_error",
["io_exception"]))
return True
def copy_binary(src, dst):
bin = open(src, "rb")
try:
bout = open(dst, "wb")
try:
buf = bin.read(8192)
while len(buf) > 0:
bout.write(buf)
buf = bin.read(8192)
finally:
bout.close()
finally:
bin.close()
#####################################################################
# Bit Lib Init #
#####################################################################
def main():
add("directory_files", 2, make_check(test_directory_files))
add("file_exists", 1, make_check(test_file_exists))
add("make_directory", 1, make_check(test_make_directory))
add("delete_file", 1, make_check(test_delete_file))
add("copy_binary", 2, make_check(test_copy_binary))