[PYTHON] Consultez les prévisions météo sur M5Stack + Google Cloud Platform
Je m'appelle Niamugi et je suis en charge du 4ème jour du Calendrier de l'Avent IoTLT 2019. "Afficher les prévisions météo sur la pile M5" est un contenu très solide, mais je voudrais le présenter car il a un mécanisme agréable et polyvalent. ** Il est populaire auprès de ma famille et je suis heureux qu'il soit utilisé quotidiennement. ** **
Apparence
Affiche les prévisions météorologiques des 3 derniers jours. La marque météorologique a été écrite par mon fils aîné.
Il sera affiché comme ceci.
- En fait, cela prend environ 30 secondes ...
Comment ça fonctionne
L'image des prévisions météo est divisée en "** générer " et " acquérir **".
- Exécutez périodiquement la "fonction de génération d'image de prévision météo" dans Cloud Scheduler de Google Cloud Platform (GCP).
- Créez une image de prévision météo avec Cloud Functions de GCP et enregistrez-la dans Google Drive.
- Depuis M5Stack, exécutez la "fonction pour obtenir l'image de la météo" dans la requête http pour obtenir l'image.
(Détails) À propos de la génération d'images
Je vais énumérer les points.
Obtenez les prévisions météo
J'obtiens des données en accédant à Prévisions météorologiques de l'Agence météorologique.
Gérer les fichiers avec Cloud Functions
Si vous souhaitez que Cloud Functions s'exécutant sur le cloud gère vos fichiers, vous pouvez les enregistrer dans le / tmp folder. En d'autres termes, en enregistrant le fichier obtenu par Google Drive dans le dossier / tmp, vous pouvez gérer le fichier de la même manière que dans l'environnement local.
Se préparer à accéder à Google Drive
Obtenez à l'avance l'ID client, le secret client et le jeton d'actualisation requis pour l'accès. J'ai déjà écrit à ce sujet sur le blog de dot studio. [Comment télécharger des données de NefryBT vers 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 )Prière de se référer à.
Utiliser Google Drive
- Ceci est un contenu assez détaillé. Nous fournirons certaines fonctions pour accéder à Google Drive.
Fonction pour obtenir le service pour accéder à Google Drive
Obtenez le service pour accéder à Google Drive à l'aide de l'ID client, du secret client et du jeton d'actualisation. J'ai fait référence à "API Google de jeton d'actualisation python: obtenir les informations d'identification du jeton de mise à jour en utilisant 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
Fonction pour rechercher par nom de fichier et obtenir un ID
Des identifiants sont attribués à chaque donnée dans Google Drive. La recherche d'identifiant est nécessaire pour acquérir et mettre à jour les données par identifiant.
def searchID(service, mimetype, nm):
"""Rechercher un identifiant correspondant sur 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
Fonction pour obtenir des données de police
Cloud Functions fonctionne sur le cloud, vous ne pourrez donc probablement pas utiliser de polices japonaises. (Je ne l'ai pas essayé) Alors récupérez la police sur Google Drive. mimetype est "application / octat-stream".
def getFontFromDrive(service, fontName):
"""Obtenez les polices de Drive et enregistrez-les dans le dossier tmp
"""
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") #Fichier
downloader = MediaIoBaseDownload(fh, request)
done = False
while done is False:
status, done = downloader.next_chunk()
return "/tmp/" + fontName
Fonction pour obtenir des données d'image
Obtenez la marque météo. Le type MIME est "image / png".
def getImageFromDrive(service, imageName):
"""Récupérez l'image de Drive et enregistrez-la dans le dossier tmp
"""
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") #Fichier
downloader = MediaIoBaseDownload(fh, request)
done = False
while done is False:
status, done = downloader.next_chunk()
return True
Fonction pour télécharger des données d'image
Téléchargez l'image de prévision météo générée sur Google Drive.
def uploadData(service, mimetype, fromData, toData, parentsID="root"):
"""Télécharger sur Drive
"""
try:
media = MediaFileUpload(fromData, mimetype=mimetype, resumable=True)
except FileNotFoundError:
return False
#Recherchez l'ID et écrasez s'il existe des données applicables.
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
Une série de flux
Utilisez la fonction préparée ci-dessus pour télécharger l'image des prévisions météorologiques sur Google Drive.
def CreateImgWeather(event, context):
""" get weatherImage and upload to drive for M5stack
"""
# 1.Obtenez un service pour accéder à Google Drive
driveService = getDriveService()
# 2.Obtenir la police
fontPath = getFontFromDrive(driveService, "meiryo.ttc")
if not fontPath:
return False
# 3.Obtenez la marque météo
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.Générer une image de prévision météo
weatherList = getWeekWeather()
ret = createImg(fontPath, "/tmp/imgWeather.jpeg ", weatherList)
if not ret:
return False
# 5.Télécharger sur Google Drive
ret = uploadData(
driveService, "image/jpeg", "/tmp/imgWeather.jpeg ", "imgWeather.jpeg "
)
if not ret:
return False
return True
(Détails) À propos de l'acquisition d'images
Côté M5Stack
Voir Source pour plus de détails.
Accédez aux fonctions Cloud Functions avec une requête HTTP POST. C'est aussi dotstudio "Lancer une requête via une communication HTTP" Je l'ai utilisé comme référence.
[nom d'hôte] = "[Nom du projet].cloudfunctions.net"
[Nom de la fonction] = "getDriveImage_M5stack";
[numéro de port] = 443;
POST /[Nom de la fonction] HTTP/1.1
Host: [nom d'hôte]:[numéro de port]
Connection: close
Content-Type: application/json;charset=utf-8
Content-Length: + [Taille des données JSON à publier]
[Données Json à publier]
Faites la demande suivante au format de données json.
{
"drive" : {
"img" : "[nom de fichier]",
"trim" : "[Numéro de division]"
}
}
En raison de la quantité de données qui peuvent être acquises en même temps, elles sont divisées en 8 parties. Par conséquent, il effectue une requête POST 8 fois.
Côté Cloud Functions
Acquiert l'image de prévision météo selon la demande POST de M5Stack. Ensuite, il renvoie les données binaires divisées en 8 parties.
Je vais mettre la source.
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():
~Identique à la génération d'image~
def searchID(service, mimetype, nm):
~Identique à la génération d'image~
def downloadData(mimetype, data):
#Obtenez un service pour accéder à Google Drive
drive_service = getDriveService()
#Rechercher une pièce d'identité
ret, id = searchID(drive_service, mimetype, data)
if not ret:
return False, None
#Rechercher des images de prévisions météorologiques
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):
"""Divisez l'image pour M5Stack. La valeur de retour correspond aux données d'image
"""
imgNumpy = 0x00
#Confirmation des données d'entrée
if _trim.isnumeric():
trimPos = int(_trim)
if trimPos <= 0 or trimPos > 8:
return False
else:
return False
#Image fractionnée
# 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),
]
#Image PIL<-Données binaires
img_pil = Image.open(BytesIO(imgBinary))
#garniture
im_crop = img_pil.crop(Trim[trimPos - 1])
#tableau numpy(RGBA) <-Image PIL
imgNumpy = np.asarray(im_crop)
return True, imgNumpy
def getBinary(img):
"""Convertir des images en données binaires
"""
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):
"""Obtenez l'image enregistrée dans googleDrive. La valeur de retour est des données binaires.
"""
img = 0x00
#Image de Drive(Données binaires)Avoir
ret, imgBinary = downloadData("image/jpeg", imgName)
if not ret:
print("...error")
return ""
print(ret, len(imgBinary))
#Diviser l'image
#* Pour M5 Stack uniquement
if trim is not None:
isGet, img = devideImage_M5stack(imgBinary, trim)
if not isGet:
return ""
#Convertir en données binaires
imgBinary = getBinary(img)
return imgBinary
def getDriveImage_M5stack(request):
imgName = ""
trim = "0"
#Demander des données(JSON)Convertir
request_json = request.get_json()
#Obtenez des informations d'accès à Google Drive
if request_json and "drive" in request_json:
imgName = request_json["drive"]["img"]
trim = request_json["drive"]["trim"]
else:
return ""
#Obtenez une image de prévisions météo découpée
ret = getDriveImg_Binary(imgName, trim)
return ret
application
L'avantage de ce mécanisme est que "** Vous pouvez l'afficher sur M5Stack si vous préparez une image **". En d'autres termes, il ne se limite pas aux prévisions météorologiques, mais peut gérer n'importe quoi, comme les horaires et les tâches. Côté M5Stack, définissez simplement le nom de l'image à acquérir. De plus, comme l'image est générée en dehors de M5Stack, il n'est pas nécessaire de toucher le programme M5Stack lorsque vous souhaitez modifier l'image.
Voici le modèle qui affiche le calendrier Google. (L'horaire est en mosaïque)
Résumé
En créant un système d'affichage d'image qui correspond à M5Stack cette fois, j'en suis venu à réfléchir à certains modèles d'application. L'affichage de M5Stack est juste la bonne taille pour la table, je voudrais donc l'utiliser de différentes manières.
J'espère que cela vous sera utile. À bientôt.
référence
[Comment télécharger des données de NefryBT vers 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 ) API Google de jeton d'actualisation python: obtenez les informations d'identification du jeton de mise à jour à l'aide de oauth2client.client Lancer une requête via la communication HTTP
Recommended Posts