# LateralContributionAdisTS.py -- Pamhyr # Copyright (C) 2023-2024 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 tools import ( trace, timer, old_pamhyr_date_to_timestamp, date_iso_to_timestamp, date_dmy_to_timestamp, ) from Model.Tools.PamhyrDB import SQLSubModel from Model.Except import NotImplementedMethodeError logger = logging.getLogger() class LateralContributionAdisTS(SQLSubModel): _sub_classes = [] def __init__(self, id: int = -1, pollutant: int = -1, name: str = "", status=None, owner_scenario=-1): super(LateralContributionAdisTS, self).__init__( id=id, status=status, owner_scenario=owner_scenario ) self._status = status self._pollutant = pollutant self._edge = None self._begin_rk = 0.0 self._end_rk = 0.0 self._data = [] self._header = ["time", "rate"] self._types = [self.time_convert, float] @classmethod def _db_create(cls, execute, ext=""): cls._db_create_lca(execute, ext) cls._db_create_lca_data(execute, ext) return cls._create_submodel(execute) @classmethod def _db_create_lca(cls, execute, ext=""): execute(f""" CREATE TABLE lateral_contribution_adists{ext}( {cls.create_db_add_pamhyr_id()}, pollutant INTEGER NOT NULL, reach INTEGER NOT NULL, begin_rk REAL NOT NULL, end_rk REAL NOT NULL, {Scenario.create_db_add_scenario()}, {Scenario.create_db_add_scenario_fk()}, FOREIGN KEY(pollutant) REFERENCES Pollutants(pamhyr_id), FOREIGN KEY(reach) REFERENCES river_reach(pamhyr_id) ) """) @classmethod def _db_create_lca_data(cls, execute, ext=""): execute(f""" CREATE TABLE lateral_contribution_data_adists{ext}( {cls.create_db_add_pamhyr_id()}, data0 TEXT NOT NULL, data1 TEXT NOT NULL, lca INTEGER, {Scenario.create_db_add_scenario()}, {Scenario.create_db_add_scenario_fk()}, FOREIGN KEY(lca) REFERENCES lateral_contribution_adists(pamhyr_id) ) """) @classmethod def _db_update(cls, execute, version, data=None): major, minor, release = version.strip().split(".") if major == "0" and int(minor) <= 1: if int(release) < 7: cls._db_create(execute) if major == "0" and int(minor) <= 2: if int(release) <= 0: cls._db_update_to_0_2_0(execute, data) return True @classmethod def _db_update_to_0_2_0(cls, execute, data): table = "lateral_contribution_adists" reachs = data['id2pid']['river_reach'] cls.update_db_add_pamhyr_id(execute, table, data) Scenario.update_db_add_scenario(execute, table) cls._db_create_lca(execute, ext="_tmp") execute( f"INSERT INTO {table}_tmp " + "(pamhyr_id, pollutant, reach, begin_rk, end_rk, scenario) " + "SELECT pamhyr_id, pollutant, edge, begin_rk, end_rk, scenario) " + f"FROM {table}" ) execute(f"DROP TABLE {table}") execute(f"ALTER TABLE {table}_tmp RENAME TO {table}") cls._db_update_to_0_2_0_set_reach_pid(execute, table, reachs) @classmethod def _db_update_to_0_2_0_data(cls, execute, data): table = "lateral_contribution_data_adists" reachs = data['id2pid']['river_reach'] cls.update_db_add_pamhyr_id(execute, table, data) Scenario.update_db_add_scenario(execute, table) cls._db_create_lca_data(execute, ext="_tmp") execute( f"INSERT INTO {table}_tmp " + "(pamhyr_id, data0, data1, lca, scenario) " + "SELECT pamhyr_id, data0, data1, lc, scenario) " + f"FROM {table}" ) execute(f"DROP TABLE {table}") execute(f"ALTER TABLE {table}_tmp RENAME TO {table}") cls._db_update_to_0_2_0_set_reach_pid(execute, table, reachs) @classmethod def _db_load(cls, execute, data=None): new = [] table = execute( "SELECT id, pollutant, edge, begin_rk, end_rk " + "FROM lateral_contribution_adists" ) if table is not None: for row in table: lc = cls( id=row[0], pollutant=row[1], status=data['status'] ) lc.edge = row[2] lc.begin_rk = row[3] lc.end_rk = row[4] values = execute( "SELECT data0," + " data1 FROM lateral_contribution_data_adists " + f"WHERE lc = '{lc.id}'" ) # Write data for v in values: data0 = lc._types[0](v[0]) data1 = lc._types[1](v[1]) # Replace data at pos ind lc._data.append((data0, data1)) new.append(lc) return new def _db_save(self, execute, data=None): execute(f"DELETE FROM lateral_contribution_adists" + f" WHERE id = {self.id}") execute(f"DELETE FROM lateral_contribution_data_adists" + f" WHERE lc = {self.id}") sql = ( "INSERT INTO " + "lateral_contribution_adists(id, " + "pollutant, edge, begin_rk, end_rk) " + "VALUES (" + f"{self.id}, {self._pollutant}, {self.edge}, " + f"{self._begin_rk}, {self._end_rk}" + ")" ) execute(sql) for d in self._data: data0 = self._db_format(str(d[0])) data1 = self._db_format(str(d[1])) sql = ( "INSERT INTO " + "lateral_contribution_data_adists(data0, data1, lc) " + f"VALUES ('{data0}', {data1}, {self.id})" ) execute(sql) return True def __len__(self): return len(self._data) @classmethod def time_convert(cls, data): if type(data) is str: if data.count("-") == 2: return date_iso_to_timestamp(data) if data.count("/") == 2: return date_dmy_to_timestamp(data) if data.count(":") == 3: return old_pamhyr_date_to_timestamp(data) if data.count(":") == 2: return old_pamhyr_date_to_timestamp("00:" + data) if data.count(".") == 1: return round(float(data)) return int(data) @property def edge(self): return self._edge @edge.setter def edge(self, edge): self._edge = edge self._status.modified() @property def header(self): return self._header.copy() @header.setter def header(self, header): self._header = header self._status.modified() @property def pollutant(self): return self._pollutant @property def data(self): return self._data.copy() @property def begin_rk(self): return self._begin_rk @begin_rk.setter def begin_rk(self, begin_rk): self._begin_rk = begin_rk self._status.modified() @property def end_rk(self): return self._end_rk @end_rk.setter def end_rk(self, end_rk): self._end_rk = end_rk self._status.modified()