Server-py/sri-pysdk/xclient/xredis.py

# update: 2021-7-14-11
import redis
import pickle


class Client(redis.StrictRedis):

    def __init__(self, host='fra-middleware-redis', port=6379, db=0, password='', channels=[]):
        """
        内部访问:
            host: fra-middleware-redis
            port: 6379
        远程访问:
            host: 118.190.217.96、192.168.20.162
            port: 7070
        """
        self.host = host
        self.port = port
        self.db = db
        super().__init__(host=self.host, port=self.port, db=self.db)

        self.channels = channels
        if self.channels:
            """
            dir(self.pubsub()):
            ['HEALTH_CHECK_MESSAGE', 'PUBLISH_MESSAGE_TYPES', 'UNSUBSCRIBE_MESSAGE_TYPES', 'channels', 'check_health', 
            'close', 'connection', 'connection_pool', 'encoder', 'execute_command', 'get_message', 'handle_message', 
            'health_check_response', 'ignore_subscribe_messages', 'listen', 'on_connect', 'parse_response', 'patterns', 
            'pending_unsubscribe_channels', 'pending_unsubscribe_patterns', 'ping', 'psubscribe', 'punsubscribe', 
            'reset', 'run_in_thread', 'shard_hint', 'subscribe', 'subscribed', 'unsubscribe']
            """
            self.channel_pubsub = self.pubsub()
            self.channel_pubsub.psubscribe(channels)
            # self.channel_listen = self.channel_pubsub.listen()

    def get_config(self):
        """
        CONFIG GET|SET: 分别用 config_get 和 config_set 实现。
        """
        return self.config_get()

    def set_config(self, name, value):
        """
        CONFIG GET|SET: 分别用 config_get 和 config_set 实现。
        """
        return self.config_set(name=name, value=value)

    def get_elements(self, key):
        """获取列表元素"""
        return self.lrange(key, 0, -1)

    def get_elements_length(self, key):
        """获取列表长度"""
        return self.llen(key)

    def pop_element(self, key, position='RightEnd'):
        """
        弹出元素
        position: 出队位置 LeftEnd 左端 RightEnd 右端
        """
        _dict = {
            'LeftEnd': 'lpop',
            'RightEnd': 'rpop',
        }
        func = getattr(self, _dict.get(position))
        pickle_data = func(key)
        element = pickle.loads(pickle_data) if pickle_data else None
        return element

    def add_element(self, key, element, position='RightEnd'):
        """
        增加元素
        position: 入队位置 LeftEnd 左端 RightEnd 右端
        """
        _dict = {
            'LeftEnd': 'lpush',
            'RightEnd': 'rpush',
        }
        func = getattr(self, _dict.get(position))
        pickle_data = pickle.dumps(element)
        elements_length = func(key, pickle_data)
        return elements_length

    def set_one(self, key, data, expire_time=None):
        """压缩存储"""
        pickle_data = pickle.dumps(data)
        return self.set(key, pickle_data, ex=expire_time)

    def get_one(self, key):
        """解压获取"""
        pickle_data = self.get(key)
        return pickle.loads(pickle_data) if pickle_data else None

    def push_one(self, channel, data):
        """
        推送一条
        """
        return self.publish(channel, pickle.dumps(data))

    def pull_one(self, channel):
        """
        拉取一条
        msg = self.channel_pubsub.parse_response(block=False, timeout=60)
        msg = self.channel_pubsub.parse_response(block=False)
            [b'pmessage', b'cctv1', b'cctv1', b'3']
        msg = self.channel_pubsub.get_message()
            {'type': 'pmessage', 'pattern': b'cctv1', 'channel': b'cctv1', 'data': b'asdfasdfasdf'}
        """
        message = self.channel_pubsub.get_message()
        channel = channel.encode()
        if message and message.get('pattern') and message.get('channel') == channel:
            return pickle.loads(message.get('data'))

    def pull_one_by_channel(self, channel):
        """
        拉取一条
        """
        message = self.channel_pubsub.get_message()
        channel = channel.encode()
        if message and message.get('pattern') and message.get('channel') == channel:
            return pickle.loads(message.get('data'))

    def pull_one_by_channels(self):
        """
        拉取一条
        """
        message = self.channel_pubsub.get_message()
        channels = [i.encode() for i in self.channels]
        if message and message.get('pattern') and message.get('channel') in channels:
            return pickle.loads(message.get('data'))

    def pull_one_by_channels_v2(self, channels):
        """
        拉取一条
        """
        message = self.channel_pubsub.get_message()
        channels = [i.encode() for i in channels]
        if message and message.get('pattern') and message.get('channel') in channels:
            return pickle.loads(message.get('data'))

    def pull_one_by_channels_v3(self, channels):
        """
        拉取一条（性能不佳）
        """
        self.channel_pubsub.psubscribe(channels)
        message = self.channel_pubsub.get_message()
        channels = [i.encode() for i in channels]
        if message and message.get('pattern') and message.get('channel') in channels:
            return pickle.loads(message.get('data'))

    def print_all(self):
        """打印"""
        for key in self.keys():
            key_type = self.type(key)

            # if key_type != b"stream":
            #     continue

            if key_type == b'string':
                value = self.get(key)
            elif key_type == b'list':
                value = self.lrange(key, 0, -1)
            elif key_type == b'set':
                value = self.smembers(key)
            elif key_type == b'zset':
                value = self.zrange(key, 0, -1)
            elif key_type == b"hash":
                value = self.hgetall(key)
            elif key_type == b"stream":
                value = self.xread({key: b"0-0"})
            else:
                value = None

            print(f"type: {key_type} | key: {key}")
            # print(f"type: {key_type} | key: {key} | value: {value}")

    def filter_keys(self):
        re_list = []
        for key in self.keys():
            if self.type(key) != b'string':
                continue
            if b'warn' not in key:
                continue
            re_list.append(key)
        return re_list

    def test(self):
        while True:
            msg = self.pull()
            if msg:
                print(msg)


