binary_legacy.go

package main

import (
	"encoding/binary"
	"encoding/json"
	"encoding/xml"
	"errors"
	"fmt"
	"io"
	"math"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"
)

func convertLegacyRootToBinary(input, output string, overwrite bool) error {
	nodes, err := findRecordNodeDirs(input)
	if err != nil {
		return err
	}
	if len(nodes) == 0 {
		return fmt.Errorf("no legacy Open Ephys Record Node was found under %s", input)
	}
	if err := os.MkdirAll(output, 0755); err != nil {
		return err
	}

	var totalWork int64
	work := make(map[string]int64, len(nodes))
	for _, node := range nodes {
		w, estimateErr := estimateNodeWork(node)
		if estimateErr != nil {
			fmt.Printf("warning: could not estimate %s exactly: %v\n", filepath.Base(node), estimateErr)
		}
		if w <= 0 {
			w = 1
		}
		work[node] = w
		totalWork += w
	}
	progress := NewProgressDisplay(totalWork)
	defer progress.Close()
	converted, skipped, failures := 0, 0, 0
	for i, node := range nodes {
		rel, relErr := filepath.Rel(input, node)
		if relErr != nil || rel == "." {
			rel = filepath.Base(node)
		}
		dst := filepath.Join(output, rel)
		progress.SetSession(fmt.Sprintf("[%d/%d] %s", i+1, len(nodes), rel), work[node])
		c, s, convertErr := convertRecordNode(node, dst, overwrite, progress)
		converted += c
		skipped += s
		if convertErr != nil {
			failures++
			progress.Messagef("  [ERROR] %v", convertErr)
		}
		progress.FinishSession()
	}
	progress.FinishAll()
	fmt.Printf("Summary: %d converted, %d skipped, %d failed\n", converted, skipped, failures)
	if failures > 0 {
		return fmt.Errorf("%d Record Node conversion(s) failed", failures)
	}
	return nil
}

type npyArrayFile struct {
	file       *os.File
	dataOffset int64
	count      int64
	elemSize   int64
}

func openNpyArray(path, expectedDescr string, elemSize int64) (*npyArrayFile, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	pre := make([]byte, 10)
	if _, err := io.ReadFull(f, pre); err != nil {
		f.Close()
		return nil, err
	}
	if string(pre[0:6]) != "\x93NUMPY" || pre[6] != 1 || pre[7] != 0 {
		f.Close()
		return nil, fmt.Errorf("unsupported or invalid NPY header: %s", path)
	}
	headerLen := int64(binary.LittleEndian.Uint16(pre[8:10]))
	header := make([]byte, headerLen)
	if _, err := io.ReadFull(f, header); err != nil {
		f.Close()
		return nil, err
	}
	headerText := string(header)
	if expectedDescr != "" && !strings.Contains(headerText, "'descr': '"+expectedDescr+"'") && !strings.Contains(headerText, "\"descr\": \""+expectedDescr+"\"") {
		f.Close()
		return nil, fmt.Errorf("NPY dtype mismatch in %s; expected %s", path, expectedDescr)
	}
	st, err := f.Stat()
	if err != nil {
		f.Close()
		return nil, err
	}
	dataOffset := 10 + headerLen
	dataBytes := st.Size() - dataOffset
	if dataBytes < 0 || elemSize <= 0 || dataBytes%elemSize != 0 {
		f.Close()
		return nil, fmt.Errorf("invalid NPY payload length in %s", path)
	}
	return &npyArrayFile{file: f, dataOffset: dataOffset, count: dataBytes / elemSize, elemSize: elemSize}, nil
}

func (a *npyArrayFile) Close() error {
	if a == nil || a.file == nil {
		return nil
	}
	return a.file.Close()
}

func (a *npyArrayFile) ReadAt(index int64, dst []byte) error {
	if index < 0 || index >= a.count || int64(len(dst)) != a.elemSize {
		return fmt.Errorf("NPY scalar read out of range")
	}
	_, err := a.file.ReadAt(dst, a.dataOffset+index*a.elemSize)
	return err
}

