summaryrefslogtreecommitdiff
path: root/src/daq/nginx-lua/s5-nginx-lua-fetch.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/daq/nginx-lua/s5-nginx-lua-fetch.py')
-rwxr-xr-xsrc/daq/nginx-lua/s5-nginx-lua-fetch.py97
1 files changed, 82 insertions, 15 deletions
diff --git a/src/daq/nginx-lua/s5-nginx-lua-fetch.py b/src/daq/nginx-lua/s5-nginx-lua-fetch.py
index f5721d3..916bb3f 100755
--- a/src/daq/nginx-lua/s5-nginx-lua-fetch.py
+++ b/src/daq/nginx-lua/s5-nginx-lua-fetch.py
@@ -162,13 +162,14 @@ class NGXLuaFetcher():
self._socket = properties['socket']
self._hostname = properties['hostname']
- self._content_id = properties['content-id']
- self._format = properties['format']
- self._quality = properties['quality']
+ self._content_id = None
+ self._format = None
+ self._quality = None
self._duration = properties['duration']
self._tags = properties['tags']
self._url = properties['url']
self._nameformat = properties['nameformat']
+ self._initStreamerIds(properties['streamer-ids'])
self._proto = None
self._conn = None
@@ -177,13 +178,54 @@ class NGXLuaFetcher():
self._looper = None
self._start_time = None
+ def _initStreamerIds(self, streamer):
+ print 'SFive: will look for the following streamer ids:'
+ self._streamer = {}
+ cs = {}
+ fs = {}
+ qs = {}
+ for s in streamer:
+ parts = s.split('/')
+ if len(parts) != 3:
+ raise ValueError('invalid streamer descriptipn "%s": must consist of 3 parts seperated by a /' % s)
+ scs = parts[0].split(',')
+ sfs = parts[1].split(',')
+ sqs = parts[2].split(',')
+ for c in scs:
+ cs[c] = 1
+ if c not in self._streamer:
+ self._streamer[c] = {}
+ for f in sfs:
+ fs[f] = 1
+ if f not in self._streamer[c]:
+ self._streamer[c][f] = {}
+ for q in sqs:
+ qs[q] = 1
+ if q not in self._streamer[c][f]:
+ self._streamer[c][f][q] = ClientList()
+ print ' %s / %s / %s' % (c, f, q)
+
+ if len(cs.keys()) == 1:
+ self._content_id = cs.keys()[0]
+ print 'SFive: only one content-id detected "%s" - will include it in init messages only' % self._content_id
+
+ if len(fs.keys()) == 1:
+ self._format = fs.keys()[0]
+ print 'SFive: only one format detected "%s" - will include it in init messages only' % self._format
+
+ if len(qs.keys()) == 1:
+ self._quality = qs.keys()[0]
+ print 'SFive: only one quality detected "%s" - will include it in init messages only' % self._quality
+
+
+
def run(self):
if self._initFetch():
# try to be aligned with current time
# this will eventually get out of sync but for now this is good enough
offset = self._duration - (time.time() % self._duration)
print 'SFive: %sZ -> will wait %0.2f seconds before starting looper (alignment)' % (datetime.datetime.utcnow().isoformat('T'), offset)
- self._looper = task.LoopingCall(self._sendUpdate)
+ self._looper = task.LoopingCall(self._sendUpdates)
reactor.callLater(offset, self._startLooper)
reactor.callWhenRunning(self._initSocket)
reactor.run()
@@ -193,9 +235,12 @@ class NGXLuaFetcher():
self._looper.start(self._duration, False)
print 'SFive: looper started at %sZ' % (self._start_time.isoformat('T'))
- def _sendUpdate(self):
+ def _sendUpdates(self):
if self._connected:
- self._sendDataset(self._start_time, self._duration, ClientList())
+ for c in self._streamer.keys():
+ for f in self._streamer[c].keys():
+ for q in self._streamer[c][f].keys():
+ self._sendDataset(self._start_time, self._duration, c, f, q)
else:
print 'SFive: not connected - just clearing stats'
@@ -281,19 +326,39 @@ class NGXLuaFetcher():
def _sendInit(self):
initdata = { "version": 1, "hostname": self._hostname,
- "streamer-id": { "content-id": self._content_id, "format": self._format, "quality": self._quality },
+ "streamer-id" : { },
"tags": self._tags }
+ if self._content_id:
+ initdata["streamer-id"]["content-id"] = self._content_id
+ if self._format:
+ initdata["streamer-id"]["format"] = self._format
+ if self._quality:
+ initdata["streamer-id"]["quality"] = self._quality
+ if len(initdata["streamer-id"].keys()) == 0:
+ del initdata["streamer-id"]
+
self._proto.sendDatagram('%s\n' % (json.dumps(initdata)));
- def _sendDataset(self, timestamp, duration, clients):
+ def _sendDataset(self, timestamp, duration, content_id, format, quality):
+ clients = self._streamer[content_id][format][quality]
data = { "start-time": timestamp.isoformat('T') + 'Z',
"duration-ms": duration * 1000,
+ "streamer-id": { },
"data": {
"clients": list(clients.values()),
"client-count": clients.getCnt(),
"bytes-sent": clients.getBytesSent()
}
}
+ if not self._content_id:
+ data["streamer-id"]["content-id"] = content_id
+ if not self._format:
+ data["streamer-id"]["format"] = format
+ if not self._quality:
+ data["streamer-id"]["quality"] = quality
+ if len(data["streamer-id"].keys()) == 0:
+ del data["streamer-id"]
+
self._proto.sendDatagram('%s\n' % (json.dumps(data)));
@@ -309,24 +374,26 @@ if __name__ == '__main__':
help='time (in seconds) between updates; defaults to 15')
parser.add_argument('--tag', '-t', dest='tags', action='append',
help='tag to be added to the statistic data, can be invoked several times')
- parser.add_argument('--content-id', '-c', dest='content-id', required=False,
- help='the content-id (i.e. av) - only used if nameformat doesn\'t contain a named match')
- parser.add_argument('--format', '-f', dest='format', required=False,
- help='the format (i.e. webm) - only used if nameformat doesn\'t contain a named match')
- parser.add_argument('--quality', '-q', dest='quality', required=False,
- help='the quality (i.e. high) - only used if nameformat doesn\'t contain a named match')
+ parser.add_argument('--streamer-id', '-S', dest='streamer-ids', action='append',
+ help='a streamer description like <content-id1>[,<content-id2>]/<format1>[,<format2>]/<quality>[,<quality>[,<quality]], can be invoked several times')
parser.add_argument('--url', '-u', dest='url', required=False,
help='the url of the nginx sfive exporert, default http://localhost/sfive')
parser.add_argument('--nameformat', '-F', dest='nameformat', required=False,
- help='a regular expression (containing named matches for content, format, quality)')
+ help='a regular expression (must contain named matches for content, format, quality)')
args = vars(parser.parse_args())
if not args['tags']:
args['tags'] = []
if not args['duration']:
args['duration'] = 15
+ else:
+ args['duration'] = int(args['duration'])
if not args['url']:
args['url'] = 'http://localhost/sfive'
if not args['nameformat']:
args['nameformat'] = '/[^/]+/(?P<format>hls|dash)/(?P<content>.+)-(?P<quality>[^-]+)/.*'
+ if not args['streamer-ids']:
+ print 'SFive: you have to specify at least one streamer-id!'
+ exit(-1)
+
fetcher = NGXLuaFetcher(args)
fetcher.run()