Implement streaming frames as they are parsed (#21)

This change builds upon previous changes to implement functionality to stream frames to a channel as they are parsed out of a dicom. This change also updates the dicomutil command line client to optionally use this new functionality. This closes #5.

Makefile

@@ -6,6 +6,10 @@ build:
  $(MAKE) test
  go build -o build/${BINARY} ./dicomutil
 
+.PHONY: build-fast
+build-fast:
+ go build -o build/${BINARY} ./dicomutil
+
 .PHONY: test
 test:
  go test ./...
@@ -21,4 +25,4 @@ release:
  $(MAKE) test
  GOOS=linux GOARCH=amd64 go build -o build/${BINARY}-linux-amd64 ./dicomutil;
  GOOS=darwin GOARCH=amd64 go build -o build/${BINARY}-darwin-amd64 ./dicomutil;
- GOOS=windows GOARCH=amd64 go build -o build/${BINARY}-windows-amd64.exe ./dicomutil;
+ GOOS=windows GOARCH=amd64 go build -o build/${BINARY}-windows-amd64.exe ./dicomutil;

dicomutil/dicomutil.go

@@ -19,11 +19,15 @@ import (
 )
 
 var (
- printMetadata = flag.Bool("print-metadata", true, "Print image metadata")
- extractImages = flag.Bool("extract-images", false, "Extract images into separate files")
- verbose = flag.Bool("verbose", false, "Activate high verbosity log operation")
+ printMetadata = flag.Bool("print-metadata", true, "Print image metadata")
+ extractImages = flag.Bool("extract-images", false, "Extract images into separate files")
+ extractImagesStream = flag.Bool("extract-images-stream", false, "Extract images using frame streaming capability")
+ verbose = flag.Bool("verbose", false, "Activate high verbosity log operation")
 )
 
+// FrameBufferSize represents the size of the *Frame buffered channel for streaming calls
+const FrameBufferSize = 100
+
 func main() {
  // Update usage docs
  flag.Usage = func() {
@@ -40,38 +44,82 @@ func main() {
  dicomlog.SetLevel(math.MaxInt32)
  }
  path := flag.Arg(0)
- p, err := dicom.NewParserFromFile(path, nil)
- if err != nil {
- log.Panic("error creating new parser", err)
- }
- parsedData, err := p.Parse(dicom.ParseOptions{DropPixelData: !*extractImages})
- if parsedData == nil || err != nil {
- log.Panicf("Error reading %s: %v", path, err)
+
+ var parsedData *dicom.DataSet
+
+ if *extractImagesStream {
+ // Stream process frames as they become available:
+ frameChannel := make(chan *dicom.Frame, FrameBufferSize)
+ p, err := dicom.NewParserFromFile(path, frameChannel)
+ if err != nil {
+ log.Panic("error creating parser", err)
+ }
+
+ // Go process frames published to frameChannel
+ var wg sync.WaitGroup
+ wg.Add(1)
+ go writeStreamingFrames(frameChannel, &wg)
+
+ // Begin parsing
+ parsedData, err = p.Parse(dicom.ParseOptions{})
+ if err != nil {
+ log.Panic("error parsing", err)
+ }
+
+ // Wait for all frames to be streamed and processed
+ wg.Wait()
+
+ } else {
+ // Non-streaming parsing:
+ p, err := dicom.NewParserFromFile(path, nil)
+ if err != nil {
+ log.Panic("error creating new parser", err)
+ }
+ parsedData, err = p.Parse(dicom.ParseOptions{DropPixelData: !*extractImages})
+ if parsedData == nil || err != nil {
+ log.Panicf("Error reading %s: %v", path, err)
+ }
+ if *extractImages {
+ for _, elem := range parsedData.Elements {
+ if elem.Tag == dicomtag.PixelData {
+ data := elem.Value[0].(dicom.PixelDataInfo)
+
+ var wg sync.WaitGroup
+ for frameIndex, frame := range data.Frames {
+ wg.Add(1)
+ go generateImage(&frame, frameIndex, &wg)
+ }
+ wg.Wait()
+
+ }
+ }
+ }
  }
+
+ // Print Metadata from parsedData if needed
  if *printMetadata {
+ log.Println(parsedData)
  for _, elem := range parsedData.Elements {
  fmt.Printf("%v\n", elem.String())
  }
  }
- if *extractImages {
- for _, elem := range parsedData.Elements {
- if elem.Tag == dicomtag.PixelData {
- data := elem.Value[0].(dicom.PixelDataInfo)
 
- var wg sync.WaitGroup
- for frameIndex, frame := range data.Frames {
- wg.Add(1)
- go generateImage(frame, frameIndex, &wg)
- }
- wg.Wait()
+ log.Println("Complete.")
+}
 
- }
- }
+func writeStreamingFrames(frameChan chan *dicom.Frame, doneWG *sync.WaitGroup) {
+ count := 0 // may not correspond to frame number
+ var wg sync.WaitGroup
+ for frame := range frameChan {
+ count++
+ wg.Add(1)
+ go generateImage(frame, count, &wg)
  }
- log.Println("Complete.")
+ wg.Wait()
+ doneWG.Done()
 }
 
-func generateImage(frame dicom.Frame, frameIndex int, wg *sync.WaitGroup) {
+func generateImage(frame *dicom.Frame, frameIndex int, wg *sync.WaitGroup) {
  if frame.IsEncapsulated {
  go generateEncapsulatedImage(frame.EncapsulatedData, frameIndex, wg)
  } else {
@@ -83,7 +131,7 @@ func generateEncapsulatedImage(frame dicom.EncapsulatedFrame, frameIndex int, wg
  defer wg.Done()
  path := fmt.Sprintf("image_%d.jpg", frameIndex) // TODO: figure out the image format
  ioutil.WriteFile(path, frame.Data, 0644)
- fmt.Printf("%s: %d bytes\n", path, len(frame.Data))
+ log.Printf("%s: %d bytes\n", path, len(frame.Data))
 }
 
 func generateNativeImage(frame dicom.NativeFrame, frameIndex int, wg *sync.WaitGroup) {
@@ -98,5 +146,5 @@ func generateNativeImage(frame dicom.NativeFrame, frameIndex int, wg *sync.WaitG
  fmt.Printf("Error while creating file: %s", err.Error())
  }
  jpeg.Encode(f, i, &jpeg.Options{Quality: 100})
- fmt.Printf("%s written \n", name)
+ log.Printf("%s written \n", name)
 }

parse.go

@@ -214,12 +214,22 @@ func (p *parser) ParseNext(options ParseOptions) *Element {
  if endOfItems {
  break
  }
- image.Frames = append(image.Frames, Frame{
+
+ // Construct frame
+ f := Frame{
  IsEncapsulated: true,
  EncapsulatedData: EncapsulatedFrame{
  Data: chunk,
  },
- })
+ }
+
+ image.Frames = append(image.Frames, f)
+ if p.frameChannel != nil {
+ p.frameChannel <- &f // write frame to channel
+ }
+ }
+ if p.frameChannel != nil {
+ close(p.frameChannel)
  }
  data = append(data, image)
  } else {
@@ -230,7 +240,7 @@ func (p *parser) ParseNext(options ParseOptions) *Element {
  return nil // TODO(suyash) investigate error handling in this library
  }
 
- image, _, err := readNativeFrames(p.decoder, p.parsedElements)
+ image, _, err := readNativeFrames(p.decoder, p.parsedElements, p.frameChannel)
 
  if err != nil {
  p.decoder.SetError(err)
@@ -507,7 +517,7 @@ func getTransferSyntax(ds *DataSet) (bo binary.ByteOrder, implicit dicomio.IsImp
 
 // readNativeFrames reads NativeData frames from a Decoder based on already parsed pixel information
 // that should be available in parsedData (elements like NumberOfFrames, rows, columns, etc)
-func readNativeFrames(d *dicomio.Decoder, parsedData *DataSet) (pixelData *PixelDataInfo, bytesRead int, err error) {
+func readNativeFrames(d *dicomio.Decoder, parsedData *DataSet, frameChan chan *Frame) (pixelData *PixelDataInfo, bytesRead int, err error) {
  image := PixelDataInfo{
  IsEncapsulated: false,
  }
@@ -581,6 +591,13 @@ func readNativeFrames(d *dicomio.Decoder, parsedData *DataSet) (pixelData *Pixel
  currentFrame.NativeData.Data[pixel] = currentPixel
  }
  image.Frames[frame] = currentFrame
+ if frameChan != nil {
+ frameChan <- &currentFrame // write the current frame to the frameChan
+ }
+ }
+
+ if frameChan != nil {
+ close(frameChan) // close frameChan if it exists
  }
 
  bytesRead = (bitsAllocated / 8) * samplesPerPixel * pixelsPerFrame * nFrames