xyzshop/plot/thdf5.py

import unittest
import redis
import h5py
import time
from datetime import datetime
from datetime import timedelta
import re
import threading
import numpy as np
from DataCenter import dataCenter
from matplotlib.figure import Figure
from PIL import Image
from io import BytesIO
from PIL import Image
import json


class DataTest(unittest.TestCase):
    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()
        result = json.dumps(days)
        print(days)

    def test_data(self):
        buf = dataCenter.draw_plot(1619020800, chname='lingzh')
        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)


if __name__ == '__main__':
    unittest.main()