bundle labels

cmd/import-labels/main.go

···
+package main
+
+import (
+	"bufio"
+	"flag"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/klauspost/compress/zstd"
+	"gopkg.in/yaml.v3"
+)
+
+type Config struct {
+	PLC struct {
+		BundleDir string `yaml:"bundle_dir"`
+	} `yaml:"plc"`
+}
+
+var CONFIG_FILE = "config.yaml"
+
+// ---------------------
+
+func main() {
+	// Define a new flag for changing the directory
+	workDir := flag.String("C", ".", "Change to this directory before running (for finding config.yaml)")
+	flag.Usage = func() { // Custom usage message
+		fmt.Fprintf(os.Stderr, "Usage: ... | %s [-C /path/to/dir]\n", os.Args[0])
+		fmt.Fprintln(os.Stderr, "Reads sorted CSV from stdin and writes compressed bundle files.")
+		flag.PrintDefaults()
+	}
+	flag.Parse() // Parse all defined flags
+
+	// Change directory if the flag was used
+	if *workDir != "." {
+		fmt.Printf("Changing working directory to %s...\n", *workDir)
+		if err := os.Chdir(*workDir); err != nil {
+			fmt.Fprintf(os.Stderr, "Error changing directory to %s: %v\n", *workDir, err)
+			os.Exit(1)
+		}
+	}
+
+	// --- REMOVED UNUSED CODE ---
+	// The csvFilePath variable and NArg check were removed
+	// as the script now reads from stdin.
+	// ---------------------------
+
+	fmt.Println("========================================")
+	fmt.Println("PLC Operation Labels Import (Go STDIN)")
+	fmt.Println("========================================")
+
+	// 1. Read config (will now read from the new CWD)
+	fmt.Printf("Loading config from %s...\n", CONFIG_FILE)
+	configData, err := os.ReadFile(CONFIG_FILE)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error reading config file: %v\n", err)
+		os.Exit(1)
+	}
+
+	var config Config
+	if err := yaml.Unmarshal(configData, &config); err != nil {
+		fmt.Fprintf(os.Stderr, "Error parsing config.yaml: %v\n", err)
+		os.Exit(1)
+	}
+
+	if config.PLC.BundleDir == "" {
+		fmt.Fprintln(os.Stderr, "Error: Could not parse plc.bundle_dir from config.yaml")
+		os.Exit(1)
+	}
+
+	finalLabelsDir := filepath.Join(config.PLC.BundleDir, "labels")
+	if err := os.MkdirAll(finalLabelsDir, 0755); err != nil {
+		fmt.Fprintf(os.Stderr, "Error creating output directory: %v\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Printf("Output Dir:     %s\n", finalLabelsDir)
+	fmt.Println("Waiting for sorted data from stdin...")
+
+	// 2. Process sorted data from stdin
+	// This script *requires* the input to be sorted by bundle number.
+
+	var currentWriter *zstd.Encoder
+	var currentFile *os.File
+	var lastBundleKey string = ""
+
+	lineCount := 0
+	startTime := time.Now()
+
+	scanner := bufio.NewScanner(os.Stdin)
+	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+
+	for scanner.Scan() {
+		line := scanner.Text()
+		lineCount++
+
+		parts := strings.SplitN(line, ",", 2)
+		if len(parts) < 1 {
+			continue // Skip empty/bad lines
+		}
+
+		bundleNumStr := parts[0]
+		bundleKey := fmt.Sprintf("%06s", bundleNumStr) // Pad with zeros
+
+		// If the bundle key is new, close the old writer and open a new one.
+		if bundleKey != lastBundleKey {
+			// Close the previous writer/file
+			if currentWriter != nil {
+				if err := currentWriter.Close(); err != nil {
+					fmt.Fprintf(os.Stderr, "Error closing writer for %s: %v\n", lastBundleKey, err)
+				}
+				currentFile.Close()
+			}
+
+			// Start the new one
+			fmt.Printf("  -> Writing bundle %s\n", bundleKey)
+			outPath := filepath.Join(finalLabelsDir, fmt.Sprintf("%s.csv.zst", bundleKey))
+
+			file, err := os.Create(outPath)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Error creating file %s: %v\n", outPath, err)
+				os.Exit(1)
+			}
+			currentFile = file
+
+			writer, err := zstd.NewWriter(file)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Error creating zstd writer: %v\n", err)
+				os.Exit(1)
+			}
+			currentWriter = writer
+			lastBundleKey = bundleKey
+		}
+
+		// Write the line to the currently active writer
+		if _, err := currentWriter.Write([]byte(line + "\n")); err != nil {
+			fmt.Fprintf(os.Stderr, "Error writing line: %v\n", err)
+		}
+
+		// Progress update
+		if lineCount%100000 == 0 {
+			elapsed := time.Since(startTime).Seconds()
+			rate := float64(lineCount) / elapsed
+			fmt.Printf("  ... processed %,d lines (%.0f lines/sec)\n", lineCount, rate)
+		}
+	}
+
+	// 3. Close the very last writer
+	if currentWriter != nil {
+		if err := currentWriter.Close(); err != nil {
+			fmt.Fprintf(os.Stderr, "Error closing final writer: %v\n", err)
+		}
+		currentFile.Close()
+	}
+
+	if err := scanner.Err(); err != nil {
+		fmt.Fprintf(os.Stderr, "Error reading stdin: %v\n", err)
+	}
+
+	totalTime := time.Since(startTime)
+	fmt.Println("\n========================================")
+	fmt.Println("Import Summary")
+	fmt.Println("========================================")
+	fmt.Printf("✓ Import completed in %v\n", totalTime)
+	fmt.Printf("Total lines processed: %,d\n", lineCount)
+}

