xyzshop/plot/testPlot.py

import time
import unittest
import logging

from refill import time_border

logging.basicConfig(filename='/var/www/html/data/log/qreader.log', level=logging.DEBUG)
log = logging.getLogger('reader')


class MyTestCase(unittest.TestCase):
    # __redis_host = '192.168.3.104'
    __redis_host = '192.168.3.46'

    def test_something(self):
        self.assertEqual(True, False)  # add assertion here

    def test_listener(self):
        try:
            from refill import queueListener
            queueListener.set_redis(self.__redis_host, '6379')
            queueListener.prepare_data()
        except Exception as ex:
            log.error(ex)
        pass

    def test_timestart(self):
        stamp = int(time.time())
        for i in range(300):
            x = stamp + i
            l = time_border(300, x, True)
            r = time_border(300, x, False)
            print('l=', l, 'r=', r)

    def testChannel(self):
        from refill import ChannelReader
        reader = ChannelReader()
        days = reader.days()
        print('days=', days)

        for day in days:
            x = reader.tuple_path(day, card_types={4, 5, 6})
            x1 = reader.tuple_path(day, {'chizeng', 'ainika'}, {4, 5, 6})
            x2 = reader.tuple_path(day, {'chizeng', 'ainika'}, {4, 5, 6}, 50)

    def test_chpainter(self):
        from refill import ChannelCumPainter

        start_time = int(time.time()) - 10 * 86400 - 3600
        end_time = int(time.time()) - 8 * 86400
        painter = ChannelCumPainter(start_time=start_time, end_time=end_time, chnames=set(), card_types={4, 5, 6})
        painter.paint()

    def test_chcov(self):
        from refill import ChannelCovPainter

        start_time = int(time.time()) - 10 * 86400 - 3600
        end_time = int(time.time()) - 10 * 86400
        painter = ChannelCovPainter(start_time=start_time, end_time=end_time, chnames=set(), card_types={4, 5, 6},filter_wave=3600)
        painter.paint()

    def testDays(self):
        from refill import MerchantReader
        from refill import ChannelReader

        try:
            chreader = ChannelReader()
            chdays = chreader.days()
            xlables = [time.strftime('%d-%H:%M:%S', time.localtime(d)) for d in chdays]
            print(xlables)

            reader = MerchantReader()
            days = reader.days()
        except Exception as ex:
            log.error(ex)
        pass

    def test_jsonLoads(self):
        import json
        try:
            str = 4
            x = json.loads(str)
            print(x)
        except Exception as ex:
            print(ex)

    def test_partial(self):
        from functools import partial
        add_five = partial()

    def test_key(self):
        def person(name, age, *, city, job):
            print(name, age, city, job)

        person('Jack', 24, city='Beijing', job='Engineer')
        person('Jack', 24, 'Beijing', job='Engineer')
        person('Jack', 24, 'Beijing', 'Engineer')

    def test_none(self):
        x = None
        len = len(x)
        print(len)

    def test_set(self):
        x = set([(4, 50), (4, 100)])
        y = set([(4, 50), (5, 100)])
        z = x | y
        print(z)

    def test_env(self):
        import os
        x = os.getenv("PYCHARM_DISPLAY_PORT", "-1")
        print(x)

    def test_matplot(self):
        import matplotlib
        x = matplotlib.__version__
        print(x)

    def test_time(self):
        x = int(time.time())
        print(x)

    def test_rpop(self):
        import redis
        import json

        pool = redis.ConnectionPool(host=self.__redis_host, port=6379, db=0)
        r = redis.Redis(connection_pool=pool)
        item = r.rpop('REFILL_MONITOR_QUEUE')

        if item is None:
            print('hello')
        else:
            try:
                val = json.loads(item)
                method = val['method']
                params = val['params']
                print(method, params)
            except Exception as ex:
                log.error(ex)

    def test_consumer(self):
        from refill import WriterConsumer

        class PrintHandler:
            def write(self, method, msg):
                log.debug(msg)
                time.sleep(0.01)

        handler = PrintHandler()
        consumer = WriterConsumer(handler,'PrintHandler')
        consumer.start()

        for i in range(100000):
            consumer.put('test', f'index = {i}')
        consumer.quit()
        consumer.join()

    def test_merge(self):
        from collections import defaultdict
        import time as time

        def merge(l,r):
            for name,ls in l.items():
                if name in r:
                    ls.extend(r[name])
                    k = set(ls)
                    r[name] = list(k)
                else:
                    r[name] = ls
            return r

        all = defaultdict(list)
        a = {'yunchonggongfs': [(4, 100), (4, 30), (4, 50)]}
        b = {'yunchonggongfs': [(4, 100), (4, 200), (4, 50)]}

        all = merge(a,all)
        all = merge(b,all)
        x = 0;




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