python工具類

socket通信工具類

# -*- coding: utf-8 -*-
import socket
import time

def tcp_server(ip, port):
    serverSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    serverSocket.bind((ip, port))
    serverSocket.listen(5)
    client, addr = serverSocket.accept()
    print('連接地址:', addr)
    while True:
        data = client.recv(1024).decode()
        print('收到{%s}的訊息{%s}' % (addr, data))
        if not data:
            break
        client.send('helloworld'.encode())
    print("{%s}斷開連接" % ip)

def tcp_client(ip, port):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect((ip, port))
    while True:
        print(client.recv(ip).decode())

def udp_server():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server_socket.bind((ip, port))
    server_socket.settimeout(10)
    while True:
        try:
            now = time.time()
            receive_data, client = server_socket.recvfrom(1024)
            print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(now)))  # 以指定格式顯示時間
            print("來自客戶端%s,發送的%s\n" % (client, receive_data))  # 列印接收的內容
        except socket.timeout:
            print("tme out")




if __name__ == '__main__':
    ip = '192.168.20.115'
    port = 5552
    tcp_server(ip, port)

http呼叫工具類

http post form表單工具類

# -*- coding: utf-8 -*-
import json
import requests
import time

def process_form_data(api_url, fields, headers):
    r = requests.post(api_url, data=fields, headers=headers)
    return r


def process(api_url, data, headers):
   response = requests.post(api_url, data=data, headers=headers)
   result = response.text
   result = json.loads(result)
   return result

def get_rtsp(begin, expiretimes):
    epid = 'system'
    uid = 'wrzs3002'
    password = 'wrzs3002-fj01'
    streamtype = 'realstream'
    puid = '200000000000000001'
    idx = '0'
    videoformat = 'rtsp'
    stream = '0'
    end = begin + expiretimes
    data = {"epid": epid, "uid": uid, "password": password, "streamtype": streamtype, "puid": puid,
         "idx": idx, "videoformat": videoformat, "stream": stream, "begin": begin,
         "end": end, "expiretimes": expiretimes}
    headers = {"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}
    api_url = "http://ip:9580/nmc/rest/rest.php"
    result = process_form_data(api_url, data, headers)
    result = json.loads(result.content)
    rtsp = result.get('response').get('url')
    rtsp = rtsp.replace('%3A', ':')
    rtsp = rtsp.replace('%2F', '/')
    rtsp = rtsp.replace('%3F', '?')
    rtsp = rtsp.replace('%3D', '=')
    return rtsp

if __name__ == '__main__':
    begin = (int)(time.time())
    expiretimes = 7200
    rtsp = get_rtsp(begin, expiretimes)
    print(rtsp)

下載excel圖片工具類

# -*- coding: utf-8 -*-
import xlrd
import urllib.request

def process_excel(file_path, save_path):
    print('開始處理excel')
    # 打開檔案
    data = xlrd.open_workbook(file_path)
    # 通過檔案名獲得作業表,獲取作業表1
    table = data.sheet_by_name('sheet1')
    print("總行數：" + str(table.nrows))
    print("總列數：" + str(table.ncols))
    for i in range(table.nrows):
        if i > 2767:
            img_path = cel_B3 = table.cell(i, 8).value
            img_name = 'drone_2020_{' + str(i) + '}_' + img_path.split('/')[-1]
            try:
                download_img(img_path, img_name, save_path)
            except:
                print('cuowu')
    # 獲取某個單元格的值，例如獲取B3單元格值
    # cel_B3 = table.cell(3, 2).value
    # print("第三行第二列的值：" + cel_B3)


def download_img(img_path, img_name, save_path):
    request = urllib.request.Request(img_path)
    try:
        response = urllib.request.urlopen(request)
        filename = save_path + '//' + img_name
        if (response.getcode() == 200):
            with open(filename, "wb") as f:
                f.write(response.read())  # 將內容寫入圖片
            return filename
    except:
        return "failed"

if __name__ == '__main__':
    file_path = 'D:\work\projects\drone\doc\火點資訊匯出.xlsx'
    save_path = 'D:\work\projects\drone\data\images'
    process_excel(file_path, save_path)