import unittest import redis import h5py import time from datetime import timedelta import re import numpy as np from DataCenter import dataCenter from MchDataCenter import mchDataCenter from SpeedDataCenter import speedDataCenter from PIL import Image import json import logging logging.basicConfig(filename='/var/www/html/data/log/qreader.log', format='%(levelname)10s %(asctime)s %(name)10s %(thread)d %(message)s', level=logging.INFO) log = logging.getLogger('starter') class DataTest(unittest.TestCase): # __redis_host = '192.168.3.104' __redis_host = '192.168.3.46' def test_parase(self): try: dataCenter.parase('succ-lingzh-1-4-50-1618184676', '1') except Exception as ex: print(ex) # docker-compose up pythoncli python -m unittest thdf5.DataTest.test_predata def test_predata(self): try: dataCenter.prepare_data() except Exception as ex: print(ex) def test_connect(self): pool = redis.ConnectionPool(host='121.89.223.81', port=57649, db=0) r = redis.Redis(connection_pool=pool) for item in r.hscan_iter('nc_channel_monitor_commit'): key = item[0] val = item[1] print(str(key, encoding="utf-8"), str(val, encoding="utf-8")) def test_regex(self): text = 'succm-lingzh-1-4-30-1618291260' result = re.split(r'-', text) pass def day(self, time): pass def test_today(self): stamp = 1617908681 stamp = int(stamp) x = time.gmtime(stamp + 8 * 3600) diff = timedelta(hours=x.tm_hour, minutes=x.tm_min, seconds=x.tm_sec) today = stamp - diff.total_seconds() y = stamp - today pass def test_days(self): days = dataCenter.days() days = days[0:30] result = json.dumps(days) print(days) def test_chplot(self): # buf = dataCenter.draw_plot(start_time=1619654400, interval=900, chname='yunling', card_type=5, amount=200) buf = dataCenter.draw_plot(start_time=1619654400, interval=900) img = Image.open(buf) img.show() def dir(self, group): result = [] for name, sub in group.items(): if isinstance(sub, h5py.Group): result.extend(self.dir(sub)) else: print(sub.name) result.append(sub.name) return result def test_h5pyr(self): try: file_name = '/var/www/html/data/stdata/data.hdf5' f = h5py.File(file_name, 'r') root = f.require_group('/') result = self.dir(group=root) f.close() except Exception as ex: print(ex) def test_fig(self): buf = dataCenter.draw() img = Image.open(buf) img.show() def test_numpy(self): x = np.arange(0, 86400) y = x.reshape((-1, 300)) y = np.sum(y, axis=1) pos = np.where(x >= 25890) x = x[pos] pos = np.where(x < 85400) x = x[pos] print(x) def test_mchdays(self): dates = mchDataCenter.days() dates.reverse() dates = dates[0:30] def test_mchpaths(self): time_stamp = int(time.time() - 86400) paths = mchDataCenter.paths(time_stamp) print(paths) def test_mchplot(self): time_stamp = int(time.time()) buf = mchDataCenter.draw_plot(start_time=time_stamp, interval=900) img = Image.open(buf) img.show() def test_curmin(self): time_sec = int(time.time()) cur_min = time_sec - time_sec % 60 print('cur_min=', cur_min) def test_ratio(self): dataCenter.set_redis('192.168.1.220','6379') dataCenter.calc_ratio() def test_mchpath(self): time_stamp = int(time.time() - 86400) mchDataCenter.set_redis('192.168.1.220','6379') mchDataCenter.mratios(time_stamp) def test_mratio(self): mchDataCenter.set_redis('192.168.1.220','6379') mchDataCenter.calc_ratio() def test_mratios(self): mchDataCenter.set_redis('192.168.1.220','6379') mchDataCenter.calc_ratios() def test_mechcounts(self): mchDataCenter.set_redis('192.168.1.220','6379') mchDataCenter.mch_counts() def test_pubRatio(self): dataCenter.set_redis('192.168.1.220','6379') dataCenter.pub_ratio() def test_speedReader(self): speedDataCenter.set_redis('192.168.1.220','6379') speedDataCenter.prepare_data() def test_calcSpeed(self): speedDataCenter.set_redis('192.168.1.220','6379') speedDataCenter.channl_speed() def test_profitPrepare(self): from refill import queueListener queueListener.set_redis(self.__redis_host,'6379') queueListener.prepare_data() def test_mamount(self): from refill import MerchantCalc calc = MerchantCalc() calc.set_redis(self.__redis_host,'6379') calc.run() def test_json(self): try: x = {'method': 'commit', 'params': {'inprice': 4.5, 'outprice': 12}} print(x) # x = [ { 'a' : 1, 'b' : 2, 'c' : 3, 'd' : 4, 'e' : 5 } ] data = json.dumps(x) print(data) y = json.loads(data) print(y) except Exception as ex: print(ex) finally: pass def test_setcache(self): r = None try: pool = redis.ConnectionPool(host='192.168.1.220', port=6379, db=0) r = redis.Redis(connection_pool=pool) r.set(f"nc_channel_ratios", "{\"time\": 1624298880, \"ratios\": {\"feinimoshu_hf-100-4-1\": [0, 0, 0, 0], \"feinimoshu_hf-200-4-1\": [0, 0, 0, 0], \"feinimoshu_hf-50-4-1\": [0, 0, 0, 0], \"feinimoshu_hf-100-6-1\": [0, 0, 0, 0], \"gftd-100-2-5\": [0, 0, 38, 77], \"gftd-200-2-5\": [0, 0, 13, 45], \"gftd-500-2-5\": [0, 0, 14, 0], \"gftdsinop-100-2-5\": [6, 0, 0, 0], \"gftdsinop-200-2-5\": [2, 0, 0, 0], \"gftdsinop-500-2-5\": [0, 0, 0, 0], \"lingzh-100-4-1\": [0, 0, 0, 0], \"lingzh-30-4-1\": [6, 0, 0, 6], \"lingzh-50-4-1\": [0, 0, 0, 0], \"lingzh-100-5-1\": [0, 0, 0, 0], \"lingzh-200-5-1\": [0, 0, 0, 0], \"lingzh-30-5-1\": [4, 0, 1, 3], \"lingzh-50-5-1\": [2, 0, 0, 3], \"lingzh-100-6-1\": [0, 0, 0, 0], \"lingzh-200-6-1\": [0, 0, 0, 0], \"lingzh-30-6-1\": [0, 0, 0, 0], \"lingzh-300-6-1\": [0, 0, 0, 0], \"lingzh-50-6-1\": [0, 0, 0, 0], \"lingzh-100-4-5\": [0, 0, 0, 0], \"lingzh-200-4-5\": [0, 0, 0, 0], \"lingzh-100-5-5\": [0, 0, 0, 0], \"lingzh-200-5-5\": [0, 0, 0, 0], \"lingzh-100-6-5\": [0, 0, 0, 0], \"lingzhoil-100-2-1\": [0, 0, 0, 0], \"lingzhoil-200-2-1\": [1, 0, 0, 1], \"moxj-30-5-1\": [5, 0, 2, 1], \"moxj-50-5-1\": [3, 0, 3, 1], \"moxj_fs-50-4-1\": [0, 0, 0, 0], \"qianqian-100-4-1\": [0, 0, 0, 0], \"qianqian-200-4-1\": [0, 0, 0, 0], \"qianqian-30-4-1\": [7, 0, 0, 7], \"qianqian-50-4-1\": [0, 0, 0, 0], \"qianqian-50-5-1\": [0, 0, 0, 0], \"qianqian-100-6-1\": [0, 0, 0, 0], \"weiyiwt-100-4-1\": [0, 0, 0, 0], \"weiyiwt-200-4-1\": [0, 0, 0, 0], \"weiyiwt-30-4-1\": [8, 0, 0, 8], \"weiyiwt-50-4-1\": [0, 0, 0, 0], \"weiyiwt-100-5-1\": [0, 0, 0, 0], \"weiyiwt-100-6-1\": [0, 0, 0, 0], \"weiyiwt-200-6-1\": [0, 0, 0, 0], \"weiyiwt-30-6-1\": [1, 0, 0, 1], \"weiyiwt-50-6-1\": [0, 0, 0, 0], \"yinteng-100-6-1\": [0, 0, 0, 0], \"yinteng-200-6-1\": [0, 0, 0, 0], \"yinteng-30-6-1\": [0, 1, 0, 0], \"yinteng-50-6-1\": [0, 0, 0, 0], \"yunling-100-4-1\": [0, 0, 0, 0], \"yunling-200-4-1\": [0, 0, 0, 0], \"yunling-30-4-1\": [8, 0, 0, 9], \"yunling-50-4-1\": [0, 0, 0, 0], \"yunling-100-5-1\": [0, 0, 0, 0], \"yunling-30-5-1\": [6, 0, 2, 3], \"yunling-50-5-1\": [4, 0, 3, 3], \"yunling-100-6-1\": [0, 0, 0, 0], \"yunling-50-6-1\": [0, 0, 0, 0], \"yunsuoyao-100-4-1\": [0, 0, 0, 0], \"yunsuoyao-30-4-1\": [1, 0, 0, 0], \"yunsuoyao-50-4-1\": [0, 0, 0, 0], \"zanzanquick-30-4-1\": [0, 0, 0, 0], \"zanzanquick-100-5-1\": [0, 0, 0, 0], \"zanzanquick-200-5-1\": [0, 0, 0, 0], \"zanzanquick-30-5-1\": [0, 0, 0, 0], \"zanzanquick-50-5-1\": [0, 0, 0, 0], \"zanzanquick-100-6-1\": [0, 0, 0, 0], \"zanzanquick-200-6-1\": [0, 0, 0, 0], \"zanzanquick-30-6-1\": [0, 0, 0, 0]}}") except Exception as ex: print(ex) if __name__ == '__main__': unittest.main()