package landscape import ( "context" "fmt" "path/filepath" "runtime" "sync" "gis/internal/raster" "golang.org/x/sync/errgroup" ) // Compute runs one node over its input rasters. See the package doc for the // concurrency model: windows are processed by a worker pool; each worker owns // its own Reader handles (GDAL datasets are not goroutine-safe) and writes are // serialised per output. Window order does not affect the result. func Compute(ctx context.Context, rio raster.IO, node Node, inputs map[string]string, outDir string, p Params) (map[string]string, error) { p = p.WithDefaults() paths := make([]string, len(node.Inputs)) for i, g := range node.Inputs { path, ok := inputs[g] if !ok || path == "" { return nil, fmt.Errorf("node %s (%s): missing input %q", node.Code, node.Name, g) } paths[i] = path } // Validate grids with one set of handles before creating anything. grid, err := checkGrids(ctx, rio, node, paths) if err != nil { return nil, err } outPaths := make(map[string]string, len(node.Outputs)) writers := make([]raster.Writer, len(node.Outputs)) for j, g := range node.Outputs { path := filepath.Join(outDir, g+".tif") w, err := rio.Create(ctx, path, grid, raster.DefaultWriterOptions()) if err != nil { closeWriters(writers[:j]) return nil, fmt.Errorf("node %s: create output %s: %w", node.Code, g, err) } writers[j] = w outPaths[g] = path } windows := raster.Windows(grid.Width, grid.Height, p.WindowSize) jobs := make(chan raster.Window) var writeMu sync.Mutex eg, gctx := errgroup.WithContext(ctx) workers := min(runtime.GOMAXPROCS(0), len(windows)) for w := 0; w < workers; w++ { eg.Go(func() error { readers, err := openAll(gctx, rio, paths) if err != nil { return err } defer closeReaders(readers) size := p.WindowSize * p.WindowSize in := make([][]float64, len(paths)) for i := range in { in[i] = make([]float64, size) } out := make([][]int16, len(writers)) for j := range out { out[j] = make([]int16, size) } for win := range jobs { n := win.Width * win.Height inView := make([][]float64, len(in)) for i, r := range readers { if err := r.ReadFloat64(win, in[i][:n]); err != nil { return fmt.Errorf("node %s: read %s window %+v: %w", node.Code, node.Inputs[i], win, err) } inView[i] = in[i][:n] } outView := make([][]int16, len(out)) for j := range out { outView[j] = out[j][:n] } node.Kernel(inView, outView, p) writeMu.Lock() for j, wr := range writers { if err := wr.WriteInt16(win, outView[j]); err != nil { writeMu.Unlock() return fmt.Errorf("node %s: write %s window %+v: %w", node.Code, node.Outputs[j], win, err) } } writeMu.Unlock() } return nil }) } eg.Go(func() error { defer close(jobs) for _, win := range windows { select { case jobs <- win: case <-gctx.Done(): return gctx.Err() } } return nil }) runErr := eg.Wait() if err := closeWriters(writers); err != nil && runErr == nil { runErr = err } if runErr != nil { return nil, runErr } return outPaths, nil } // checkGrids opens every input once, verifies all grids match the first, and // returns that grid. func checkGrids(ctx context.Context, rio raster.IO, node Node, paths []string) (raster.Grid, error) { readers, err := openAll(ctx, rio, paths) if err != nil { return raster.Grid{}, err } defer closeReaders(readers) ref := readers[0].Grid() for i := 1; i < len(readers); i++ { if g := readers[i].Grid(); !rio.SameGrid(ref, g) { return raster.Grid{}, fmt.Errorf("node %s: input %s grid (%dx%d, %v) differs from %s (%dx%d, %v); inputs must be aligned", node.Code, node.Inputs[i], g.Width, g.Height, g.GeoTransform, node.Inputs[0], ref.Width, ref.Height, ref.GeoTransform) } } return ref, nil } func openAll(ctx context.Context, rio raster.IO, paths []string) ([]raster.Reader, error) { readers := make([]raster.Reader, 0, len(paths)) for _, p := range paths { r, err := rio.Open(ctx, p) if err != nil { closeReaders(readers) return nil, err } readers = append(readers, r) } return readers, nil } func closeReaders(rs []raster.Reader) { for _, r := range rs { if r != nil { _ = r.Close() } } } func closeWriters(ws []raster.Writer) error { var first error for _, w := range ws { if w == nil { continue } if err := w.Close(); err != nil && first == nil { first = err } } return first }