func (a *npyArrayFile) ReadAllBytes() ([]byte, error) {
	if a.count > int64(int(^uint(0)>>1))/a.elemSize {
		return nil, errors.New("NPY array is too large for this build")
	}
	data := make([]byte, int(a.count*a.elemSize))
	_, err := a.file.ReadAt(data, a.dataOffset)
	return data, err
}

func readNpyInt64All(path string) ([]int64, error) {
	a, err := openNpyArray(path, "<i8", 8)
	if err != nil {
		return nil, err
	}
	defer a.Close()
	raw, err := a.ReadAllBytes()
	if err != nil && err != io.EOF {
		return nil, err
	}
	out := make([]int64, a.count)
	for i := range out {
		out[i] = int64(binary.LittleEndian.Uint64(raw[i*8 : i*8+8]))
	}
	return out, nil
}

func readNpyInt16All(path string) ([]int16, error) {
	a, err := openNpyArray(path, "<i2", 2)
	if err != nil {
		return nil, err
	}
	defer a.Close()
	raw, err := a.ReadAllBytes()
	if err != nil && err != io.EOF {
		return nil, err
	}
	out := make([]int16, a.count)
	for i := range out {
		out[i] = int16(binary.LittleEndian.Uint16(raw[i*2 : i*2+2]))
	}
	return out, nil
}

func readNpyFloat64All(path string) ([]float64, error) {
	a, err := openNpyArray(path, "<f8", 8)
	if err != nil {
		return nil, err
	}
	defer a.Close()
	raw, err := a.ReadAllBytes()
	if err != nil && err != io.EOF {
		return nil, err
	}
	out := make([]float64, a.count)
	for i := range out {
		out[i] = math.Float64frombits(binary.LittleEndian.Uint64(raw[i*8 : i*8+8]))
	}
	return out, nil
}

func findSingleBinaryRecording(path string) (string, OEBin, error) {
	st, err := os.Stat(path)
	if err != nil {
		return "", OEBin{}, err
	}
	if !st.IsDir() {
		if strings.EqualFold(filepath.Base(path), "structure.oebin") {
			return readBinaryRecording(filepath.Dir(path))
		}
		return "", OEBin{}, fmt.Errorf("Binary input must be a recording directory or structure.oebin")
	}
	var structures []string
	err = filepath.WalkDir(path, func(candidate string, entry os.DirEntry, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		if !entry.IsDir() && strings.EqualFold(entry.Name(), "structure.oebin") {
			structures = append(structures, candidate)
		}
		return nil
	})
	if err != nil {
		return "", OEBin{}, err
	}
	if len(structures) != 1 {
		return "", OEBin{}, fmt.Errorf("expected exactly one Binary recording (structure.oebin) under %s; found %d", path, len(structures))
	}
	return readBinaryRecording(filepath.Dir(structures[0]))
}

func readBinaryRecording(dir string) (string, OEBin, error) {
	data, err := os.ReadFile(filepath.Join(dir, "structure.oebin"))
	if err != nil {
		return "", OEBin{}, err
	}
	var info OEBin
	if err := json.Unmarshal(data, &info); err != nil {
		return "", OEBin{}, fmt.Errorf("cannot parse structure.oebin: %w", err)
	}
	if len(info.Continuous) == 0 {
		return "", OEBin{}, errors.New("structure.oebin contains no continuous streams")
	}
	return dir, info, nil
}

func convertBinaryToLegacy(input, output string, overwrite bool) error {
	dir, info, err := findSingleBinaryRecording(input)
	if err != nil {
		return err
	}
	if err := prepareDirectoryOutput(output, overwrite); err != nil {
		return err
	}
	var totalWork int64
	streamWork := make([]int64, len(info.Continuous))
	for i, stream := range info.Continuous {
		continuousDir := filepath.Join(dir, "continuous", filepath.FromSlash(strings.TrimSuffix(stream.FolderName, "/")))
		if stat, statErr := os.Stat(filepath.Join(continuousDir, "continuous.dat")); statErr == nil {
			streamWork[i] = stat.Size()
			totalWork += stat.Size()
		}
	}
	if totalWork <= 0 {
		totalWork = 1
	}
	progress := NewProgressDisplay(totalWork)
	defer progress.Close()
	progress.SetSession("Binary recording -> legacy Open Ephys", totalWork)

	root := XMLNode{XMLName: xmlName("EXPERIMENT"), Attr: []xml.Attr{
		xmlAttr("format_version", "0.4"), xmlAttr("number", "1"),
	}}
	recording := XMLNode{XMLName: xmlName("RECORDING"), Attr: []xml.Attr{xmlAttr("number", "1")}}
	for streamIndex, stream := range info.Continuous {
		if stream.NumChannels <= 0 || len(stream.Channels) != stream.NumChannels {
			return fmt.Errorf("stream %q has inconsistent channel metadata", stream.StreamName)
		}
		continuousDir := filepath.Join(dir, "continuous", filepath.FromSlash(strings.TrimSuffix(stream.FolderName, "/")))
		work := streamWork[streamIndex]
		if work <= 0 {
			work = 1
		}
		progress.SetTask(stream.StreamName, work)
		if err := exportBinaryStreamToLegacy(continuousDir, output, stream, streamIndex, info.Events, &recording, progress); err != nil {
			return fmt.Errorf("stream %q: %w", stream.StreamName, err)
		}
		progress.FinishTask()
	}
	progress.FinishSession()
	progress.FinishAll()
	root.Children = append(root.Children, recording)
	if err := writeXMLFile(filepath.Join(output, "structure.openephys"), root); err != nil {
		return err
	}
	copyIfExists(filepath.Join(dir, "settings.xml"), filepath.Join(output, "settings.xml"))
	return nil
}

