Add ticks to --devs

Fixes #278

Signed-off-by: Vadim Markovtsev <vadim@sourced.tech>

internal/pb/pb.proto

@@ -136,7 +136,10 @@ message TickDevs {
 
 message DevsAnalysisResults {
     map<int32, TickDevs> ticks = 1;
+    // developer identities, the indexes correspond to TickDevs' keys.
     repeated string dev_index = 2;
+    // how long each tick is, as an int64 nanosecond count (Go's time.Duration)
+    int64 tick_size = 8;
 }
 
 message Sentiment {

leaves/devs.go

@@ -1,10 +1,12 @@
 package leaves
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"sort"
 	"strings"
+	"time"
 
 	"github.com/gogo/protobuf/proto"
 	"gopkg.in/src-d/go-git.v4"
@@ -31,6 +33,8 @@ type DevsAnalysis struct {
 	ticks map[int]map[int]*DevTick
 	// reversedPeopleDict references IdentityDetector.ReversedPeopleDict
 	reversedPeopleDict []string
+	// tickSize references TicksSinceStart.tickSize
+	tickSize time.Duration
 
 	l core.Logger
 }
@@ -43,6 +47,8 @@ type DevsResult struct {
 
 	// reversedPeopleDict references IdentityDetector.ReversedPeopleDict
 	reversedPeopleDict []string
+	// tickSize references TicksSinceStart.tickSize
+	tickSize time.Duration
 }
 
 // DevTick is the statistics for a development tick and a particular developer.
@@ -103,6 +109,9 @@ func (devs *DevsAnalysis) Configure(facts map[string]interface{}) error {
 	if val, exists := facts[identity.FactIdentityDetectorReversedPeopleDict].([]string); exists {
 		devs.reversedPeopleDict = val
 	}
+	if val, exists := facts[items.FactTickSize].(time.Duration); exists {
+		devs.tickSize = val
+	}
 	return nil
 }
 
@@ -119,6 +128,9 @@ func (devs *DevsAnalysis) Description() string {
 // Initialize resets the temporary caches and prepares this PipelineItem for a series of Consume()
 // calls. The repository which is going to be analysed is supplied as an argument.
 func (devs *DevsAnalysis) Initialize(repository *git.Repository) error {
+	if devs.tickSize == 0 {
+		return errors.New("tick size must be specified")
+	}
 	devs.l = core.NewLogger()
 	devs.ticks = map[int]map[int]*DevTick{}
 	devs.OneShotMergeProcessor.Initialize()
@@ -178,6 +190,7 @@ func (devs *DevsAnalysis) Finalize() interface{} {
 	return DevsResult{
 		Ticks:              devs.ticks,
 		reversedPeopleDict: devs.reversedPeopleDict,
+		tickSize:           devs.tickSize,
 	}
 }
 
@@ -234,6 +247,7 @@ func (devs *DevsAnalysis) Deserialize(pbmessage []byte) (interface{}, error) {
 	result := DevsResult{
 		Ticks:              ticks,
 		reversedPeopleDict: message.DevIndex,
+		tickSize:           time.Duration(message.TickSize),
 	}
 	return result, nil
 }
@@ -242,6 +256,19 @@ func (devs *DevsAnalysis) Deserialize(pbmessage []byte) (interface{}, error) {
 func (devs *DevsAnalysis) MergeResults(r1, r2 interface{}, c1, c2 *core.CommonAnalysisResult) interface{} {
 	cr1 := r1.(DevsResult)
 	cr2 := r2.(DevsResult)
+	if cr1.tickSize != cr2.tickSize {
+		return fmt.Errorf("mismatching tick sizes (r1: %d, r2: %d) received",
+			cr1.tickSize, cr2.tickSize)
+	}
+	t01 := items.FloorTime(c1.BeginTimeAsTime(), cr1.tickSize)
+	t02 := items.FloorTime(c2.BeginTimeAsTime(), cr2.tickSize)
+	t0 := t01
+	if t02.Before(t0) {
+		t0 = t02
+	}
+	offset1 := int(t01.Sub(t0) / cr1.tickSize)
+	offset2 := int(t02.Sub(t0) / cr2.tickSize)
+
 	merged := DevsResult{}
 	var mergedIndex map[string]identity.MergedIndex
 	mergedIndex, merged.reversedPeopleDict = identity.MergeReversedDictsIdentities(
@@ -249,6 +276,7 @@ func (devs *DevsAnalysis) MergeResults(r1, r2 interface{}, c1, c2 *core.CommonAn
 	newticks := map[int]map[int]*DevTick{}
 	merged.Ticks = newticks
 	for tick, dd := range cr1.Ticks {
+		tick += offset1
 		newdd, exists := newticks[tick]
 		if !exists {
 			newdd = map[int]*DevTick{}
@@ -279,6 +307,7 @@ func (devs *DevsAnalysis) MergeResults(r1, r2 interface{}, c1, c2 *core.CommonAn
 		}
 	}
 	for tick, dd := range cr2.Ticks {
+		tick += offset2
 		newdd, exists := newticks[tick]
 		if !exists {
 			newdd = map[int]*DevTick{}
@@ -357,11 +386,13 @@ func (devs *DevsAnalysis) serializeText(result *DevsResult, writer io.Writer) {
 	for _, person := range result.reversedPeopleDict {
 		fmt.Fprintf(writer, "  - %s\n", yaml.SafeString(person))
 	}
+	fmt.Fprintln(writer, "  tick_size:", int(result.tickSize.Seconds()))
 }
 
 func (devs *DevsAnalysis) serializeBinary(result *DevsResult, writer io.Writer) error {
 	message := pb.DevsAnalysisResults{}
 	message.DevIndex = result.reversedPeopleDict
+	message.TickSize = int64(result.tickSize)
 	message.Ticks = map[int32]*pb.TickDevs{}
 	for tick, devs := range result.Ticks {
 		dd := &pb.TickDevs{}

leaves/devs_test.go

@@ -3,6 +3,7 @@ package leaves
 import (
 	"bytes"
 	"testing"
+	"time"
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/stretchr/testify/assert"
@@ -18,6 +19,7 @@ import (
 
 func fixtureDevs() *DevsAnalysis {
 	d := DevsAnalysis{}
+	d.tickSize = 24 * time.Hour
 	d.Initialize(test.Repository)
 	people := [...]string{"one@srcd", "two@srcd"}
 	d.reversedPeopleDict = people[:]
@@ -67,13 +69,17 @@ func TestDevsConfigure(t *testing.T) {
 	devs := DevsAnalysis{}
 	facts := map[string]interface{}{}
 	facts[ConfigDevsConsiderEmptyCommits] = true
-	devs.Configure(facts)
-	assert.Equal(t, devs.ConsiderEmptyCommits, true)
+	facts[items.FactTickSize] = 3 * time.Hour
+	assert.NoError(t, devs.Configure(facts))
+	assert.True(t, devs.ConsiderEmptyCommits)
+	assert.Equal(t, 3*time.Hour, devs.tickSize)
 }
 
 func TestDevsInitialize(t *testing.T) {
 	d := fixtureDevs()
 	assert.NotNil(t, d.ticks)
+	d = &DevsAnalysis{}
+	assert.Error(t, d.Initialize(test.Repository))
 }
 
 func TestDevsConsumeFinalize(t *testing.T) {
@@ -279,6 +285,7 @@ func TestDevsFinalize(t *testing.T) {
 	x := devs.Finalize().(DevsResult)
 	assert.Equal(t, x.Ticks, devs.ticks)
 	assert.Equal(t, x.reversedPeopleDict, devs.reversedPeopleDict)
+	assert.Equal(t, 24*time.Hour, devs.tickSize)
 }
 
 func TestDevsFork(t *testing.T) {
@@ -310,6 +317,7 @@ func TestDevsSerialize(t *testing.T) {
   people:
   - "one@srcd"
   - "two@srcd"
+  tick_size: 86400
 `, buffer.String())
 
 	buffer = &bytes.Buffer{}
@@ -318,6 +326,7 @@ func TestDevsSerialize(t *testing.T) {
 	msg := pb.DevsAnalysisResults{}
 	assert.Nil(t, proto.Unmarshal(buffer.Bytes(), &msg))
 	assert.Equal(t, msg.DevIndex, devs.reversedPeopleDict)
+	assert.Equal(t, int64(24*time.Hour), msg.TickSize)
 	assert.Len(t, msg.Ticks, 2)
 	assert.Len(t, msg.Ticks[1].Devs, 2)
 	assert.Equal(t, msg.Ticks[1].Devs[0], &pb.DevTick{
@@ -360,6 +369,7 @@ func TestDevsMergeResults(t *testing.T) {
 	r1 := DevsResult{
 		Ticks:              map[int]map[int]*DevTick{},
 		reversedPeopleDict: people1[:],
+		tickSize:           24 * time.Hour,
 	}
 	r1.Ticks[1] = map[int]*DevTick{}
 	r1.Ticks[1][0] = &DevTick{10, ls(20, 30, 40), map[string]items.LineStats{"Go": ls(12, 13, 14)}}
@@ -373,6 +383,7 @@ func TestDevsMergeResults(t *testing.T) {
 	r2 := DevsResult{
 		Ticks:              map[int]map[int]*DevTick{},
 		reversedPeopleDict: people2[:],
+		tickSize:           22 * time.Hour,
 	}
 	r2.Ticks[1] = map[int]*DevTick{}
 	r2.Ticks[1][0] = &DevTick{10, ls(20, 30, 40), map[string]items.LineStats{"Go": ls(12, 13, 14)}}
@@ -387,7 +398,10 @@ func TestDevsMergeResults(t *testing.T) {
 		100, ls(200, 300, 400), map[string]items.LineStats{"Go": ls(62, 63, 64)}}
 
 	devs := fixtureDevs()
-	rm := devs.MergeResults(r1, r2, nil, nil).(DevsResult)
+	c1 := core.CommonAnalysisResult{BeginTime: 1556224895}
+	assert.IsType(t, assert.AnError, devs.MergeResults(r1, r2, &c1, &c1))
+	r2.tickSize = r1.tickSize
+	rm := devs.MergeResults(r1, r2, &c1, &c1).(DevsResult)
 	peoplerm := [...]string{"1@srcd", "2@srcd", "3@srcd"}
 	assert.Equal(t, rm.reversedPeopleDict, peoplerm[:])
 	assert.Len(t, rm.Ticks, 4)
@@ -408,4 +422,29 @@ func TestDevsMergeResults(t *testing.T) {
 		identity.AuthorMissing: {
 			100 * 2, ls(200*2, 300*2, 400*2), map[string]items.LineStats{"Go": ls(94, 96, 98)}},
 	})
+
+	c2 := core.CommonAnalysisResult{BeginTime: 1556224895 + 24*3600}
+	rm = devs.MergeResults(r1, r2, &c1, &c2).(DevsResult)
+	assert.Len(t, rm.Ticks, 5)
+	assert.Equal(t, rm.Ticks[1], map[int]*DevTick{
+		0: {10, ls(20, 30, 40), map[string]items.LineStats{"Go": ls(12, 13, 14)}},
+		1: {1, ls(2, 3, 4), map[string]items.LineStats{"Go": ls(22, 23, 24)}},
+	})
+	assert.Equal(t, rm.Ticks[2], map[int]*DevTick{
+		2: {10, ls(20, 30, 40), map[string]items.LineStats{"Go": ls(12, 13, 14)}},
+		0: {1, ls(2, 3, 4), map[string]items.LineStats{"Go": ls(22, 23, 24)}},
+	})
+	assert.Equal(t, rm.Ticks[3], map[int]*DevTick{
+		2:                      {11, ls(21, 31, 41), map[string]items.LineStats{"Go": ls(32, 33, 34)}},
+		identity.AuthorMissing: {100, ls(200, 300, 400), map[string]items.LineStats{"Go": ls(42, 43, 44)}},
+	})
+	assert.Equal(t, rm.Ticks[10], map[int]*DevTick{
+		0:                      {11, ls(21, 31, 41), map[string]items.LineStats{}},
+		identity.AuthorMissing: {100, ls(200, 300, 400), map[string]items.LineStats{"Go": ls(32, 33, 34)}},
+	})
+	assert.Equal(t, rm.Ticks[11], map[int]*DevTick{
+		1:                      {10, ls(20, 30, 40), map[string]items.LineStats{"Go": ls(42, 43, 44)}},
+		2:                      {11, ls(21, 31, 41), map[string]items.LineStats{"Go": ls(52, 53, 54)}},
+		identity.AuthorMissing: {100, ls(200, 300, 400), map[string]items.LineStats{"Go": ls(62, 63, 64)}},
+	})
 }