My name is Niamugi and I am in charge of the 4th day of IoTLT Advent Calendar 2019. "Displaying the weather forecast on M5Stack" is a very solid content, but I would like to introduce it because it has a nice and versatile mechanism. ** It is popular with my family, and I am happy that it is used on a daily basis. ** **
Displays the weather forecast for the last 3 days. I had my eldest son write the weather mark.
It will be displayed like this.
The weather forecast image is divided into "** generate " and " acquire **".
I will enumerate the points.
Data is obtained by accessing Weather Forecast of Japan Meteorological Agency.
If you want Cloud Functions running on the cloud to work with your files, you can save them in the / tmp folder. In other words, by saving the file acquired by Google Drive in the / tmp folder, you can handle the file in the same way as in the local environment.
Obtain the client ID, client secret, and refresh token required for access in advance. I wrote about this on the blog of dot studio before. [How to upload data from NefryBT to Google Drive](https://dotstud.io/blog/update-nefrybt-to-googledrive/#%E3%82%A2%E3%83%83%E3%83%97% E3% 83% AD% E3% 83% BC% E3% 83% 89% E3% 81% BE% E3% 81% A7% E3% 81% AE% E6% 89% 8B% E9% A0% 86 )Please refer to.
Get the service to access Google Drive using the client ID, client secret, and refresh token. I referred to "python refresh token Google API: get credentials from update token using oauth2client.client".
def getDriveService():
CLIENT_ID = os.getenv("drive_client_id")
CLIENT_SECRET = os.getenv("drive_client_secret")
REFRESH_TOKEN = os.getenv("drive_refresh_token")
creds = client.OAuth2Credentials(
access_token=None,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
refresh_token=REFRESH_TOKEN,
token_expiry=None,
token_uri=GOOGLE_TOKEN_URI,
user_agent=None,
revoke_uri=None,
)
http = creds.authorize(httplib2.Http())
creds.refresh(http)
service = build("drive", "v3", credentials=creds, cache_discovery=False)
return service
Each data in Google Drive is assigned an ID. Since data is acquired and updated by ID, it is necessary to search for the ID.
def searchID(service, mimetype, nm):
"""Find a matching ID from Drive
"""
query = ""
if mimetype:
query = "mimeType='" + mimetype + "'"
page_token = None
while True:
response = (
service.files()
.list(
q=query,
spaces="drive",
fields="nextPageToken, files(id, name)",
pageToken=page_token,
)
.execute()
)
for file in response.get("files", []):
if file.get("name") == nm:
return True, file.get("id")
page_token = response.get("nextPageToken", None)
if page_token is None:
break
Since Cloud Functions runs on the cloud, I think Japanese fonts probably won't work. (I haven't tried it) So get the font from Google Drive. The mimetype is "application / octet-stream".
def getFontFromDrive(service, fontName):
"""Get fonts from Drive and save them in tmp folder
"""
ret, id = searchID(service, "application/octet-stream", fontName)
if not ret:
return None
request = service.files().get_media(fileId=id)
fh = io.FileIO("/tmp/" + fontName, "wb") #File
downloader = MediaIoBaseDownload(fh, request)
done = False
while done is False:
status, done = downloader.next_chunk()
return "/tmp/" + fontName
Get the weather mark. mimetype is "image / png".
def getImageFromDrive(service, imageName):
"""Get the image from Drive and save it in the tmp folder
"""
ret, id = searchID(service, "image/png", imageName)
if not ret:
return False
request = service.files().get_media(fileId=id)
fh = io.FileIO("/tmp/" + imageName, "wb") #File
downloader = MediaIoBaseDownload(fh, request)
done = False
while done is False:
status, done = downloader.next_chunk()
return True
Upload the generated weather forecast image to Google Drive.
def uploadData(service, mimetype, fromData, toData, parentsID="root"):
"""Upload to Drive
"""
try:
media = MediaFileUpload(fromData, mimetype=mimetype, resumable=True)
except FileNotFoundError:
return False
#Search for ID and overwrite if there is applicable data.
ret, id = searchID(service, mimetype, toData)
if ret:
file_metadata = {"name": toData}
file = (
service.files()
.update(fileId=id, body=file_metadata, media_body=media, fields="id")
.execute()
)
else:
file_metadata = {"name": toData, "parents": [parentsID]}
file = (
service.files()
.create(body=file_metadata, media_body=media, fields="id")
.execute()
)
return True
Use the function prepared above to upload the weather forecast image to Google Drive.
def CreateImgWeather(event, context):
""" get weatherImage and upload to drive for M5stack
"""
# 1.Get a service to access Google Drive
driveService = getDriveService()
# 2.Get font
fontPath = getFontFromDrive(driveService, "meiryo.ttc")
if not fontPath:
return False
# 3.Get the weather mark
if not getImageFromDrive(driveService, "noImage.png "):
return False
if not getImageFromDrive(driveService, "fine.png "):
return False
if not getImageFromDrive(driveService, "cloud.png "):
return False
if not getImageFromDrive(driveService, "rain.png "):
return False
if not getImageFromDrive(driveService, "snow.png "):
return False
# 4.Generate weather forecast image
weatherList = getWeekWeather()
ret = createImg(fontPath, "/tmp/imgWeather.jpeg ", weatherList)
if not ret:
return False
# 5.Upload to Google Drive
ret = uploadData(
driveService, "image/jpeg", "/tmp/imgWeather.jpeg ", "imgWeather.jpeg "
)
if not ret:
return False
return True
For details, refer to Source.
Access the Cloud Functions functions with an http POST request. This is also dotstudio's "Throw a request via HTTP communication" I used it as a reference.
[hostname] = "[Project name].cloudfunctions.net"
[Function name] = "getDriveImage_M5stack";
[port number] = 443;
POST /[Function name] HTTP/1.1
Host: [hostname]:[port number]
Connection: close
Content-Type: application/json;charset=utf-8
Content-Length: + [Size of json data to post]
[Json data to post]
Make the following request in json data format.
{
"drive" : {
"img" : "[file name]",
"trim" : "[Split number]"
}
}
Due to the amount of data that can be acquired at one time, it is divided into 8 parts. Therefore, it makes a POST request 8 times.
Get the weather forecast image according to the POST request from M5Stack. Then, it returns the binary data divided into 8 parts.
I will put the sauce on it.
import sys
import os
import io
from io import BytesIO
import numpy as np
from PIL import Image
import httplib2
from googleapiclient.discovery import build
from oauth2client import client, GOOGLE_TOKEN_URI
from apiclient.http import MediaIoBaseDownload
def getDriveService():
~Same as image generation~
def searchID(service, mimetype, nm):
~Same as image generation~
def downloadData(mimetype, data):
#Get a service to access Google Drive
drive_service = getDriveService()
#Search for ID
ret, id = searchID(drive_service, mimetype, data)
if not ret:
return False, None
#Search for weather forecast images
request = drive_service.files().get_media(fileId=id)
fh = io.BytesIO()
downloader = MediaIoBaseDownload(fh, request)
done = False
while done is False:
status, done = downloader.next_chunk()
return True, fh.getvalue()
def devideImage_M5stack(imgBinary, _trim):
"""Split the image for M5Stack. The return value is image data
"""
imgNumpy = 0x00
#Confirmation of input data
if _trim.isnumeric():
trimPos = int(_trim)
if trimPos <= 0 or trimPos > 8:
return False
else:
return False
#Image split
# 1 2 3 4
# 5 6 7 8
Trim = [
(0, 0, 80, 120),
(80, 0, 160, 120),
(160, 0, 240, 120),
(240, 0, 320, 120),
(0, 120, 80, 240),
(80, 120, 160, 240),
(160, 120, 240, 240),
(240, 120, 320, 240),
]
#PIL image<-Binary data
img_pil = Image.open(BytesIO(imgBinary))
#trimming
im_crop = img_pil.crop(Trim[trimPos - 1])
#numpy array(RGBA) <-PIL image
imgNumpy = np.asarray(im_crop)
return True, imgNumpy
def getBinary(img):
"""Convert images to binary data
"""
ret = ""
pilImg = Image.fromarray(np.uint8(img))
output = io.BytesIO()
pilImg.save(output, format="JPEG")
ret = output.getvalue()
return ret
def getDriveImg_Binary(imgName, trim):
"""Get the image saved in googleDrive. The return value is binary data.
"""
img = 0x00
#Image from Drive(Binary data)Get
ret, imgBinary = downloadData("image/jpeg", imgName)
if not ret:
print("...error")
return ""
print(ret, len(imgBinary))
#Split the image
#* For M5Stack only
if trim is not None:
isGet, img = devideImage_M5stack(imgBinary, trim)
if not isGet:
return ""
#Convert to binary data
imgBinary = getBinary(img)
return imgBinary
def getDriveImage_M5stack(request):
imgName = ""
trim = "0"
#Request data(JSON)Convert
request_json = request.get_json()
#Get access information to Google Drive
if request_json and "drive" in request_json:
imgName = request_json["drive"]["img"]
trim = request_json["drive"]["trim"]
else:
return ""
#Get a trimmed weather forecast image
ret = getDriveImg_Binary(imgName, trim)
return ret
The good thing about this mechanism is that "** You can display it on M5Stack if you prepare an image **". In other words, it is not limited to weather forecasts, but can handle anything such as schedules and tasks. On the M5Stack side, just set the image name to be acquired. Also, since the image is generated outside of M5Stack, there is no need to touch the M5Stack program when you want to modify the image.
The following is the pattern that displayed the Google calendar. (The schedule is mosaic)
Now that we have created an image display system that matches M5Stack, we have come to think of some application patterns. The display of M5Stack is just the right size for the table, so I would like to use it in various ways.
I hope it will be helpful for you. See you soon.
[How to upload data from NefryBT to Google Drive](https://dotstud.io/blog/update-nefrybt-to-googledrive/#%E3%82%A2%E3%83%83%E3%83%97% E3% 83% AD% E3% 83% BC% E3% 83% 89% E3% 81% BE% E3% 81% A7% E3% 81% AE% E6% 89% 8B% E9% A0% 86 ) python refresh token Google API: use oauth2client.client to get credentials from update token Throw a request via HTTP communication
Recommended Posts