Python使用阿里云OSS服务

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

前言：

在远程搭建了一个平台，通过改远程平台进行数据的采集，需要将数据内容传送至本地进行处理；为了实现该功能，考虑了阿里云的OSS对象储存的服务。

40G包月只需1元:-)
甚至还有客服致电给你，说有问题可直接通过电话联系对方，15星好评

OSS安装

关于账号注册，开通服务等等功能直接去阿里云的官方进行相应操作即可

安装python-devel
- win：此过程不需要，在安装Python时已经包含了；
- Debian/Ubuntu：apt-get install python-dev
安装OSS：pip3 install oss2 -i https://pypi.tuna.tsinghua.edu.cn/simple/
- 安装完成后可用python进行验证：
In [1]: import oss2In [2]: oss2.__version__ Out[2]: '2.8.0'

OSS使用

创建存储空间：

方式一：直接去网站页面进行创建即可；
方式二：通过代码创建
import oss2 auth = oss2.Auth('yourAccessKeyId', 'yourAccessKeySecret') # Endpoint以杭州为例，其它Region请按实际情况填写。yourBucketName就是你要创建的Bucket bucket = oss2.Bucket(auth, 'http://oss-cn-hangzhou.aliyuncs.com', 'yourBucketName') bucket.create_bucket(oss2.models.BUCKET_ACL_PRIVATE)
yourBucketName为你的Bucket名称，若已存在则会报错；
yourAccessKeySecret，别告诉别哈：）

连接OSS：

auth = oss2.Auth('yourAccessKeyId', 'yourAccessKeySecret') bucket = oss2.Bucket(auth, 'http://oss-cn-hangzhou.aliyuncs.com', 'yourBucketName')

上传文件：

# 此处上传了图片文件 image_path = './test.jpg' put_result = bucket.put_object_from_file('test.jpg', image_path) if put_result.status == 200: # 若此时的status状态为200，则说明上传成功；在你选择的Bucket中已经存在文件名为'test.jpg'的文件了 print('put success')

下载文件：

# param1:oss上bucket中的文件名 # param2:保存在当地的文件路径+文件名 get_result = bucket.get_object_to_file('test01.jpg', './get/test.jpg') if get_result == 200: print("get sucess")

整体代码：

根据触发信号获取摄像头的图片；
获取摄像头的实时图片；
将先图片保存到本地->上传阿里云->在本地进行删除

import oss2 import time import cv2 import os import RPi.GPIO as GPIOclass Camera(object): """摄像的初始化，配置""" def __init__(self, channel): self.capture = cv2.VideoCapture(channel) self.fps = int(self.capture.get(cv2.CAP_PROP_FPS)) self.video_height = int(self.capture.get(cv2.CAP_PROP_FRAME_HEIGHT)) self.video_width = int(self.capture.get(cv2.CAP_PROP_FRAME_WIDTH)) self.capture.set(cv2.CAP_PROP_FRAME_WIDTH, self.video_width) self.capture.set(cv2.CAP_PROP_FRAME_HEIGHT, self.video_height) self.capture.set(cv2.CAP_PROP_FPS, self.fps) # 设置摄像头的缓存图片为1，为了能获取实时的图像 self.capture.set(cv2.CAP_PROP_BUFFERSIZE, 1) def get_pic(self, image_dir=None, image_name=None, is_show=False, is_write=False): """ 写入照片 """ if self.capture.isOpened(): # 读取缓存中的图片，起到清空缓存的作用 abandon_ret, abandon_frame = self.capture.read() # 用于处理/上传的图片 ret, frame = self.capture.read() # 判断图片信息是否读取成功 if ret is False: print('get picture failed') return None # 是否需要显示 if is_show is True: cv2.imshow('image', frame) cv2.waitKey(200) # 是否需要写入到文件中 if is_write is True: cv2.imwrite(image_dir + '/' + image_name, frame) print('get picture success') return frame def release_camera(self): """ 释放摄像机资源 """ self.capture.release() cv2.destroyAllWindows()class oss(object): """对象存储类，将图片传至阿里云端""" def __init__(self): self.auth = oss2.Auth('LTAIkA49HkgVYUsW', '**********************') self.bucket = oss2.Bucket(self.auth, 'http://oss-cn-hangzhou.aliyuncs.com', '20190731') def put_image(self, image_dir, image_name): put_result = self.bucket.put_object_from_file(image_name, image_dir + '/' + image_name) # 状态为200表示成功的上传 if put_result.status == 200: print('put success') def get_image(self): passclass gpio(object): """触发信号的控制""" event_flag = False # 判断是否有电平触发信号 def __init__(self, pin): """GPIO初始化""" self.pin = pin # 设置引脚应用模式 GPIO.setmode(GPIO.BCM) GPIO.setup(self.pin, GPIO.IN, pull_up_down=GPIO.PUD_DOWN) # 设计引脚触发模式为：上升沿触发模式，防抖的时间为1000ms GPIO.add_event_detect(self.pin, GPIO.RISING, callback=gpio.callback, bouncetime=1000) @staticmethod def callback(flag): """回调函数""" gpio.event_flag = True def stop_event(self): GPIO.remove_event_detect(self.pin) def start_event(self): GPIO.add_event_detect(self.pin, GPIO.RISING, callback=gpio.callback, bouncetime=1000) @staticmethod def release(): """GPIO释放引脚资源""" GPIO.cleanup()if __name__ == "__main__": picture = None sleep_time = 0 picture_path = './put' picture_name = '.jpg' gpio_pin = 18 # oss配置 oss_server = oss() # 开启摄像头 camera = Camera(0) # GPIO初始化 event = gpio(gpio_pin) while True: if gpio.event_flag is True: gpio.event_flag = False # 标志位置位 event.stop_event() # 关闭事件触发 picture_name = time.strftime('%Y-%m-%d-%H-%M-%S', time.localtime(time.time())) + '.jpg' # 获取图片 picture = camera.get_pic(picture_path, picture_name, is_show=True, is_write=True) if picture is not None: pass # oss_server.put_image(picture_path, picture_name) else: camera.release_camera() camera = Camera(0) # 删除照片，防止内存爆炸 os.remove(picture_path + '/' + picture_name) event.start_event() # 开启事件触发 # time.sleep(sleep_time) gpio.release()