go.mod

···
 	gopkg.in/yaml.v3 v3.0.1
 )
 
-require github.com/klauspost/compress v1.18.1 // indirect
+require github.com/klauspost/compress v1.18.1
 
 require (
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d

internal/api/handlers.go

···
 	})
 }
 
+func (s *Server) handleGetBundleLabels(w http.ResponseWriter, r *http.Request) {
+	resp := newResponse(w)
+
+	bundleNum, err := getBundleNumber(r)
+	if err != nil {
+		resp.error("invalid bundle number", http.StatusBadRequest)
+		return
+	}
+
+	labels, err := s.bundleManager.GetBundleLabels(r.Context(), bundleNum)
+	if err != nil {
+		resp.error(err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	resp.json(map[string]interface{}{
+		"bundle": bundleNum,
+		"count":  len(labels),
+		"labels": labels,
+	})
+}
+
 // ===== UTILITY FUNCTIONS =====
 
 func normalizeEndpoint(endpoint string) string {

internal/api/server.go

···
 	api.HandleFunc("/plc/bundles/{number}", s.handleGetPLCBundle).Methods("GET")
 	api.HandleFunc("/plc/bundles/{number}/dids", s.handleGetPLCBundleDIDs).Methods("GET")
 	api.HandleFunc("/plc/bundles/{number}/download", s.handleDownloadPLCBundle).Methods("GET")
+	api.HandleFunc("/plc/bundles/{number}/labels", s.handleGetBundleLabels).Methods("GET")
 
 	// PLC history/metrics
 	api.HandleFunc("/plc/history", s.handleGetPLCHistory).Methods("GET")

internal/plc/manager.go

···
 
 import (
 	"context"
+	"encoding/csv"
 	"fmt"
 	"io"
+	"os"
+	"path/filepath"
 	"sort"
+	"strconv"
+	"strings"
 	"time"
 
 	"github.com/atscan/atscand/internal/log"
 	"github.com/atscan/atscand/internal/storage"
+	"github.com/klauspost/compress/zstd"
 	plcbundle "tangled.org/atscan.net/plcbundle"
 )
 
···
 
 	return history, nil
 }
+
+// GetBundleLabels reads labels from a compressed CSV file for a specific bundle
+func (bm *BundleManager) GetBundleLabels(ctx context.Context, bundleNum int) ([]*PLCOpLabel, error) {
+	// Define the path to the labels file
+	labelsDir := filepath.Join(bm.bundleDir, "labels")
+	labelsFile := filepath.Join(labelsDir, fmt.Sprintf("%06d.csv.zst", bundleNum))
+
+	// Check if file exists
+	if _, err := os.Stat(labelsFile); os.IsNotExist(err) {
+		log.Verbose("No labels file found for bundle %d at %s", bundleNum, labelsFile)
+		// Return empty, not an error
+		return []*PLCOpLabel{}, nil
+	}
+
+	// Open the Zstd-compressed file
+	file, err := os.Open(labelsFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open labels file: %w", err)
+	}
+	defer file.Close()
+
+	// Create a Zstd reader
+	zstdReader, err := zstd.NewReader(file)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create zstd reader: %w", err)
+	}
+	defer zstdReader.Close()
+
+	// Create a CSV reader
+	csvReader := csv.NewReader(zstdReader)
+	// We skipped the header, so no header read needed
+	// Set FieldsPerRecord to 7 for validation
+	//csvReader.FieldsPerRecord = 7
+
+	var labels []*PLCOpLabel
+
+	// Read all records
+	for {
+		// Check for context cancellation
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+
+		record, err := csvReader.Read()
+		if err == io.EOF {
+			break // End of file
+		}
+		if err != nil {
+			log.Error("Error reading CSV record in %s: %v", labelsFile, err)
+			continue // Skip bad line
+		}
+
+		// Parse the CSV record (which is []string)
+		label, err := parseLabelRecord(record)
+		if err != nil {
+			log.Error("Error parsing CSV data for bundle %d: %v", bundleNum, err)
+			continue // Skip bad data
+		}
+
+		labels = append(labels, label)
+	}
+
+	return labels, nil
+}
+
+// parseLabelRecord converts a new format CSV record into a PLCOpLabel struct
+func parseLabelRecord(record []string) (*PLCOpLabel, error) {
+	// New format: 0:bundle, 1:position, 2:cid(short), 3:size, 4:confidence, 5:labels
+	if len(record) != 6 {
+		err := fmt.Errorf("invalid record length: expected 6, got %d", len(record))
+		// --- ADDED LOG ---
+		log.Warn("Skipping malformed CSV line: %v (data: %s)", err, strings.Join(record, ","))
+		// ---
+		return nil, err
+	}
+
+	// 0:bundle
+	bundle, err := strconv.Atoi(record[0])
+	if err != nil {
+		// --- ADDED LOG ---
+		log.Warn("Skipping malformed CSV line: 'bundle' column: %v (data: %s)", err, strings.Join(record, ","))
+		// ---
+		return nil, fmt.Errorf("parsing 'bundle': %w", err)
+	}
+
+	// 1:position
+	position, err := strconv.Atoi(record[1])
+	if err != nil {
+		// --- ADDED LOG ---
+		log.Warn("Skipping malformed CSV line: 'position' column: %v (data: %s)", err, strings.Join(record, ","))
+		// ---
+		return nil, fmt.Errorf("parsing 'position': %w", err)
+	}
+
+	// 2:cid(short)
+	shortCID := record[2]
+
+	// 3:size
+	size, err := strconv.Atoi(record[3])
+	if err != nil {
+		// --- ADDED LOG ---
+		log.Warn("Skipping malformed CSV line: 'size' column: %v (data: %s)", err, strings.Join(record, ","))
+		// ---
+		return nil, fmt.Errorf("parsing 'size': %w", err)
+	}
+
+	// 4:confidence
+	confidence, err := strconv.ParseFloat(record[4], 64)
+	if err != nil {
+		// --- ADDED LOG ---
+		log.Warn("Skipping malformed CSV line: 'confidence' column: %v (data: %s)", err, strings.Join(record, ","))
+		// ---
+		return nil, fmt.Errorf("parsing 'confidence': %w", err)
+	}
+
+	// 5:labels
+	detectors := strings.Split(record[5], ";")
+
+	label := &PLCOpLabel{
+		Bundle:     bundle,
+		Position:   position,
+		CID:        shortCID,
+		Size:       size,
+		Confidence: confidence,
+		Detectors:  detectors,
+	}
+
+	return label, nil
+}

internal/plc/types.go

···
 	Type     string
 	Endpoint string
 }
