Add TileJoin into the cli-vector

Author: Wentao-Kuang

package-lock.json

@@ -25,7 +25,7 @@
     "@linzjs/style": "^5.4.0",
     "@types/aws-lambda": "^8.10.75",
     "@types/geojson": "^7946.0.7",
-    "@types/node": "^20.14.8",
+    "@types/node": "^22.15.24",
     "@types/sinon": "^17.0.2",
     "conventional-github-releaser": "^3.1.5",
     "cross-env": "^7.0.3",

package.json

@@ -43,10 +43,9 @@
     "@types/mapbox__geojson-area": "^0.2.6",
     "@types/p-limit": "^2.1.0",
     "@types/polylabel": "^1.1.3",
+    "@types/tar-stream": "^2.2.2",
     "cmd-ts": "^0.12.1",
-    "stac-ts": "^1.0.0",
-    "@types/better-sqlite3": "^7.6.9",
-    "@types/tar-stream": "^2.2.2"
+    "stac-ts": "^1.0.0"
   },
   "publishConfig": {
     "access": "public"
@@ -61,9 +60,9 @@
     "geojson": "^0.5.0",
     "p-limit": "^6.2.0",
     "polylabel": "^2.0.1",
+    "sqlite": "^5.1.1",
+    "tar-stream": "^2.2.0",
     "zod": "^3.24.4",
-    "zod-geojson": "^0.0.3",
-    "better-sqlite3": "^9.4.3",
-    "tar-stream": "^2.2.0"
+    "zod-geojson": "^0.0.3"
   }
 }

packages/cli-vector/package.json

