ags_2_db_algorithm.py

# -*- coding: utf-8 -*-

"""
/***************************************************************************
 AGSTools
								 A QGIS plugin
 This plugin parses an AGS file and creates a database from it
 Generated by Plugin Builder: http://g-sherman.github.io/Qgis-Plugin-Builder/
							  -------------------
		begin                : 2023-04-19
		copyright            : (C) 2025 by Oliver Burdekin / burdGIS
		email                : info@burdgis.com
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   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 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/
"""

__author__ = 'Oliver Burdekin / burdGIS'
__date__ = '2023-04-19'
__copyright__ = '(C) 2025 by Oliver Burdekin / burdGIS'


# This will get replaced with a git SHA1 when you do a git archive

__revision__ = '$Format:%H$'

from qgis.PyQt.QtCore import QCoreApplication, QSettings, QVariant
from qgis.core import (QgsApplication,
					   QgsSettings,
					   QgsProcessingAlgorithm,
					   QgsProcessingParameterFile,
					   QgsProcessingParameterFileDestination,
					   QgsProcessingParameterCrs,
					   QgsCoordinateReferenceSystem,
					   QgsVectorLayer,
					   QgsVectorLayer,
					   QgsField,
					   QgsFields,
					   QgsFeature,
					   QgsGeometry,
					   QgsPointXY,
					   QgsVectorFileWriter,
					   QgsProject,
					   QgsCoordinateTransform,
					   QgsProcessingException,
					   QgsDataSourceUri,
					   QgsMapLayer,
					   QgsProviderRegistry
						)
from qgis.utils import iface
from io import StringIO
# import sqlite3
import os