+
+// PLCOpLabel holds metadata from the label CSV file
+type PLCOpLabel struct {
+	Bundle     int      `json:"bundle"`
+	Position   int      `json:"position"`
+	CID        string   `json:"cid"`
+	Size       int      `json:"size"`
+	Confidence float64  `json:"confidence"`
+	Detectors  []string `json:"detectors"`
+}

utils/import-labels.js

···
+import { file, write } from "bun";
+import { join } from "path";
+import { mkdir } from "fs/promises";
+import { init, compress } from "@bokuweb/zstd-wasm";
+
+// --- Configuration ---
+const CSV_FILE = process.argv[2];
+const CONFIG_FILE = "config.yaml";
+const COMPRESSION_LEVEL = 5; // zstd level 1-22 (5 is a good balance)
+// ---------------------
+
+if (!CSV_FILE) {
+  console.error("Usage: bun run utils/import-labels.js <path-to-csv-file>");
+  process.exit(1);
+}
+
+console.log("========================================");
+console.log("PLC Operation Labels Import (Bun + WASM)");
+console.log("========================================");
+
+// 1. Read and parse config
+console.log(`Loading config from ${CONFIG_FILE}...`);
+const configFile = await file(CONFIG_FILE).text();
+const config = Bun.YAML.parse(configFile);
+const bundleDir = config?.plc?.bundle_dir;
+
+if (!bundleDir) {
+  console.error("Error: Could not parse plc.bundle_dir from config.yaml");
+  process.exit(1);
+}
+
+const FINAL_LABELS_DIR = join(bundleDir, "labels");
+await mkdir(FINAL_LABELS_DIR, { recursive: true });
+
+console.log(`CSV File:       ${CSV_FILE}`);
+console.log(`Output Dir:     ${FINAL_LABELS_DIR}`);
+console.log("");
+
+// 2. Initialize Zstd WASM module
+await init();
+
+// --- Pass 1: Read entire file into memory and group by bundle ---
+console.log("Pass 1/2: Reading and grouping all lines by bundle...");
+console.warn("This will use a large amount of RAM!");
+
+const startTime = Date.now();
+const bundles = new Map(); // Map<string, string[]>
+let lineCount = 0;
+
+const inputFile = file(CSV_FILE);
+const fileStream = inputFile.stream();
+const decoder = new TextDecoder();
+let remainder = "";
+
+for await (const chunk of fileStream) {
+  const text = remainder + decoder.decode(chunk);
+  const lines = text.split("\n");
+  remainder = lines.pop() || "";
+
+  for (const line of lines) {
+    if (line === "") continue;
+    lineCount++;
+
+    if (lineCount === 1 && line.startsWith("bundle,")) {
+      continue; // Skip header
+    }
+
+    const firstCommaIndex = line.indexOf(",");
+    if (firstCommaIndex === -1) {
+      console.warn(`Skipping malformed line: ${line}`);
+      continue;
+    }
+    const bundleNumStr = line.substring(0, firstCommaIndex);
+    const bundleKey = bundleNumStr.padStart(6, "0");
+
+    // Add line to the correct bundle's array
+    if (!bundles.has(bundleKey)) {
+      bundles.set(bundleKey, []);
+    }
+    bundles.get(bundleKey).push(line);
+  }
+}
+// Note: We ignore any final `remainder` as it's likely an empty line
+
+console.log(`Finished reading ${lineCount.toLocaleString()} lines.`);
+console.log(`Found ${bundles.size} unique bundles.`);
+
+// --- Pass 2: Compress and write each bundle ---
+console.log("\nPass 2/2: Compressing and writing bundle files...");
+let i = 0;
+for (const [bundleKey, lines] of bundles.entries()) {
+  i++;
+  console.log(`  (${i}/${bundles.size}) Compressing bundle ${bundleKey}...`);
+
+  // Join all lines for this bundle into one big string
+  const content = lines.join("\n");
+  
+  // Compress the string
+  const compressedData = compress(Buffer.from(content), COMPRESSION_LEVEL);
+
+  // Write the compressed data to the file
+  const outPath = join(FINAL_LABELS_DIR, `${bundleKey}.csv.zst`);
+  await write(outPath, compressedData);
+}
+
+// 3. Clean up
+const totalTime = (Date.now() - startTime) / 1000;
+console.log("\n========================================");
+console.log("Import Summary");
+console.log("========================================");
+console.log(`✓ Import completed in ${totalTime.toFixed(2)} seconds.`);
+console.log(`Total lines processed: ${lineCount.toLocaleString()}`);
+console.log(`Label files are stored in: ${FINAL_LABELS_DIR}`);

