Skip to content

Commit

Permalink
...
Browse files Browse the repository at this point in the history
  • Loading branch information
@leendertvanwolfswinkel
leendertvanwolfswinkel committed Dec 23, 2024
1 parent 320abce commit 46fc455
Show file tree
Hide file tree
Showing 2 changed files with 148 additions and 149 deletions.
1 change: 0 additions & 1 deletion dependencies.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,6 @@
Dependency("alembic", "alembic", "==1.8.*", False),
Dependency("threedigrid", "threedigrid", "==2.2.*", False),
Dependency("threedi-schema", "threedi_schema", "==0.230.0.dev0", False),
# Dependency("threedi-modelchecker", "threedi_modelchecker", "==2.14.1", False),
Dependency("threedidepth", "threedidepth", "==0.6.3", False),
Dependency("click", "click", ">=8.0", False),
Dependency("packaging", "packaging", "", False),
Expand Down
296 changes: 148 additions & 148 deletions processing/schematisation_algorithms.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
# from threedi_results_analysis.processing.deps.guess_indicator import guess_indicators_utils

from threedi_schema import ThreediDatabase
from threedi_modelchecker import ThreediModelChecker
# from threedi_modelchecker import ThreediModelChecker
from threedi_schema import errors
# from threedi_results_analysis.processing.download_hydx import download_hydx

Expand Down Expand Up @@ -130,153 +130,153 @@ def createInstance(self):
return MigrateAlgorithm()


class CheckSchematisationAlgorithm(QgsProcessingAlgorithm):
"""
Run the schematisation checker
"""

INPUT = "INPUT"
OUTPUT = "OUTPUT"
ADD_TO_PROJECT = "ADD_TO_PROJECT"

def initAlgorithm(self, config):
self.addParameter(
QgsProcessingParameterFile(
self.INPUT, self.tr("3Di Spatialite"), extension="sqlite"
)
)

self.addParameter(
QgsProcessingParameterFileDestination(
self.OUTPUT, self.tr("Output"), fileFilter="csv"
)
)

self.addParameter(
QgsProcessingParameterBoolean(
self.ADD_TO_PROJECT, self.tr("Add result to project"), defaultValue=True
)
)

def processAlgorithm(self, parameters, context, feedback):
self.add_to_project = self.parameterAsBoolean(
parameters, self.ADD_TO_PROJECT, context
)
self.output_file_path = None
input_filename = self.parameterAsFile(parameters, self.INPUT, context)
threedi_db = get_threedi_database(filename=input_filename, feedback=feedback)
if not threedi_db:
return {self.OUTPUT: None}
try:
model_checker = ThreediModelChecker(threedi_db)
except errors.MigrationMissingError:
feedback.pushWarning(
"The selected 3Di model does not have the latest migration. Please "
"migrate your model to the latest version."
)
return {self.OUTPUT: None}
schema = threedi_db.schema
schema.set_spatial_indexes()
generated_output_file_path = self.parameterAsFileOutput(
parameters, self.OUTPUT, context
)
self.output_file_path = f"{os.path.splitext(generated_output_file_path)[0]}.csv"
session = model_checker.db.get_session()
session.model_checker_context = model_checker.context
total_checks = len(model_checker.config.checks)
progress_per_check = 100.0 / total_checks
checks_passed = 0
try:
with open(self.output_file_path, "w", newline="") as output_file:
writer = csv.writer(output_file)
writer.writerow(
[
"level",
"error_code",
"id",
"table",
"column",
"value",
"description",
]
)
for i, check in enumerate(model_checker.checks(level="info")):
model_errors = check.get_invalid(session)
for error_row in model_errors:
writer.writerow(
[
check.level.name,
check.error_code,
error_row.id,
check.table.name,
check.column.name,
getattr(error_row, check.column.name),
check.description(),
]
)
checks_passed += 1
feedback.setProgress(int(checks_passed * progress_per_check))
except PermissionError:
# PermissionError happens for example when a user has the file already open
# with Excel on Windows, which locks the file.
feedback.pushWarning(
f"Not enough permissions to write the file '{self.output_file_path}'.\n\n"
"The file may be used by another program. Please close all "
"other programs using the file or select another output "
"file."
)
return {self.OUTPUT: None}

return {self.OUTPUT: self.output_file_path}

def postProcessAlgorithm(self, context, feedback):
if self.add_to_project:
if self.output_file_path:
result_layer = QgsVectorLayer(
self.output_file_path, "3Di schematisation errors"
)
QgsProject.instance().addMapLayer(result_layer)
return {self.OUTPUT: self.output_file_path}

def name(self):
"""
Returns the algorithm name, used for identifying the algorithm. This
string should be fixed for the algorithm, and must not be localised.
The name should be unique within each provider. Names should contain
lowercase alphanumeric characters only and no spaces or other
formatting characters.
"""
return "check_schematisation"

def displayName(self):
"""
Returns the translated algorithm name, which should be used for any
user-visible display of the algorithm name.
"""
return self.tr("Check Schematisation")

