# Ensemble.py -- Pamhyr # Copyright (C) 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 . # -*- coding: utf-8 -*- import struct import logging import itertools import numpy as np from Model.Tools.PamhyrDB import SQLSubModel from Model.Scenario import Scenario from Model.Ensembles.Distribution import Uniform logger = logging.getLogger() class Ensemble(SQLSubModel): _sub_classes = [] def __init__(self, id: int = -1, name: str = "", distribution=None, params=[0., 100.], data_type=None, target_data=None, status=None, owner_scenario=-1): super(Ensemble, self).__init__( id=id, status=status, owner_scenario=owner_scenario ) self._name = name self._data_type = data_type self._distribution = distribution self._params = params self._target_data = target_data @classmethod def _db_create(cls, execute, ext=""): execute(f""" CREATE TABLE ensemble{ext} ( {cls.create_db_add_pamhyr_id()}, deleted BOOLEAN NOT NULL DEFAULT FALSE, name TEXT NOT NULL, type VARCHAR(8) NOT NULL, distribution INTEGER NOT NULL, distribution_name VARCHAR(32) NOT NULL, params BLOB NOT NULL, params_len INTEGER NOT NULL, data_pid INTEGER 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"] distributions = data["ens_distributions"] stricklers = data["stricklers"] loaded = data['loaded_pid'] table = execute( "SELECT pamhyr_id, deleted, name, type, " + "distribution, distribution_name, params, params_len, data_pid, " + "scenario " + "FROM ensemble " + 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) data_type = next(it) fid = next(it) fname = next(it) params = next(it) params_len = next(it) data_pid = next(it) owner_scenario = next(it) def fn(f): return f._name == fname if fid != -1: def fn(f): return f._pamhyr_id == fid distribution = next( filter(fn, distributions.lst), None ) target_data = next( filter( lambda s: s._pamhyr_id == data_pid, stricklers.lst ), None ) rdata = list(cls._decode_params(params, params_len)) new_ensemble = cls( id, name=name, distribution=distribution, params=rdata, data_type=data_type, target_data=target_data, status=data["status"], owner_scenario=owner_scenario ) if deleted: new_ensemble.set_as_deleted() loaded.add(id) new.append(new_ensemble) return new def _db_save(self, execute, data=None): if not self.must_be_saved(): return True execute( "DELETE FROM ensemble " + f"WHERE pamhyr_id = {self.pamhyr_id} " + f"AND scenario = {self._status.scenario_id}" ) bparams, length = self._encode_params() if self._distribution is None: fid = -1 fname = "NULL" else: fid = self._distribution._pamhyr_id fname = self._distribution._name if self._distribution._type == "generic": fid = -1 data_pid = -1 if self._target_data is not None: data_pid = self._target_data._pamhyr_id execute( "INSERT INTO " + "ensemble(pamhyr_id, deleted, name, type, " + " distribution, distribution_name, params, params_len, " + " data_pid, scenario) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", self.pamhyr_id, self.is_deleted(), self._name, self._data_type, fid, fname, bparams, length, data_pid, self._status.scenario_id ) return True def _encode_params(self): length = len(self._params) data_format = ">" + ''.join(itertools.repeat("d", length)) return struct.pack(data_format, *self._params), length @classmethod def _decode_params(cls, bparams, length): data_format = ">" + ''.join(itertools.repeat("d", length)) return struct.unpack(data_format, bparams) def random(self, nb): return np.nditer( self._distribution.random( self._params, nb ) ) @property def name(self): return self._name @name.setter def name(self, name): self._name = name self._status.modified() @property def distribution(self): return self._distribution @distribution.setter def distribution(self, distribution): self._distribution = distribution self._status.modified() @property def params(self): return self._params @params.setter def params(self, params): self._params = params self._status.modified() @property def data_type(self): return self._data_type @data_type.setter def data_type(self, data_type): self._data_type = data_type self._status.modified() @property def target_data(self): return self._target_data @target_data.setter def target_data(self, target_data): self._target_data = target_data self._status.modified() def get_data_dict(self, use_ids=False): target_data = self.target_data distribution = self.distribution if use_ids: if self.target_data: target_data = self.target_data.pamhyr_id if self.distribution: distribution = self.distribution.name return { "name": self.name, "data_type": self.data_type, "target_data": target_data, "distribution": distribution, "params": self.params, } def set_data_dict(self, data): self.name = data["name"] self.data_type = data["data_type"] self.target_data = data["target_data"] self.distribution = data["distribution"] self.params = data["params"]