class AGS2DBAlgorithm(QgsProcessingAlgorithm):
	"""
	This is an example algorithm that takes a vector layer and
	creates a new identical one.

	It is meant to be used as an example of how to create your own
	algorithms and explain methods and variables used to do it. An
	algorithm like this will be available in all elements, and there
	is not need for additional work.

	All Processing algorithms should extend the QgsProcessingAlgorithm
	class.
	"""

	# Constants used to refer to parameters and outputs. They will be
	# used when calling the algorithm from another algorithm, or when
	# calling from the QGIS console.

	OUTPUT = 'OUTPUT'
	INPUT = 'INPUT'
	CRS = 'CRS'

	def initAlgorithm(self, config):
		"""
		Here we define the inputs and output of the algorithm, along
		with some other properties.
		"""

		# We add the input vector features source. It can have any kind of
		# geometry.
		self.addParameter(
			QgsProcessingParameterFile(
				self.INPUT,
				self.tr('Input File'),
				
			)
		)

		self.addParameter(
			QgsProcessingParameterCrs(
				self.CRS,
				'Coordinate Reference System',
				defaultValue=QgsCoordinateReferenceSystem('EPSG:27700')
			)
		)


		# We add a feature sink in which to store our processed features (this
		# usually takes the form of a newly created vector layer when the
		# algorithm is run in QGIS).
		self.addParameter(
			QgsProcessingParameterFileDestination(
				self.OUTPUT,
				self.tr('Output File'),
				fileFilter='GeoPackage (*.gpkg);;SpatiaLite (*.sqlite)'
				
			)
		)

	def parse_ags_file(self, file_contents):

		def is_numeric(value):
			"""Utility function to check if value can be converted to a number."""
			try:
				float(value)
				return True
			except ValueError:
				return False

		lines = file_contents.split('\n')
		data = {}
		current_group = None
		headers = []

		for line in lines:
			if not line.strip():
				continue

			temp = line.strip().split('","')
			temp = [item.strip('"') for item in temp]

			if temp[0] == 'GROUP':
				current_group = temp[1]
				data[current_group] = []  # Initialize the group
			elif temp[0] == 'HEADING':
				headers = temp[1:]
			elif temp[0] == 'UNIT':
				if current_group and headers:
					unit_values = temp[1:]
					# Pad missing values with empty strings
					if len(unit_values) < len(headers):
						unit_values.extend([''] * (len(headers) - len(unit_values)))
					# Validate and skip invalid unit rows
					if all(not value.strip() for value in unit_values):  # All empty or whitespace
						continue
					data[f"{current_group}_units"] = dict(zip(headers, unit_values))
			elif temp[0] == 'DATA':
				# Add data rows to the current group
				if current_group and headers:
					record = dict(zip(headers, temp[1:]))
					data[current_group].append(record)

		# Detect column types for each group independently
		column_types = {}
		for group_name, records in data.items():
			if group_name.endswith("_units"):
				continue

			if records:
				group_headers = list(records[0].keys())
				column_types[group_name] = {}
				for header in group_headers:
					all_numeric = all(is_numeric(record.get(header)) for record in records if record.get(header))
					column_types[group_name][header] = "REAL" if all_numeric else "TEXT"

		return data, column_types

	def createLOCAFeatures(self, records, column_types, crs, feedback):
		"""
		Create a memory point layer for LOCA features using available coordinate fields.
		- Checks for LOCA_NATE/LOCA_NATN, LOCA_LOCX/LOCA_LOCY, LOCA_LAT/LOCA_LON in that order.
		- Logs if features or all fail to have coordinates.
		- Reprojects if using LAT/LON from EPSG:4326 to target CRS.
		"""
		# Determine which coordinate fields are present and use them in priority:
		coord_priority = [
			('LOCA_NATE', 'LOCA_NATN'),
			('LOCA_LOCX', 'LOCA_LOCY'),
			('LOCA_LAT', 'LOCA_LON')
		]
		
		chosen_coords = None
		for pair in coord_priority:
			# Check if both fields exist
			if all(p in column_types['LOCA'] for p in pair):
				x_field, y_field = pair
				# Check if there's at least one record with non-empty coordinates
				has_data = False
				for record in records:
					x_val = record.get(x_field)
					y_val = record.get(y_field)
					if x_val and x_val.strip() != '' and y_val and y_val.strip() != '':
						has_data = True
						break

				if has_data:
					chosen_coords = pair
					break

		# Create the layer
		layer = QgsVectorLayer("Point?crs={}".format(crs.authid()), "LOCA", "memory")
		data_provider = layer.dataProvider()

		# Add fields
		fields = QgsFields()
		for field_name, field_type in column_types['LOCA'].items():
			qgis_field_type = QVariant.Double if field_type == "REAL" else QVariant.String
			fields.append(QgsField(field_name, qgis_field_type))
		data_provider.addAttributes(fields)
		layer.updateFields()

		if chosen_coords is None:
			feedback.reportError("No recognized coordinate fields found in LOCA. Creating empty point layer.")
			return layer

		x_field, y_field = chosen_coords

		# Set up coordinate transform if needed
		# If we're using LAT/LON, we assume they're in EPSG:4326 and must transform to chosen CRS
		transform = None
		if chosen_coords == ('LOCA_LAT', 'LOCA_LON'):
			source_crs = QgsCoordinateReferenceSystem('EPSG:4326')
			transform = QgsCoordinateTransform(source_crs, crs, QgsProject.instance())

		features_with_coords = 0
		for record in records:
			# Get attributes
			attrs = []
			for header, ftype in column_types['LOCA'].items():
				val = record.get(header)
				if ftype == "REAL":
					if val is None or val.strip() == '':
						attrs.append(None)
					else:
						attrs.append(float(val))
				else:
					attrs.append(val if val is not None else None)

			# Attempt to get coordinates
			x_val = record.get(x_field)
			y_val = record.get(y_field)
			if x_val is None or x_val.strip() == '' or y_val is None or y_val.strip() == '':
				# No coordinates for this feature
				feedback.pushInfo(f"LOCA feature missing coordinates: {record}")
				# We still add the feature but without geometry
				
				f = QgsFeature()
				f.setAttributes(attrs)
				f.setGeometry(None)
				data_provider.addFeature(f)
				continue

			# Convert to float
			x = float(x_val)
			y = float(y_val)

			# Transform if LAT/LON
			if transform:
				pt = transform.transform(QgsPointXY(y, x))  # careful with order: LAT = Y, LON = X
			else:
				pt = QgsPointXY(x, y)

			f = QgsFeature()
			f.setAttributes(attrs)
			f.setGeometry(QgsGeometry.fromPointXY(pt))
			data_provider.addFeature(f)
			features_with_coords += 1

		layer.updateExtents()

		if features_with_coords == 0:
			feedback.reportError("No LOCA features had valid coordinates. The LOCA layer will have no geometries.")

		return layer

	def create_database_connection(self, output_path, feedback):
		# Only works for gpkg
		md = QgsProviderRegistry.instance().providerMetadata("ogr")
		conn = md.createConnection(output_path, {})
		conn_name = os.path.splitext(os.path.basename(output_path))[0]
		md.saveConnection(conn, conn_name)


		# Refresh database connections in the Browser Panel
		iface.browserModel().refresh()

	def add_svg_paths(self, feedback):
	
		svg_path = os.path.join(os.path.dirname(__file__), 'styles', 'svg')
		svg_paths = QgsApplication.svgPaths()
		if svg_path not in svg_paths:
			svg_paths.append(svg_path)
			QgsApplication.setDefaultSvgPaths(svg_paths)
			feedback.pushInfo("Added custom SVG path for symbols.")

	def loadLayerAndApplyStyle(self, output_path, table_name, qml_path, feedback):
		# Load the layer using OGR provider. GPK and SpatiaLite are supported.

		layer_path = f"{output_path}|layername={table_name}"
		layer = QgsVectorLayer(layer_path, table_name, "ogr")

		if not layer.isValid():
			feedback.pushInfo(f"Table '{table_name}' cannot be added.")
			return None

		QgsProject.instance().addMapLayer(layer)
		feedback.pushInfo(f"Added '{table_name}' layer to the project.")

		if layer.isValid() and os.path.exists(qml_path):
			layer.loadNamedStyle(qml_path)
			layer.triggerRepaint()
			feedback.pushInfo("Applied QML style to the LOCA layer.")

		if iface:
			iface.layerTreeView().refreshLayerSymbology(layer.id())

		return layer

	def processAlgorithm(self, parameters, context, feedback):
		# Define the output path
		output_path = self.parameterAsFileOutput(parameters, self.OUTPUT, context)
		ext = os.path.splitext(output_path)[1].lower()
		feedback.pushInfo(f"Writing all groups GeoPackage to: {output_path}")

		# Remove existing GeoPackage if it exists
		if os.path.exists(output_path):
			os.remove(output_path)
			feedback.pushInfo(f"Removed existing GeoPackage at: {output_path}")

		# Load AGS4 file contents
		ags_file_path = self.parameterAsFile(parameters, self.INPUT, context)
		with open(ags_file_path, 'r') as ags_file:
			file_contents = ags_file.read()

		# Get chosen CRS
		crs = self.parameterAsCrs(parameters, self.CRS, context)

		# Parse the AGS file
		data, column_types = self.parse_ags_file(file_contents)

		first_layer = True
		transform_context = context.transformContext()

		for group_name, records in data.items():
			if group_name.endswith("_units"):
				continue  # Skip unit tables

			feedback.pushInfo(f"Processing group: {group_name}")

			if group_name.upper() == "LOCA":
				# Create spatial LOCA layer
				layer = self.createLOCAFeatures(records, column_types, crs, feedback)

			else:
				# Non-spatial layer
				layer = QgsVectorLayer("NoGeometry", group_name, "memory")
				data_provider = layer.dataProvider()

				# Add fields
				fields = QgsFields()
				for field_name, field_type in column_types[group_name].items():
					qgis_field_type = QVariant.Double if field_type == "REAL" else QVariant.String
					fields.append(QgsField(field_name, qgis_field_type))
				data_provider.addAttributes(fields)
				layer.updateFields()

				# Populate layer
				for record in records:
					feature = QgsFeature()
					attrs = []
					for header, ftype in column_types[group_name].items():
						val = record.get(header)
						if ftype == "REAL":
							if val is None or val.strip() == '':
								attrs.append(None)
							else:
								attrs.append(float(val))
						else:
							attrs.append(val if val is not None else None)
					feature.setAttributes(attrs)
					data_provider.addFeature(feature)
				layer.updateExtents()

			# Export layer
			options = QgsVectorFileWriter.SaveVectorOptions()
			options.driverName = "GPKG"
			if ext == '.gpkg':
				options.driverName = "GPKG"
			elif ext == '.sqlite':
				options.driverName = "SQLite"
				options.driverOptions = ["SPATIALITE=YES"]
			else:
				raise QgsProcessingException("Unsupported file format selected.")

			options.layerName = group_name
			options.fileEncoding = "UTF-8"
			# Geometry type will be derived automatically for V3 method

			if first_layer:
				options.actionOnExistingFile = QgsVectorFileWriter.CreateOrOverwriteFile
			else:
				options.actionOnExistingFile = QgsVectorFileWriter.CreateOrOverwriteLayer

			error, newFilename, newLayer, errorMessage = QgsVectorFileWriter.writeAsVectorFormatV3(
				layer,
				output_path,
				transform_context,
				options
			)

			if error == QgsVectorFileWriter.NoError:
				feedback.pushInfo(f"Successfully wrote group '{group_name}' to GeoPackage.")
			else:
				feedback.reportError(f"Error writing group '{group_name}' to GeoPackage: {errorMessage}")

			first_layer = False

		feedback.pushInfo("ALL GROUPS WRITTEN - DB CREATED")

		# After the writing is done, call the helper functions:
		self.add_svg_paths(feedback)



		qml_path = os.path.join(os.path.dirname(__file__), 'styles', 'loca_spatial.qml')

    	# Load and style the LOCA layer
		self.loadLayerAndApplyStyle(output_path, "LOCA", qml_path, feedback)

		self.create_database_connection(output_path, feedback)

		return {self.OUTPUT: output_path}
	
	def processing_log(self, message):
		"""
		Logs a message to the Processing log.
		"""
		self.logMessage(message)

	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 'AGS2DB'

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

	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 ''

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

	def createInstance(self):
		return AGS2DBAlgorithm()