orchestrator: Generate MPEG7 perceptual hashes for fast verification

core/capabilities.go

@@ -35,6 +35,7 @@ const (
 	Capability_GOP
 	Capability_AuthToken
 	Capability_SceneClassification
+	Capability_MPEG7VideoSignature
 )
 
 var capFormatConv = errors.New("capability: unknown format")
@@ -57,6 +58,7 @@ func DefaultCapabilities() []Capability {
 		Capability_ProfileH264ConstrainedHigh,
 		Capability_GOP,
 		Capability_AuthToken,
+		Capability_MPEG7VideoSignature,
 	}
 }
 

core/orchestrator.go

@@ -396,7 +396,8 @@ type TranscodeData struct {
 // TranscodedSegmentData contains encoded data for a profile
 type TranscodedSegmentData struct {
 	Data   []byte
-	Pixels int64 // Encoded pixels
+	PHash  []byte // Perceptual hash data (maybe nil)
+	Pixels int64  // Encoded pixels
 }
 
 type SegChanData struct {
@@ -581,6 +582,12 @@ func (n *LivepeerNode) transcodeSeg(config transcodeConfig, seg *stream.HLSSegme
 				string(md.ManifestID), md.AuthToken.SessionId, seg.SeqNo, len(tSegments[i].Data))
 			return terr(fmt.Errorf("ZeroSegments"))
 		}
+		if md.CalcPerceptualHash && tSegments[i].PHash == nil {
+			glog.Errorf("Could not find perceptual hash for manifestID=%s sessionID=%s seqNo=%d profile=%v",
+				string(md.ManifestID), md.AuthToken.SessionId, seg.SeqNo, md.Profiles[i].Name)
+			// FIXME: Return the error once everyone has upgraded their nodes
+			// return terr(fmt.Errorf("MissingPerceptualHash"))
+		}
 		glog.V(common.DEBUG).Infof("Transcoded segment manifestID=%s sessionID=%s seqNo=%d profile=%s len=%d",
 			string(md.ManifestID), md.AuthToken.SessionId, seg.SeqNo, md.Profiles[i].Name, len(tSegments[i].Data))
 		hash := crypto.Keccak256(tSegments[i].Data)

core/streamdata.go

@@ -47,17 +47,18 @@ func (s *StreamParameters) StreamID() string {
 }
 
 type SegTranscodingMetadata struct {
-	ManifestID       ManifestID
-	Fname            string
-	Seq              int64
-	Hash             ethcommon.Hash
-	Profiles         []ffmpeg.VideoProfile
-	OS               *net.OSInfo
-	Duration         time.Duration
-	Caps             *Capabilities
-	AuthToken        *net.AuthToken
-	DetectorEnabled  bool
-	DetectorProfiles []ffmpeg.DetectorProfile
+	ManifestID         ManifestID
+	Fname              string
+	Seq                int64
+	Hash               ethcommon.Hash
+	Profiles           []ffmpeg.VideoProfile
+	OS                 *net.OSInfo
+	Duration           time.Duration
+	Caps               *Capabilities
+	AuthToken          *net.AuthToken
+	DetectorEnabled    bool
+	DetectorProfiles   []ffmpeg.DetectorProfile
+	CalcPerceptualHash bool
 }
 
 func (md *SegTranscodingMetadata) Flatten() []byte {
@@ -109,15 +110,16 @@ func NetSegData(md *SegTranscodingMetadata) (*net.SegData, error) {
 	}
 	// Generate serialized segment info
 	segData := &net.SegData{
-		ManifestId:       []byte(md.ManifestID),
-		Seq:              md.Seq,
-		Hash:             md.Hash.Bytes(),
-		Storage:          storage,
-		Duration:         int32(md.Duration / time.Millisecond),
-		Capabilities:     md.Caps.ToNetCapabilities(),
-		AuthToken:        md.AuthToken,
-		DetectorEnabled:  md.DetectorEnabled,
-		DetectorProfiles: detectorProfiles,
+		ManifestId:         []byte(md.ManifestID),
+		Seq:                md.Seq,
+		Hash:               md.Hash.Bytes(),
+		Storage:            storage,
+		Duration:           int32(md.Duration / time.Millisecond),
+		Capabilities:       md.Caps.ToNetCapabilities(),
+		AuthToken:          md.AuthToken,
+		DetectorEnabled:    md.DetectorEnabled,
+		DetectorProfiles:   detectorProfiles,
+		CalcPerceptualHash: md.CalcPerceptualHash,
 		// Triggers failure on Os that don't know how to use FullProfiles/2/3
 		Profiles: []byte("invalid"),
 	}

core/transcoder.go

@@ -36,7 +36,7 @@ func (lt *LocalTranscoder) Transcode(md *SegTranscodingMetadata) (*TranscodeData
 		Accel: ffmpeg.Software,
 	}
 	profiles := md.Profiles
-	opts := profilesToTranscodeOptions(lt.workDir, ffmpeg.Software, profiles)
+	opts := profilesToTranscodeOptions(lt.workDir, ffmpeg.Software, profiles, md.CalcPerceptualHash)
 	if md.DetectorEnabled {
 		opts = append(opts, detectorsToTranscodeOptions(lt.workDir, ffmpeg.Software, md.DetectorProfiles)...)
 	}
@@ -77,7 +77,7 @@ func (nv *NvidiaTranscoder) Transcode(md *SegTranscodingMetadata) (*TranscodeDat
 		Device: nv.device,
 	}
 	profiles := md.Profiles
-	out := profilesToTranscodeOptions(WorkDir, ffmpeg.Nvidia, profiles)
+	out := profilesToTranscodeOptions(WorkDir, ffmpeg.Nvidia, profiles, md.CalcPerceptualHash)
 	if md.DetectorEnabled {
 		out = append(out, detectorsToTranscodeOptions(WorkDir, ffmpeg.Nvidia, md.DetectorProfiles)...)
 	}
@@ -189,7 +189,21 @@ func resToTranscodeData(res *ffmpeg.TranscodeResults, opts []ffmpeg.TranscodeOpt
 				glog.Error("Cannot read transcoded output for ", oname)
 				return nil, err
 			}
-			segments = append(segments, &TranscodedSegmentData{Data: o, Pixels: res.Encoded[i].Pixels})
+			// Extract perceptual hash if calculated
+			var s []byte = nil
+			if opts[i].CalcSign {
+				sigfile := oname + ".bin"
+				s, err = ioutil.ReadFile(sigfile)
+				if err != nil {
+					glog.Error("Cannot read perceptual hash at ", sigfile)
+					return nil, err
+				}
+				err = os.Remove(sigfile)
+				if err != nil {
+					glog.Error("Cannot delete perceptual hash after reading: ", sigfile)
+				}
+			}
+			segments = append(segments, &TranscodedSegmentData{Data: o, Pixels: res.Encoded[i].Pixels, PHash: s})
 			os.Remove(oname)
 		} else {
 			detections = append(detections, res.Encoded[i].DetectData)
@@ -203,14 +217,15 @@ func resToTranscodeData(res *ffmpeg.TranscodeResults, opts []ffmpeg.TranscodeOpt
 	}, nil
 }
 
-func profilesToTranscodeOptions(workDir string, accel ffmpeg.Acceleration, profiles []ffmpeg.VideoProfile) []ffmpeg.TranscodeOptions {
+func profilesToTranscodeOptions(workDir string, accel ffmpeg.Acceleration, profiles []ffmpeg.VideoProfile, calcPHash bool) []ffmpeg.TranscodeOptions {
 	opts := make([]ffmpeg.TranscodeOptions, len(profiles), len(profiles))
 	for i := range profiles {
 		o := ffmpeg.TranscodeOptions{
 			Oname:        fmt.Sprintf("%s/out_%s.tempfile", workDir, common.RandName()),
 			Profile:      profiles[i],
 			Accel:        accel,
 			AudioEncoder: ffmpeg.ComponentOptions{Name: "copy"},
+			CalcSign:     calcPHash,
 		}
 		opts[i] = o
 	}

core/transcoder_test.go

@@ -151,6 +151,23 @@ func TestResToTranscodeData(t *testing.T) {
 	assert.Equal(int64(300), tData.Segments[1].Pixels)
 	assert.True(fileDNE(file1.Name()))
 	assert.True(fileDNE(file2.Name()))
+
+	// Test signature file
+	res = &ffmpeg.TranscodeResults{Encoded: make([]ffmpeg.MediaInfo, 1)}
+	pHash := []byte{4, 2, 0, 6, 9}
+
+	file1, err = ioutil.TempFile(tempDir, "foo")
+	require.Nil(err)
+	ioutil.WriteFile(file1.Name()+".bin", pHash, 0664)
+
+	opts = make([]ffmpeg.TranscodeOptions, 1)
+	opts[0].Oname = file1.Name()
+	opts[0].CalcSign = true
+
+	tData, err = resToTranscodeData(res, opts)
+	assert.Nil(err)
+	assert.Equal(tData.Segments[0].PHash, pHash)
+	assert.True(fileDNE(file1.Name()))
 }
 
 func TestProfilesToTranscodeOptions(t *testing.T) {
@@ -166,12 +183,12 @@ func TestProfilesToTranscodeOptions(t *testing.T) {
 
 	// Test 0 profiles
 	profiles := []ffmpeg.VideoProfile{}
-	opts := profilesToTranscodeOptions(workDir, ffmpeg.Software, profiles)
+	opts := profilesToTranscodeOptions(workDir, ffmpeg.Software, profiles, false)
 	assert.Equal(0, len(opts))
 
 	// Test 1 profile
 	profiles = []ffmpeg.VideoProfile{ffmpeg.P144p30fps16x9}
-	opts = profilesToTranscodeOptions(workDir, ffmpeg.Software, profiles)
+	opts = profilesToTranscodeOptions(workDir, ffmpeg.Software, profiles, false)
 	assert.Equal(1, len(opts))
 	assert.Equal("foo/out_bar.tempfile", opts[0].Oname)
 	assert.Equal(ffmpeg.Software, opts[0].Accel)
@@ -180,7 +197,7 @@ func TestProfilesToTranscodeOptions(t *testing.T) {
 
 	// Test > 1 profile
 	profiles = []ffmpeg.VideoProfile{ffmpeg.P144p30fps16x9, ffmpeg.P240p30fps16x9}
-	opts = profilesToTranscodeOptions(workDir, ffmpeg.Software, profiles)
+	opts = profilesToTranscodeOptions(workDir, ffmpeg.Software, profiles, false)
 	assert.Equal(2, len(opts))
 
 	for i, p := range profiles {
@@ -191,9 +208,14 @@ func TestProfilesToTranscodeOptions(t *testing.T) {
 	}
 
 	// Test different acceleration value
-	opts = profilesToTranscodeOptions(workDir, ffmpeg.Nvidia, profiles)
+	opts = profilesToTranscodeOptions(workDir, ffmpeg.Nvidia, profiles, false)
 	assert.Equal(2, len(opts))
 
+	// Test signature calculation
+	opts = profilesToTranscodeOptions(workDir, ffmpeg.Nvidia, profiles, true)
+	assert.True(opts[0].CalcSign)
+	assert.True(opts[1].CalcSign)
+
 	for i, p := range profiles {
 		assert.Equal("foo/out_bar.tempfile", opts[i].Oname)
 		assert.Equal(ffmpeg.Nvidia, opts[i].Accel)