2021年12月7日火曜日
Fireabase CLI を使って Realtime Database のデータをエクスポートする
2021年12月6日月曜日
SQLAlchemy で JSON 型に upsert する方法
概要
JSON 側でフィールドがない場合には追加しある場合は更新します
結論としては update + json_set で OK です
環境
- macOS 11.6.1
- MySQL 8.0.26
- flask-sqlalchemy 2.5.1
準備
CREATE DATABASE test;
USE test;
CREATE TABLE `user` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`profile` json DEFAULT NULL,
PRIMARY KEY (`id`)
);
サンプルコード
"""flask_sqlalchemyを使ってJSON型をテストするクラス."""
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root@localhost/test?charset=utf8' # noqa: E501
db = SQLAlchemy(app)
ma = Marshmallow()
class User(db.Model):
"""Userテーブルのモデル."""
id = db.Column(db.Integer, primary_key=True)
profile = db.Column(db.JSON())
class UserResponseSchema(ma.Schema):
"""Userテーブル用のレスポンスデータスキーム."""
id = ma.String(attribute="id")
profile = ma.Dict(attribute="profile")
class UserTable():
"""Userテーブルを操作するクラス."""
def insert(self, profile):
"""profileを登録します."""
p = User(profile=profile)
db.session.add(p)
db.session.commit()
def select_all(self):
"""全件取得."""
return User.query.filter().all()
def upsert(self, id, profile):
"""フィールドがない場合は登録します."""
user_query = User.query.filter(User.id == id)
for key, value in profile.items():
user_query.update(
{"profile": db.func.json_set(
User.profile,
"$." + key,
value)}, synchronize_session='fetch')
db.session.commit()
if __name__ == '__main__':
# テストレコード登録一度のみ
# user_table = UserTable()
# profile = {"name": "hawksnowlog"}
# user_table.insert(profile)
# レコード取得
user_table = UserTable()
records = user_table.select_all()
ret = UserResponseSchema(many=True).dump(records)
print(ret)
# レコード追加
new_profile = {'age': 10, 'name': 'snowlog'}
user_table.upsert(ret[0]["id"], new_profile)
2021年12月3日金曜日
2021年12月2日木曜日
SqlAlchemy で外部キーとして参照しているテーブルの逆参照する方法
概要
db.relationship で外部キーの参照ができます
外部キーとして紐付いている先から参照元のデータを取得したケースがあると思います
今回はその方法を紹介します
環境
- macOS 11.6.1
- Python 3.8.3
準備
- pipenv install Flask-Migrate Flask-SQLAlchemy Flask-Marshmallow Flask mysqlclient
my_app/database.py
Item は必ずどれかの User に紐付いています
User は Item に紐付いていないケースがあります
item = ma.Nested(ItemSchema, many=True) で参照元を参照します
item の部分の変数名は item = db.relationship("Item", backref="user") で使用している変数名と同じにする必要があります
同じにしないと SqlAlchemy が参照元を辿れず値がうまく取得できません
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
db = SQLAlchemy()
ma = Marshmallow()
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128))
age = db.Column(db.Integer)
__table_args__ = (
db.UniqueConstraint('name', 'age', name='unique_name_age'),
)
item = db.relationship("Item", backref="user")
def __repr__(self):
return "%s,%s,%i" % (self.id, self.name, self.age)
class Item(db.Model):
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey("user.id"))
name = db.Column(db.String(20))
class ItemSchema(ma.Schema):
class Meta:
fields = ("id", "name")
class UserSchema(ma.Schema):
class Meta:
fields = ("id", "name", "age", "item")
item = ma.Nested(ItemSchema, many=True)
my_app/__init__.py
User の情報を取得するためのアプリです
from flask import Flask, jsonify
from flask_migrate import Migrate
from my_app.database import db, ma
from my_app.lib import UserCRUD
app = Flask(__name__)
app.debug = True
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql+mysqldb://root:@localhost/test?charset=utf8"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db.init_app(app)
migrate = Migrate(app, db)
ma.init_app(app)
@app.route('/<id>')
def select(id=None):
crud = UserCRUD()
user = crud.select(id)
return jsonify(user)
my_app/lib/__init__.py
User テーブルを操作するためのクラスです
from my_app.database import db, User, UserSchema
class UserCRUD():
def select(self, id):
user = User.query.get(id)
return UserSchema(many=False).dump(user)
DB マイグレート
-
mysql -u root -p -e "create database test;"
- FLASK_APP=my_app pipenv run flask db init
- FLASK_APP=my_app pipenv run flask db migrate -m “Create item table”
- FLASK_APP=my_app pipenv run flask db upgrade
insert into user values (1, 'hawk', 10);
insert into user values (2, 'snowlog', 20);
insert into item values (1, 1, 'apple');
insert into item values (2, 1, 'grape');
insert into item values (3, 1, 'banana');
動作確認
- FLASK_APP=my_app pipenv run flask run
- curl localhost:5000/1
{"age":10,"id":1,"item":[{"id":1,"name":"apple"},{"id":2,"name":"grape"},{"id":3,"name":"banana"}],"name":"hawk"}
- curl localhost:5000/2
{"age":20,"id":2,"item":[],"name":"snowlog"}
こんな感じで User から Item を参照することができます
Item クラスの特定のフィールドだけ返したい場合
post_dump を使うといいかなと思います
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from marshmallow import post_dump
db = SQLAlchemy()
ma = Marshmallow()
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128))
age = db.Column(db.Integer)
__table_args__ = (
db.UniqueConstraint('name', 'age', name='unique_name_age'),
)
item = db.relationship("Item", backref="user")
def __repr__(self):
return "%s,%s,%i" % (self.id, self.name, self.age)
class Item(db.Model):
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey("user.id"))
name = db.Column(db.String(20))
class ItemSchema(ma.Schema):
class Meta:
fields = ("id", "name")
class UserSchema(ma.Schema):
class Meta:
fields = ("id", "name", "age", "item")
item = ma.Nested(ItemSchema, many=True)
best_item_name = ma.String(default="")
@post_dump
def set_best_item(self, data, many, **kwargs):
if data["item"]:
data["best_item_name"] = data["item"][0]["name"]
del data["item"]
return data
最後に
User.query.get(id) の結果から item を参照することはできるのですが Schema で逆参照を使う場合に少し工夫が必要です
参考サイト
(Mac) Firebase CLI をインストールする
概要
macOS に firebase CLI をインストールしてセットアップする方法を紹介します
環境
- macOS 11.6.1
- firebase cli 9.23.0
インストール
オムニバスインストールできるようなスクリプトが準備されているのでそれを使います
管理者権限のパスワードが求められるので入力します
- curl -sL https://firebase.tools | bash
-- Checking for existing firebase-tools on PATH...
-- Checking your machine type...
-- Downloading binary from https://firebase.tools/bin/macos/latest
######################################################################## 100.0%
-- Setting permissions on binary...
-- Checking your PATH variable...
-- firebase-tools@9.23.0 is now installed
-- All Done!
動作確認
ログインできるか確認しましょう
- firebase login
ブラウザが開くのでログインしましょう
firebase cli に oauth で権限を与えれば OK です
- firebase projects:list
でプロジェクトの一覧が表示されれば OK です
参考サイト
2021年11月30日火曜日
RaspberryPi4 の ds64-shell に python をインストールする
概要
ds64 環境にはデフォルトではインストールされていないのでインストールが必要です
環境
- RaspberryPi4 8GRAM
- RaspberryPiOS 5.10.17 armv7l
インストール
- ds64-shell
- sudo apt -y update
- sudo apt -y install python3
- sudo apt -y install python3-pip
- pip3 install pigpio RPi.GPIO
ただ意味はないかも
「RuntimeError: Not running on a RPi!」というエラーになり gpio などの制御はできないようです
2021年11月29日月曜日
RaspberyPi でフルカラーLEDを制御してみた
概要
PWM を使ってフルカラーLEDでいろんな色を点滅させてみました
環境
- RaspberryPi4 8GRAM
- RaspberryPiOS 5.10.17 armv7l
- Python 2.7.16
- pigpio 1.78
サンプルコード
後述でポイントを紹介しています
処理の概要はコメントにて記載しています
- vim test_full_color_led.py
# -*- coding: utf-8 -*-
"""フルカラーのLEDを制御するモジュール."""
import RPi.GPIO as GPIO
import time
import logging
from logging import getLogger
class FullColorLEDTest():
"""フルカラーのLEDをテストするクラス."""
# 各ピン定義
RED = 11
GREEN = 12
BLUE = 13
def __init__(self):
"""ピンを初期化します."""
logging.basicConfig(level=logging.INFO)
self.logger = getLogger(__name__)
# RGB のカラーコードをここに記載します
self.colors = [
# red, green, blue, yellow, purple, cyan になります
0xFF0000, 0x00FF00, 0x0000FF, 0xFFFF00, 0xFF00FF, 0x00FFFF
]
pin_def = {
self.RED: 2000, # red
self.GREEN: 2000, # green
self.BLUE: 5000, # blue
}
GPIO.setmode(GPIO.BOARD)
# 制御するピンを管理するdict
self.pins = {
self.RED: None,
self.GREEN: None,
self.BLUE: None,
}
# 各ピンの初期化
for pin_num, freq in pin_def.items():
GPIO.setup(pin_num, GPIO.OUT)
pin = GPIO.PWM(pin_num, freq)
pin.start(0)
self.pins[pin_num] = pin
self.logger.info('Started pwm pins for a full color led.')
def calc_duty(self, x, in_min, in_max, out_min, out_max):
"""デューティー比を計算します."""
return (x - in_min) * (out_max - out_min) / (in_max - in_min) + out_min
def blink_color(self, col):
"""各ピンに色を設定し指定の色を光らせます."""
# 左ビットシフト
red = (col & 0x110000) >> 16
green = (col & 0x001100) >> 8
blue = (col & 0x000011) >> 0
# デューティー比に変換 (0-100)
red = self.calc_duty(red, 0, 255, 0, 100)
green = self.calc_duty(green, 0, 255, 0, 100)
blue = self.calc_duty(blue, 0, 255, 0, 100)
# ピンにデューティー比を設定
self.pins[self.RED].ChangeDutyCycle(red)
self.pins[self.GREEN].ChangeDutyCycle(green)
self.pins[self.BLUE].ChangeDutyCycle(blue)
def blink(self):
"""フルカラーLEDを点滅させます."""
try:
self.logger.info('Started blinking a full color led.')
while True:
for col in self.colors:
self.blink_color(col)
time.sleep(0.5)
except KeyboardInterrupt:
pass
finally:
self.cleanup()
def cleanup(self):
"""クリーンアップ処理を実行します."""
for pin in self.pins.values():
pin.stop()
GPIO.cleanup()
self.logger.info('The pin cleanup process was completed.')
if __name__ == "__main__":
test = FullColorLEDTest()
test.blink()
説明
今回使用したフルカラーLEDは赤青緑で3ピン必要になるのでピンを管理する dict を準備しました (この辺りはクラスにしてもいいかも)
ピンの情報と周波数も別途 dict で定義しています
init でピンの初期化を行ったあとは blink で各カラーを順番に点滅させます
点滅させたい色は self.colors で定義していので他の色を点滅させたい場合はここに追加していきます
定義したカラーコードを実際にはデューティー比に変換する必要があります
それは calc_duty で行っておりカラーコードを 0 から 100 の範囲に変換する処理を行っています
あとは変換した値を ChangeDutyCycle で各ピンに対して設定して上げれば OK です
回路
- 11 -> 赤のピン
- 12 -> 緑のピン
- 13 -> 青のピン
- 6 (GND) -> カソード
に接続するだけです
各色を管理するピンは 220 オームの抵抗を挟んで接続しています
動作確認
- python test_full_color_led.py
で実行しましょう
以下のように順番に点滅すれば成功です
最後に
フルカラーLED と PWM を使えば好きな色の LED を光らせることができます
少しお値段が高いのが難点ですが便利です
各ピンの出力周波数に関しては参考にさせていただいたサイトの値をそのまま拝借しているだけなのでフルカラーLED の種類によっては調整の必要があるかもしれません
参考サイト
登録:
投稿 (Atom)