decode.go

//go:build ignore

// decode.go — 獨立單檔 .pga 解碼器（高速、多核心、低記憶體）。
//
// 這支檔案自帶完整格式解碼，不依賴專案任何套件，可單獨帶走使用。
// 因帶 //go:build ignore，不會被 `go build ./...` 納入（不與 main.go 衝突）。
//
// 用法:
//
//	go run decode.go [flags] <file.pga | 目錄>...
//	go build -o decode decode.go && ./decode backup/          # 編成 binary 更快
//
// Flags:
//
//	-csv          輸出 CSV: timestamp,station,pga（串流、低記憶體、逐檔有序）
//	-workers N    解碼平行度（預設 = CPU 核心數；僅驗證/測速模式）
//	-q            安靜：只印總結
//
// 預設（無 -csv）：多核心平行解碼＋驗證，印每檔摘要與總吞吐量 —— 即「解碼測試」。
//
// 檔案格式（小端序）：
//
//	Header:  magic "PGA1"(4) | version u8 | scale u32
//	Chunk*:  startUnix i64 | rowCount u16 | stationCount u16 | compLen u32 | payload[compLen]
//	  payload(raw DEFLATE 解開後):
//	    stations:   uvarint count, 之後 count×(uvarint nameLen | name bytes)
//	    timestamps: rowCount × uvarint(與前一列的秒差；第一列為與 startUnix 的差)
//	    values:     第一列 uvarint(整數化值)；其餘列 uvarint(zigzag(本列-前一列))；0=缺席
package main

import (
	"bufio"
	"bytes"
	"compress/flate"
	"encoding/binary"
	"flag"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"runtime"
	"sort"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
)

const magic = "PGA1"

var locTW = time.FixedZone("UTC+8", 8*3600)

func main() {
	csv := flag.Bool("csv", false, "輸出 CSV(timestamp,station,pga) 到 stdout（串流、低記憶體）")
	workers := flag.Int("workers", runtime.NumCPU(), "解碼平行度（驗證/測速模式）")
	quiet := flag.Bool("q", false, "只印總結")
	flag.Parse()

	files := expandArgs(flag.Args())
	if len(files) == 0 {
		fmt.Fprintln(os.Stderr, "用法: decode [-csv] [-workers N] [-q] <file.pga|目錄>...")
		os.Exit(2)
	}

	if *csv {
		streamCSV(files)
		return
	}
	benchmark(files, *workers, *quiet)
}

// ===== 每個 goroutine 專屬的可重用緩衝（零共享、熱路徑免鎖免配置）=====
type worker struct {
	comp     []byte        // 壓縮位元組
	raw      *bytes.Buffer // 解壓後 payload
	fr       io.ReadCloser // 可重置的 flate reader
	stations [][]byte      // 指向 raw 的零拷貝測站名
	ts       []int64       // 每列時間戳
	prev     []int64       // 差量還原用
	vals     []int64       // 目前列各站值
	checksum uint64        // 觸發實際解碼、順帶當一致性指紋
}

func newWorker() *worker {
	return &worker{raw: new(bytes.Buffer), fr: flate.NewReader(bytes.NewReader(nil))}
}

type fileMeta struct {
	path      string
	rows      int
	values    int   // 非零值數量
	stations  int   // 最後一塊的測站數
	first     int64 // 最早時間戳
	last      int64 // 最晚時間戳
	compBytes int64 // 壓縮輸入位元組
	err       error
}

// decodeFile 串流解碼單一檔案；每列呼叫 onRow(可為 nil)。
func (w *worker) decodeFile(path string, onRow func(ts int64, stations [][]byte, vals []int64)) fileMeta {
	m := fileMeta{path: path, first: 1<<63 - 1}
	f, err := os.Open(path)
	if err != nil {
		m.err = err
		return m
	}
	defer f.Close()

	var hdr [9]byte
	if _, err := io.ReadFull(f, hdr[:]); err != nil {
		m.err = fmt.Errorf("讀檔頭: %w", err)
		return m
	}
	if string(hdr[:4]) != magic {
		m.err = fmt.Errorf("非 PGA 檔（magic=%q）", hdr[:4])
		return m
	}
	scale := binary.LittleEndian.Uint32(hdr[5:9])
	m.compBytes += 9

	var ch [16]byte
	for {
		n, err := io.ReadFull(f, ch[:])
		if err == io.EOF || (err == io.ErrUnexpectedEOF && n == 0) {
			break // 正常結束
		}
		if err != nil {
			m.err = fmt.Errorf("讀塊頭: %w", err)
			return m
		}
		startUnix := int64(binary.LittleEndian.Uint64(ch[0:8]))
		rowCount := int(binary.LittleEndian.Uint16(ch[8:10]))
		stationCount := int(binary.LittleEndian.Uint16(ch[10:12]))
		compLen := int(binary.LittleEndian.Uint32(ch[12:16]))
		m.compBytes += int64(16 + compLen)

		if cap(w.comp) < compLen {
			w.comp = make([]byte, compLen)
		}
		w.comp = w.comp[:compLen]
		if _, err := io.ReadFull(f, w.comp); err != nil {
			m.err = fmt.Errorf("讀壓縮內容: %w", err)
			return m
		}
		if err := w.inflate(w.comp); err != nil {
			m.err = fmt.Errorf("解壓: %w", err)
			return m
		}
		if err := w.parseChunk(w.raw.Bytes(), startUnix, rowCount, stationCount, scale, onRow, &m); err != nil {
			m.err = err
			return m
		}
	}
	return m
}

