kpi: Establish data model for DRB.UEActive

Similarly to DRB.IPLatDl 3GPP says to use average number of active UEs
during a period for DRB.UEActive. Which means that we'll also need to
use statistical profile for this value to be able to aggregate
measurements. But contrary to DRB.IPLatDl, DRB.UEActive should be
statistical profile with time-based sampling because the averaging
happened over time, not over arbitrary general samples.

-> Introduce StatT type, that is similar to Stat, to represent such
   time-based profiling and use it to define DRB.UEActive; Teach
   Calc.aggregate to handle aggregation of such time-based statistical
   profiles via

	    a₁⋅δt₁ + a₂·δt₂
	A = ───────────────
	       δt₁ + δt₂

   formula.

This patch is partly based on the following patch by Paul Graydon:
paul.graydon/xlte@eb9f1fa9

kpi.py

@@ -137,6 +137,24 @@ class Stat(np.void):
             ('max', dtype),
             ('n',   np.int64)]))
 
+# StatT[dtype] represents result of statistical profiling with time-based sampling
+# for a value with specified dtype.
+#
+# It is organized as NumPy structured scalar with avg, min and max fields.
+#
+# NOTE contrary to Stat there is no n field and containing Measurement.X.δT
+#      should be taken to know during which time period the profile was collected.
+#
+# It is used inside Measurement for e.g. DRB.UEActive .
+class StatT(np.void):
+    # _dtype_for returns dtype that StatT[dtype] will use.
+    @classmethod
+    def _dtype_for(cls, dtype):
+        return np.dtype((cls, [
+            ('avg', np.float64),    # see avg note in Stat
+            ('min', dtype),
+            ('max', dtype)]))
+
 
 # Measurement represents set of measured values and events observed and counted
 # during one particular period of time.
@@ -175,6 +193,7 @@ class Measurement(np.void):
     Tcc    = np.int32   # cumulative counter
     Ttime  = np.float64 # time is represented in seconds since epoch
     S  = Stat ._dtype_for   # statistical profile with arbitrary sampling
+    St = StatT._dtype_for   # statistical profile with time-based sampling
 
     # _dtype defines measured values and events.
     _dtype = np.dtype([
@@ -200,6 +219,8 @@ class Measurement(np.void):
         ('DRB.PdcpSduBitrateUl.QCI',        np.float64),# bit/s     4.4.1.1                 NOTE not kbit/s
         ('DRB.PdcpSduBitrateDl.QCI',        np.float64),# bit/s     4.4.1.2                 NOTE not kbit/s
 
+        ('DRB.UEActive',                 St(np.int32)), # 1         4.4.2.4  36.314:4.1.3.3
+
         ('DRB.IPLatDl.QCI',               S(Ttime)),    # s         4.4.5.1  32.450:6.3.2   NOTE not ms
 
         # DRB.IPThpX.QCI = DRB.IPVolX.QCI / DRB.IPTimeX.QCI         4.4.6.1-2 32.450:6.3.1
@@ -225,7 +246,7 @@ class Measurement(np.void):
         ('PEE.Energy',                      np.float64),# J         4.12.2                  NOTE not kWh
     ])
 
-    del S
+    del S, St
 
 
 # Interval is NumPy structured scalar that represents [lo,hi) interval.
@@ -304,6 +325,15 @@ def __new__(cls, min, avg, max, n, dtype=np.float64):
     s['n']   = n
     return s
 
+# StatT() creates new StatT instance with specified values and dtype.
+@func(StatT)
+def __new__(cls, min, avg, max, dtype=np.float64):
+    s = _newscalar(cls, cls._dtype_for(dtype))
+    s['min'] = min
+    s['avg'] = avg
+    s['max'] = max
+    return s
+
 
 # _all_qci expands <name>.QCI into <name>.sum and [] of <name>.<qci> for all possible qci values.
 # TODO remove and use direct array access (after causes are expanded into array too)
@@ -405,12 +435,25 @@ def __repr__(s):
     return "Stat(%s, %s, %s, %s, dtype=%s)" % (_vstr(s['min']), _vstr(s['avg']),
                 _vstr(s['max']), _vstr(s['n']), s['min'].dtype)
 
+# __repr__ returns StatT(min, avg, max, dtype=...)
+# NA values are represented as "ø".
+@func(StatT)
+def __repr__(s):
+    return "StatT(%s, %s, %s, dtype=%s)" % (_vstr(s['min']), _vstr(s['avg']),
+                _vstr(s['max']), s['min'].dtype)
+
 # __str__ returns "<min avg max>·n"
 # NA values are represented as "ø".
 @func(Stat)
 def __str__(s):
     return "<%s %s %s>·%s" % (_vstr(s['min']), _vstr(s['avg']), _vstr(s['max']), _vstr(s['n']))
 
+# __str__ returns "<min avg max>"
+# NA values are represented as "ø".
+@func(StatT)
+def __str__(s):
+    return "<%s %s %s>" % (_vstr(s['min']), _vstr(s['avg']), _vstr(s['max']))
+
 
 # _vstr returns string representation of scalar or subarray v.
 def _vstr(v):  # -> str
@@ -805,6 +848,13 @@ def aggregate(calc): # -> ΣMeasurement
             if isinstance(v, np.number):
                 Σf['value'] += v
 
+            elif isinstance(v, StatT):
+                Σv['min'] = xmin(Σv['min'], v['min'])
+                Σv['max'] = xmax(Σv['max'], v['max'])
+                # TODO better sum everything and then divide as a whole to avoid loss of precision
+                Σv['avg'], _ = xavg(Σv['avg'], m['X.Tstart'] - Σ['X.Tstart'] - Σf['τ_na'],
+                                     v['avg'], m['X.δT'])
+
             elif isinstance(v, Stat):
                 Σv['min'] = xmin(Σv['min'], v['min'])
                 Σv['max'] = xmax(Σv['max'], v['max'])

