mirror of https://gitlab.com/pamhyr/pamhyr2
Ensemble: Function: Add base classes.
parent
9b1fb4e7d2
commit
9456cf32fc
|
|
@ -0,0 +1,191 @@
|
|||
# Function.py -- Pamhyr
|
||||
# Copyright (C) 2023-2026 INRAE
|
||||
#
|
||||
# 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 3 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, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import struct
|
||||
import logging
|
||||
import itertools
|
||||
|
||||
from tools import logger_exception
|
||||
|
||||
from Model.Except import NotImplementedMethodeError
|
||||
from Model.Tools.PamhyrDB import SQLSubModel
|
||||
from Model.Scenario import Scenario
|
||||
|
||||
logger = logging.getLogger()
|
||||
|
||||
|
||||
class Function(SQLSubModel):
|
||||
_sub_classes = []
|
||||
|
||||
def __init__(self, id: int = -1,
|
||||
name: str = "", script: str = "",
|
||||
status=None, owner_scenario=-1):
|
||||
super(Function, self).__init__(
|
||||
id=id, status=status,
|
||||
owner_scenario=owner_scenario
|
||||
)
|
||||
|
||||
self._name = name
|
||||
self._type = "generic"
|
||||
self._script = script
|
||||
|
||||
@classmethod
|
||||
def _db_create(cls, execute, ext=""):
|
||||
execute(f"""
|
||||
CREATE TABLE ensemble_function{ext} (
|
||||
{cls.create_db_add_pamhyr_id()},
|
||||
deleted BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
name TEXT NOT NULL,
|
||||
script TEXT NOT NULL,
|
||||
{Scenario.create_db_add_scenario()},
|
||||
{Scenario.create_db_add_scenario_fk()},
|
||||
PRIMARY KEY(pamhyr_id, scenario)
|
||||
)
|
||||
""")
|
||||
|
||||
if ext == "_tmp":
|
||||
return True
|
||||
|
||||
return cls._create_submodel(execute)
|
||||
|
||||
@classmethod
|
||||
def _db_update(cls, execute, version, data=None):
|
||||
major, minor, release = version.strip().split(".")
|
||||
if major == "0" and int(minor) < 2:
|
||||
cls._db_create(execute)
|
||||
elif major == "0" and int(minor) == 2:
|
||||
if int(release) < 8:
|
||||
cls._db_create(execute)
|
||||
|
||||
return cls._update_submodel(execute, version, data)
|
||||
|
||||
@classmethod
|
||||
def _db_load(cls, execute, data=None):
|
||||
new = []
|
||||
scenario = data["scenario"]
|
||||
loaded = data['loaded_pid']
|
||||
|
||||
table = execute(
|
||||
"SELECT pamhyr_id, deleted, name, script, scenario " +
|
||||
"FROM ensemble_function " +
|
||||
f"WHERE scenario = {scenario.id} " +
|
||||
f"AND pamhyr_id NOT IN ({', '.join(map(str, loaded))})"
|
||||
)
|
||||
|
||||
for row in table:
|
||||
it = iter(row)
|
||||
|
||||
id = next(it)
|
||||
deleted = (next(it) == 1)
|
||||
name = next(it)
|
||||
script = next(it)
|
||||
owner_scenario = next(it)
|
||||
|
||||
new_function = cls(
|
||||
id, name=name, script=script,
|
||||
status=data["status"],
|
||||
owner_scenario=owner_scenario
|
||||
)
|
||||
if deleted:
|
||||
new_function.set_as_deleted()
|
||||
|
||||
loaded.add(id)
|
||||
new.append(new_function)
|
||||
|
||||
return new
|
||||
|
||||
def _db_save(self, execute, data=None):
|
||||
if self._type != "custom":
|
||||
return True
|
||||
|
||||
if not self.must_be_saved():
|
||||
return True
|
||||
|
||||
execute(
|
||||
"DELETE FROM ensemble_function " +
|
||||
f"WHERE pamhyr_id = {self.pamhyr_id} " +
|
||||
f"AND scenario = {self._status.scenario_id}"
|
||||
)
|
||||
|
||||
brange, length = self._encode_range()
|
||||
|
||||
execute(
|
||||
"INSERT INTO " +
|
||||
"ensemble_function(pamhyr_id, deleted, " +
|
||||
" name, script, scenario) " +
|
||||
"VALUES (?, ?, ?, ?, ?, )",
|
||||
self.pamhyr_id, self.is_deleted(),
|
||||
self._name, self._script,
|
||||
self._status.scenario_id
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
def _data_traversal(self,
|
||||
predicate=lambda obj, data: True,
|
||||
modifier=lambda obj, data: None,
|
||||
data={}):
|
||||
if predicate(self, data):
|
||||
modifier(self, data)
|
||||
|
||||
def get_sample(self, r):
|
||||
raise NotImplementedMethodeError(cls, cls._load)
|
||||
|
||||
|
||||
###########################
|
||||
# FUNCTION IMPLEMENTATION #
|
||||
###########################
|
||||
|
||||
class Uniform(Function):
|
||||
_sub_classes = []
|
||||
|
||||
def __init__(self, id: int = -1,
|
||||
name: str = "uniform", script: str = "",
|
||||
status=None, owner_scenario=-1):
|
||||
super(Custom, self).__init__(
|
||||
id=id, status=status,
|
||||
name=name, script=script,
|
||||
owner_scenario=owner_scenario
|
||||
)
|
||||
|
||||
def get_sample(self, dist, ens_range):
|
||||
return range(ens_range[0], ens_range[1])
|
||||
|
||||
|
||||
class Custom(Function):
|
||||
_sub_classes = []
|
||||
|
||||
def __init__(self, id: int = -1,
|
||||
name: str = "", script: str = "",
|
||||
status=None, owner_scenario=-1):
|
||||
super(Custom, self).__init__(
|
||||
id=id, status=status,
|
||||
name=name, script=script,
|
||||
owner_scenario=owner_scenario
|
||||
)
|
||||
|
||||
self._type = "custom"
|
||||
|
||||
def get_sample(self, dist, ens_range):
|
||||
try:
|
||||
# Run script to (re)define the sample function
|
||||
f = eval(self._script)
|
||||
return f(dist, ens_range)
|
||||
except Exception as e:
|
||||
logger_exception(e)
|
||||
return None # TODO: Raise helpful exception
|
||||
|
|
@ -0,0 +1,57 @@
|
|||
# FunctionList.py -- Pamhyr
|
||||
# Copyright (C) 2023-2026 INRAE
|
||||
#
|
||||
# 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 3 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, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from tools import trace, timer
|
||||
|
||||
from Model.Tools.PamhyrListExt import PamhyrModelList
|
||||
from Model.Ensembles.Function import Function
|
||||
|
||||
|
||||
class FunctionList(PamhyrModelList):
|
||||
_sub_classes = [Function]
|
||||
|
||||
@classmethod
|
||||
def _db_load(cls, execute, data=None):
|
||||
new = cls(status=data['status'])
|
||||
|
||||
new._lst = [Uniform]
|
||||
new._lst += Function._db_load(
|
||||
execute, data
|
||||
)
|
||||
|
||||
return new
|
||||
|
||||
def _db_save(self, execute, data=None):
|
||||
execute(
|
||||
"DELETE FROM ensemble_function " +
|
||||
f"WHERE scenario = {self._status.scenario_id}"
|
||||
)
|
||||
|
||||
if data is None:
|
||||
data = {}
|
||||
|
||||
for ens in self._lst:
|
||||
ens._db_save(execute, data=data)
|
||||
|
||||
return True
|
||||
|
||||
def new(self, index):
|
||||
r = Custom(status=self._status)
|
||||
self._lst.insert(index, r)
|
||||
self._status.modified()
|
||||
return r
|
||||
Loading…
Reference in New Issue