func (w *worker) inflate(comp []byte) error {
	if err := w.fr.(flate.Resetter).Reset(bytes.NewReader(comp), nil); err != nil {
		return err
	}
	w.raw.Reset()
	_, err := w.raw.ReadFrom(w.fr)
	return err
}

func (w *worker) parseChunk(raw []byte, startUnix int64, rowCount, stationCount int, scale uint32,
	onRow func(ts int64, stations [][]byte, vals []int64), m *fileMeta) error {

	pos := 0
	readUv := func() (uint64, bool) {
		v, n := binary.Uvarint(raw[pos:])
		if n <= 0 {
			return 0, false
		}
		pos += n
		return v, true
	}

	sc64, ok := readUv()
	if !ok || int(sc64) != stationCount {
		return fmt.Errorf("測站數不符（頭=%d, payload=%d）", stationCount, sc64)
	}
	sc := stationCount

	// 測站名：零拷貝指向 raw
	w.stations = w.stations[:0]
	for i := 0; i < sc; i++ {
		nl, ok := readUv()
		if !ok || pos+int(nl) > len(raw) {
			return fmt.Errorf("測站名越界")
		}
		w.stations = append(w.stations, raw[pos:pos+int(nl)])
		pos += int(nl)
	}

	// 時間戳
	if cap(w.ts) < rowCount {
		w.ts = make([]int64, rowCount)
	}
	w.ts = w.ts[:rowCount]
	prevT := startUnix
	for i := 0; i < rowCount; i++ {
		d, ok := readUv()
		if !ok {
			return fmt.Errorf("時間戳解碼失敗")
		}
		prevT += int64(d)
		w.ts[i] = prevT
	}

	// 數值（差量還原）
	if cap(w.prev) < sc {
		w.prev = make([]int64, sc)
		w.vals = make([]int64, sc)
	}
	w.prev = w.prev[:sc]
	w.vals = w.vals[:sc]
	for ci := range w.prev {
		w.prev[ci] = 0
	}

	for ri := 0; ri < rowCount; ri++ {
		for ci := 0; ci < sc; ci++ {
			u, ok := readUv()
			if !ok {
				return fmt.Errorf("數值解碼失敗 @row=%d col=%d", ri, ci)
			}
			var cur int64
			if ri == 0 {
				cur = int64(u)
			} else {
				cur = w.prev[ci] + unzigzag(u)
			}
			w.prev[ci] = cur
			w.vals[ci] = cur
			if cur != 0 {
				w.checksum += uint64(cur)
				m.values++
			}
		}
		if onRow != nil {
			onRow(w.ts[ri], w.stations, w.vals)
		}
	}

	// meta
	m.rows += rowCount
	m.stations = sc
	if w.ts[0] < m.first {
		m.first = w.ts[0]
	}
	if w.ts[rowCount-1] > m.last {
		m.last = w.ts[rowCount-1]
	}
	_ = scale
	return nil
}

