add channel money consume monitor

plot/ch_consume.py

@@ -0,0 +1,19 @@
+from refill import opt_parse, ChSpeedControlCalc
+import signal as sig
+import sys,getopt
+
+import logging
+logging.basicConfig(filename='/var/www/html/data/log/speed.log',
+                    format='%(levelname)10s  %(asctime)s  %(name)10s %(thread)d %(message)s',
+                    level=logging.DEBUG)
+logger = logging.getLogger('channel_speed_ratio')
+
+if __name__ == '__main__':
+    try:
+        rhost, rport = opt_parse()
+        calc = ChSpeedControlCalc()
+        calc.set_redis(rhost, rport)
+        sig.signal(sig.SIGINT, lambda: calc.stop())
+        calc.run()
+    except Exception as ex:
+        logger.error(ex)

plot/refill/ChConsumeCalc.py

@@ -0,0 +1,46 @@
+from .ChannelCalc import ChannelCalc, detail_pathes
+from .algorithm import calc_chspeed, calc_chratio, calc_commit, calc_unback_time
+from .DataStream import EChPosmap as pos_map
+from .DataStream import ch_calc_cfgs, mktime
+from .ChannelReader import ChannelReader
+import time as time
+import json
+import redis
+
+import logging
+
+logger = logging.getLogger('ChSpeedCalc')
+
+
+class ChConsumeCalc(ChannelCalc):
+    def _calc_handler(self, rclient):
+        logger.debug('_calc_handler')
+        start_period, ratio_period, speed_period, monitor_period, cdf_speed_period = ch_calc_cfgs()
+
+        reader = self._reader()
+        end_time = int(time.time())
+        # end_time = mktime('2022-11-01 11:10:00')
+        days, start_time, end_time = self.calc_time(reader, end_time - start_period, end_time)
+
+        day_stamp = days[0]
+        tuple_pathes = reader.many_tuple_path(days, card_types=set([1, 2, 4, 5, 6, 7]))
+        gen = detail_pathes(reader, tuple_pathes, days)
+
+        start_pos = start_time - day_stamp
+        end_pos = end_time - day_stamp
+
+        mins = speed_period / 60
+        result = dict()
+        for _name, _card_type, _spec, _data in gen:
+            speed = calc_chspeed(_data, pos_map, end_pos - speed_period, end_pos) / mins
+            speed = int(speed)
+            ratio, commit, notify_time, succ_time = calc_chratio(_data, pos_map, end_pos - ratio_period, end_pos)
+            key = f'{_name}-{_spec}-{_card_type}'
+            result[key] = [speed, ratio, commit, notify_time, 0]
+            logger.debug("%s-%d-%d speed=%d ratio=%.5f commit=%d notify_time=%.5f succ_time=%.5f",
+                         _name, _card_type, _spec, speed, ratio, commit, notify_time, succ_time)
+        if len(result) != 0:
+            rclient.set(f"nc_refill_channel_control_model", json.dumps(result))
+            rclient.publish('refill', json.dumps({'type': 'channel_control', 'value': 0}))
+
+        return 60