Python3远程监控程序实现

我不是女神ヾ 2022-05-21 10:49 326阅读 0赞

## 简述 ##

一开始觉得这个很有趣，然后就想来做一个来玩一下

*  使用语言： Python3
 *  使用工具：opencv视频监控 + socket数据传输技术

**程序检验：** 这里我考虑了一下，发现还是没有必要实现封装成可执行文件。还是直接就放代码吧。（先放代码，以后再做解释）  
本程序，经过本人修改，保证可以使用

--------------------

**使用要求：**

*  Sender代码必须要在一台有摄像头的电脑上运行起来。然后把数据编码，压缩之后，再传给另外一个电脑
 *  Reciever作为接受端，没什么特别的要求。
 *  两个电脑都必须要按转好numpy + opencv （安装方法的话，在我的opencv文集中有一篇就是写这个的）
 *  对了，我的接受端关闭操作是（输入键盘中的 Esc）这样就可以退出监控。
 *  发送端是关闭不了这个监控的
 *  至于这个IP地址的设置：我写的都是接受端的地址。端口随便设置的

**局限性：**

*  本程序目前只能在局域网内进行实时监控。

**改进思路：**

*  如果想扩展成一个广域网上的。（就可以用一个广域网上的服务器做中转站）
 *  而且，上述的方法，如果能找到一台肉鸡，还可以避免探测到对应的监控对象的ip地址

--------------------

## 代码 ##

**Sender(发送端代码)**

import socket
    import struct
    import time
    import cv2
    import numpy
    
    
    class Config(object):
        def __init__(self):
            self.TargetIP = ('192.168.199.121', 6666)
            self.resolution = (640, 480)  # 分辨率
            self.img_fps = 15  # each second send pictures
            self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server.connect(self.TargetIP)
            self.img = ''
            self.img_data = ''
    
        def RT_Image(self):
            camera = cv2.VideoCapture(0)
            img_param = [int(cv2.IMWRITE_JPEG_QUALITY), self.img_fps]
    
            while True:
                time.sleep(0.1)  # sleep for 0.1 seconds
                _, self.img = camera.read()
    
                self.img = cv2.resize(self.img, self.resolution)
    
                _, img_encode = cv2.imencode('.jpg', self.img, img_param)
                img_code = numpy.array(img_encode)
                self.img_data = img_code.tostring()  # bytes data
                try:
    
                    packet = struct.pack(b'lhh', len(self.img_data), self.resolution[0],
                                         self.resolution[1])
                    self.server.send(packet)
                    self.server.send(self.img_data)
    
                except Exception as e:
                    print(e.args)
                    camera.release()
                    return
    
    
    if __name__ == '__main__':
        config = Config()
        config.RT_Image()

**Reciever 接受端代码：**

import socket
    import cv2
    import struct
    import numpy
    import threading
    
    
    class Camera_Connect_Object(object):
        def __init__(self, TargetIP=('', 6666)):
            self.TargetIP = TargetIP
            self.resolution = (640, 480)
            self.src = 888 + 15
            self.interval = 0
            self.img_fps = 15
    
            self.Server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.Server.bind(self.TargetIP)
            self.Server.listen(5)
    
        def RT_Image(self):
            self.client, self.addr = self.Server.accept()
            self.name = self.addr[0] + " Camera"
            print(self.name)
            while True:
                # time.sleep(0.3) # sleep for 0.3 seconds
                tempdata = self.client.recv(8)
                if len(tempdata) == 0:
                    print("+1")
                    continue
                info = struct.unpack('lhh', tempdata)
                buf_size = int(info[0])
    
                if buf_size:
                    try:
                        self.buf = b""
                        self.temp_buf = self.buf
                        while buf_size:
                            self.temp_buf = self.client.recv(buf_size)
                            buf_size -= len(self.temp_buf)
                            self.buf += self.temp_buf
                        data = numpy.fromstring(self.buf, dtype='uint8')
    
                        self.image = cv2.imdecode(data, 1)
                        cv2.imshow(self.name, self.image)
                    except Exception as e:
                        print(e.args)
                        pass
                    finally:
                        if cv2.waitKey(10) == 27:
                            self.client.close()
                            cv2.destroyAllWindows()
                            break
    
        def Get_data(self):
            showThread = threading.Thread(target=self.RT_Image)
            showThread.start()
            showThread.join()
    
    
    if __name__ == '__main__':
        camera = Camera_Connect_Object()
        camera.Get_data()