# InitialConditionsTemperature.py -- Pamhyr # Copyright (C) 2023-2025 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 logging from functools import reduce from tools import trace, timer, old_pamhyr_date_to_timestamp from Model.Tools.PamhyrDB import SQLSubModel from Model.Except import NotImplementedMethodeError from Model.Scenario import Scenario from Model.InitialConditionsTemperature.InitialConditionsTemperatureSpec \ import ICTemperatureSpec logger = logging.getLogger() class InitialConditionsTemperature(SQLSubModel): _sub_classes = [ ICTemperatureSpec, ] def __init__(self, id: int = -1, name: str = "default", status=None, owner_scenario=-1): super(InitialConditionsTemperature, self).__init__( id=id, status=status, owner_scenario=owner_scenario ) self._status = status self._name = name self._temperature = None self._data = [] @classmethod def _db_create(cls, execute, ext=""): execute(f""" CREATE TABLE initial_conditions_temperature{ext}( {cls.create_db_add_pamhyr_id()}, deleted BOOLEAN NOT NULL DEFAULT FALSE, name TEXT NOT NULL, temperature REAL NOT NULL, {Scenario.create_db_add_scenario()}, {Scenario.create_db_add_scenario_fk()} ) """) return cls._create_submodel(execute) @classmethod def _db_update(cls, execute, version, data=None): major, minor, release = version.strip().split(".") if major == "0": if int(minor) < 2 or (int(minor) == 2 and int(release) < 7): table_name = "initial_conditions_temperature" if not cls.is_table_exists(execute, table_name): cls._db_create(execute) return cls._update_submodel(execute, version, data) @classmethod def _db_load(cls, execute, data=None): new = [] status = data['status'] scenario = data["scenario"] loaded = data['loaded_pid'] if scenario is None: return new table = execute( "SELECT pamhyr_id, deleted, " + "name, temperature, scenario " + "FROM initial_conditions_temperature " + f"WHERE scenario = {scenario.id} " + f"AND pamhyr_id NOT IN ({', '.join(map(str, loaded))})" ) if table is not None: for row in table: it = iter(row) pid = next(it) deleted = (next(it) == 1) name = next(it) temperature = next(it) owner_scenario = next(it) ic = cls( id=pid, name=name, status=status, owner_scenario=owner_scenario ) if deleted: ic.set_as_deleted() ic.temperature = temperature data['ic_default_id'] = pid ic._data = ICTemperatureSpec._db_load(execute, data) loaded.add(pid) new.append(ic) data["scenario"] = scenario.parent new += cls._db_load(execute, data) data["scenario"] = scenario return new def _db_save(self, execute, data=None): if not self.must_be_saved(): return True execute( "DELETE FROM initial_conditions_temperature " + f"WHERE pamhyr_id = {self.id} " + f"AND scenario = {self._status.scenario_id}" ) temperature = self._temperature if \ self._temperature is not None else 0.0 sql = ( "INSERT INTO " + "initial_conditions_temperature(" + "pamhyr_id, deleted, name, temperature, " + "scenario" + ") " + "VALUES (" + f"{self.id}, {self.is_deleted()}, " + f"'{self._db_format(self._name)}', " + f"{temperature}, {self._status.scenario_id}" + ")" ) execute(sql) data['ic_default_id'] = self.id execute( "DELETE FROM initial_conditions_temperature_spec " + f"WHERE ic_default = {self.id} " + f"AND scenario = {self._status.scenario_id}" ) for ic_spec in self._data: ic_spec._db_save(execute, data) return True def __len__(self): return len(self._data) @property def name(self): return self._name @name.setter def name(self, name): self._name = name self.modified() @property def temperature(self): return self._temperature @temperature.setter def temperature(self, temperature): self._temperature = temperature self.modified() def new(self, index): n = ICTemperatureSpec(status=self._status) self._data.insert(index, n) self.modified() return n def delete(self, data): list( map( lambda x: x.set_as_deleted(), data ) ) self.modified() def delete_i(self, indexes): list( map( lambda e: e[1].set_as_deleted(), filter( lambda e: e[0] in indexes, enumerate(self._data) ) ) ) self.modified() def insert(self, index, data): if data in self._data: self.undelete([data]) else: self._data.insert(index, data) self.modified() def undelete(self, lst): for x in lst: x.set_as_not_deleted() self.modified() def split_reach(self, reach, profile, reach1, reach2): parts = [] for new_reach in (reach1, reach2): rks = new_reach.reach.get_rk() if rks: parts.append((new_reach.id, min(rks), max(rks))) specifications = [ specification for specification in self._data if (not specification.is_deleted() and specification.reach == reach.id) ] clones = [] for specification in specifications: begin = specification.start_rk end = specification.end_rk if begin is None or end is None: continue lower, upper = sorted((begin, end)) reverse = begin > end for new_reach, part_lower, part_upper in parts: clipped_lower = max(lower, part_lower) clipped_upper = min(upper, part_upper) if clipped_lower >= clipped_upper: continue start_rk = clipped_upper if reverse else clipped_lower end_rk = clipped_lower if reverse else clipped_upper clones.append(specification.cloned_for( new_reach, start_rk, end_rk )) if clones: self._data.extend(clones) self.modified()