uopi/internal/datasource/epics/archive.go

package epics

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/uopi/uopi/internal/datasource"
)

// archiveResponse is the top-level JSON structure returned by the EPICS Archive
// Appliance getData.json endpoint.
//
// Example response shape:
//
//	[
//	  {
//	    "meta": { "name": "PV:NAME", "EGU": "A" },
//	    "data": [
//	      { "secs": 1700000000, "nanos": 123000000, "val": 3.14, "severity": 0, "status": 0 },
//	      ...
//	    ]
//	  }
//	]
type archiveResponse []struct {
	Meta struct {
		Name string `json:"name"`
		EGU  string `json:"EGU"`
	} `json:"meta"`
	Data []archivePoint `json:"data"`
}

type archivePoint struct {
	Secs     int64   `json:"secs"`
	Nanos    int64   `json:"nanos"`
	Val      any     `json:"val"`
	Severity int     `json:"severity"`
	Status   int     `json:"status"`
}

// fetchArchiveHistory queries the EPICS Archive Appliance JSON API for
// historical data for the given PV in the time range [start, end].
func fetchArchiveHistory(ctx context.Context, archiveURL, signal string, start, end time.Time, maxPoints int) ([]datasource.Value, error) {
	// Build the request URL.
	// Format: GET {archiveURL}/retrieval/data/getData.json?pv={name}&from={start}&to={end}&maxSamples={n}
	reqURL, err := url.Parse(archiveURL)
	if err != nil {
		return nil, fmt.Errorf("epics archive: invalid archive URL %q: %w", archiveURL, err)
	}
	reqURL = reqURL.JoinPath("retrieval", "data", "getData.json")

	q := reqURL.Query()
	q.Set("pv", signal)
	q.Set("from", start.UTC().Format(time.RFC3339Nano))
	q.Set("to", end.UTC().Format(time.RFC3339Nano))
	if maxPoints > 0 {
		q.Set("maxSamples", strconv.Itoa(maxPoints))
	}
	reqURL.RawQuery = q.Encode()

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil)
	if err != nil {
		return nil, fmt.Errorf("epics archive: building request: %w", err)
	}
	req.Header.Set("Accept", "application/json")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("epics archive: HTTP request failed: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("epics archive: unexpected status %d for %q", resp.StatusCode, signal)
	}

	var ar archiveResponse
	if err := json.NewDecoder(resp.Body).Decode(&ar); err != nil {
		return nil, fmt.Errorf("epics archive: decoding response: %w", err)
	}

	if len(ar) == 0 {
		return nil, nil
	}

	points := ar[0].Data
	out := make([]datasource.Value, 0, len(points))

	// EPICS Archive Appliance uses the POSIX epoch (unlike the CA wire protocol
	// which uses the EPICS epoch).  The JSON API returns POSIX seconds.
	for _, p := range points {
		ts := time.Unix(p.Secs, p.Nanos).UTC()

		var q datasource.Quality
		switch p.Severity {
		case 0:
			q = datasource.QualityGood
		case 1, 2:
			q = datasource.QualityUncertain
		default:
			q = datasource.QualityBad
		}

		data := coerceArchiveValue(p.Val)

		out = append(out, datasource.Value{
			Timestamp: ts,
			Data:      data,
			Quality:   q,
		})
	}

	return out, nil
}

// searchArchivePVs queries the Archive Appliance management API for PV names
// matching the given pattern (glob style, e.g. "PV:*").
func searchArchivePVs(ctx context.Context, archiveURL, pattern string) ([]string, error) {
	if archiveURL == "" {
		return nil, nil
	}

	// Build the request URL.
	// Format: GET {archiveURL}/mgmt/bpl/getAllPVs?pv={pattern}
	reqURL, err := url.Parse(archiveURL)
	if err != nil {
		return nil, fmt.Errorf("epics archive: invalid archive URL %q: %w", archiveURL, err)
	}
	reqURL = reqURL.JoinPath("mgmt", "bpl", "getAllPVs")

	q := reqURL.Query()
	if pattern == "" {
		pattern = "*"
	}
	// AA expects glob patterns. If no glob characters, wrap in stars.
	if !strings.ContainsAny(pattern, "*?[]") {
		pattern = "*" + pattern + "*"
	}
	q.Set("pv", pattern)
	reqURL.RawQuery = q.Encode()

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil)
	if err != nil {
		return nil, fmt.Errorf("epics archive: building request: %w", err)
	}

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("epics archive: HTTP request failed: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("epics archive: unexpected status %d for search %q", resp.StatusCode, pattern)
	}

	var pvs []string
	if err := json.NewDecoder(resp.Body).Decode(&pvs); err != nil {
		return nil, fmt.Errorf("epics archive: decoding response: %w", err)
	}

	return pvs, nil
}

// coerceArchiveValue converts the raw JSON value (which json.Unmarshal decodes
// as float64, string, bool, []interface{}, or nil) into one of the types
// accepted by datasource.Value.Data.
func coerceArchiveValue(v any) any {
	if v == nil {
		return float64(0)
	}
	switch t := v.(type) {
	case float64:
		return t
	case string:
		return t
	case bool:
		return t
	case []interface{}:
		// Waveform: convert to []float64.
		out := make([]float64, 0, len(t))
		for _, elem := range t {
			if f, ok := elem.(float64); ok {
				out = append(out, f)
			}
		}
		return out
	default:
		return float64(0)
	}
}