kpi_test.py

@@ -21,7 +21,7 @@
 from __future__ import print_function, division, absolute_import
 
 from xlte.kpi import Calc, MeasurementLog, Measurement, ΣMeasurement, Interval, \
-                     Stat, NA, isNA, Σqci, Σcause, nqci
+                     Stat, StatT, NA, isNA, Σqci, Σcause, nqci
 import numpy as np
 from pytest import raises
 
@@ -58,6 +58,12 @@ def test_Measurement():
     assert m['S1SIG.ConnEstabAtt'] == 123
     m['RRC.ConnEstabAtt.sum'] = 17
     assert m['RRC.ConnEstabAtt.sum'] == 17
+    m['DRB.UEActive']['min'] = 1
+    m['DRB.UEActive']['avg'] = 6
+    m['DRB.UEActive']['max'] = 8
+    assert m['DRB.UEActive']['min'] == 1
+    assert m['DRB.UEActive']['avg'] == 6
+    assert m['DRB.UEActive']['max'] == 8
     m['DRB.IPVolDl.QCI'][:] = 0
     m['DRB.IPVolDl.5'] = 55
     m['DRB.IPVolDl.7'] = NA(m['DRB.IPVolDl.7'].dtype)
@@ -83,7 +89,8 @@ def test_Measurement():
 
 
     # str/repr
-    assert repr(m) == "Measurement(RRC.ConnEstabAtt.sum=17, DRB.IPLatDl.QCI={3:<0.0 33.0 0.0>·123 4:<0.0 44.0 0.0>·432 8:<0.0 ø 0.0>·ø}, DRB.IPVolDl.QCI={5:55 7:ø 9:99}, S1SIG.ConnEstabAtt=123)"
+    assert repr(m) == "Measurement(RRC.ConnEstabAtt.sum=17, DRB.UEActive=<1 6.0 8>, DRB.IPLatDl.QCI={3:<0.0 33.0 0.0>·123 4:<0.0 44.0 0.0>·432 8:<0.0 ø 0.0>·ø}, DRB.IPVolDl.QCI={5:55 7:ø 9:99}, S1SIG.ConnEstabAtt=123)"
+    assert repr(m['DRB.UEActive'])  == "StatT(1, 6.0, 8, dtype=int32)"
     assert repr(m['DRB.IPLatDl.3']) == "Stat(0.0, 33.0, 0.0, 123, dtype=float64)"
     s = str(m)
     assert s[0]  == '('
@@ -91,6 +98,7 @@ def test_Measurement():
     v = s[1:-1].split(', ')
     vok = ['ø'] * len(m._dtype0.names)
     vok[m.dtype.names.index("RRC.ConnEstabAtt.sum")]   = "17"
+    vok[m.dtype.names.index("DRB.UEActive")]           = "<1 6.0 8>"
     vok[m.dtype.names.index("S1SIG.ConnEstabAtt")]     = "123"
     vok[m.dtype.names.index("DRB.IPVolDl.QCI")]        = "{5:55 7:ø 9:99}"
     vok[m.dtype.names.index("DRB.IPLatDl.QCI")]        = "{3:<0.0 33.0 0.0>·123 4:<0.0 44.0 0.0>·432 8:<0.0 ø 0.0>·ø}"
@@ -523,6 +531,7 @@ def test_Calc_aggregate():
     m1['X.δT']      = 2
     m1['S1SIG.ConnEstabAtt'] = 12                               # Tcc
     m1['ERAB.SessionTimeUE'] = 1.2                              # Ttime
+    m1['DRB.UEActive']       = StatT(1, 3.7, 5)                 # StatT
     m1['DRB.IPLatDl.7']      = Stat(4*ms, 7.32*ms, 25*ms, 17)   # Stat
 
     m2 = Measurement()
@@ -530,6 +539,7 @@ def test_Calc_aggregate():
     m2['X.δT']      = 3
     m2['S1SIG.ConnEstabAtt'] = 11
     m2['ERAB.SessionTimeUE'] = 0.7
+    m2['DRB.UEActive']       = StatT(2, 3.2, 7)
     m2['DRB.IPLatDl.7']      = Stat(3*ms, 5.23*ms, 11*ms, 11)
 
     mlog.append(m1)
@@ -551,6 +561,9 @@ def test_Calc_aggregate():
     assert M['ERAB.SessionTimeUE']['value'] == 1.2 + 0.7
     assert M['ERAB.SessionTimeUE']['τ_na']  == 5
 
+    assert M['DRB.UEActive']['value']   == StatT(1, (3.7*2 + 3.2*3)/(2+3), 7)
+    assert M['DRB.UEActive']['τ_na']    == 5
+
     assert M['DRB.IPLatDl.7']['value']  == Stat(3*ms, (7.32*17 + 5.23*11)/(17+11)*ms, 25*ms, 17+11)
     assert M['DRB.IPLatDl.7']['τ_na']   == 5
 
@@ -564,7 +577,7 @@ def test_Calc_aggregate():
         assert f['τ_na'] == 10
     for name in M.dtype.names:
         if name not in ('X.Tstart', 'X.δT', 'S1SIG.ConnEstabAtt',
-                        'ERAB.SessionTimeUE', 'DRB.IPLatDl.7'):
+                        'ERAB.SessionTimeUE', 'DRB.UEActive', 'DRB.IPLatDl.7'):
             _(name)