QueueをつかったPython マルチスレッド
マルチスレッドのテストプログラム。
Listをもらって、加算して、結果をグローバルのリストに書き込む。
import threading import queue import time commonList=[] q=queue.Queue() def worker(): """ マルチスレッドで走らせる関数 Queueからデータをもらう。 結果をcommonListに書き込む 計算内容や書き込む場所はqに書いてある """ global q global commonList print("thread Start") while True: time.sleep(0.1) #Ctrl+Cで終了させるおまじない。 item = q.get() if item is None: print("thread None") break commonList.append((item.ID,sum(item.values))) q.task_done() print("thread End") def main(): global q global commonList num_worker_threads=3 a=Job([1,2,3],0) b=Job([3,4,5],1) c=Job([6,7,8],2) d=Job([9,0,1],3) e=Job([2,3,4],4) f=Job([5,6,7,8],5) threadItem=[a,b,c,d,e,f] threads=[] #スレッドを立てる。作業はあとでQueueで渡す。 for i in range(num_worker_threads): t = threading.Thread(target=worker) t.start() threads.append(t) #itemを投入。データを渡す。 for item in threadItem: q.put(item) # block until all tasks are done qが空になるのを待っている。 q.join() #スレッド停止命令(None)の投入 for i in range(num_worker_threads): q.put(None) #スレッドの終了まち for t in threads: t.join() #結果の表示 print(commonList) class Job(): #xはリスト,yは書き込み先のリストのインデックス def __init__(self,x,y): self.values=x.copy() self.ID=y if __name__ == '__main__': main()
結果
C:\Python34>python multiThreadTest.py thread Start thread Start thread Start thread None thread End thread None thread End thread None thread End [(0, 6), (1, 12), (2, 21), (3, 10), (4, 9), (5, 26)]
DBに特定のデータがあるかどうかを確認する
DB内にデータがあるかどうか確認する。
あった時は更新(update)、なかった時は挿入(insert)するように処理すればよい。
cur.execute(sql,data)に渡した時にSQL文がうまく展開されない?問題で困った。
# -*- coding: utf-8 import mysql.connector import datetime config = { 'user': 'root', 'password': 'password', 'host': 'localhost', 'database':'testdb', 'charset':'utf8' } cnx = mysql.connector.connect(**config) cur = cnx.cursor(buffered=True) #tableの削除と作成 sql = 'DROP TABLE IF EXISTS TEST_UPDATE;' cur.execute(sql) cnx.commit() sql = 'CREATE TABLE IF NOT EXISTS TEST_UPDATE (\ id INT UNSIGNED NOT NULL AUTO_INCREMENT,\ time DATETIME DEFAULT NULL, \ value1 INT UNSIGNED DEFAULT NULL ,\ value2 INT UNSIGNED DEFAULT NULL ,\ primary key(id),\ unique (time));' cur.execute(sql) cnx.commit() #初期データのインサート sql = 'INSERT INTO TEST_UPDATE (time, value1,value2) VALUES (%s, %s, %s);' data= ('2017-5-01T05:50:00','1','1') cur.execute(sql, data) sql = 'INSERT INTO TEST_UPDATE (time, value1,value2) VALUES (%s, %s, %s);' data= ('2017-5-01T05:50:05','2','2') cur.execute(sql, data) cnx.commit() #この時点でのテーブルの中身 #mysql> select * from test_update; #+----+---------------------+--------+--------+ #’ id | time | value1 | value2 | #+----+---------------------+--------+--------+ #| 1 | 2017-05-01 05:50:00 | 1 | 1 | #| 2 | 2017-05-01 05:50:05 | 2 | 2 | #+----+---------------------+--------+--------+ #2 rows in set (0.00 sec) #データの有無の確認1 sql= "select exists(select * from test_update where time=%s);" #data=(datetime.datetime(2017,5,1,5,50,00),) data= ('2017-5-01T05:50:00',)#なぜかカンマが必要。 try: cur.execute(sql,data) except mysql.connector.Error as err: print(cur.statement)#実行したSQL文の確認 raise #print(cur.fetchone()) #あった時…(1,) #なかった時…(0,) if cur.fetchone()[0]==0: print("なかった") else: print("あった") #データの有無の確認2 #テーブル名をdataで渡そうとすると、クォートがついてしまってうまく動かない。 tablename='test_update' sql= "select exists(select * from " + tablename +" where time=%s);" data= ('2017-5-01T05:50:05',) try: cur.execute(sql,data) except mysql.connector.Error as err: print(cur.statement)#実行したSQL文の確認 raise if cur.fetchone()[0]==0: print("なかった") else: print("あった") #データの有無の確認3 tablename='test_update' sql= "select exists(select * from " + tablename +" where time=%s);" data= ('2017-5-01T05:50:15',) try: cur.execute(sql,data) except mysql.connector.Error as err: print(cur.statement)#実行したSQL文の確認 raise if cur.fetchone()[0]==0: print("なかった") else: print("あった") cur.close() cnx.close() print('end')
pythonのunittestのコード
ちょっと前に書いた記事compute-cucco.hatenablog.comのテストコード。というか抽象クラスは関係ないので、ただのテストコード。
raiseに対するテストコードの書き方が分からない。。。
# -*- coding: utf-8 -*- import unittest import movingCalcs from movingCalcs import factory class TestmovingCalcs(unittest.TestCase): """test class of movingCalcs.py """ def test_add(self): """test method for Add """ node = factory("Add") value1 = 1 value2 = 2 expected = 3 actual = node.calc(value1, value2) self.assertEqual(expected, actual) def test_sub(self): """test method for sub """ node = factory("Sub") value1 = 1 value2 = 2 expected = -1 actual = node.calc(value1, value2) self.assertEqual(expected, actual) def test_zero(self): """test method for Zero """ node = factory("Zero") value1 = 1 value2 = 2 expected = 0 actual = node.calc(value1, value2) self.assertEqual(expected, actual) def test_other(self): """test method for other """ with self.assertRaises(factory): node = factory("Other") if __name__ == "__main__": unittest.main()
Jenkinsのインストール(CentOS on raspberry pi3)
インストール手順は以下にアリ。
RedHat Repository for Jenkins
追記:Javaがない、といってJenkinsの起動に失敗します。動作確認はまた後日。
# sudo wget -O /etc/yum.repos.d/jenkins.repo https://pkg.jenkins.io/redhat-stable/jenkins.repo
sudo: wget: コマンドが見つかりません
wgetがない、と言われたので以下に書き換え。
sudo curl -o /etc/yum.repos.d/jenkins.repo https://pkg.jenkins.io/redhat-stable/jenkins.repo
中身を確認すると。
# more /etc/yum.repos.d/jenkins.repo
[jenkins]
name=Jenkins-stable
baseurl=http://pkg.jenkins.io/redhat-stable
gpgcheck=1
後は指示通り。まとめると以下。
sudo curl -o /etc/yum.repos.d/jenkins.repo https://pkg.jenkins.io/redhat-stable/jenkins.repo
sudo rpm --import https://pkg.jenkins.io/redhat-stable/jenkins.io.key
yum install jenkins
Raspberry Pi3にCentOS
CentOSが入るらしいのでやってみた。
wi-fiのドライバを別途ダウンロードする必要があるので、Raspberry Pi3に有線LANが必要。
ダウンロード
More download choicesへ。
http://wiki.centos.org/Download
AltArch Releasesの中に、RaspberryPi3 (img)があるのでこれを使う。
上記を解凍して(7zipなど)、イメージ書き込みツールでSDカードに書き込み。ここはraspbianと同じ。
ログイン
ログインID、パスワードはroot/centos。ここに書いてある。
https://wiki.centos.org/SpecialInterestGroup/AltArch/Arm32/RaspberryPi3
rootのパスワードの変更
#passwd
ディスクの拡張
ディスクの拡張は以下。/root/READMEに書いてある。
If you want to automatically resize your / partition, just type the following (as root user):
/usr/local/bin/rootfs-expand
Wifiの認識
Wifiの認識は以下。ファイルを落として再起動。これをやってもwifiのアダプタを認識するだけなので、設定は別。
For wifi on the rpi3, just proceed with those steps :
curl --location https://github.com/RPi-Distro/firmware-nonfree/raw/master/brcm80211/brcm/brcmfmac43430-sdio.bin > /usr/lib/firmware/brcm/brcmfmac43430-sdio.bin
curl --location https://github.com/RPi-Distro/firmware-nonfree/raw/master/brcm80
211/brcm/brcmfmac43430-sdio.txt > /usr/lib/firmware/brcm/brcmfmac43430-sdio.txtsystemctl reboot
Wifiの設定(raspberry Pi3にcentos7を入れてみた - Qiitaより)
[root@centos-rpi3 ~]# nmcli d
デバイス タイプ 状態 接続
eth0 ethernet 接続済み eth0
wlan0 wifi 管理無し --
lo loopback 管理無し --
$ nmcli d wifi
# -> Wifiアクセスポイントのリストが一覧されます
$ nmcli d wifi connect
キーボードの設定(raspberry Pi3にcentos7を入れてみた - Qiitaより)
centos
$ localectl set-keymap jp106
$ localectl set-keymap jp-OADG109A##System Localeを変更もしておきます
$ localectl set-locale LANG=ja_JP.utf8##localectlで確認
$ localectl
System Locale: LANG=ja_JP.utf8
VC Keymap: jp-OADG109A
X11 Layout: jp
X11 Model: jp106
X11 Options: terminate:ctrl_alt_bksp
Abstract Base Classのテスト
Abstract Base Classのテスト。
メソッド名が同じで、引数が異なるオブジェクトの生成と実行のサンプルです。
メソッド名を共通で固定したいので、Abstract Base Classを使ってみました。
変数名にnameを使っていたら、「TypeError: 'str' object is not callable」と怒られたのでclassNameとしています。
# -*- coding: utf-8 -*- from abc import ABCMeta, abstractmethod # 抽象クラス class CalcNode(metaclass=ABCMeta): @abstractmethod def className(self): pass @abstractmethod def calc(self): pass #抽象クラスの中身。Zero,Add,Subの3つのクラス class classZero(CalcNode): def __init__(self): self.name = "Zero" def className(self): return self.name def calc(self,val1=1,val2=2): return 0 class classAdd(CalcNode): def __init__(self): self.name= "Add" def className(self): return self.name def calc(self,val1=1,val2=2): return val1+val2 class classSub(CalcNode): def __init__(self): self.name= "Sub" def className(self): return self.name def calc(self,val1=1,val2=2): return val1-val2 def factory(className): if className=="Zero": return classZero() elif className=="Add": return classAdd() elif className=="Sub": return classSub() else: raise def main(): nodes=(factory("Add"), factory("Zero"), factory("Sub")) for node in nodes: print(node.className()) print(node.calc(val1=2,val2=5)) print(node.calc(val2=5,val1=2)) #入れ替えても大丈夫 print(node.calc(2,5)) #変数名を省いても大丈夫 print(node.calc(val1=100)) #片方だけ入れても大丈夫 print(node.calc(val2=100)) #片方だけ入れても大丈夫 print(node.calc()) #何も入れなくても大丈夫 #main()を実行する if __name__== '__main__': main()