Commit 90aa72ef authored by Lucas Bourneuf's avatar Lucas Bourneuf

major overhaul refactor fixing bugs

- Script is now a living class
- biseau.script module now does everything Script-related
- the overall things are clearer than before, and functional, and
documented, it was, and still is, necessary
- modifying a script source code is now working, and tested \o/
parent 5735ac81
from .module_loader import Script
from .script import Script
from .core import run, single_image_from_filenames, multiple_images_from_filenames, gif_from_filenames, compile_to_single_image, compile_to_images, compile_context_to_dots
from .module_loader import build_scripts_from_file, build_scripts_from_dir
......@@ -7,7 +7,6 @@ import os
import io
import time
import json
import clyngor
import tempfile
from PIL import Image
from collections import OrderedDict
......@@ -16,6 +15,7 @@ from . import Script
from . import asp_to_dot
from . import dot_writer
from . import module_loader
from biseau.script import solve_context, Script
EXT_TO_TYPE = utils.reverse_dict({
......@@ -87,25 +87,10 @@ def yield_run(scripts:[Script], initial_context:str='', verbosity:int=0) -> (str
context = initial_context
for idx, script in enumerate(scripts, start=1):
script_start = time.time()
if script.input_mode is str:
input_data = context
else: # need a solving
assert script.input_mode is iter
input_data = solve_context(context)
if verbosity >= 2:
print('RUN run_on…', end='', flush=True)
new_context = script.run_on(input_data, **script.options_values)
if verbosity >= 2:
# if verbosity >= 3:
# print('NEW CONTEXT:', new_context)
if script.erase_context:
context = new_context
context += '\n' + new_context
context =, context, verbosity)
script_time = round(time.time() - script_start, 2)
if verbosity >= 1:
print(f"SCRIPT {idx}: {} add {len(new_context.splitlines())} lines to the context in {script_time}s.")
print(f"SCRIPT {idx}: {} added {len(context.splitlines())} lines to the context in {script_time}s.")
yield context, script_time
run_time = round(time.time() - run_start, 2)
if verbosity >= 1:
......@@ -120,10 +105,6 @@ def run(scripts:[Script], initial_context:str='', verbosity:int=0) -> str:
return out
def solve_context(context:str, *, nb_model:int=0) -> clyngor.Answers:
return clyngor.solve(inline=context, nb_model=nb_model, options='--project').by_predicate.careful_parsing.int_not_parsed
def compile_to_single_image(context:str, outfile:str=None, dotfile:str=None,
nb_model:int=0, return_image:bool=True,
dot_prog:str='dot', verbosity:int=0) -> Image or None:
This diff is collapsed.
"""Definitions and functions applied to Script object.
The Script object waits for many parameters, detailed in its constructor,
and its main interface is the run_on method.
The run_on method, constituting the main API interface, IS A NIGHTMARE to generate.
From the Script point of view, the inner _run_on method (gotten from the module,
like JSON), maps a context/models and options to a string/generator of string.
The call_script function is uniformizing all of these in a single interface.
When the source_code attribute of a Script is modified, this run a machinery
whose goal is to build the _run_on method.
This machinery is not runned if run_on is given in the Script constructor.
The machinery is basically an extraction of the source code depending
of the language (python, asp, python file or asp file) and generate a standard
_run_on method based on that.
But because the script can also specify a (new) name, a (new) value for erase_context
and other attributes, that are parsed in module_loader.build_script_from_module,
it is necessary to call that function on the new module (induced by a source_code change).
Therefore, a new Script instance is built, from which all attributes are copied in the current instance.
The code is on the verge to lock itself in an infinite loop.
I'm sorry for the ones coming after.
import clyngor
from biseau import utils
class Script:
def __init__(self, name:str, tags:set, description:str, module, options, options_values, input_mode, incompatible, spec_inputs, spec_outputs, inputs, outputs, source_code, language, erase_context, *, run_on:callable or None=None): = str(name) # human readable name
self.tags = set(map(str, tags)) # set of tags identifying the script
self.description = str(description) # human readable and high level description of the script
self.module = module # reference to the module itself
self.options = tuple(options) # list of (name, type, default, description) describing each option
self.options_values = dict(options_values) # mutable mapping allowing to set options value to be used
self.input_mode = input_mode # define if run_on must receive the context or the resulting ASP models
self.incompatible = set(incompatible) # list of incompatibles modules
self.spec_inputs = spec_inputs # function in module to call to get the inputs knowing the parameters
self.spec_outputs = spec_outputs # function in module to call to get the outputs knowing the parameters
self.inputs = inputs # function in module to call to get all possible inputs
self.outputs = outputs # function in module to call to get all possible outputs
self.language = str(language) # string indicating the language implementing the source code
self.erase_context = bool(erase_context) # true if the script erase the context (default: false, context is kept)
# creation of run_on function
self.__source_code = str(source_code) if source_code else None # don't trigger the setter
assert source_code != 'None', "this already happened, it was because previous line called str() on source_code, which could be None"
self._run_on = run_on
if not self._run_on:
self._run_on = compile_run_on(self.language, self.__source_code, self._update_from_module)
def run_on(self, context:str, *args, **kwargs):
return call_script(self, context, *args, **kwargs)
def source_code(self) -> str:
return self.__source_code
def source_code(self, new_source:str):
self.__source_code = str(new_source)
if self._run_on._source != self.source_code:
self._run_on = compile_run_on(self.language, self.source_code, self._update_from_module)
def _update_from_module(self, module):
"""Change all self attributes based on those of given module,
especially language and source_code defining run_on
This shouldn't be called when run_on is provided.
from biseau.module_loader import build_script_from_module
script = build_script_from_module(module, defaults=vars(self))
for key in vars(self):
if getattr(script, key, None) is not None:
setattr(self, key, getattr(script, key))
if self._run_on:
self._run_on._source = self.source_code
# print('\tFROM:', {k: v for k, v in vars(module).items() if not k.startswith('_')})
# print('\t TO :', {k: v for k, v in vars(self).items() if not k.startswith('_')})
def call(script, context:str, *args, **kwargs):
return call_script(script, context, *args, **kwargs)
class Module:
"""Placeholder for a module/namespace.
It's assumed safe to make them hashable on content. Also, the hashable
property is only used during validation and initial core treatments.
def __hash__(self):
return hash(tuple(self.__dict__.values()))
def from_dict(mapping:dict) -> object:
"Create Module instance with given dictionnary as members"
module = Module()
for key, val in mapping.items():
setattr(module, key, val)
return module
def solve_context(context:str, *, nb_model:int=0) -> clyngor.Answers:
"""Uniformized way to solve an ASP context"""
return clyngor.solve(inline=context, nb_model=nb_model, options='--project').by_predicate.careful_parsing.int_not_parsed
def call_script(script, context:str, verbosity:int=0):
"""Uniformized call to a Script object"""
if script.input_mode is str:
input_data = context
else: # need a solving
assert script.input_mode is iter
input_data = solve_context(context)
if verbosity >= 2: print('RUN run_on… ', end='', flush=True)
new_context = utils.join_on_genstr(script._run_on)(input_data, **script.options_values)
if verbosity >= 2: print('OK!')
if verbosity >= 3: print('NEW CONTEXT:', new_context)
return new_context + ('\n' if script.erase_context else ('\n' + context))
def build_module_from_python_code(pycode:str, **module_options:dict):
"""Low level function, expecting some python code in string, and returning a namespace/module.
env = {}
exec(pycode, env)
print(f'Imported Python error:\n', traceback.format_exc())
# env.update(module_options)
env = {**module_options, **env}
env['source'] = pycode
# print('OPTIONS:', module_options, {k: v for k, v in env.items() if not k.startswith('_')})
module = Module.from_dict({k: v for k, v in env.items() if not k.startswith('__')})
# print('MODULE:', {k: v for k, v in vars(module).items() if not k.startswith('_')})
return module
def compile_run_on(language:str, code:str, updater:callable=None) -> callable:
"""Return a runner on given language/code, call given updater with a module
when a new python module is found
module = None
if language == 'python':
# we need to compile the source code to retrieve the run_on function
module = build_module_from_python_code(code)
run_on = module.run_on
elif language == 'python file':
with open(code) as fd:
source =
module = build_module_from_python_code(source)
run_on = module.run_on
elif language == 'asp':
def run_on(context:str):
return code
elif language == 'asp file':
def run_on(context:str):
with open(code) as fd:
raise NotImplementedError(f"Language {language} is not supported.")
if updater and module: updater(module)
run_on._source = code
return run_on
......@@ -9,6 +9,7 @@ from collections import defaultdict
HANDLED_COLORS = {'red', 'green', 'blue'}
def color_from_colors(colors:set) -> str:
"""Return one color that represents the given ones"""
if all(color in HANDLED_COLORS for color in colors):
......@@ -58,10 +59,10 @@ def run_python_code(code:str, namespace:dict=None) -> dict:
>>> run_python_code('a = 1')['a']
>>> a = 1
>>> run_python_code('a = 2')['a']
>>> b = 1
>>> run_python_code('b = 2')['b']
>>> a
>>> b
......@@ -2,6 +2,7 @@
import itertools
from collections import defaultdict
from pydot import graph_from_dot_data
import biseau
from biseau import run, compile_context_to_dots
from biseau.module_loader import build_scripts_from_asp_code
......@@ -45,3 +46,47 @@ def _graphdict_from_dot(dot:str) -> dict:
for edge in graph.get_edges():
yield dict(dictgraph)
def test_python_builder():
script = biseau.module_loader.build_script_from_json({
'name': 'test-python',
'python': 'def run_on(context):\n return "a."\n',
context = script.run_on('b :- a.')
expected = 'a.\nb :- a.'
print('CONTEXT :', context)
print('EXPECTED:', expected)
assert context == expected
assert == 'test-python'
def test_python_builder_with_internal_code_modifiers():
script = biseau.module_loader.build_script_from_json({
'name': 'test-python',
'python': 'NAME = "test-python-from-code"\nOUTPUTS = {"a/0"}\nERASE_CONTEXT = True\ndef run_on(context):\n return "a."\n',
context = script.run_on('b :- a.')
expected = 'a.\n'
print('CONTEXT :', context)
print('EXPECTED:', expected)
assert context == expected
assert == 'test-python-from-code' # shallowed by json
assert script.outputs() == {'a/0'}
def test_python_builder_with_modifications():
script = biseau.module_loader.build_script_from_json({
'name': 'test-python',
'python': 'def run_on(context):\n yield "a."\n',
assert script.run_on('b :- a.') == script.run_on('b :- a.')
# change the python code, verifying that the changes are really made
script.source_code = script.source_code.replace('a', 'b')
assert script.run_on('a :- b.') == 'b.\na :- b.'
# change of language and code
script.language = 'asp' # no direct effect
script.source_code = 'b.' # this one triggers the script.run_on replacement
assert script.run_on('a :- b.') == 'b.\na :- b.'
# name is conserved between switches
assert == 'test-python'
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment