Refactor for signal access from parser.

decode/decode.go

@@ -110,7 +110,7 @@ func NewDecoder(cfg PacketConfig, fastMag bool) (d Decoder) {
 }
 
 // Decode accepts a sample block and performs various DSP techniques to extract a packet.
-func (d Decoder) Decode(input []byte) (pkts [][]byte) {
+func (d Decoder) Decode(input []byte) []int {
 	// Shift buffers to append new block.
 	copy(d.IQ, d.IQ[d.Cfg.BlockSize<<1:])
 	copy(d.Signal, d.Signal[d.Cfg.BlockSize:])
@@ -135,39 +135,12 @@ func (d Decoder) Decode(input []byte) (pkts [][]byte) {
 	// Pack the quantized signal into slices for searching.
 	d.Pack(d.Quantized[:d.Cfg.BlockSize2], d.slices)
 
-	// Get a list of indexes the preamble exists at.
-	indexes := d.Search(d.slices, d.preamble)
-
-	// We will likely find multiple instances of the message so only keep
-	// track of unique instances.
-	seen := make(map[string]bool)
-
-	// For each of the indexes the preamble exists at.
-	for _, qIdx := range indexes {
-		// Check that we're still within the first sample block. We'll catch
-		// the message on the next sample block otherwise.
-		if qIdx > d.Cfg.BlockSize {
-			continue
-		}
-
-		// Packet is 1 bit per byte, pack to 8-bits per byte.
-		for pIdx := 0; pIdx < d.Cfg.PacketSymbols; pIdx++ {
-			d.pkt[pIdx>>3] <<= 1
-			d.pkt[pIdx>>3] |= d.Quantized[qIdx+(pIdx*d.Cfg.SymbolLength2)]
-		}
-
-		// Store the packet in the seen map and append to the packet list.
-		pktStr := fmt.Sprintf("%02X", d.pkt)
-		if !seen[pktStr] {
-			seen[pktStr] = true
-			pkts = append(pkts, make([]byte, len(d.pkt)))
-			copy(pkts[len(pkts)-1], d.pkt)
-		}
-	}
-	return
+	// Return a list of indexes the preamble exists at.
+	return d.Search(d.slices, d.preamble)
 }
 
-// A Demodulator knows how to demodulate an array of uint8 IQ samples into an array of float64 samples.
+// A Demodulator knows how to demodulate an array of uint8 IQ samples into an
+// array of float64 samples.
 type Demodulator interface {
 	Execute([]byte, []float64)
 }
@@ -289,6 +262,40 @@ func (d Decoder) Search(slices [][]byte, preamble []byte) (indexes []int) {
 	return
 }
 
+// Given a list of indeces the preamble exists at, sample the appropriate bits
+// of the signal's bit-decision. Pack bits of each index into an array of byte
+// arrays and return.
+func (d Decoder) Slice(indices []int) (pkts [][]byte) {
+	// We will likely find multiple instances of the message so only keep
+	// track of unique instances.
+	seen := make(map[string]bool)
+
+	// For each of the indices the preamble exists at.
+	for _, qIdx := range indices {
+		// Check that we're still within the first sample block. We'll catch
+		// the message on the next sample block otherwise.
+		if qIdx > d.Cfg.BlockSize {
+			continue
+		}
+
+		// Packet is 1 bit per byte, pack to 8-bits per byte.
+		for pIdx := 0; pIdx < d.Cfg.PacketSymbols; pIdx++ {
+			d.pkt[pIdx>>3] <<= 1
+			d.pkt[pIdx>>3] |= d.Quantized[qIdx+(pIdx*d.Cfg.SymbolLength2)]
+		}
+
+		// Store the packet in the seen map and append to the packet list.
+		pktStr := fmt.Sprintf("%02X", d.pkt)
+		if !seen[pktStr] {
+			seen[pktStr] = true
+			pkts = append(pkts, make([]byte, len(d.pkt)))
+			copy(pkts[len(pkts)-1], d.pkt)
+		}
+	}
+
+	return
+}
+
 func NextPowerOf2(v int) int {
 	return 1 << uint(math.Ceil(math.Log2(float64(v))))
 }

idm/idm.go

@@ -18,7 +18,6 @@ package idm
 
 import (
 	"encoding/binary"
-	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -52,10 +51,20 @@ func NewPacketConfig(symbolLength int) (cfg decode.PacketConfig) {
 }
 
 type Parser struct {
+	decode.Decoder
 	crc.CRC
 }
 
-func NewParser() (p Parser) {
+func (p Parser) Dec() decode.Decoder {
+	return p.Decoder
+}
+
+func (p Parser) Cfg() decode.PacketConfig {
+	return p.Decoder.Cfg
+}
+
+func NewParser(symbolLength int, fastMag bool) (p Parser) {
+	p.Decoder = decode.NewDecoder(NewPacketConfig(symbolLength), fastMag)
 	p.CRC = crc.NewCRC("CCITT", 0xFFFF, 0x1021, 0x1D0F)
 	return
 }
@@ -152,42 +161,56 @@ func (idm IDM) Record() (r []string) {
 	return
 }
 
-func (p Parser) Parse(data parse.Data) (msg parse.Message, err error) {
-	var idm IDM
-
-	if residue := p.Checksum(data.Bytes[4:92]); residue != p.Residue {
-		err = fmt.Errorf("packet checksum failed: 0x%04X", residue)
-		return
+func (p Parser) Parse(indices []int) (msgs []parse.Message) {
+	seen := make(map[string]bool)
+
+	for _, pkt := range p.Decoder.Slice(indices) {
+		if s := string(pkt); !seen[s] {
+			seen[s] = true
+		} else {
+			continue
+		}
+
+		data := parse.NewDataFromBytes(pkt)
+
+		// If the checksum fails, bail.
+		if residue := p.Checksum(data.Bytes[4:92]); residue != p.Residue {
+			continue
+		}
+
+		var idm IDM
+		idm.Preamble = binary.BigEndian.Uint32(data.Bytes[0:4])
+		idm.PacketTypeID = data.Bytes[4]
+		idm.PacketLength = data.Bytes[5]
+		idm.HammingCode = data.Bytes[6]
+		idm.ApplicationVersion = data.Bytes[7]
+		idm.ERTType = data.Bytes[8] & 0x0F
+		idm.ERTSerialNumber = binary.BigEndian.Uint32(data.Bytes[9:13])
+		idm.ConsumptionIntervalCount = data.Bytes[13]
+		idm.ModuleProgrammingState = data.Bytes[14]
+		idm.TamperCounters = data.Bytes[15:21]
+		idm.AsynchronousCounters = binary.BigEndian.Uint16(data.Bytes[21:23])
+		idm.PowerOutageFlags = data.Bytes[23:29]
+		idm.LastConsumptionCount = binary.BigEndian.Uint32(data.Bytes[29:33])
+
+		offset := 264
+		for idx := range idm.DifferentialConsumptionIntervals {
+			interval, _ := strconv.ParseUint(data.Bits[offset:offset+9], 2, 9)
+			idm.DifferentialConsumptionIntervals[idx] = uint16(interval)
+			offset += 9
+		}
+
+		idm.TransmitTimeOffset = binary.BigEndian.Uint16(data.Bytes[86:88])
+		idm.SerialNumberCRC = binary.BigEndian.Uint16(data.Bytes[88:90])
+		idm.PacketCRC = binary.BigEndian.Uint16(data.Bytes[90:92])
+
+		// If the meter id is 0, bail.
+		if idm.ERTSerialNumber == 0 {
+			continue
+		}
+
+		msgs = append(msgs, idm)
 	}
 
-	idm.Preamble = binary.BigEndian.Uint32(data.Bytes[0:4])
-	idm.PacketTypeID = data.Bytes[4]
-	idm.PacketLength = data.Bytes[5]
-	idm.HammingCode = data.Bytes[6]
-	idm.ApplicationVersion = data.Bytes[7]
-	idm.ERTType = data.Bytes[8] & 0x0F
-	idm.ERTSerialNumber = binary.BigEndian.Uint32(data.Bytes[9:13])
-	idm.ConsumptionIntervalCount = data.Bytes[13]
-	idm.ModuleProgrammingState = data.Bytes[14]
-	idm.TamperCounters = data.Bytes[15:21]
-	idm.AsynchronousCounters = binary.BigEndian.Uint16(data.Bytes[21:23])
-	idm.PowerOutageFlags = data.Bytes[23:29]
-	idm.LastConsumptionCount = binary.BigEndian.Uint32(data.Bytes[29:33])
-
-	offset := 264
-	for idx := range idm.DifferentialConsumptionIntervals {
-		interval, _ := strconv.ParseUint(data.Bits[offset:offset+9], 2, 9)
-		idm.DifferentialConsumptionIntervals[idx] = uint16(interval)
-		offset += 9
-	}
-
-	idm.TransmitTimeOffset = binary.BigEndian.Uint16(data.Bytes[86:88])
-	idm.SerialNumberCRC = binary.BigEndian.Uint16(data.Bytes[88:90])
-	idm.PacketCRC = binary.BigEndian.Uint16(data.Bytes[90:92])
-
-	if idm.ERTSerialNumber == 0 {
-		return idm, errors.New("invalid meter id")
-	}
-
-	return idm, nil
+	return
 }

parse/parse.go

@@ -5,6 +5,8 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/bemasher/rtlamr/decode"
+
 	"github.com/bemasher/rtlamr/csv"
 )
 
@@ -37,14 +39,16 @@ func NewDataFromBits(data string) (d Data) {
 }
 
 type Parser interface {
-	Parse(Data) (Message, error)
+	Parse([]int) []Message
+	Dec() decode.Decoder
+	Cfg() decode.PacketConfig
 }
 
 type Message interface {
+	csv.Recorder
 	MsgType() string
 	MeterID() uint32
 	MeterType() uint8
-	csv.Recorder
 }
 
 type LogMessage struct {

recv.go

@@ -27,7 +27,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/bemasher/rtlamr/decode"
 	"github.com/bemasher/rtlamr/idm"
 	"github.com/bemasher/rtlamr/parse"
 	"github.com/bemasher/rtlamr/scm"
@@ -42,24 +41,21 @@ var rcvr Receiver
 
 type Receiver struct {
 	rtltcp.SDR
-	d decode.Decoder
 	p parse.Parser
 }
 
 func (rcvr *Receiver) NewReceiver() {
 	switch strings.ToLower(*msgType) {
 	case "scm":
-		rcvr.d = decode.NewDecoder(scm.NewPacketConfig(*symbolLength), *fastMag)
-		rcvr.p = scm.NewParser()
+		rcvr.p = scm.NewParser(*symbolLength, *fastMag)
 	case "idm":
-		rcvr.d = decode.NewDecoder(idm.NewPacketConfig(*symbolLength), *fastMag)
-		rcvr.p = idm.NewParser()
+		rcvr.p = idm.NewParser(*symbolLength, *fastMag)
 	default:
 		log.Fatalf("Invalid message type: %q\n", *msgType)
 	}
 
 	if !*quiet {
-		rcvr.d.Cfg.Log()
+		rcvr.p.Cfg().Log()
 		log.Println("CRC:", rcvr.p)
 	}
 
@@ -95,7 +91,7 @@ func (rcvr *Receiver) NewReceiver() {
 	}
 
 	if !sampleRateFlagSet {
-		rcvr.SetSampleRate(uint32(rcvr.d.Cfg.SampleRate))
+		rcvr.SetSampleRate(uint32(rcvr.p.Cfg().SampleRate))
 	}
 	if !gainFlagSet {
 		rcvr.SetGainMode(true)
@@ -115,7 +111,7 @@ func (rcvr *Receiver) Run() {
 		tLimit = time.After(*timeLimit)
 	}
 
-	block := make([]byte, rcvr.d.Cfg.BlockSize2)
+	block := make([]byte, rcvr.p.Cfg().BlockSize2)
 
 	start := time.Now()
 	for {
@@ -134,26 +130,22 @@ func (rcvr *Receiver) Run() {
 			}
 
 			pktFound := false
-			for _, pkt := range rcvr.d.Decode(block) {
-				scm, err := rcvr.p.Parse(parse.NewDataFromBytes(pkt))
-				if err != nil {
-					// log.Println(err)
-					continue
-				}
+			indices := rcvr.p.Dec().Decode(block)
 
-				if len(meterID) > 0 && !meterID[uint(scm.MeterID())] {
+			for _, pkt := range rcvr.p.Parse(indices) {
+				if len(meterID) > 0 && !meterID[uint(pkt.MeterID())] {
 					continue
 				}
 
-				if len(meterType) > 0 && !meterType[uint(scm.MeterType())] {
+				if len(meterType) > 0 && !meterType[uint(pkt.MeterType())] {
 					continue
 				}
 
 				var msg parse.LogMessage
 				msg.Time = time.Now()
 				msg.Offset, _ = sampleFile.Seek(0, os.SEEK_CUR)
-				msg.Length = rcvr.d.Cfg.BufferLength << 1
-				msg.Message = scm
+				msg.Length = rcvr.p.Cfg().BufferLength << 1
+				msg.Message = pkt
 
 				if encoder == nil {
 					// A nil encoder is just plain-text output.
@@ -183,7 +175,7 @@ func (rcvr *Receiver) Run() {
 
 			if pktFound {
 				if *sampleFilename != os.DevNull {
-					_, err = sampleFile.Write(rcvr.d.IQ)
+					_, err = sampleFile.Write(rcvr.p.Dec().IQ)
 					if err != nil {
 						log.Fatal("Error writing raw samples to file:", err)
 					}