Merge pull request #279 from vmarkovtsev/master

Ticks and --devs

Author: vmarkovtsev

Makefile

@@ -25,7 +25,7 @@ internal/pb/pb.pb.go: internal/pb/pb.proto ${GOPATH}/bin/protoc-gen-gogo.exe
 endif
 
 python/labours/pb_pb2.py: internal/pb/pb.proto
-	protoc --python_out python/hercules --proto_path=internal/pb internal/pb/pb.proto
+	protoc --python_out python/labours --proto_path=internal/pb internal/pb/pb.proto
 
 cmd/hercules/plugin_template_source.go: cmd/hercules/plugin.template
 	cd cmd/hercules && go generate

cmd/hercules/combine.go

@@ -59,7 +59,10 @@ var combineCmd = &cobra.Command{
 			bar.Increment()
 			anotherResults, anotherMetadata, errs := loadMessage(fileName, &repos)
 			if anotherMetadata != nil {
-				mergeResults(mergedResults, mergedMetadata, anotherResults, anotherMetadata, only)
+				mergeErrs := mergeResults(mergedResults, mergedMetadata, anotherResults, anotherMetadata, only)
+				for _, err := range mergeErrs {
+					errs = append(errs, err.Error())
+				}
 			}
 			allErrors[fileName] = errs
 			debug.FreeOSMemory()
@@ -173,7 +176,8 @@ func mergeResults(mergedResults map[string]interface{},
 	mergedCommons *hercules.CommonAnalysisResult,
 	anotherResults map[string]interface{},
 	anotherCommons *hercules.CommonAnalysisResult,
-	only string) {
+	only string) []error {
+	var errors []error
 	for key, val := range anotherResults {
 		if only != "" && key != only {
 			continue
@@ -185,13 +189,18 @@ func mergeResults(mergedResults map[string]interface{},
 		}
 		item := hercules.Registry.Summon(key)[0].(hercules.ResultMergeablePipelineItem)
 		mergedResult = item.MergeResults(mergedResult, val, mergedCommons, anotherCommons)
-		mergedResults[key] = mergedResult
+		if err, isErr := mergedResult.(error); isErr {
+			errors = append(errors, fmt.Errorf("could not merge %s: %v", item.Name(), err))
+		} else {
+			mergedResults[key] = mergedResult
+		}
 	}
 	if mergedCommons.CommitsNumber == 0 {
 		*mergedCommons = *anotherCommons
 	} else {
 		mergedCommons.Merge(anotherCommons)
 	}
+	return errors
 }
 
 func getOptionsString() string {

internal/pb/pb.pb.go

@@ -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 {

internal/pb/pb.proto

@@ -123,11 +123,12 @@ func (ticks *TicksSinceStart) Consume(deps map[string]interface{}) (map[string]i
 	if index == 0 {
 		// first iteration - initialize the file objects from the tree
 		// our precision is 1 day
-		*ticks.tick0 = commit.Committer.When
-		if ticks.tick0.Unix() < 631152000 { // 01.01.1990, that was 30 years ago
+		tick0 := commit.Committer.When
+		if tick0.Unix() < 631152000 { // 01.01.1990, that was 30 years ago
 			ticks.l.Warnf("suspicious committer timestamp in %s > %s: %d",
-				ticks.remote, commit.Hash.String(), ticks.tick0.Unix())
+				ticks.remote, commit.Hash.String(), tick0.Unix())
 		}
+		*ticks.tick0 = FloorTime(tick0, ticks.TickSize)
 	}
 
 	tick := int(commit.Committer.When.Sub(*ticks.tick0) / ticks.TickSize)
@@ -163,6 +164,16 @@ func (ticks *TicksSinceStart) Fork(n int) []core.PipelineItem {
 	return core.ForkCopyPipelineItem(ticks, n)
 }
 
+// FloorTime is the missing implementation of time.Time.Floor() - round to the nearest less than or equal.
+func FloorTime(t time.Time, d time.Duration) time.Time {
+	// We have check if the regular rounding resulted in Floor() + d.
+	result := t.Round(d)
+	if result.After(t) {
+		result = result.Add(-d)
+	}
+	return result
+}
+
 func init() {
 	core.Registry.Register(&TicksSinceStart{})
 }

internal/plumbing/ticks.go

@@ -49,6 +49,7 @@ func TestTicksSinceStartRegistration(t *testing.T) {
 
 func TestTicksSinceStartConsume(t *testing.T) {
 	tss := fixtureTicksSinceStart()
+	tss.TickSize = time.Second
 	deps := map[string]interface{}{}
 	commit, _ := test.Repository.CommitObject(plumbing.NewHash(
 		"cce947b98a050c6d356bc6ba95030254914027b1"))
@@ -58,10 +59,25 @@ func TestTicksSinceStartConsume(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, 0, res[DependencyTick].(int))
 	assert.Equal(t, 0, tss.previousTick)
+	assert.Equal(t, 2016, tss.tick0.Year())
+	assert.Equal(t, time.Month(12), tss.tick0.Month())
+	assert.Equal(t, 12, tss.tick0.Day())
 	assert.Equal(t, 18, tss.tick0.Hour())   // 18 UTC+1
 	assert.Equal(t, 30, tss.tick0.Minute()) // 30
 	assert.Equal(t, 29, tss.tick0.Second()) // 29
 
+	tss = fixtureTicksSinceStart()
+	res, err = tss.Consume(deps)
+	assert.Nil(t, err)
+	assert.Equal(t, 0, res[DependencyTick].(int))
+	assert.Equal(t, 0, tss.previousTick)
+	assert.Equal(t, 2016, tss.tick0.Year())
+	assert.Equal(t, time.Month(12), tss.tick0.Month())
+	assert.Equal(t, 12, tss.tick0.Day())
+	assert.Equal(t, 1, tss.tick0.Hour()) // UTC+1
+	assert.Equal(t, 0, tss.tick0.Minute())
+	assert.Equal(t, 0, tss.tick0.Second())
+
 	commit, _ = test.Repository.CommitObject(plumbing.NewHash(
 		"fc9ceecb6dabcb2aab60e8619d972e8d8208a7df"))
 	deps[core.DependencyCommit] = commit
@@ -123,9 +139,9 @@ func TestTicksSinceStartConsumeWithTickSize(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, 0, res[DependencyTick].(int))
 	assert.Equal(t, 0, tss.previousTick)
-	assert.Equal(t, 18, tss.tick0.Hour())   // 18 UTC+1
-	assert.Equal(t, 30, tss.tick0.Minute()) // 30
-	assert.Equal(t, 29, tss.tick0.Second()) // 29
+	assert.Equal(t, 18, tss.tick0.Hour())  // 18 UTC+1
+	assert.Equal(t, 0, tss.tick0.Minute()) // 30
+	assert.Equal(t, 0, tss.tick0.Second()) // 29
 
 	commit, _ = test.Repository.CommitObject(plumbing.NewHash(
 		"fc9ceecb6dabcb2aab60e8619d972e8d8208a7df"))
@@ -160,7 +176,7 @@ func TestTicksCommits(t *testing.T) {
 	tss.commits[0] = []plumbing.Hash{plumbing.NewHash(
 		"cce947b98a050c6d356bc6ba95030254914027b1")}
 	commits := tss.commits
-	tss.Initialize(test.Repository)
+	assert.NoError(t, tss.Initialize(test.Repository))
 	assert.Len(t, tss.commits, 0)
 	assert.Equal(t, tss.commits, commits)
 }

internal/plumbing/ticks_test.go

@@ -133,15 +133,15 @@ type BurndownResult struct {
 	// The rest of the elements are equal the number of line removals by the corresponding
 	// authors in reversedPeopleDict: 2 -> 0, 3 -> 1, etc.
 	PeopleMatrix DenseHistory
-	// The size of each tick.
-	TickSize time.Duration
 
 	// The following members are private.
 
 	// reversedPeopleDict is borrowed from IdentityDetector and becomes available after
 	// Pipeline.Initialize(facts map[string]interface{}). Thus it can be obtained via
 	// facts[FactIdentityDetectorReversedPeopleDict].
 	reversedPeopleDict []string
+	// tickSize references TicksSinceStart.tickSize
+	tickSize time.Duration
 	// sampling and granularity are copied from BurndownAnalysis and stored for service purposes
 	// such as merging several results together.
 	sampling    int
@@ -571,7 +571,7 @@ func (analyser *BurndownAnalysis) Finalize() interface{} {
 		FileOwnership:      fileOwnership,
 		PeopleHistories:    peopleHistories,
 		PeopleMatrix:       peopleMatrix,
-		TickSize:           analyser.tickSize,
+		tickSize:           analyser.tickSize,
 		reversedPeopleDict: analyser.reversedPeopleDict,
 		sampling:           analyser.Sampling,
 		granularity:        analyser.Granularity,
@@ -613,7 +613,7 @@ func (analyser *BurndownAnalysis) Deserialize(pbmessage []byte) (interface{}, er
 		GlobalHistory: convertCSR(msg.Project),
 		FileHistories: map[string]DenseHistory{},
 		FileOwnership: map[string]map[int]int{},
-		TickSize:      time.Duration(msg.GetTickSize()),
+		tickSize:      time.Duration(msg.TickSize),
 
 		granularity: int(msg.Granularity),
 		sampling:    int(msg.Sampling),
@@ -649,17 +649,17 @@ func (analyser *BurndownAnalysis) MergeResults(
 	r1, r2 interface{}, c1, c2 *core.CommonAnalysisResult) interface{} {
 	bar1 := r1.(BurndownResult)
 	bar2 := r2.(BurndownResult)
-	if bar1.TickSize != bar2.TickSize {
+	if bar1.tickSize != bar2.tickSize {
 		return fmt.Errorf("mismatching tick sizes (r1: %d, r2: %d) received",
-			bar1.TickSize, bar2.TickSize)
+			bar1.tickSize, bar2.tickSize)
 	}
 	// for backwards-compatibility, if no tick size is present set to default
-	analyser.tickSize = bar1.TickSize
+	analyser.tickSize = bar1.tickSize
 	if analyser.tickSize == 0 {
 		analyser.tickSize = items.DefaultTicksSinceStartTickSize * time.Hour
 	}
 	merged := BurndownResult{
-		TickSize: analyser.tickSize,
+		tickSize: analyser.tickSize,
 	}
 	if bar1.sampling < bar2.sampling {
 		merged.sampling = bar1.sampling
@@ -683,6 +683,7 @@ func (analyser *BurndownAnalysis) MergeResults(
 				bar1.GlobalHistory, bar2.GlobalHistory,
 				bar1.granularity, bar1.sampling,
 				bar2.granularity, bar2.sampling,
+				bar1.tickSize,
 				c1, c2)
 		}()
 	}
@@ -706,6 +707,7 @@ func (analyser *BurndownAnalysis) MergeResults(
 						m1, m2,
 						bar1.granularity, bar1.sampling,
 						bar2.granularity, bar2.sampling,
+						bar1.tickSize,
 						c1, c2,
 					)
 				}(i)
@@ -755,8 +757,11 @@ func (analyser *BurndownAnalysis) MergeResults(
 	return merged
 }
 
-func (analyser *BurndownAnalysis) roundTime(unix int64, dir bool) int {
-	ticks := float64(unix) / analyser.tickSize.Seconds()
+func roundTime(t time.Time, d time.Duration, dir bool) int {
+	if !dir {
+		t = items.FloorTime(t, d)
+	}
+	ticks := float64(t.Unix()) / d.Seconds()
 	if dir {
 		return int(math.Ceil(ticks))
 	}
@@ -766,7 +771,8 @@ func (analyser *BurndownAnalysis) roundTime(unix int64, dir bool) int {
 // mergeMatrices takes two [number of samples][number of bands] matrices,
 // resamples them to ticks so that they become square, sums and resamples back to the
 // least of (sampling1, sampling2) and (granularity1, granularity2).
-func (analyser *BurndownAnalysis) mergeMatrices(m1, m2 DenseHistory, granularity1, sampling1, granularity2, sampling2 int,
+func (analyser *BurndownAnalysis) mergeMatrices(
+	m1, m2 DenseHistory, granularity1, sampling1, granularity2, sampling2 int, tickSize time.Duration,
 	c1, c2 *core.CommonAnalysisResult) DenseHistory {
 	commonMerged := c1.Copy()
 	commonMerged.Merge(c2)
@@ -783,19 +789,19 @@ func (analyser *BurndownAnalysis) mergeMatrices(m1, m2 DenseHistory, granularity
 		granularity = granularity2
 	}
 
-	size := analyser.roundTime(commonMerged.EndTime, true) -
-		analyser.roundTime(commonMerged.BeginTime, false)
+	size := roundTime(commonMerged.EndTimeAsTime(), tickSize, true) -
+		roundTime(commonMerged.BeginTimeAsTime(), tickSize, false)
 	perTick := make([][]float32, size+granularity)
 	for i := range perTick {
 		perTick[i] = make([]float32, size+sampling)
 	}
 	if len(m1) > 0 {
 		addBurndownMatrix(m1, granularity1, sampling1, perTick,
-			analyser.roundTime(c1.BeginTime, false)-analyser.roundTime(commonMerged.BeginTime, false))
+			roundTime(c1.BeginTimeAsTime(), tickSize, false)-roundTime(commonMerged.BeginTimeAsTime(), tickSize, false))
 	}
 	if len(m2) > 0 {
 		addBurndownMatrix(m2, granularity2, sampling2, perTick,
-			analyser.roundTime(c2.BeginTime, false)-analyser.roundTime(commonMerged.BeginTime, false))
+			roundTime(c2.BeginTimeAsTime(), tickSize, false)-roundTime(commonMerged.BeginTimeAsTime(), tickSize, false))
 	}
 
 	// convert daily to [][]int64
@@ -992,7 +998,7 @@ func addBurndownMatrix(matrix DenseHistory, granularity, sampling int, accPerTic
 func (analyser *BurndownAnalysis) serializeText(result *BurndownResult, writer io.Writer) {
 	fmt.Fprintln(writer, "  granularity:", result.granularity)
 	fmt.Fprintln(writer, "  sampling:", result.sampling)
-	fmt.Fprintln(writer, "  tick_size:", result.TickSize)
+	fmt.Fprintln(writer, "  tick_size:", int(result.tickSize.Seconds()))
 	yaml.PrintMatrix(writer, result.GlobalHistory, 2, "project", true)
 	if len(result.FileHistories) > 0 {
 		fmt.Fprintln(writer, "  files:")
@@ -1045,7 +1051,7 @@ func (analyser *BurndownAnalysis) serializeBinary(result *BurndownResult, writer
 	message := pb.BurndownAnalysisResults{
 		Granularity: int32(result.granularity),
 		Sampling:    int32(result.sampling),
-		TickSize:    int64(result.TickSize),
+		TickSize:    int64(result.tickSize),
 	}
 	if len(result.GlobalHistory) > 0 {
 		message.Project = pb.ToBurndownSparseMatrix(result.GlobalHistory, "project")