-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Dev christiaan sprint 2 Finished
- Loading branch information
Showing
145 changed files
with
11,327 additions
and
2,681 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| import asyncio | ||
| from pathlib import Path | ||
| from typing import Optional | ||
|
|
||
| from otlmow_converter.OtlmowConverter import OtlmowConverter | ||
| from otlmow_model.OtlmowModel.Helpers.generated_lists import get_hardcoded_class_dict | ||
|
|
||
| from Domain.logger.OTLLogger import OTLLogger | ||
| from GUI.screens.RelationChange_elements.RelationChangeHelpers import RelationChangeHelpers | ||
|
|
||
|
|
||
| class Helpers: | ||
| all_OTL_asset_types_dict = {} | ||
|
|
||
| @classmethod | ||
| def create_external_typeURI_options(cls): | ||
| all_type_uris = get_hardcoded_class_dict() | ||
| for uri, info in all_type_uris.items(): | ||
| abbr_type_uri = RelationChangeHelpers.get_abbreviated_typeURI(uri, add_namespace=True) | ||
| screen_name = info['label'] | ||
| if "#" in abbr_type_uri: | ||
| abbr_type_uri_split = abbr_type_uri.split("#") | ||
| screen_name = "#".join([screen_name, abbr_type_uri_split[0]]) | ||
|
|
||
| cls.all_OTL_asset_types_dict[screen_name] = uri | ||
| cls.all_OTL_asset_types_dict = Helpers.sort_nested_dict(cls.all_OTL_asset_types_dict) | ||
|
|
||
| @classmethod | ||
| def sort_nested_dict(cls, dictionary, by='keys'): | ||
| """Recursively sorts a dictionary by keys or values.""" | ||
| if not isinstance(dictionary, dict): | ||
| if isinstance(dictionary, list): | ||
| return sorted(dictionary, key=lambda relation_object: relation_object.typeURI) | ||
| return dictionary | ||
|
|
||
| # Sort the current dictionary | ||
| if by == 'keys': | ||
| sorted_dict = { | ||
| k: cls.sort_nested_dict(v, by=by) | ||
| for k, v in sorted(dictionary.items()) | ||
| } | ||
|
|
||
| elif by == 'values': | ||
| sorted_dict = { | ||
| k: cls.sort_nested_dict(v, by=by) | ||
| for k, v in sorted(dictionary.items(), key=lambda item: item[1]) | ||
| } | ||
|
|
||
| else: | ||
| raise ValueError("Invalid sort parameter. Use 'keys' or 'values'.") | ||
|
|
||
| return sorted_dict | ||
|
|
||
| @classmethod | ||
| def equal_paths(cls,path1: Optional[Path], path2: Optional[Path]): | ||
| if path1 and path2: | ||
| return path1.name == path2.name | ||
| elif path1 or path2: | ||
| return False | ||
| else: | ||
| return True | ||
|
|
||
| @classmethod | ||
| def converter_from_file_to_object(cls,file_path,**kwargs): | ||
| OTLLogger.logger.debug(f"Execute OtlmowConverter.from_file_to_objects({file_path.name})", | ||
| extra={"timing_ref": f"file_to_objects_{file_path.stem}"}) | ||
| object_lists = list(OtlmowConverter.from_file_to_objects(file_path,**kwargs)) | ||
| OTLLogger.logger.debug(f"Execute OtlmowConverter.from_file_to_objects({file_path.name}) ({len(object_lists)} objects)", | ||
| extra={"timing_ref": f"file_to_objects_{file_path.stem}"}) | ||
| return object_lists |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,163 @@ | ||
| import logging | ||
| import os | ||
| import re | ||
| import subprocess | ||
| import sys | ||
| from pathlib import Path | ||
|
|
||
| from Domain import global_vars | ||
| from Domain.logger.OTLLogger import OTLLogger | ||
| from Exceptions.FDOToolboxNotInstalledError import FDOToolboxNotInstalledError | ||
|
|
||
| ROOT_DIR = Path(Path(__file__).absolute()).parent.parent | ||
| sys.path.insert(0,str(ROOT_DIR.absolute()))# needed for python to import project files | ||
|
|
||
| from Exceptions.FDOToolboxProcessError import FDOToolboxProcessError | ||
| from Exceptions.WrongFileTypeError import WrongFileTypeError | ||
| from GUI.translation.GlobalTranslate import GlobalTranslate | ||
|
|
||
|
|
||
| class SDFHandler: | ||
|
|
||
|
|
||
|
|
||
| @classmethod | ||
| def _get_classes_from_SDF_file(cls, sdf_file_path:Path) -> list[str]: | ||
|
|
||
| sdf_file_path_str = sdf_file_path.absolute() | ||
| command = f'"{global_vars.FDO_toolbox_path_str}" list-classes --from-file "{sdf_file_path_str}"' | ||
|
|
||
| output, error = cls.run_command(command=command) | ||
|
|
||
| if error: | ||
| cls._filter_out_coordinate_system_not_installed_error(command=command, error=error) | ||
|
|
||
| if output: | ||
| return output.split("\n") | ||
| else: | ||
| return [] | ||
|
|
||
| @classmethod | ||
| def _filter_out_coordinate_system_not_installed_error(cls, command:str, error:str) -> None: | ||
| # The following error occurs even though the program works properly | ||
| # Here we filter out this error to be able to capture and use other critical errors | ||
| pattern = re.compile( | ||
| r'\(\d+\) DefaultDir: "C:\\ProgramData\\Autodesk\\Geospatial Coordinate Systems" is not a directory! Install the Coordinate System library into this directory or set MENTOR_DICTIONARY_PATH to where they are currently installed\.\n' | ||
| r'\(\d+\) MgCoordinateSystemInitializationFailedException caught in CCoordinateSystemCatalog constructor - The coordinate system initialization failed\.\n\n' | ||
| r'- MgCoordinateSystemCatalog\.SetDefaultDictionaryDirAndFileNames\(\) line 603 file \.\.\\CoordinateSystem\\CoordSysCatalog\.cpp\n' | ||
| r'- MgCoordinateSystemCatalog\.GetDefaultDictionaryDir\(\) line 263 file \.\.\\CoordinateSystem\\CoordSysCatalog\.cpp' | ||
| ) | ||
| filtered_error, instance_count = pattern.subn("", error) | ||
| if filtered_error: | ||
| raise FDOToolboxProcessError(language=GlobalTranslate._, command=command, | ||
| fdo_toolbox_error=filtered_error) | ||
|
|
||
|
|
||
| @classmethod | ||
| def _validate_SDF_file(cls, sdf_filepath) -> None: | ||
| if not sdf_filepath.exists(): | ||
| raise FileNotFoundError(f'{sdf_filepath} is not a valid path. File does not exist.') | ||
| if sdf_filepath.suffix != ".sdf": | ||
| raise WrongFileTypeError(language=GlobalTranslate._, | ||
| expected_filetype_name="SDF-file", | ||
| expected_filetype_suffix=".sdf") | ||
|
|
||
| @classmethod | ||
| def _get_objects_from_class(cls, sdf_filepath, sdf_class) -> str: | ||
|
|
||
| sdf_file_path_str = sdf_filepath.absolute() | ||
| command = f'"{global_vars.FDO_toolbox_path_str}" query-features --class "{sdf_class}" --from-file "{sdf_file_path_str}" --format CSV' | ||
|
|
||
| output, error = cls.run_command(command) | ||
| if error: | ||
| cls._filter_out_coordinate_system_not_installed_error(command, error) | ||
|
|
||
| output = output.replace("_",".") | ||
| output = output.replace("Geometry", "geometry") | ||
| output = output.replace("XYZ", "Z") | ||
|
|
||
| return output + "\n" # added for compatibility with old OTL-wizard csv generation | ||
|
|
||
| @classmethod | ||
| def convert_SDF_to_CSV(cls, sdf_filepath:Path, csv_output_path:Path) -> list[str]: | ||
|
|
||
| OTLLogger.logger.debug( | ||
| f"Executing Domain.SDFHandler.SDFHandler.convert_SDF_to_CSV for {sdf_filepath.name}", | ||
| extra={"timing_ref": f"validate_{sdf_filepath.name}"}) | ||
|
|
||
| output_csv_filepath_list = [] | ||
|
|
||
| cls._check_if_FDOToolbox_is_installed() | ||
| cls._validate_SDF_file(sdf_filepath) | ||
|
|
||
| # format the expected output csv files based on the classes in the sdf file | ||
| output_classes: list[str] = SDFHandler._get_classes_from_SDF_file(sdf_filepath) | ||
| if ((os.path.exists(csv_output_path) and os.path.isdir(csv_output_path)) or | ||
| csv_output_path.suffix == ""): | ||
| csv_output_path_is_dir = True | ||
| output_basepath: str = str(csv_output_path) | ||
| os.makedirs(csv_output_path, exist_ok=True) | ||
| else: | ||
| csv_output_path_is_dir = False | ||
| output_basepath: str = os.path.splitext(csv_output_path)[0] | ||
| os.makedirs(csv_output_path.parent, exist_ok=True) | ||
|
|
||
| for otlclass in output_classes: | ||
|
|
||
| objects_str:str = SDFHandler._get_objects_from_class(sdf_filepath=sdf_filepath, sdf_class=otlclass) | ||
|
|
||
|
|
||
|
|
||
| # build absolute path to csv of output | ||
| if csv_output_path_is_dir: | ||
| filepath_of_output_csv_for_one_class = str( | ||
| Path(output_basepath) / f"{otlclass}.csv") | ||
| else: | ||
| filepath_of_output_csv_for_one_class = output_basepath + otlclass + ".csv" | ||
|
|
||
| with open(filepath_of_output_csv_for_one_class,mode='w+') as output_csv_file: | ||
| output_csv_file.write(objects_str) | ||
|
|
||
| logging.debug(f"created file: {filepath_of_output_csv_for_one_class}") | ||
| output_csv_filepath_list.append(filepath_of_output_csv_for_one_class) | ||
|
|
||
| OTLLogger.logger.debug( | ||
| f"Executing Domain.SDFHandler.SDFHandler.convert_SDF_to_CSV for {sdf_filepath.name}", | ||
| extra={"timing_ref": f"validate_{sdf_filepath.name}"}) | ||
|
|
||
| return output_csv_filepath_list | ||
|
|
||
| @classmethod | ||
| def run_command(cls,command): | ||
| # Using subprocess.run to execute the command without opening a terminal window | ||
| result = subprocess.run(command, shell=True, stdout=subprocess.PIPE, | ||
| stderr=subprocess.PIPE, text=True) | ||
| return result.stdout.strip(), result.stderr.strip() | ||
|
|
||
| @classmethod | ||
| def _check_if_FDOToolbox_is_installed(cls): | ||
| if not os.path.exists(global_vars.FDO_toolbox_path_str): | ||
| raise FDOToolboxNotInstalledError(GlobalTranslate._) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| logger = logging.getLogger() | ||
| logger.setLevel(logging.DEBUG) | ||
| settings = {"language": "ENGLISH"} | ||
| root = Path(Path(__file__).absolute()).parent.parent | ||
| GlobalTranslate(settings, root/ 'locale') | ||
| # sdf_path_input = Path("C:\\Users\\chris\\Documents\\job_related\\wegen_en_verkeer\\new_python_otl_wizard\\testData\\Ruben_SDF_test\\DA-2024-03992_export\\download.sdf") | ||
| sdf_path_input = Path( | ||
| "C:\\Users\\chris\\Documents\\job_related\\wegen_en_verkeer\\new_python_otl_wizard\\testData\\Ruben_SDF_test\\DA-2025-00023_export\\download.sdf") | ||
|
|
||
| # sdf_file_classes = SDFHandler.get_classes_from_SDF_file(sdf_file_path=sdf_path_input) | ||
| # sdf_objects = SDFHandler._get_objects_from_class(sdf_file_path=sdf_path_input,sdf_class="OTL_Brandblusser") | ||
| sdf_objects = SDFHandler._get_objects_from_class(sdf_filepath=sdf_path_input, sdf_class="OTL_Voedingskabel") | ||
| print(sdf_objects) | ||
|
|
||
| # write output to test files | ||
| # output = root / "UnitTests\\test_files\\output_ref\\output_get_object_from_class_DA-2025-00023_export.txt" | ||
| # output_csv = root / "UnitTests\\test_files\\output_test\\convert_SDF_to_CSV_DA-2025-00023_export\\test_DA-2024-03992_export.csv" | ||
| # output_csv = root / "C:\\Users\\chris\\Documents\\job_related\\wegen_en_verkeer\\new_python_otl_wizard\\testData\\Ruben_SDF_test\\DA-2025-00023_export\\christiaan_omzetting" | ||
| # SDFHandler.convert_SDF_to_CSV(sdf_path_input,output_csv) | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.