func exportBinaryStreamToLegacy(continuousDir, output string, stream OEBinContinuous, streamIndex int, allEvents []OEBinEvent, recording *XMLNode, progress *ProgressDisplay) error {
	dat, err := os.Open(filepath.Join(continuousDir, "continuous.dat"))
	if err != nil {
		return err
	}
	defer dat.Close()
	stat, err := dat.Stat()
	if err != nil {
		return err
	}
	rowBytes := int64(stream.NumChannels) * 2
	if rowBytes <= 0 || stat.Size()%rowBytes != 0 {
		return fmt.Errorf("continuous.dat size is not divisible by %d", rowBytes)
	}
	totalSamples := stat.Size() / rowBytes
	recordCount := (totalSamples + samplesPerRecord - 1) / samplesPerRecord

	sampleNumbers, err := openNpyArray(filepath.Join(continuousDir, "sample_numbers.npy"), "<i8", 8)
	if err != nil {
		return fmt.Errorf("sample_numbers.npy: %w", err)
	}
	defer sampleNumbers.Close()
	timestamps, err := openNpyArray(filepath.Join(continuousDir, "timestamps.npy"), "<f8", 8)
	if err != nil {
		return fmt.Errorf("timestamps.npy: %w", err)
	}
	defer timestamps.Close()
	if sampleNumbers.count < totalSamples || timestamps.count < totalSamples {
		return fmt.Errorf("sample_numbers.npy/timestamps.npy contain fewer values than continuous.dat")
	}

	safeStream := sanitizeComponent(stream.StreamName)
	channelFiles := make([]*os.File, stream.NumChannels)
	channelNames := make([]string, stream.NumChannels)
	for i, channel := range stream.Channels {
		name := fmt.Sprintf("100_%s_%s.continuous", safeStream, sanitizeComponent(channel.ChannelName))
		channelNames[i] = name
		f, createErr := os.Create(filepath.Join(output, name))
		if createErr != nil {
			return createErr
		}
		channelFiles[i] = f
		if createErr := writeLegacyContinuousHeader(f, channel, stream); createErr != nil {
			return createErr
		}
	}
	closeChannels := func() {
		for _, f := range channelFiles {
			if f != nil {
				_ = f.Close()
			}
		}
	}
	defer closeChannels()

	timestampName := fmt.Sprintf("100_%s.timestamps", safeStream)
	tsFile, err := os.Create(filepath.Join(output, timestampName))
	if err != nil {
		return err
	}
	defer tsFile.Close()
	var scalar [8]byte
	row := make([]byte, int(rowBytes)*samplesPerRecord)
	record := make([]byte, recordSize)
	for rec := int64(0); rec < recordCount; rec++ {
		start := rec * samplesPerRecord
		remaining := totalSamples - start
		count := int64(samplesPerRecord)
		if remaining < count {
			count = remaining
		}
		n, err := dat.ReadAt(row[:int(count*rowBytes)], start*rowBytes)
		if err != nil && err != io.EOF {
			return err
		}
		if err := sampleNumbers.ReadAt(start, scalar[:]); err != nil {
			return err
		}
		sampleStart := int64(binary.LittleEndian.Uint64(scalar[:]))
		if err := timestamps.ReadAt(start, scalar[:]); err != nil {
			return err
		}
		timestamp := math.Float64frombits(binary.LittleEndian.Uint64(scalar[:]))
		binary.LittleEndian.PutUint64(scalar[:], math.Float64bits(timestamp))
		if _, err := tsFile.Write(scalar[:]); err != nil {
			return err
		}

		for channel := 0; channel < stream.NumChannels; channel++ {
			for i := range record {
				record[i] = 0
			}
			binary.LittleEndian.PutUint64(record[0:8], uint64(sampleStart))
			binary.LittleEndian.PutUint16(record[8:10], uint16(samplesPerRecord))
			binary.BigEndian.PutUint16(record[10:12], 1)
			for sample := 0; sample < samplesPerRecord; sample++ {
				value := int16(0)
				if int64(sample) < count {
					off := (sample*stream.NumChannels + channel) * 2
					value = int16(binary.LittleEndian.Uint16(row[off : off+2]))
				}
				binary.BigEndian.PutUint16(record[12+sample*2:14+sample*2], uint16(value))
			}
			copy(record[recordSize-10:], recordMarker[:])
			if _, err := channelFiles[channel].Write(record); err != nil {
				return err
			}
		}
		if progress != nil {
			progress.Add(int64(n))
		}
	}

	eventEntries := eventMetadataForStream(stream, streamIndex, allEvents)
	eventFilename := ""
	if len(eventEntries) > 0 {
		eventFilename = fmt.Sprintf("100_%s.events", safeStream)
		if err := exportBinaryEventsToLegacy(continuousDir, output, eventEntries, eventFilename); err != nil {
			return err
		}
	}
	streamNode := XMLNode{XMLName: xmlName("STREAM"), Attr: []xml.Attr{
		xmlAttr("name", stream.StreamName),
		xmlAttr("source_node_id", strconv.Itoa(stream.SourceProcessorID)),
		xmlAttr("source_node_name", stream.SourceProcessorName),
		xmlAttr("sample_rate", strconv.FormatFloat(stream.SampleRate, 'g', -1, 64)),
	}}
	for i, channel := range stream.Channels {
		streamNode.Children = append(streamNode.Children, XMLNode{XMLName: xmlName("CHANNEL"), Attr: []xml.Attr{
			xmlAttr("name", channel.ChannelName),
			xmlAttr("bitVolts", strconv.FormatFloat(channel.BitVolts, 'g', -1, 64)),
			xmlAttr("type", "continuous"),
			xmlAttr("filename", channelNames[i]),
		}})
	}
	streamNode.Children = append(streamNode.Children, XMLNode{XMLName: xmlName("TIMESTAMPS"), Attr: []xml.Attr{xmlAttr("filename", timestampName)}})
	if eventFilename != "" {
		streamNode.Children = append(streamNode.Children, XMLNode{XMLName: xmlName("EVENTS"), Attr: []xml.Attr{xmlAttr("filename", eventFilename)}})
	}
	recording.Children = append(recording.Children, streamNode)
	return nil
}

func writeLegacyContinuousHeader(w io.Writer, channel OEBinChannel, stream OEBinContinuous) error {
	header := fmt.Sprintf("header.version = 0.4\nformat = 'Open Ephys Data Format'\nversion = 0.4\ndescription = 'Open Ephys Data Format'\ndate_created = '%s'\nchannel = '%s'\nchannelType = 'continuous'\nsampleRate = %s\nblockLength = %d\nbitVolts = %s\n",
		time.Now().Format("2006-01-02 15:04:05"), channel.ChannelName, strconv.FormatFloat(stream.SampleRate, 'g', -1, 64), samplesPerRecord, strconv.FormatFloat(channel.BitVolts, 'g', -1, 64))
	buf := make([]byte, headerBytes)
	copy(buf, header)
	_, err := w.Write(buf)
	return err
}

