LC Model

src/Model/LateralContributionsAdisTS/LateralContributionAdisTS.py

+# 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 <https://www.gnu.org/licenses/>.
+
+# -*- 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 = []
+    _id_cnt = 0
+
+    def __init__(self, id: int = -1, pollutant: int = -1, name: str = "", status=None):
+        super(LateralContributionAdisTS, self).__init__()
+
+        self._status = status
+
+        if id == -1:
+            self.id = LateralContributionAdisTS._id_cnt
+        else:
+            self.id = id
+
+        self._pollutant = pollutant
+        self._edge = None
+        self._begin_kp = 0.0
+        self._end_kp = 0.0
+        self._data = []
+        self._header = []
+        self._types = [float, float]
+
+        LateralContributionAdisTS._id_cnt = max(
+            LateralContributionAdisTS._id_cnt + 1, self.id)
+
+    @classmethod
+    def _db_create(cls, execute):
+        execute("""
+          CREATE TABLE lateral_contribution_adists(
+            id INTEGER NOT NULL PRIMARY KEY,
+            pollutant INTEGER NOT NULL,
+            edge INTEGER NOT NULL,
+            begin_kp REAL NOT NULL,
+            end_kp REAL NOT NULL,
+            FOREIGN KEY(pollutant) REFERENCES Pollutants(id),
+            FOREIGN KEY(edge) REFERENCES river_reach(id)
+          )
+        """)
+
+        execute("""
+          CREATE TABLE lateral_contribution_data_adists(
+            id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+            data0 TEXT NOT NULL,
+            data1 TEXT NOT NULL,
+            lc INTEGER,
+            FOREIGN KEY(lc) REFERENCES lateral_contribution(id)
+          )
+        """)
+
+        return cls._create_submodel(execute)
+
+    @classmethod
+    def _db_update(cls, execute, version):
+        return True
+
+    @classmethod
+    def _db_load(cls, execute, data=None):
+        new = []
+
+        table = execute(
+            "SELECT id, pollutant, edge, begin_kp, end_kp " +
+            "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_kp = row[3]
+                lc.end_kp = 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 WHERE id = {self.id}")
+        execute(f"DELETE FROM lateral_contribution_data_adists WHERE lc = {self.id}")
+
+        node = -1
+        if self._node is not None:
+            node = self._node
+
+        sql = (
+            "INSERT INTO " +
+            "lateral_contribution_adists(id, pollutant, edge, begin_kp, end_kp) " +
+            "VALUES (" +
+            f"{self.id}, {self._pollutant}, " +
+            f"{self._begin_kp}, {self._end_kp}" +
+            ")"
+        )
+        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_kp(self):
+        return self._begin_kp
+
+    @begin_kp.setter
+    def begin_kp(self, begin_kp):
+        self._begin_kp = begin_kp
+        self._status.modified()
+
+    @property
+    def end_kp(self):
+        return self._end_kp
+
+    @end_kp.setter
+    def end_kp(self, end_kp):
+        self._end_kp = end_kp
+        self._status.modified()
+

src/Model/LateralContributionsAdisTS/LateralContributionsAdisTSList.py

+# LateralContributionsAdisTSList.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 <https://www.gnu.org/licenses/>.
+
+# -*- coding: utf-8 -*-
+
+from copy import copy
+from tools import trace, timer
+
+from Model.Tools.PamhyrList import PamhyrModelList
+from Model.Except import NotImplementedMethodeError
+
+from Model.LateralConditionsAdisTS.LateralConditionAdisTS import LateralConditionAdisTS
+
+class LateralConditionsAdisTSList(PamhyrModelList):
+    _sub_classes = [
+        LateralConditionAdisTS,
+    ]
+
+    @classmethod
+    def _db_load(cls, execute, data=None):
+        new = cls(status=data['status'])
+
+        if data is None:
+            data = {}
+
+        new._lst = LateralConditionAdisTS._db_load(
+            execute, data
+        )
+
+        return new
+
+    def _db_save(self, execute, data=None):
+        execute("DELETE FROM lateral_contribution_adists")
+        execute("DELETE FROM lateral_condition_data_adists")
+
+        if data is None:
+            data = {}
+
+        for lc in self._lst:
+            lc._db_save(execute, data=data)
+
+        return True
+
+    def new(self, index, pollutant):
+        n = LateralConditionAdisTS(pollutant=pollutant, status=self._status)
+        self._lst.insert(index, n)
+        self._status.modified()
+        return n

src/Model/River.py

@@ -49,6 +49,7 @@ from Model.OutputKpAdists.OutputKpListAdists import OutputKpAdistsList
 from Model.Pollutants.PollutantsList import PollutantsList
 from Model.InitialConditionsAdisTS.InitialConditionsAdisTSList import InitialConditionsAdisTSList
 from Model.BoundaryConditionsAdisTS.BoundaryConditionsAdisTSList import BoundaryConditionsAdisTSList
+from Model.LateralConditionsAdisTS.LateralConditionsAdisTSList import LateralConditionsAdisTSList
 
 
 class RiverNode(Node, SQLSubModel):
@@ -237,6 +238,7 @@ class River(Graph, SQLSubModel):
         PollutantsList,
         InitialConditionsAdisTSList,
         BoundaryConditionsAdisTSList,
+        LateralConditionsAdisTSList,
     ]
 
     def __init__(self, status=None):
@@ -264,6 +266,7 @@ class River(Graph, SQLSubModel):
         self._Pollutants = PollutantsList(status=self._status)
         self._InitialConditionsAdisTS = InitialConditionsAdisTSList(status=self._status)
         self._BoundaryConditionsAdisTS = BoundaryConditionsAdisTSList(status=self._status)
+        self._LateralConditionsAdisTS = LateralConditionsAdisTSList(status=self._status)
 
     @classmethod
     def _db_create(cls, execute):
@@ -348,6 +351,8 @@ class River(Graph, SQLSubModel):
 
         new._BoundaryConditionsAdisTS = BoundaryConditionsAdisTSList._db_load(execute, data)
 
+        new._LateralConditionsAdisTS = LateralConditionsAdisTSList._db_load(execute, data)
+
         return new
 
     def _db_save(self, execute, data=None):
@@ -371,6 +376,7 @@ class River(Graph, SQLSubModel):
         objs.append(self._Pollutants)
         objs.append(self._InitialConditionsAdisTS)
         objs.append(self._BoundaryConditionsAdisTS)
+        objs.append(self._LateralConditionsAdisTS)
 
         self._save_submodel(execute, objs, data)
         return True
@@ -510,6 +516,10 @@ Last export at: @date."""
     def boundary_conditions_adists(self):
         return self._BoundaryConditionsAdisTS
 
+    @property
+    def lateral_conditions_adists(self):
+        return self._LateralConditionsAdisTS
+
     def get_params(self, solver):
         if solver in self._parameters:
             return self._parameters[solver]