# -*- coding: utf-8 -*- class RingBuffer(): def __init__(self,bufferSize): self.size = bufferSize self.pc= 0 #次に使うバッファの番号 self.data=[None]*self.size #固定長さのリスト self.valid=False #リングバッファがいっぱいになったらTrueにす…

multiprocessingのサンプルコード。 マルチコア処理してほしい＆処理には共通の情報を利用する、という条件あり。 メンバ変数の書き換えは、returnには反映されるが、実行のたびに1に戻っている感じ。 # -*- coding: utf-8 -*- from multiprocessing import …

threading.Threadを使って引数持つ関数を実行すると、以下のようなエラーになることがある。 run()に直接関数を記述せず、すでにある関数をrun()の中で実行しようとすると起きる。 "TypeError: function1() got multiple values for argument 'arg1'" 原因は…

ファイルの最後の1行を読みたい時。 # -*- coding: utf-8 file_name="some_file.csv" with open(file_name,'r') as f: num_lines = sum(1 for line in f) f.seek(0) for line in range(0, num_lines-1): f.readline() line = f.readline() print(line)

マルチスレッドのテストプログラム。 Listをもらって、加算して、結果をグローバルのリストに書き込む。 import threading import queue import time commonList=[] q=queue.Queue() def worker(): """ マルチスレッドで走らせる関数 Queueからデータをもら…

DB内にデータがあるかどうか確認する。 あった時は更新(update)、なかった時は挿入(insert)するように処理すればよい。cur.execute(sql,data)に渡した時にSQL文がうまく展開されない？問題で困った。 # -*- coding: utf-8 import mysql.connector import dat…

ちょっと前に書いた記事compute-cucco.hatenablog.comのテストコード。というか抽象クラスは関係ないので、ただのテストコード。 raiseに対するテストコードの書き方が分からない。。。 # -*- coding: utf-8 -*- import unittest import movingCalcs from mo…

mysqlでのテーブルの1対多のjoin。TABLE_SECには、5秒ごとのデータが入っている。 TABLE_HOURには、1時間ごとのデータが入っている。10:00:06の時点では、10:00:00からの5秒間のデータは確定しているが、 10:00:00から1時間のデータは確定していない。 その…

pythonで、逐次入力に対して移動平均を計算する。 移動平均幅は、コンストラクタの引数に与えておく。以下、ソース。 # -*- coding: utf-8 -*- class MovingAverage: """逐次入力に対して移動平均を計算するためのクラス""" def __init__(self,size): # コン…