utils/import-labels.sh

···
+#!/bin/bash
+# import-labels-v4-sorted-pipe.sh
+
+set -e
+
+if [ $# -lt 1 ]; then
+    echo "Usage: ./utils/import-labels-v4-sorted-pipe.sh <csv-file>"
+    exit 1
+fi
+
+CSV_FILE="$1"
+CONFIG_FILE="config.yaml"
+
+[ ! -f "$CSV_FILE" ] && echo "Error: CSV file not found" && exit 1
+[ ! -f "$CONFIG_FILE" ] && echo "Error: config.yaml not found" && exit 1
+
+# Extract bundle directory path
+BUNDLE_DIR=$(grep -A 5 "^plc:" "$CONFIG_FILE" | grep "bundle_dir:" | sed 's/.*bundle_dir: *"//' | sed 's/".*//' | head -1)
+
+[ -z "$BUNDLE_DIR" ] && echo "Error: Could not parse plc.bundle_dir from config.yaml" && exit 1
+
+FINAL_LABELS_DIR="$BUNDLE_DIR/labels"
+
+echo "========================================"
+echo "PLC Operation Labels Import (Sorted Pipe)"
+echo "========================================"
+echo "CSV File:       $CSV_FILE"
+echo "Output Dir:     $FINAL_LABELS_DIR"
+echo ""
+
+# Ensure the final directory exists
+mkdir -p "$FINAL_LABELS_DIR"
+
+echo "Streaming, sorting, and compressing on the fly..."
+echo "This will take time. `pv` will show progress of the TAIL command."
+echo "The `sort` command will run after `pv` is complete."
+echo ""
+
+# This is the single-pass pipeline
+tail -n +2 "$CSV_FILE" | \
+    pv -l -s $(tail -n +2 "$CSV_FILE" | wc -l) | \
+    sort -t, -k1,1n | \
+    awk -F',' -v final_dir="$FINAL_LABELS_DIR" '
+    # This awk script EXPECTS input sorted by bundle number (col 1)
+    BEGIN {
+        # last_bundle_num tracks the bundle we are currently writing
+        last_bundle_num = -1
+        # cmd holds the current zstd pipe command
+        cmd = ""
+    }
+    {
+        current_bundle_num = $1
+        
+        # Check if the bundle number has changed
+        if (current_bundle_num != last_bundle_num) {
+            
+            # If it changed, and we have an old pipe open, close it
+            if (last_bundle_num != -1) {
+                close(cmd)
+            }
+            
+            # Create the new pipe command, writing to the final .zst file
+            outfile = sprintf("%s/%06d.csv.zst", final_dir, current_bundle_num)
+            cmd = "zstd -T0 -o " outfile
+            
+            # Update the tracker
+            last_bundle_num = current_bundle_num
+            
+            # Print progress to stderr
+            printf "  -> Writing bundle %06d\n", current_bundle_num > "/dev/stderr"
+        }
+        
+        # Print the current line ($0) to the open pipe
+        # The first time this runs for a bundle, it opens the pipe
+        # Subsequent times, it writes to the already-open pipe
+        print $0 | cmd
+    }
+    # END block: close the very last pipe
+    END {
+        if (last_bundle_num != -1) {
+            close(cmd)
+        }
+        printf "  Finished. Total lines: %d\n", NR > "/dev/stderr"
+    }'
+
+echo ""
+echo "========================================"
+echo "Import Summary"
+echo "========================================"
+echo "✓ Import completed successfully!"
+echo "Label files are stored in: $FINAL_LABELS_DIR"