Python Tips

文字列を16進数にエンコード・デコード

from binascii import hexlify, unhexlify

# str -> hex
def str2hex(str_):
    return hexlify(str_.encode('utf8'))

# hex -> str
def hex2str(hex_):
    return unhexlify(hex_).decode('utf8')

UNIXタイムスタンプの取得

#from datetime import datetime

#datetime型を渡すとタイムスタンプで返す デフォルトは現在時刻
#def get_timestamp(datetime_=datetime.now()):
#    return round(datetime_.timestamp())

上のやり方は間違い。

from datetime import datetime

#datetime型を渡すとタイムスタンプで返す デフォルトは現在時刻
def get_timestamp(datetime_=None):
    if datetime_ is None:
        datetime_ = datetime.now()
    return round(datetime_.timestamp())

ローカルIPアドレスの取得

from socket import socket, AF_INET, SOCK_DGRAM
from contextlib import closing

def get_ip():
    with closing(socket(AF_INET, SOCK_DGRAM)) as sock:
        sock.connect(('8.8.8.8', 80))
        ip = sock.getsockname()[0]
    return ip

UDP関連

・ip: 自分のローカルIPアドレス
・addr: 送り先のIPアドレス
・bufsizeはいつも4096で設定してる

from socket import *
from contextlib import closing

#ユニキャスト送信
def send(addr, port, data):
    with closing(socket(AF_INET, SOCK_DGRAM)) as sock:
        sock.sendto(data, (addr, port))
    return 1

#ユニキャスト受信
def recv(ip, port, bufsize=4096):
    with closing(socket(AF_INET, SOCK_DGRAM)) as sock:
        sock.bind((ip, port))
        data, addr = sock.recvfrom(bufsize)
    return [data, addr[0]]


#マルチキャスト送信
def multi_send(ip, group_addr, port, data):
    with closing(socket(AF_INET, SOCK_DGRAM)) as sock:
        sock.setsockopt(IPPROTO_IP, IP_MULTICAST_IF, inet_aton(ip))
        sock.sendto(data, (group_addr, port))
    return 1

#マルチキャスト受信
def multi_recv(ip, group_addr, port, bufsize=4096):
    with closing(socket(AF_INET, SOCK_DGRAM)) as sock:
        sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
        sock.bind(('', port))
        sock.setsockopt(IPPROTO_IP, IP_ADD_MEMBERSHIP,
                        inet_aton(group_addr)+inet_aton(ip))

        data, addr = sock.recvfrom(bufsize)
    return [data, addr[0]]

マルチスレッド処理

・マルチスレッドで動かすときによく使う関数たち

from threading import Thread, Event, active_count

#マルチスレッドで動かしたい関数
def function(arg):
    print(arg)

thread = Thread(target=function, args=['ヽ(*゚д゚)ノ',])


thread.start()#マルチスレッド開始

thread.join()#スレッドが死ぬまで待機 タイムアウトを浮動小数点数で設定できる

thread_state = thread.is_alive()#スレッドが生きていればTrueを返す


flag = Event()#スレッド間のフラグ処理に使う

flag.set()#flagにTrueをセット

flag.clear()#flagにFalseをセット

flag.wait()#flagがTrueになるまで待機 タイムアウトを浮動小数点数で設定できる

flag_state = flag.is_set()#flagがTrueだったらTrueを返す


active_count()#生存中のスレッド数を返す

OpenWeatherMapから天気情報を取得

・requestsをインストールしとく
・appidはOpenWeatherMapに登録して入手
・公式のAPI一覧に詳しく載ってる(https://openweathermap.org/api)
・無料で得られる情報は2時間ごとの更新
・1分間で60回までリクエストできる

import requests

#緯度経度から現在の天気情報を取得
def get_current_weatherdata_latlon(appid, lat, lon):
    params = {'appid': appid, 'lat': lat, 'lon': lon}
    res = requests.get('http://api.openweathermap.org/data/2.5/weather', params)
    return res.json()

#緯度経度から天気予報情報を取得(5日間分、3時間毎)
def get_forecast_weatherdata_latlon(appid, lat, lon):
    params = {'appid': appid, 'lat': lat, 'lon': lon}
    res = requests.get('http://api.openweathermap.org/data/2.5/forecast', params)
    return res.json()

TwitterAPI関連

・requestsとrequests_oauthlibをインストールしとく
oauth認証の各キーはtwitterデベロッパーに登録して取得。
・各APIごとに呼び出し制限回数は異なる。制限は15分でリセットされる。
・公式のAPI一覧に詳しく載ってる。(https://dev.twitter.com/docs)
・とりあえずよく使うものだけ

import requests
from requests_oauthlib import OAuth1

oauth_keys = {}
oauth_keys['consumerkey'] = '***************'
oauth_keys['consumersecret'] = '***************'
oauth_keys['accesstoken'] = '***************'
oauth_keys['accesstokensecret'] = '***************'

class TwitterCliant:
    def __init__(self, oauth_keys):
        self.auth = OAuth1(oauth_keys['consumerkey'], oauth_keys['consumersecret'],
                           oauth_keys['accesstoken'], oauth_keys['accesstokensecret'])

    #ツイートを上げる
    def update(self, text):
        url = "https://api.twitter.com/1.1/statuses/update.json"
        data = {"status": text}
        res = requests.post(url, data=data, auth=self.auth)
        return res.json()

    #ツイートIDを指定してリプライ
    def reply(self, id_, text):
        url = "https://api.twitter.com/1.1/statuses/update.json"
        data = {"in_reply_to_status_id": id_, "status": text}
        res = requests.post(url, data=data, auth=self.auth)
        return res.json()

    #ツイートIDを指定してリツイート
    def retweet(self, id_):
        url = "https://api.twitter.com/1.1/statuses/retweet/%s.json"%id_
        res = requests.post(url, auth=self.auth)
        return res.json()

    #ツイートIDを指定してツイート削除
    def delete(self, id_):
        url = "https://api.twitter.com/1.1/statuses/destroy/%s.json"%id_
        res = requests.post(url, auth=self.auth)
        return res.json()

    #ホームタイムラインの取得
    def timeline(self, count):
        url = "https://api.twitter.com/1.1/statuses/home_timeline.json"
        params = {"count": count}
        res = requests.get(url, params=params, auth=self.auth)
        return res.json()

    #キーワードでツイート検索(キーワードはリストで渡す)
    def search(self, keys, count):
        url = "https://api.twitter.com/1.1/search/tweets.json"
        query = ' AND '.join(keys)
        params = {"q": query, "count": count}
        res = requests.get(url, params=params, auth=self.auth)
        return res.json()