@@ -65,7 +65,7 @@ export const CreateCommand = command({
 
     // if applicable, combine all mbtiles files into a single mbtiles file
     if (args.join === true) {
-      const mbtileFiles = items.map((item) => item.tmpPaths.mbtiles.pathname);
+      const mbtileFiles = items.map((item) => item.tmpPaths.mbtiles);
       logger.info({ joining: mbtileFiles.length }, 'JoinMbtiles:Start');
 
       const joinedFile = new URL(`joined.mbtiles`, TmpPath);
@@ -130,7 +130,7 @@ async function downloadSourceFile(
 
   const layer = options.layer;
 
-  logger.info({ source: layer.source }, 'DownloadSourceFile: Start');
+  logger.info({ source: layer.source, id: layer.id, name: layer.name }, 'DownloadSourceFile: Start');
 
   if (!(await fsa.exists(tmpPaths.source.path))) {
     // TODO: We don't acturally need to head file from lds-cache here.
@@ -147,7 +147,7 @@ async function downloadSourceFile(
     });
   }
 
-  logger.info({ destination: tmpPaths.source.path }, 'DownloadSourceFile: End');
+  logger.info({ destination: tmpPaths.source.path, id: layer.id, name: layer.name }, 'DownloadSourceFile: End');
 }
 
 /**
@@ -178,36 +178,42 @@ async function createMbtilesFile(
   /**
    * Convert the source file into an ndjson
    */
-  logger.info({ source: tmpPaths.source.path }, '[1/5] Convert source file to ndjson: Start');
+  logger.info({ source: tmpPaths.source.path, dataset: layer.name }, '[1/5] Convert source file to ndjson: Start');
   if (!(await fsa.exists(tmpPaths.ndjson))) {
     await ogr2ogrNDJson(tmpPaths.source.path, tmpPaths.ndjson, logger);
   }
-  logger.info({ destination: tmpPaths.ndjson }, '[1/5] Convert source file to ndjson: End');
+  logger.info({ destination: tmpPaths.ndjson, dataset: layer.name }, '[1/5] Convert source file to ndjson: End');
 
   /**
    * Parse the ndjson file and apply the generalization options
    */
-  logger.info({ source: tmpPaths.ndjson }, '[2/5] Generalise ndjson features: Start');
+  logger.info({ source: tmpPaths.ndjson, dataset: layer.name }, '[2/5] Generalise ndjson features: Start');
   let metrics: Metrics | null = null;
   if (!(await fsa.exists(tmpPaths.genNdjson))) {
     metrics = await generalize(tmpPaths.ndjson, tmpPaths.genNdjson, options, logger);
     if (metrics.output === 0) throw new Error(`Failed to generalize ndjson file ${tmpPaths.ndjson.href}`);
   }
-  logger.info({ destination: tmpPaths.genNdjson }, '[2/5] Generalise ndjson features: End');
+  logger.info({ destination: tmpPaths.genNdjson, dataset: layer.name }, '[2/5] Generalise ndjson features: End');
 
   /**
    * Transform the generalized ndjson file to an mbtiles file
    */
-  logger.info({ source: tmpPaths.genNdjson }, '[3/5] Transform generalised ndjson into mbtiles: Start');
+  logger.info(
+    { source: tmpPaths.genNdjson, dataset: layer.name },
+    '[3/5] Transform generalised ndjson into mbtiles: Start',
+  );
   if (!(await fsa.exists(tmpPaths.mbtiles))) {
     await tippecanoe(tmpPaths.genNdjson, tmpPaths.mbtiles, layer, logger);
   }
-  logger.info({ destination: tmpPaths.mbtiles }, '[3/5] Transform generalised ndjson into mbtiles: End');
+  logger.info(
+    { destination: tmpPaths.mbtiles, dataset: layer.name },
+    '[3/5] Transform generalised ndjson into mbtiles: End',
+  );
 
   /**
    * Copy the mbtiles file to the same directory as the Vector Stac Item file
    */
-  logger.info({ source: tmpPaths.mbtiles }, '[4/5] Copy mbtiles to stac location: Start');
+  logger.info({ source: tmpPaths.mbtiles, dataset: layer.name }, '[4/5] Copy mbtiles to stac location: Start');
   if (!(await fsa.exists(tmpPaths.mbtilesCopy))) {
     await fsa.write(tmpPaths.mbtilesCopy, fsa.readStream(tmpPaths.mbtiles));
 
@@ -216,12 +222,12 @@ async function createMbtilesFile(
       throw new Error(`Failed to write the mbtiles file to ${tmpPaths.mbtilesCopy.href}`);
     }
   }
-  logger.info({ destination: tmpPaths.mbtilesCopy }, '[4/5] Copy mbtiles to stac location: End');
+  logger.info({ destination: tmpPaths.mbtilesCopy, dataset: layer.name }, '[4/5] Copy mbtiles to stac location: End');
 
   /**
    * Update the Vector Stac Item file
    */
-  logger.info({ source: tmpPaths.origin }, '[5/5] Update stac: Start');
+  logger.info({ source: tmpPaths.origin, dataset: layer.name }, '[5/5] Update stac: Start');
 
   // Update 'cache' flag to 'true' now that the mbtiles file exists
   layer.cache!.exists = true;

packages/cli-vector/src/cli/cli.create.ts

@@ -56,12 +56,12 @@ export const ExtractCommand = command({
     const smallLayers = [];
     const largeLayers = [];
     let total = 0;
-    const allFiles: string[] = [];
+    const allFiles = [];
     const vectorStac = new VectorStac(logger);
     for (const schema of schemas) {
       for (const layer of schema.layers) {
         if (layer.cache == null) throw new Error(`Fail to prepare cache path for layer ${schema.name}:${layer.id}`);
-        allFiles.push(layer.cache.path.href);
+        allFiles.push({ path: layer.cache.path.href });
 
         // Skip if the layer is already processed in cache
         if (layer.cache.exists) {

packages/cli-vector/src/cli/cli.extract.ts

@@ -1,31 +1,35 @@
-import { Epsg, TileMatrixSets } from '@basemaps/geo';
-import { fsa, LogType, Url } from '@basemaps/shared';
+import { TileMatrixSets } from '@basemaps/geo';
+import { fsa, isArgo, LogType, Url, UrlArrayJsonFile } from '@basemaps/shared';
 import { CliId, CliInfo } from '@basemaps/shared/build/cli/info.js';
 import { getLogger, logArguments } from '@basemaps/shared/build/cli/log.js';
-import { command, option, string } from 'cmd-ts';
-import { basename } from 'path';
+import { command, option, optional, string } from 'cmd-ts';
+import { mkdir } from 'fs/promises';
+import path, { basename, dirname } from 'path';
 import { createGzip } from 'zlib';
 
 import { createStacFiles } from '../stac.js';
 import { toTarIndex } from '../transform/covt.js';
 import { toTarTiles } from '../transform/mbtiles.to.ttiles.js';
 import { tileJoin } from '../transform/tippecanoe.js';
 
-async function fromFile(path: URL): Promise<string[]> {
-  const toProcess = await fsa.readJson(path);
-  const paths: string[] = [];
-  if (!Array.isArray(toProcess)) throw new Error(`File ${path.href} is not an array`);
-  for (const task of toProcess) {
-    if (typeof task !== 'string') throw new Error(`File ${path.href} is not an array of strings`);
-    const sourceFile = new URL(task);
-    if (sourceFile.protocol === 'file:') {
-      paths.push(sourceFile.pathname);
+async function download(filePaths: URL[], outputPath: string, logger: LogType): Promise<URL[]> {
+  const paths: URL[] = [];
+  for (const file of filePaths) {
+    if (file.protocol === 'file:') {
+      paths.push(file);
     } else {
-      const fileName = basename(sourceFile.pathname);
-      if (fileName == null) throw new Error(`Unsupported source pathname ${sourceFile.pathname}`);
-      const localFile = `tmp/join/${fileName}`;
-      const stream = fsa.readStream(sourceFile);
-      await fsa.write(fsa.toUrl(localFile), stream);
+      const fileName = basename(file.pathname);
+      if (fileName == null) throw new Error(`Unsupported source pathname ${file.pathname}`);
+      const localFile = fsa.toUrl(`${outputPath}/downloads/${fileName}`);
+      if (await fsa.exists(localFile)) {
+        logger.info({ file: file, localFile, fileName }, 'Download:FileExists');
+      } else {
+        const stats = await fsa.head(file);
+        logger.info({ file: file, localFile, fileName, size: stats?.size }, 'Download:Start');
+        const stream = fsa.readStream(file);
+        await fsa.write(localFile, stream);
+        logger.info({ file: file, localFile, fileName }, 'Download:End');
+      }
       paths.push(localFile);
     }
   }
@@ -36,8 +40,8 @@ async function fromFile(path: URL): Promise<string[]> {
  * Upload output file into s3 bucket
  *
  */
-async function upload(file: URL, bucketPath: string, logger: LogType): Promise<URL> {
-  logger.info({ file: file }, 'Load:Start');
+async function upload(file: URL, bucketPath: URL, logger: LogType): Promise<URL> {
+  logger.info({ file: file.href }, 'Load:Start');
   let filename = basename(file.pathname);
   let stream = fsa.readStream(file);
 
@@ -47,9 +51,11 @@ async function upload(file: URL, bucketPath: string, logger: LogType): Promise<U
     stream = stream.pipe(createGzip());
   }
 
-  // Upload to s3
-  let path = fsa.toUrl(`${bucketPath}/${CliId}/${filename}`);
-  if (filename.endsWith('catalog.json')) path = fsa.toUrl(`${bucketPath}/${filename}`); // Upload catalog to root directory
+  // Upload to s3 or copy to local
+  let path = new URL(`topographic/${CliId}/${filename}`, bucketPath);
+  logger.info({ file: file, path: path }, 'Load:Path');
+  if (path.protocol === 'file:') await mkdir(dirname(path.pathname), { recursive: true });
+  if (filename.endsWith('catalog.json')) path = new URL(filename, bucketPath); // Upload catalog to root directory
   await fsa.write(path, stream);
   logger.info({ file: file, path: path }, 'Load:Finish');
   return path;
@@ -58,7 +64,7 @@ async function upload(file: URL, bucketPath: string, logger: LogType): Promise<U
 export const JoinArgs = {
   ...logArguments,
   fromFile: option({
-    type: Url,
+    type: UrlArrayJsonFile,
     long: 'from-file',
     description: 'Path to JSON file containing array of paths to mbtiles.',
   }),
@@ -80,13 +86,13 @@ export const JoinArgs = {
     type: string,
     long: 'title',
     description: 'Title for the output etl data in the STAC file',
-    defaultValue: () => 'Topographic',
+    defaultValue: () => 'Topographic V2',
     defaultValueIsSerializable: true,
   }),
   target: option({
-    type: string,
+    type: optional(Url),
     long: 'target',
-    description: 'Path of target location, could be local or s3',
+    description: 'Path of target location to upload the processed file, could be local or s3',
   }),
 };
 
@@ -97,32 +103,49 @@ export const JoinCommand = command({
   args: JoinArgs,
   async handler(args) {
     const logger = getLogger(this, args, 'cli-vector');
-    const filePaths = await fromFile(args.fromFile);
-
-    logger.info({ files: filePaths.length }, 'JoinMbtiles: Start');
-
-    const outputMbtiles = fsa.toUrl(`tmp/${args.filename}.mbtiles`).pathname;
+    const outputPath = path.resolve('tmp/join/');
+    await mkdir(outputPath, { recursive: true });
+    const tileMatrix = TileMatrixSets.find(args.tileMatrix);
+    if (tileMatrix == null) throw new Error(`Tile matrix ${args.tileMatrix} is not supported`);
+    const bucketPath = new URL(`vector/${tileMatrix.projection.code}/`, args.target ?? outputPath);
+    const filePaths = await download(args.fromFile, outputPath, logger);
 
+    const outputMbtiles = path.join(outputPath, `${args.filename}.mbtiles`);
+    logger.info({ files: filePaths.length, outputMbtiles }, 'JoinMbtiles: Start');
     await tileJoin(filePaths, outputMbtiles, logger);
+    logger.info({ files: filePaths.length, outputMbtiles }, 'JoinMbtiles: End');
 
-    const outputCotar = fsa.toUrl(`tmp/${args.filename}.covt`);
-
+    const outputCotar = path.join(outputPath, `${args.filename}.tar.co`);
+    logger.info({ mbtiles: outputMbtiles, outputCotar }, 'ToTartTiles: Start');
     await toTarTiles(outputMbtiles, outputCotar, logger);
+    logger.info({ mbtiles: outputMbtiles, outputCotar }, 'ToTartTiles: End');
 
-    const [outputIndex, outputTar] = await toTarIndex(outputCotar, 'tmp/', args.filename, logger);
+    const outputIndex = path.join(outputPath, `${args.filename}.tar.index`);
+    logger.info({ cotar: outputCotar, outputIndex }, 'toTarIndex: Start');
+    await toTarIndex(outputCotar, outputIndex, logger);
+    logger.info({ cotar: outputCotar, outputIndex }, 'toTarIndex: End');
 
-    const tileMatrix = TileMatrixSets.find(args.tileMatrix);
-    if (tileMatrix == null) throw new Error(`Tile matrix ${args.tileMatrix} is not supported`);
-    const stacFiles = await createStacFiles(filePaths, args.target, args.filename, tileMatrix, args.title, logger);
+    logger.info({ target: bucketPath, tileMatrix: tileMatrix.identifier }, 'CreateStac: Start');
+    const stacFiles = await createStacFiles(args.fromFile, bucketPath, args.filename, tileMatrix, args.title, logger);
+    logger.info({ cotar: outputCotar, outputIndex }, 'CreateStac: End');
 
     // Upload output to s3
-    const bucketPath = `${args.target}/vector/${Epsg.Google.code.toString()}/${args.filename}`;
-    await upload(fsa.toUrl(outputMbtiles), bucketPath, logger);
-    await upload(fsa.toUrl(outputTar), bucketPath, logger);
-    await upload(fsa.toUrl(outputIndex), bucketPath, logger);
-    // Upload stac Files
-    for (const file of stacFiles) {
-      await upload(file, bucketPath, logger);
+    logger.info({ target: bucketPath, tileMatrix: tileMatrix.identifier }, 'Upload: Start');
+    if (args.target) {
+      await upload(fsa.toUrl(outputMbtiles), bucketPath, logger);
+      await upload(fsa.toUrl(outputCotar), bucketPath, logger);
+      await upload(fsa.toUrl(outputIndex), bucketPath, logger);
+      // Upload stac Files
+      for (const file of stacFiles) {
+        await upload(file, bucketPath, logger);
+      }
+      logger.info({ target: bucketPath, tileMatrix: tileMatrix.identifier }, 'Upload: End');
+    }
+
+    // Write output target for argo tasks to create pull request
+    if (isArgo()) {
+      const target = new URL(`topographic/${CliId}/${args.filename}.tar.co`, bucketPath);
+      await fsa.write(fsa.toUrl('/tmp/target'), JSON.stringify([target]));
     }
   },
 });

packages/cli-vector/src/cli/cli.join.ts

@@ -23,6 +23,7 @@ export const zSimplify = z.object({
 export const zLayer = z.object({
   id: z.string(),
   name: z.string(),
+  version: z.number().optional(),
   source: z.string(),
   tags: zTags,
   attributes: zAttributes.optional(),