Merge pull request #151 from Patowhiz/pwa

Implements importing of large files and adds gaps analysis on data availability

Author: Patowhiz

back-end/api/package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "api",
-  "version": "preview-2.0.1",
+  "version": "preview-2.0.3",
   "description": "Climsoft API",
   "author": "Climsoft Foundation",
   "private": true,
@@ -32,7 +32,7 @@
     "connect-pg-simple": "^10.0.0",
     "duckdb-async": "^1.1.0",
     "express-session": "^1.18.1",
-    "mariadb": "^3.4.0",
+    "mariadb": "^3.4.5",
     "pg": "^8.15.6",
     "reflect-metadata": "^0.1.13",
     "rxjs": "^7.8.1",

back-end/api/package.json

@@ -14,7 +14,7 @@ import { SqlScriptsLoaderService } from 'src/sql-scripts/sql-scripts-loader.serv
 
 @Injectable()
 export class MigrationsService {
-  private readonly SUPPORTED_DB_VERSION: string = "0.0.2"; // TODO. Should come from a versioning file. 
+  private readonly SUPPORTED_DB_VERSION: string = "0.0.3"; // TODO. Should come from a versioning file. 
   private readonly logger = new Logger(MigrationsService.name);
 
   constructor(
@@ -84,6 +84,7 @@ export class MigrationsService {
     await this.sqlScriptsService.addEntryDatetimeTriggerToDB();
     await this.sqlScriptsService.addLogsTriggersToDB();
     await this.sqlScriptsService.addQCTestsFunctionsToDB();
+     await this.sqlScriptsService.addDataAvailabilityFunctionsToDB();
   }
 
   private async seedFirstUser() {

back-end/api/src/migrations/migrations.service.ts

@@ -34,17 +34,16 @@ export interface V4ElementModel {
 export interface V4StationModel {
     stationId: string;
     stationName: string;
-    wmoid: string;
-    icaoid: string;
-    wsi: string;
+    wmoid: string | null;
+    icaoid: string | null;
+    wsi: string | null;
     longitude: number;
     latitude: number;
     elevation: string;
-    qualifier: string;
+    qualifier: string | null;
     stationOperational: boolean;
-    openingDatetime: string;
-    closingDatetime: string;
-    authority: string;
+    openingDatetime: string | null;
+    closingDatetime: string | null; 
 }
 
 @Injectable()
@@ -120,7 +119,8 @@ export class ClimsoftV4WebSyncSetUpService {
                 password: AppConfig.v4DbCredentials.password,
                 database: AppConfig.v4DbCredentials.databaseName,
                 port: AppConfig.v4DbCredentials.port,
-                dateStrings: true,
+                dateStrings: true, 
+                charset: 'utf8mb4',
             });
 
             // Clear any previous conflicts
@@ -298,7 +298,6 @@ export class ClimsoftV4WebSyncSetUpService {
         try {
             conn = await this.v4DBPool.getConnection();
             const rows: V4StationModel[] = await conn.query("SELECT stationId as stationId, stationName as stationName, wmoid as wmoid, icaoid as icaoid, wsi as wsi, longitude as longitude, latitude as latitude, elevation as elevation, qualifier as qualifier, stationOperational as stationOperational, openingDatetime as openingDatetime, closingDatetime as closingDatetime, authority as authority FROM station");
-            //console.log('station rows: ', rows[0]);
             return rows;
         } catch (error) {
             console.error('Setting up V4 stations failed: ', error);
@@ -330,6 +329,23 @@ export class ClimsoftV4WebSyncSetUpService {
                 v4Station.stationName = `${v4Station.stationName}_${(i + 1)}`;
             }
 
+            // Some climsoft version 4 installations have the below columns storing null bytes instead of nulls
+            // So ignore such null bytes
+            //----------------------------------------------
+            if(v4Station.wmoid !== null && v4Station.wmoid.startsWith('\x00')){
+                v4Station.wmoid = null;
+            }
+
+             if(v4Station.wsi !== null && v4Station.wsi.startsWith('\x00')){
+                v4Station.wsi = null;
+            }
+
+             if(v4Station.icaoid !== null && v4Station.icaoid.startsWith('\x00')){
+                v4Station.icaoid = null;
+            }
+
+            //----------------------------------------------
+
             // Make sure the wmo id is unique. V5 doesn't accept duplicates like v4 model
             if (v4Station.wmoid !== null && v5Dtos.find(item => item.wmoId === v4Station.wmoid)) {
                 v4Station.wmoid = `${v4Station.wmoid}_${(i + 1)}`;
@@ -370,7 +386,18 @@ export class ClimsoftV4WebSyncSetUpService {
             v5Dtos.push(dto);
         }
 
-        await this.stationsService.bulkPut(v5Dtos, userId);
+        // Save on batches of a thousand just incase there are many stations
+        const batchSize: number = 1000;
+        for (let i = 0; i < v5Dtos.length; i += batchSize) {
+            const batch = v5Dtos.slice(i, i + batchSize);
+            try {
+                await this.stationsService.bulkPut(batch, userId);
+            } catch (error) {
+                //console.error(error);
+                console.log('Error saving stations: ', batch);
+                return false;
+            }
+        }
 
         // Important to do this just incase observations were not being saved to v4 database due to lack of stations or changes in v4 configuration
         this.setupV4StationsChecking();

back-end/api/src/observation/services/climsoft-v4-web-sync-set-up.service.ts

@@ -1,4 +1,4 @@
-import { BadRequestException, Injectable } from '@nestjs/common';
+import { BadRequestException, Injectable, Logger } from '@nestjs/common';
 import { CreateObservationDto } from '../dtos/create-observation.dto';
 import { ObservationsService } from './observations.service';
 import { SourceTemplatesService } from 'src/metadata/source-templates/services/source-templates.service';
@@ -17,6 +17,7 @@ import { SourceTypeEnum } from 'src/metadata/source-templates/enums/source-type.
 
 @Injectable()
 export class ObservationImportService {
+    private readonly logger = new Logger(ObservationImportService.name);
     // Enforce these fields to always match CreateObservationDto properties naming. Important to ensure objects returned by duckdb matches the dto structure. 
     private readonly STATION_ID_PROPERTY_NAME: keyof CreateObservationDto = "stationId";
     private readonly ELEMENT_ID_PROPERTY_NAME: keyof CreateObservationDto = "elementId";
@@ -59,7 +60,7 @@ export class ObservationImportService {
             }
 
         } catch (error) {
-            console.error("File Import Failed: " , error)
+            console.error("File Import Failed: ", error)
             throw new BadRequestException("Error: File Import Failed: " + error);
         } finally {
             this.fileIOService.deleteFile(tmpFilePathName);
@@ -100,29 +101,31 @@ export class ObservationImportService {
 
         //console.log("alterSQLs: ", alterSQLs);
 
-        let startTime = new Date().getTime();
         // Execute the duckdb DDL SQL commands
+        let startTime = new Date().getTime();
         await this.fileIOService.duckDb.exec(alterSQLs);
-        console.log("DuckDB alters took: ", new Date().getTime() - startTime);
+        this.logger.log(`DuckDB alters took ${new Date().getTime() - startTime} milliseconds`);
 
         if (sourceDef.scaleValues) {
             startTime = new Date().getTime();
             // Scale values if indicated, execute the scale values SQL
             await this.fileIOService.duckDb.exec(await this.getScaleValueSQL(tmpObsTableName));
-            console.log("DuckDB scaling took: ", new Date().getTime() - startTime);
+            this.logger.log(`DuckDB scaling took ${new Date().getTime() - startTime} milliseconds`);
         }
 
-        startTime = new Date().getTime();
         // Get the rows of the columns that match the dto properties
+        startTime = new Date().getTime();
         const rows = await this.fileIOService.duckDb.all(`SELECT ${this.STATION_ID_PROPERTY_NAME}, ${this.ELEMENT_ID_PROPERTY_NAME}, ${this.SOURCE_ID_PROPERTY_NAME}, ${this.level}, ${this.DATE_TIME_PROPERTY_NAME}, ${this.INTERVAL_PROPERTY_NAME}, ${this.VALUE_PROPERTY_NAME}, ${this.FLAG_PROPERTY_NAME}, ${this.COMMENT_PROPERTY_NAME} FROM ${tmpObsTableName};`);
-
-        console.log("DuckDB fetch rows took: ", new Date().getTime() - startTime);
+        this.logger.log(`DuckDB fetch rows took ${new Date().getTime() - startTime} milliseconds`);
 
         // Delete the table 
+        startTime = new Date().getTime();
         await this.fileIOService.duckDb.run(`DROP TABLE ${tmpObsTableName};`);
+        this.logger.log(`DuckDB drop table took ${new Date().getTime() - startTime} milliseconds`);
 
         // Save the rows into the database
-        await this.observationsService.bulkPut(rows as CreateObservationDto[], userId);
+        // TODO. Note, no need await. All current active ingestion processes will be tagged and show on the ingestion monitoring page
+        this.observationsService.bulkPut(rows as CreateObservationDto[], userId);
     }
 
     private getAlterStationColumnSQL(source: CreateImportTabularSourceDTO, tableName: string, stationId?: string): string {

back-end/api/src/observation/services/observation-import.service.ts

@@ -254,6 +254,7 @@ export class ObservationsService {
     public async bulkPut(createObservationDtos: CreateObservationDto[], userId: number, qcStatus = QCStatusEnum.NONE, ignoreV4Saving: boolean = false): Promise<void> {
         let startTime = new Date().getTime();
 
+        // Transform dtos to entities
         const obsEntities: ObservationEntity[] = [];
         for (const dto of createObservationDtos) {
             const entity: ObservationEntity = this.observationRepo.create({
@@ -274,18 +275,18 @@ export class ObservationsService {
 
             obsEntities.push(entity);
         }
-
-
         this.logger.log(`DTO transformation took: ${(new Date().getTime() - startTime)} milliseconds`);
 
+        // Save in batches of 1000 to minimise excess payload errors when saving to postgres
+        this.logger.log(`Saving ${obsEntities.length} entities from user - ${userId}`);
         startTime = new Date().getTime();
-
         const batchSize = 1000; // batch size of 1000 seems to be safer (incase there are comments) and faster.
         for (let i = 0; i < obsEntities.length; i += batchSize) {
             const batch = obsEntities.slice(i, i + batchSize);
             await this.insertOrUpdateObsValues(this.observationRepo, batch);
+            this.logger.log(`${batch.length} entities from user - ${userId} successfully saved!`);
         }
-        this.logger.log(`Saving entities took: ${(new Date().getTime() - startTime)} milliseconds`);
+        this.logger.log(`Saving entities from user - ${userId}, took: ${(new Date().getTime() - startTime)} milliseconds`);
 
         if (!ignoreV4Saving) {
             // Initiate saving to version 4 database as well
@@ -543,7 +544,7 @@ export class ObservationsService {
             filter.toDate ?? null                                     // p_to_date timestamptz
         ];
 
-        const sql = `SELECT *  FROM func_data_availaibility_details($1, $2, $3, $4, $5, $6)`;
+        const sql = `SELECT * FROM func_data_availaibility_details($1, $2, $3, $4, $5, $6)`;
 
         const rows = await this.dataSource.query(sql, params);
 

back-end/api/src/observation/services/observations.service.ts

@@ -1,4 +1,5 @@
 import { Injectable, StreamableFile } from '@nestjs/common';
+import os from "os";
 import * as fs from 'node:fs';
 import * as path from 'node:path';
 import { Database } from "duckdb-async";

back-end/api/src/shared/services/file-io.service.ts

@@ -61,9 +61,7 @@ BEGIN
 										o.station_id, 
 										o.element_id, 
 										o.level, 
-										o."interval", 
-										MIN(o.date_time) AS from_date, 
-										MAX(o.date_time) AS to_date
+										o."interval"
 									FROM observations o
 									WHERE %s
 									GROUP BY o.station_id, o.element_id, o.level, o."interval"
@@ -74,7 +72,7 @@ BEGIN
 										og.element_id, 
 										og.level, og.
 										"interval", 
-										generate_series(og.from_date, og.to_date, (og."interval" || ' minutes')::interval) AS date_time
+										generate_series($5, $6, (og."interval" || ' minutes')::interval) AS date_time
 									FROM observation_groups og
 					),
 					infilled_data AS (

…ilability/data-availiability-details.sql …/data-availiability-details-function.sqlback-end/api/src/sql-scripts/data-availability/data-availiability-details.sql renamed to back-end/api/src/sql-scripts/data-availability/data-availiability-details-function.sql back-end/api/src/sql-scripts/data-availability/data-availiability-details.sql renamed to back-end/api/src/sql-scripts/data-availability/data-availiability-details-function.sql

@@ -30,7 +30,7 @@ export class SqlScriptsLoaderService {
             await this.dataSource.query(sql);
             this.logger.log('default entry date time triggers added');
         } catch (error) {
-            this.logger.error(`Developer error in adding entry date time triggers: ${error}`);
+            this.logger.error(`Developer error in adding entry date time triggers`);
             throw new Error(error);
         }
     }
@@ -75,11 +75,29 @@ export class SqlScriptsLoaderService {
             await this.dataSource.query(sql);
             this.logger.log('qc tests functions added');
         } catch (error) {
-            this.logger.error(`Developer error in adding qc tests functions: ${error}`);
+            this.logger.error(`Developer error in adding qc tests functions`);
             throw new Error(error);
         }
     }
 
+     /**
+     * Used by the migrations service
+     */
+    public async addDataAvailabilityFunctionsToDB() {
+        try {
+            // Get the script directory from absolute path of this service file
+            // For windows platform, replace the backslashes with forward slashes.
+            const scriptsDirPath: string = this.getScriptsDirectoryPath().replaceAll("\\", "\/");
+            const entryDatetimeScriptsDirPath: string = `${scriptsDirPath}/data-availability/data-availiability-details-function.sql`
+            const sql: string = await this.fileIOService.readFile(entryDatetimeScriptsDirPath, 'utf8');
+            //console.log('ENTRY DATE TIME SQL:', sql);
+            await this.dataSource.query(sql);
+            this.logger.log('data availability functions added');
+        } catch (error) {
+            this.logger.error(`Developer error in adding data availability functions`);
+            throw new Error(error);
+        }
+    }