if __name__ == '__main__':
    # --- init ---
    # db0 = Client(host='192.168.30.49', port=7070, channels=['cctv1', 'cctv2'])
    db0 = Client(host='192.168.30.59', port=7070)
    # db0 = Client(host='192.168.1.190', port=7070)
    # db0 = Client(host='fra-middleware-redis', port=6379)

    # --- test ---
    out = db0.get_elements_length('tasklist1')
    print(out)

    # --- test --
    # db0.set_config('client-output-buffer-limit', 'normal 0 0 0 slave 268435456 67108864 60 pubsub 0 0 0')
    # out = db0.get_config()
    # print(out)

    # --- test ---
    # db0.push_one('cctv1', 'asdfasdfasdf')

    # --- test ---
    # db0.set_object('compare_face_task', [1, 2, 3])

    # --- test ---
    # out = db0.get_object('compare_face_task')
    # print(out)

    # --- test ---
    # db0.hset(name='test001', key='test002', value='test003')
    # out = db0.hgetall(name='test001')
    # print(out)

    # --- test ---
    # db0.print_all()

    # --- test ---
    # db0.flushdb()

    # --- test ---
    # import pickle
    # out = db0.get('real_cam')
    # out = pickle.loads(out)
    # print(out)

    # --- test ---
    # import pickle
    # out = db0.get('webcam')
    # out = pickle.loads(out)
    # print(out)

    # --- test ---
    # import pickle
    # config = {
    #     'aass': 12,
    #     '1': {
    #        '2': {
    #        }
    #     }
    # }
    #
    # config = pickle.dumps(config)
    # db0.set('webcam', config, ex=1000)
    # out = db0.get('webcam')
    # out = pickle.loads(out)
    # print(out)

    # --- test ---
    # out = db0.filter_keys()
    # print(out)
    # out = db0.get(out[0])
    # print(out)

    # --- test ---
    # db0.delete('warn:192.168.1.68:1604928475')
    # db0.delete('warn:192.168.1.68:1604928695')
    # db0.delete('warn:192.168.1.68:1604928479')
    # db0.delete('warn:192.168.1.68:1604928680')
    # db0.delete('warn:192.168.1.68:1604928491')
    # db0.print_all()

    # --- test ---
    # import pickle
    # data = pickle.dumps(0)
    # db0.set('warn:192.168.1.68:<last_at>', data, ex=1000*86400)
    # out = db0.get(f'warn:192.168.1.68:<last_at>')
    # print(out)

    # --- test ---
    # import pickle
    # data = {
    #     'last_at': 112233,
    #     'image': pickle.dumps({}),
    # }
    # db0.hmset(name='192.168.1.68:admin:HuaWei123:<find_at>', mapping=data)
    # out = db0.hmget(name='192.168.1.68:admin:HuaWei123:<find_at>', keys='last_at')
    # print(out)
    # out = db0.hget(name='192.168.1.68:admin:HuaWei123:<find_at>', key='last_at')
    # print(out)

    # --- test ---
    # db0.set('food', b'mutton', ex=1000)
    # print(type(db0.get('food')))
    # print(repr(db0.get('food')))
    # db0.print_all()

    # --- test ---
    # data = db0.get('192.168.1.68:admin:HuaWei123')
    # import numpy as np
    # data = np.frombuffer(data)
    # print(repr(data))

    # --- test ---
    # data = db0.get('192.168.1.68:admin:HuaWei123')
    # import pickle
    # data = pickle.loads(data)
    # import cv2
    # cv2.imshow('data', data)
    # cv2.waitKey(0)