func eventMetadataForStream(stream OEBinContinuous, streamIndex int, events []OEBinEvent) []OEBinEvent {
	_ = streamIndex
	var out []OEBinEvent
	for _, event := range events {
		if event.StreamName == stream.StreamName || event.StreamName == "" {
			out = append(out, event)
		}
	}
	return out
}

func exportBinaryEventsToLegacy(continuousDir, output string, entries []OEBinEvent, filename string) error {
	binaryRoot := filepath.Dir(filepath.Dir(continuousDir))
	path := filepath.Join(output, filename)
	f, err := os.Create(path)
	if err != nil {
		return err
	}
	defer f.Close()
	header := make([]byte, headerBytes)
	copy(header, fmt.Sprintf("header.version = 0.4\nformat = 'Open Ephys Data Format'\nversion = 0.4\ndescription = 'Open Ephys event data'\n"))
	if _, err := f.Write(header); err != nil {
		return err
	}
	for _, entry := range entries {
		folder := filepath.FromSlash(strings.TrimSuffix(entry.FolderName, "/"))
		eventDir := filepath.Join(binaryRoot, "events", folder)
		samples, err := readNpyInt64All(filepath.Join(eventDir, "sample_numbers.npy"))
		if err != nil {
			return err
		}
		states, err := readNpyInt16All(filepath.Join(eventDir, "states.npy"))
		if err != nil {
			return err
		}
		if len(samples) != len(states) {
			return fmt.Errorf("event sample/state array lengths differ in %s", eventDir)
		}
		processorID := uint8(0)
		name := filepath.Base(filepath.Dir(eventDir))
		if strings.HasPrefix(name, "Legacy_Event-") {
			id := strings.TrimPrefix(name, "Legacy_Event-")
			id = strings.SplitN(id, ".", 2)[0]
			if n, parseErr := strconv.Atoi(id); parseErr == nil && n >= 0 && n <= 255 {
				processorID = uint8(n)
			}
		}
		for i, sample := range samples {
			line := int(states[i])
			state := uint8(0)
			if line > 0 {
				state = 1
			} else {
				line = -line
			}
			if line < 1 {
				line = 1
			}
			if line > 255 {
				line = 255
			}
			record := make([]byte, eventRecordSize)
			binary.LittleEndian.PutUint64(record[0:8], uint64(sample))
			binary.LittleEndian.PutUint16(record[8:10], 0)
			record[11] = processorID
			record[12] = state
			record[13] = byte(line - 1)
			binary.LittleEndian.PutUint16(record[14:16], 1)
			if _, err := f.Write(record); err != nil {
				return err
			}
		}
	}
	return nil
}

func prepareDirectoryOutput(path string, overwrite bool) error {
	if st, err := os.Stat(path); err == nil {
		if !overwrite {
			return fmt.Errorf("output already exists: %s (use --overwrite to replace it)", path)
		}
		if !st.IsDir() {
			return fmt.Errorf("output must be a directory: %s", path)
		}
		if err := os.RemoveAll(path); err != nil {
			return err
		}
	} else if !os.IsNotExist(err) {
		return err
	}
	return os.MkdirAll(path, 0755)
}

func xmlName(local string) xml.Name {
	return xml.Name{Local: local}
}

func xmlAttr(name, value string) xml.Attr {
	return xml.Attr{Name: xml.Name{Local: name}, Value: value}
}

func writeXMLFile(path string, node XMLNode) error {
	f, err := os.Create(path)
	if err != nil {
		return err
	}
	defer f.Close()
	enc := xml.NewEncoder(f)
	if _, err := f.WriteString("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"); err != nil {
		return err
	}
	if err := enc.Encode(node); err != nil {
		return err
	}
	return enc.Flush()
}