Implementation of Python3 Remote Monitoring Program
- 2021-07-18 08:35:38
- OfStack
Brief introduction
I started to think this was very interesting, and then I wanted to make one to play one
Language used: Python3
Tools: opencv Video Surveillance + socket Data Transmission Technology
Program verification: Here I consider 1, found that there is still no need to implement encapsulation into executable files. Let's just put the code. (Put the code first and explain it later.)
This procedure has been modified by myself to ensure that it can be used
Use requirements:
The Sender code must run on a computer with a camera. Then the data is encoded, compressed and transmitted to another computer
Reciever, as the receiver, has no special requirements.
Both computers must press numpy + opencv. (If you install it, there is one article in my opencv collection that writes about it.)
By the way, my receiver shutdown operation is (Esc in the input keyboard) so that you can exit the monitoring.
The sender can't turn off this monitoring
As for the setting of this IP address: I write the address of the receiver. The port is randomly set
Limitations:
At present, this program can only be monitored in real time in LAN.
Ideas for improvement:
If you want to expand into a wide area network. (You can use a server on a wide area network as a relay station)
Moreover, if one broiler can be found, the ip address of the corresponding monitoring object can be avoided by the above method
Code
Sender (sender code)
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) # Resolution
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 recipient code:
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()