// ===== 預設模式：多核心平行解碼＋測速 =====
func benchmark(files []string, workers int, quiet bool) {
	if workers < 1 {
		workers = 1
	}
	start := time.Now()

	jobs := make(chan string, workers)
	metas := make(chan fileMeta, workers)
	var wg sync.WaitGroup
	var checksum uint64

	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			w := newWorker()
			for path := range jobs {
				m := w.decodeFile(path, nil) // nil：只解碼驗證、不輸出
				metas <- m
			}
			atomic.AddUint64(&checksum, w.checksum)
		}()
	}
	go func() {
		for _, f := range files {
			jobs <- f
		}
		close(jobs)
	}()
	go func() { wg.Wait(); close(metas) }()

	var totRows, totVals, okN, failN int
	var totBytes int64
	var firstTS int64 = 1<<63 - 1
	var lastTS int64
	for m := range metas {
		if m.err != nil {
			failN++
			fmt.Fprintf(os.Stderr, "✗ %s: %v\n", m.path, m.err)
			continue
		}
		okN++
		totRows += m.rows
		totVals += m.values
		totBytes += m.compBytes
		if m.first < firstTS {
			firstTS = m.first
		}
		if m.last > lastTS {
			lastTS = m.last
		}
		if !quiet {
			fmt.Printf("%-40s  %4d 列 × %3d 站  %s ~ %s\n",
				filepath.Base(m.path), m.rows, m.stations,
				tfmt(m.first), tfmt(m.last))
		}
	}

	el := time.Since(start)
	secs := el.Seconds()
	fmt.Printf("\n解碼完成：%d 檔成功", okN)
	if failN > 0 {
		fmt.Printf("、%d 檔失敗", failN)
	}
	fmt.Printf("　workers=%d　耗時 %s\n", workers, el.Round(time.Millisecond))
	if firstTS <= lastTS {
		fmt.Printf("時間範圍：%s ~ %s (UTC+8)\n", tfmt(firstTS), tfmt(lastTS))
	}
	fmt.Printf("總計：%d 列、%s 個值、輸入 %.2f MB\n",
		totRows, human(totVals), float64(totBytes)/1e6)
	if secs > 0 {
		fmt.Printf("吞吐量：%s 值/秒、%s 列/秒、%.1f MB/秒\n",
			human(int(float64(totVals)/secs)), human(int(float64(totRows)/secs)),
			float64(totBytes)/1e6/secs)
	}
}

// ===== -csv：串流輸出（低記憶體、逐檔有序）=====
func streamCSV(files []string) {
	out := bufio.NewWriterSize(os.Stdout, 1<<20)
	defer out.Flush()
	out.WriteString("timestamp,station,pga\n")

	w := newWorker()
	var line []byte
	var tsCache int64 = -1
	var tsStr []byte

	for _, path := range files {
		m := w.decodeFile(path, func(ts int64, stations [][]byte, vals []int64) {
			if ts != tsCache { // 同一秒的時間字串只格式化一次
				tsStr = time.Unix(ts, 0).In(locTW).AppendFormat(tsStr[:0], time.RFC3339)
				tsCache = ts
			}
			for ci := range stations {
				v := vals[ci]
				if v == 0 {
					continue // 缺席
				}
				line = line[:0]
				line = append(line, tsStr...)
				line = append(line, ',')
				line = append(line, stations[ci]...)
				line = append(line, ',')
				line = appendFixed4(line, v)
				line = append(line, '\n')
				out.Write(line)
			}
		})
		if m.err != nil {
			fmt.Fprintf(os.Stderr, "✗ %s: %v\n", path, m.err)
		}
	}
}

// ===== 小工具 =====
func unzigzag(u uint64) int64 { return int64(u>>1) ^ -int64(u&1) }

// appendFixed4 假設 scale=10000（4 位小數）；把整數化值格式化為 d.dddd。
func appendFixed4(b []byte, v int64) []byte {
	b = strconv.AppendInt(b, v/10000, 10)
	f := v % 10000
	b = append(b, '.', byte('0'+f/1000), byte('0'+f/100%10), byte('0'+f/10%10), byte('0'+f%10))
	return b
}

func tfmt(unix int64) string {
	return time.Unix(unix, 0).In(locTW).Format("2006-01-02T15:04:05")
}

func human(n int) string {
	switch {
	case n >= 1_000_000_000:
		return fmt.Sprintf("%.2fG", float64(n)/1e9)
	case n >= 1_000_000:
		return fmt.Sprintf("%.2fM", float64(n)/1e6)
	case n >= 1_000:
		return fmt.Sprintf("%.1fK", float64(n)/1e3)
	default:
		return strconv.Itoa(n)
	}
}

// expandArgs 展開參數：目錄 → 其中所有 *.pga（排序）；檔案照原樣。
func expandArgs(args []string) []string {
	var out []string
	for _, a := range args {
		info, err := os.Stat(a)
		if err != nil {
			fmt.Fprintf(os.Stderr, "略過 %s: %v\n", a, err)
			continue
		}
		if info.IsDir() {
			matches, _ := filepath.Glob(filepath.Join(a, "*.pga"))
			sort.Strings(matches)
			out = append(out, matches...)
		} else {
			out = append(out, a)
		}
	}
	return out
}