243 lines
9.8 KiB
Python
243 lines
9.8 KiB
Python
"""
|
|
Silique (https://www.silique.fr)
|
|
Copyright (C) 2024
|
|
|
|
distribued with GPL-2 or later license
|
|
|
|
This program is free software; you can redistribute it and/or modify
|
|
it under the terms of the GNU General Public License as published by
|
|
the Free Software Foundation; either version 2 of the License, or
|
|
(at your option) any later version.
|
|
|
|
This program is distributed in the hope that it will be useful,
|
|
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
GNU General Public License for more details.
|
|
|
|
You should have received a copy of the GNU General Public License
|
|
along with this program; if not, write to the Free Software
|
|
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
|
"""
|
|
from typing import List
|
|
from re import findall
|
|
|
|
from tiramisu import undefined, Calculation
|
|
from tiramisu.error import PropertiesOptionError, LeadershipError, ConfigError
|
|
|
|
from .object_model import CONVERT_OPTION
|
|
|
|
|
|
class UserDatas:
|
|
def __init__(self, config) -> None:
|
|
self.config = config
|
|
|
|
def user_datas(self, user_datas: List[dict]):
|
|
self.values = {}
|
|
self.errors = []
|
|
self.warnings = []
|
|
self._populate_values(user_datas)
|
|
self._auto_configure_dynamics()
|
|
self._populate_config()
|
|
self._populate_error_warnings()
|
|
return {
|
|
"errors": self.errors,
|
|
"warnings": self.warnings,
|
|
}
|
|
|
|
def _populate_values(self, user_datas):
|
|
for datas in user_datas:
|
|
options = datas.get("options", {})
|
|
for name, data in datas.get("values", {}).items():
|
|
self.values[name] = {
|
|
"values": data,
|
|
"options": options.copy(),
|
|
}
|
|
self.errors.extend(datas.get("errors", []))
|
|
self.warnings.extend(datas.get("warnings", []))
|
|
|
|
def _get_variable(self, config):
|
|
for subconfig in config:
|
|
if subconfig.isoptiondescription():
|
|
yield from self._get_variable(subconfig)
|
|
else:
|
|
yield subconfig
|
|
|
|
def _auto_configure_dynamics(self):
|
|
cache = {}
|
|
added = []
|
|
for path, data in list(self.values.items()):
|
|
value = data["values"]
|
|
try:
|
|
option = self.config.option(path)
|
|
option.name()
|
|
except (ConfigError, PropertiesOptionError):
|
|
pass
|
|
except AttributeError:
|
|
self._not_found_is_dynamic(self.config, path, cache, added)
|
|
|
|
def _not_found_is_dynamic(self, config, path, cache, added):
|
|
"""if path is not found, check if parent is a dynamic family
|
|
"""
|
|
current_path = ""
|
|
identifiers = []
|
|
# get parent
|
|
for name in path.split(".")[:-1]:
|
|
if current_path:
|
|
current_path += "."
|
|
current_path += name
|
|
if current_path in cache:
|
|
config, identifier = cache[current_path]
|
|
identifiers.append(identifier)
|
|
continue
|
|
tconfig = config.option(name)
|
|
try:
|
|
tconfig.group_type()
|
|
# object exists, so current config is the temporary config
|
|
config = tconfig
|
|
if config.isdynamic(only_self=True):
|
|
identifiers.append(config.identifiers()[-1])
|
|
except AttributeError:
|
|
# try to found the good dynamic family
|
|
for tconfig in config.list(uncalculated=True):
|
|
if not tconfig.isdynamic(only_self=True):
|
|
# it's not a dynamic variable
|
|
continue
|
|
identifier = self._get_identifier(
|
|
tconfig.name(), name
|
|
)
|
|
if identifier is None:
|
|
# it's a dynamic variable but doesn't match the current name
|
|
continue
|
|
dynamic_variable = tconfig.information.get(
|
|
"dynamic_variable",
|
|
None,
|
|
)
|
|
if not dynamic_variable:
|
|
# it's the good dynamic variable but it's not linked to a variable
|
|
# so cannot change the variable
|
|
continue
|
|
option_type = self.config.option(
|
|
dynamic_variable
|
|
).information.get("type")
|
|
dyn_options_values = self.config.option(dynamic_variable).get().impl_getdefault()
|
|
if "{{ identifier }}" in dynamic_variable:
|
|
for s in identifiers:
|
|
dynamic_variable = dynamic_variable.replace(
|
|
"{{ identifier }}", str(s), 1
|
|
)
|
|
if dynamic_variable not in self.values and not dyn_options_values:
|
|
self.values[dynamic_variable] = {"values": []}
|
|
added.append(dynamic_variable)
|
|
elif dynamic_variable not in added:
|
|
continue
|
|
config = tconfig
|
|
identifiers.append(identifier)
|
|
typ = CONVERT_OPTION.get(option_type, {}).get(
|
|
"func"
|
|
)
|
|
if typ:
|
|
identifier = typ(identifier)
|
|
if (
|
|
identifier
|
|
not in self.values[dynamic_variable]["values"]
|
|
):
|
|
self.values[dynamic_variable]["values"].append(
|
|
identifier
|
|
)
|
|
cache[current_path] = config, identifier
|
|
break
|
|
|
|
def _populate_config(self):
|
|
while self.values:
|
|
value_is_set = False
|
|
for option in self._get_variable(self.config):
|
|
path = option.path()
|
|
if path not in self.values:
|
|
continue
|
|
options = self.values[path].get("options", {})
|
|
value = self.values[path]["values"]
|
|
needs_convert = options.get("needs_convert", False)
|
|
|
|
# converted value
|
|
if option.ismulti():
|
|
if options.get("multi_separator") and not isinstance(value, list):
|
|
value = value.split(options["multi_separator"])
|
|
self.values[path]["values"] = value
|
|
if option.issubmulti():
|
|
value = [[val] for val in value]
|
|
if needs_convert:
|
|
if option.issubmulti():
|
|
for idx, val in enumerate(value):
|
|
value[idx] = [convert_value(option, v) for v in val]
|
|
else:
|
|
value = [convert_value(option, val) for val in value]
|
|
self.values[path]["values"] = value
|
|
self.values[path]["options"]["needs_convert"] = False
|
|
elif needs_convert:
|
|
value = convert_value(option, value)
|
|
index = option.index()
|
|
if index is not None:
|
|
if not isinstance(value, list) or index >= len(value):
|
|
continue
|
|
value = value[index]
|
|
try:
|
|
option.value.set(value)
|
|
value_is_set = True
|
|
# value is correctly set, remove variable to the set
|
|
if index is not None:
|
|
# if it's a follower waiting for all followers are sets
|
|
self.values[path]["values"][index] = undefined
|
|
if set(self.values[path]["values"]) == {undefined}:
|
|
self.values.pop(path)
|
|
else:
|
|
self.values.pop(path)
|
|
except Exception:
|
|
if path != option.path():
|
|
self.values[option.path()] = self.values.pop(path)
|
|
if not value_is_set:
|
|
break
|
|
|
|
def _get_identifier(self, true_name, name) -> str:
|
|
if true_name == "{{ identifier }}":
|
|
return name
|
|
regexp = true_name.replace("{{ identifier }}", "(.*)")
|
|
finded = findall(regexp, name)
|
|
if len(finded) != 1 or not finded[0]:
|
|
return None
|
|
return finded[0]
|
|
|
|
def _populate_error_warnings(self):
|
|
# we don't find variable, apply value just to get error or warning messages
|
|
for path, data in self.values.items():
|
|
try:
|
|
option = self.config.option(path)
|
|
value = data["values"]
|
|
if option.isfollower():
|
|
for index, val in enumerate(value):
|
|
if val is undefined:
|
|
continue
|
|
self.config.option(path, index).value.set(val)
|
|
else:
|
|
option.value.set(value)
|
|
except AttributeError as err:
|
|
self.errors.append(str(err))
|
|
except (ValueError, LeadershipError) as err:
|
|
self.errors.append(str(err))
|
|
except PropertiesOptionError as err:
|
|
self.warnings.append(str(err))
|
|
|
|
|
|
def convert_value(option, value):
|
|
if value == "":
|
|
return None
|
|
option_type = option.information.get("type")
|
|
if option_type == 'choice':
|
|
choices = option.value.list()
|
|
if value not in choices and isinstance(value, str):
|
|
# FIXME add other tests (boolean, float, ...)
|
|
if value.isnumeric() and int(value) in choices:
|
|
value = int(value)
|
|
func = CONVERT_OPTION.get(option_type, {}).get("func")
|
|
if func:
|
|
return func(value)
|
|
return value
|