def group(self):
"""
Returns the name of the group this algorithm belongs to. This string
should be localised.
"""
return self.tr(self.groupId())

def groupId(self):
"""
Returns the unique ID of the group this algorithm belongs to. This
string should be fixed for the algorithm, and must not be localised.
The group id should be unique within each provider. Group id should
contain lowercase alphanumeric characters only and no spaces or other
formatting characters.
"""
return "Schematisation"

def tr(self, string):
return QCoreApplication.translate("Processing", string)

def createInstance(self):
return CheckSchematisationAlgorithm()
# class CheckSchematisationAlgorithm(QgsProcessingAlgorithm):
# """
# Run the schematisation checker
# """

# INPUT = "INPUT"
# OUTPUT = "OUTPUT"
# ADD_TO_PROJECT = "ADD_TO_PROJECT"

# def initAlgorithm(self, config):
# self.addParameter(
# QgsProcessingParameterFile(
# self.INPUT, self.tr("3Di Spatialite"), extension="sqlite"
# )
# )

# self.addParameter(
# QgsProcessingParameterFileDestination(
# self.OUTPUT, self.tr("Output"), fileFilter="csv"
# )
# )

# self.addParameter(
# QgsProcessingParameterBoolean(
# self.ADD_TO_PROJECT, self.tr("Add result to project"), defaultValue=True
# )
# )

# def processAlgorithm(self, parameters, context, feedback):
# self.add_to_project = self.parameterAsBoolean(
# parameters, self.ADD_TO_PROJECT, context
# )
# self.output_file_path = None
# input_filename = self.parameterAsFile(parameters, self.INPUT, context)
# threedi_db = get_threedi_database(filename=input_filename, feedback=feedback)
# if not threedi_db:
# return {self.OUTPUT: None}
# try:
# model_checker = ThreediModelChecker(threedi_db)
# except errors.MigrationMissingError:
# feedback.pushWarning(
# "The selected 3Di model does not have the latest migration. Please "
# "migrate your model to the latest version."
# )
# return {self.OUTPUT: None}
# schema = threedi_db.schema
# schema.set_spatial_indexes()
# generated_output_file_path = self.parameterAsFileOutput(
# parameters, self.OUTPUT, context
# )
# self.output_file_path = f"{os.path.splitext(generated_output_file_path)[0]}.csv"
# session = model_checker.db.get_session()
# session.model_checker_context = model_checker.context
# total_checks = len(model_checker.config.checks)
# progress_per_check = 100.0 / total_checks
# checks_passed = 0
# try:
# with open(self.output_file_path, "w", newline="") as output_file:
# writer = csv.writer(output_file)
# writer.writerow(
# [
# "level",
# "error_code",
# "id",
# "table",
# "column",
# "value",
# "description",
# ]
# )
# for i, check in enumerate(model_checker.checks(level="info")):
# model_errors = check.get_invalid(session)
# for error_row in model_errors:
# writer.writerow(
# [
# check.level.name,
# check.error_code,
# error_row.id,
# check.table.name,
# check.column.name,
# getattr(error_row, check.column.name),
# check.description(),
# ]
# )
# checks_passed += 1
# feedback.setProgress(int(checks_passed * progress_per_check))
# except PermissionError:
# # PermissionError happens for example when a user has the file already open
# # with Excel on Windows, which locks the file.
# feedback.pushWarning(
# f"Not enough permissions to write the file '{self.output_file_path}'.\n\n"
# "The file may be used by another program. Please close all "
# "other programs using the file or select another output "
# "file."
# )
# return {self.OUTPUT: None}

# return {self.OUTPUT: self.output_file_path}

# def postProcessAlgorithm(self, context, feedback):
# if self.add_to_project:
# if self.output_file_path:
# result_layer = QgsVectorLayer(
# self.output_file_path, "3Di schematisation errors"
# )
# QgsProject.instance().addMapLayer(result_layer)
# return {self.OUTPUT: self.output_file_path}

# def name(self):
# """
# Returns the algorithm name, used for identifying the algorithm. This
# string should be fixed for the algorithm, and must not be localised.
# The name should be unique within each provider. Names should contain
# lowercase alphanumeric characters only and no spaces or other
# formatting characters.
# """
# return "check_schematisation"

# def displayName(self):
# """
# Returns the translated algorithm name, which should be used for any
# user-visible display of the algorithm name.
# """
# return self.tr("Check Schematisation")

# def group(self):
# """
# Returns the name of the group this algorithm belongs to. This string
# should be localised.
# """
# return self.tr(self.groupId())

# def groupId(self):
# """
# Returns the unique ID of the group this algorithm belongs to. This
# string should be fixed for the algorithm, and must not be localised.
# The group id should be unique within each provider. Group id should
# contain lowercase alphanumeric characters only and no spaces or other
# formatting characters.
# """
# return "Schematisation"

# def tr(self, string):
# return QCoreApplication.translate("Processing", string)

# def createInstance(self):
# return CheckSchematisationAlgorithm()


# class ImportSufHydAlgorithm(QgsProcessingAlgorithm):
Expand Down

0 comments on commit 46fc455

Please sign in to comment.