強化学習を勉強したい

普段私は業務でAI/機械学習に触れています。
ただ今までは教師あり学習を中心に勉強してきたのもあり、教師なし学習や強化学習についてあまり触れたことが無いなと思いました。
そこで昨日、強化学習ハンズオンに最適なこちらの記事を拝見し、実際自分もやってみようと。

やるにしても何かテーマが欲しいなぁと思いつつ記事を眺めていると、エージェントの動きに「あれこれ酔っ払いみたいじゃね？」と感じてしまいました。
こちらの記事では、始点から終点までの最適経路を求める強化学習をテーマとしていました。
エージェントは80％の確率で希望の方向に進み、10％の確率で左方向、10％の確率で右方向へ行ってしまう設定になっています。

これを使えば「酔っ払いの挙動を再現できるのでは・・・？」と思い、ハンズオンがてら実験してみました。
コロナ前、居酒屋でべろべろになった記憶を思い出しながらご覧ください。
こちらの記事もむちゃくちゃ便利ですのでご覧になってくださいね。

サンプルコード

ほぼこちらの記事にある通りで、酔っ払い変数beroberoを導入するために一部改変しています。
定義したメソッドの中身については↑の記事にあるGithubを参照ください。
改変した箇所だけ取り出します。

#抽象クラス
class MDP:
    #MDP:マルコフ決定過程(Markov decision processes
    #引数に酔っ払い度合いberoberoを定義
    def __init__(self, init, actlist, terminals, gamma=.9,berobero=0.1):
        #init:初期状態
        #actlist:行動
        #erminals:終了状態
        #gamma:割引関数
        self.init = init
        self.actlist = actlist
        self.terminals = terminals

#具象クラス
class GridMDP(MDP):
    #引数に酔っ払い度合いberoberoを定義
    def __init__(self, grid, terminals, init=(0, 0), gamma=.9,berobero=0.1):
        #gridは場を定義する行列
        grid.reverse()  # because we want row 0 on bottom, not on top                                                                                                  
        MDP.__init__(self, init, actlist=orientations,
                     terminals=terminals, gamma=gamma,berobero=berobero)
        self.grid = grid
        self.berobero=berobero
        self.rows = len(grid)
        self.cols = len(grid[0])

    #遷移確率と次の行動のリスト
    #berobero=0がしらふ、0.1がほろよい、0.3でべろべろなイメージ
    #berobero=0.5で確実に蟹歩きになるので一周回って大丈夫的な
    def T(self, state, action):
        if action is None:
            return [(0.0, state)]
        else:
            return [(1-2*self.berobero, self.go(state, action)),
                    (self.berobero, self.go(state, turn_right(action))),
                    (self.berobero, self.go(state, turn_left(action)))]

問題設定

始点から終点までエージェントが進むときに、報酬が最大になるような進み方を探します。
エージェントは一定の確率で希望とは違う方向にすすむ。beroberoの確率で左、beroberoの確率で右、1-2*beroberoの確率で希望通りの方向に進めます。
なので、berobero=0がしらふ、0.1がほろよい、0.3でべろべろなイメージです。
berobero=0.5で確実にカニ歩きになるので「一周回って大丈夫な人」をイメージしてください。

酔っぱらった時は壁沿いを歩く

実際に自分が酔っぱらった時、壁に手をついて壁沿いに移動したりしませんか？
座敷なんかだと「間失礼しまーす」と通るのが怖いので、最短で壁までいって壁に沿って移動するイメージがあります。
これをちょっとシミュレーションしてみたいと思います。

出口の近くに人がいる時

まずはこちらの記事の例を使って検証してみます。
お店には柱が一本あり、出口のそばに人が一人いるパターンを考えます。
人にぶつからずに出口に出られる最適な経路を導きます。

酔っ払いなのであれば可能な限り右にすすんで人にぶつかるリスクは負いたくないものです。
なので直観的には上にいって突き当たってそのまま右にいくのが良さそうです。

lossはそこに行ったときにもらう報酬で、-0.5にすると「移動するごとに-0.5される」みたいな部屋になります。
報酬のマイナスが大きいほど可能な限り最短で移動しようとします。
「早く出口に行かないと吐いちゃうぅぅぅ」みたいな設計はlossを使えばできそうですね。(今回はやりません)

#もともとのパターン
loss = 0
grid=[
    [loss, loss, loss, +1],
    [loss, None,  loss, -1],
    [loss, loss,loss,loss]
]

#左下から数える
sequential_decision_environment = GridMDP(grid,terminals=[(3, 2), (3, 1)],berobero=0.1)

pi = best_policy(sequential_decision_environment, value_iteration(sequential_decision_environment, .01))

print_table(sequential_decision_environment.to_arrows(pi))

ほろよいパターン

berobero=0.1の場合はこちら。
見方としては「そのマスからどっち向きに移動するのがもっとも妥当か」を矢印にして出力しています。
この場合、「始点から上へ行き右へ行くルートが最適だ」という結果になります。
直観的にも妥当そうですね。

べろべろパターン

berobero=0.3の場合はこちら。
面白いのは最初に左を向こうとするんですよね。
意地でも右方向には行かないという意志を感じます。
終点に入る直前も上方向を向くことで「右方向、最悪左に戻ってもよし」みたいにしているのが面白い。

カニ歩きパターン

berobero=0.5の場合がこちら。
もうこれ「やべぇ酔いすぎた・・・横方向に着実にすすんでいこ・・・」という強い意志を感じます。
右を向いて横歩きし、上を向いて横歩きすることで難なくゴール。
前に進めないことを逆手に取った悟りの境地を感じますね。

4つの座敷卓の飲み会から脱出する時

さて、いよいよ実践編です。
4つの卓の飲み会の奥の席から脱出することを考えます。
4つの卓を疑似的に4人の人に見立て、「間失礼しまーす」しながら出口へ行くか、遠回りでも壁沿いに行くかを見てみます。

ほろよいパターン

berobero=0.1の場合はこちら。
この場合は「間失礼しまーす」のパターンですね。
酔っぱらっていないので間を抜けても大丈夫だろう的な気持ちを感じます。

べろべろパターン

berobero=0.3の場合はこちら。
これです！！壁伝いに行こうとする強い意志を感じますね！！
酔っぱらったときはとにかく壁沿いにすすんで無難に行くべし！！

ふと思ったんですがberobero=0.1にしてこれにならない理由って何故なんでしょう？
loss=0なので遠回りするデメリットは無いはずですし、これなら確実です。
ご存じの方はご教示いただけると嬉しいです！

カニ歩きパターン

berobero=0.5の場合がこちら。
前に進めないことを逆手にとってカニ歩きしているのがなんか面白いですね
べろべろになりすぎて危機感感じて冷静になっている姿がありありと浮かびます。

遠回りにもリスクがあるとき

先ほどは遠回りにリスクがありませんでしたが、今回は少しリスクをつけてみます。
近道で行こうが遠回りで行こうが「間失礼しまーす」が生じ、遠回りすると道が太くより安全に右側に進めるシチュエーションを考えます。
こんな居酒屋あるかい。

ほろよいパターン

berobero=0.1の場合はこちら。
この場合は右向きに「間失礼しまーす」のパターンですね。

スタート地点で下方向を向いている現象って何故でしょう・・・？
「いやだぁあ出口行きたくなｱｱｱｱｲｨ」的な厭世的な嘆きを感じます。

べろべろパターン

berobero=0.3の場合はこちら。
安全択を取ったパターンですね。
None一つ上の場所で左を向くことで「右には絶対行かない」という強い意志を感じます。

カニ歩きパターン

berobero=0.5の場合がこちら。
カニ歩きを習得すれば人込みなぞ何のその。

さいごに

今回は強化学習ハンズオンが目的だったので条件検討はまだまだ甘いかもしれません。
特にloss=0で設定したので、そこが変わることによる挙動変化もみたいですね。
また今回は価値反復法という手法をモデルにしましたが、Q-learningについても実装してみたい。

そしてお酒はほどほどに楽しく飲みましょう！！！！
最後まで読んでいただきありがとうございました！
是非是非LGTMしてくれると嬉しいです。

はじめに

Pythonをお使いの方なら既知ですが、Pythonでは型による強制力はありません。
型アノテーション（typing）がPython3.5で追加されるまでは型の記述もできませんでした。
コメントや変数名で空気を読むしかできないわけです。
（個人的に辞書は最悪でした）

なので型アノテーションが実装されてからは本当に助かっています。
もう癖になっているので書かない方が違和感があります。
（同じ思いの方も多いかと思います笑）

ただ依然として強制力はなく、mypyやVSCode拡張のPylanceでチェックができるに留まっています。
第三者にモジュールとして提供する場合はIFの型チェックはできないわけで実装の中でケアしないといけないです。
こういったケア（チェック処理）をわざわざ1つ1つに対して書くのは手間で本質的ではないので，
簡潔に書きたいというのがモチベーションです。

アプローチ

Pythonにはデコレータという便利な機能があります。
デコレータを使うと関数の実行前に処理を実行できるので、
チェック処理をこの中で行ってあげるというアプローチになります。

なので各関数ではデコレータだけ書いてしまえばOKということになります。

実装

以下のようなデコレータ関数を定義します。
error引数には型アノテーションの不一致があった際のErrorを指定できます。
check_all_collectionはCollection型の引数をチェックする際に全件チェックするかどうかを指定できます。

"""
引数の型をチェックするデコレータ定義ファイル
"""
import functools
import inspect
from typing import Any, Union, Callable, _GenericAlias


def check_args_type(error: Exception = TypeError, check_all_collection: bool = False):
    """
    引数の型がアノテーションの型と一致しているかチェックを行うデコレータ関数
    Args:
        error: 不一致時のエラークラス
        check_all_collection: コレクション型の中身を全てチェックするか
    """
    def _decorator(func: Callable):
        @functools.wraps(func)
        def args_type_check_wrapper(*args, **kwargs):
            sig = inspect.signature(func)
            try:
                for arg_key, arg_val in sig.bind(*args, **kwargs).arguments.items():
                    # アノテーションがタイプでない/空の場合は判定しない
                    annotation = sig.parameters[arg_key].annotation
                    if not isinstance(annotation, type) and not isinstance(annotation, _GenericAlias):
                        continue
                    if annotation == inspect._empty:
                        continue

                    # 一致判定
                    # Generic系のタイプだった場合は派生形・一部が一致していればOK
                    is_match = __check_generic_alias(annotation, arg_val, check_all_collection)
                    if not is_match:
                        message = f"引数'{arg_key}'の型が正しくありません。annotaion：{annotation} request:{type(arg_val)}"
                        raise error(message)
            except TypeError as exc:
                raise error("引数の型か数が一致しません。") from exc
            return func(*args, **kwargs)
        return args_type_check_wrapper
    return _decorator

def __check_generic_alias(
    annotation: Union[_GenericAlias, type],
    request: Any,
    check_all_collection: bool = False
):
    """
    GenericAliasの型チェック
    Args:
        annotation: アノテーションタイプ
        request: リクエスト
        check_all_collection: コレクション型の中身を全てチェックするか
    """
    # Anyの場合は型チェックしない
    if annotation == Any:
        return True

    # 型チェック
    request_type = type(request)
    if isinstance(annotation, _GenericAlias):
        if annotation.__origin__ == request_type:    # for collection ...list, dict, set
            # -----------
            # list
            # -----------
            if annotation.__origin__ == list and request:
                _annotation = annotation.__args__[0]
                if check_all_collection:    # 全件チェックの場合は1つずつ確認
                    for _request in request:
                        is_match = __check_generic_alias(
                            _annotation, _request, check_all_collection
                        )
                        if not is_match:
                            return False
                    return True

                else:   # 全件チェックでない場合は先頭を取り出して確認
                    return __check_generic_alias(
                        _annotation, request[0], check_all_collection
                    )

            # -----------
            # dict
            # -----------
            if annotation.__origin__ == dict and request:
                _annotation_key = annotation.__args__[0]
                _annotation_value = annotation.__args__[1]
                if check_all_collection:    # 全件チェックの場合は1つずつ確認
                    for _request in request.keys():
                        is_match = __check_generic_alias(
                            _annotation_key, _request, check_all_collection
                        )
                        if not is_match:
                            return False
                    for _request in request.values():
                        is_match = __check_generic_alias(
                            _annotation_value, _request, check_all_collection
                        )
                        if not is_match:
                            return False
                    return True

                else:   # 全件チェックでない場合は先頭を取り出して確認
                    is_match_key = __check_generic_alias(
                        _annotation_key, list(request.keys())[0], check_all_collection
                    )
                    is_match_value = __check_generic_alias(
                        _annotation_value, list(request.values())[0], check_all_collection
                    )
                    is_match = is_match_key and is_match_value
                    return is_match

            # 中身が存在してない場合，originがあっていればOKとする
            if not request:
                return True

        else:
            # list/dictの場合はoriginが一致していないとエラーとする
            origin = annotation.__origin__
            if origin == list or origin == dict:
                return False
            # それ以外は再帰的にチェック
            else:
                for arg in annotation.__args__:
                    is_match = __check_generic_alias(arg, request)
                    if is_match:
                        return True
    else:
        # BoolはintのサブクラスなのでissubclassでTureとなる
        # 本来意味合いが違うのでNGとしたい
        if request_type == bool and annotation == int:
            return False
        return issubclass(request_type, annotation)
    return False

使用例はその1。

# 一番シンプルなパターン
@check_args_type()
def test(value: int, is_valid: bool) -> float:
    """
    （省略）
    """
    return 0.0

def main():
    # OK
    result = test(5, True)

    # NG -> TypeError
    result = test(0.0, False)

    # NG2 -> TypeError
    result = test(1, "True")

使用例その2。

# Collectionの中身を全てチェックするパターン
@check_args_type(check_all_collection=True)
def test2(value: List[int]) -> List[float]:
    """
    （省略）
    """
    return [0.0]

def main():
    # OK
    result = test2([0, 5, 10, 20])

    # NG -> TypeError
    result = test([0.0, 5.0, 10.0, 20.0])

    # NG2 -> TypeError
    result = test([0, 5, "test"])

Enumやgeneratorなど考慮不足の型があるかと思いますが、基本的な型であれば網羅できているかと思います。
（必要であれば追加していく形でお願いします）

まとめ

関数実行時に引数の型アノテーションをチェックしてエラーにする方法を紹介しました。
これで型による強制力が発揮できます。
厳密性が求められる場面（IFなどの境界）では使えるかなと思います。

PS）
契約プログラミングのように値まで確認できれば最高なので、拡張予定です。

はじめに

Twitterで一時期流行していた 100 Days Of Code なるものを先日知りました。本記事は、初学者である私が100日の学習を通してどの程度成長できるか記録を残すこと、アウトプットすることを目的とします。誤っている点、読みにくい点多々あると思います。ご指摘いただけると幸いです！

今回学習する教材

• Effective Python
  • 8章構成
  • 本章216ページ

今日の進捗

• 進行状況：56-60ページ
• 第3章：クラスと継承
• 本日学んだことの中で、よく忘れるところ、知らなかったところを書いていきます。

辞書やタプルで記録管理するよりもヘルパークラスを使う

辞書やタプルで記録管理するのは、使いやすい反面、複雑になってしまうと可読性を著しく下げてしまう。
一例として、あるお店の売り上げを記録するクラスを定義する。

class SimpleSalesRecord(object):
    def __init__(self):
        self._record = {}

    def add_shop_id(self, shop_id):
        self._record[shop_id] = []

    def report_record(self, shop_id, price):
        self._record[shop_id].append(price)

    def average_price(self, shop_id):
        records = self._record[shop_id]
        return sum(records) / len(records)

record = SimpleSalesRecord()
record.add_shop_id(111)
record.report_record(111, 90)
record.report_record(111, 80)
print(record.average_price(111))

出力結果

850.0

このクラスを拡張して、商品ごとの売上を管理するようにする場合を考える。

class ByitemSalesRecord(object):
    def __init__(self):
        self._record = {}

    def add_shop_id(self, shop_id):
        self._record[shop_id] = {}         # リストから辞書に変更

    def report_record(self, shop_id, item, price):
        by_item = self._record[shop_id]
        record_list = by_item.setdefault(item, [])
        record_list.append(price)

    def average_price(self, shop_id):
        by_item = self._record[shop_id]
        total, count = 0, 0
        for prices in by_item.values():
            total += sum(prices)
            count += len(prices)
        return total / count

record = ByitemSalesRecord()
record.add_shop_id(111)                     # shop_idが111の店を追加
record.report_record(111, 'Apple', 100)     # 111店の売り上げを追加
record.report_record(111, 'Apple', 120)     # 111店の売り上げを追加

print(record.average_price(111))   # 111店の売り上げの平均を算出

出力結果

Apple: 110.0
Orange: 90.0

さらに、賞味期限が近いものは値引きするという機能を実装する。

class DiscountSalesRecord(object):
    def __init__(self):
        self._record = {}

    def add_shop_id(self, shop_id):
        self._record[shop_id] = {}         # リストから辞書に変更

    def report_record(self, shop_id, item, price, weight):
        by_item = self._record[shop_id]
        record_list = by_item.setdefault(item, [])
        record_list.append((price, weight))

    def average_price(self, shop_id):
        by_item = self._record[shop_id]
        price_sum, price_count = 0, 0
        for item, prices in by_item.items():
            item_avg, total_weight = 0, 0
            for price, weight in prices:
                price_sum += price * weight
                price_count += 1
        return price_sum / price_count

record = DiscountSalesRecord()
record.add_shop_id(111)                          # shop_idが111の店を追加
record.report_record(111, 'Apple', 100, 0.8)     # 位置引数が何を意味するのか明確ではない
record.report_record(111, 'Apple', 120, 1) 
print(record.average_price(111))                 # 111店の売り上げの平均を算出

average_priceメソッドでは、ループの中にループがあり、読みにくくなっている。このような複雑さが生じた場合は、辞書とタプルからクラス階層にシフトすべき頃合いである。

クラスへのリファクタリング

上記のコードをクラスへリファクタリングしたコードは以下のようになる。
collectionモジュールのnamedtupleは通常のタプルと異なり、キーワード引数でも指定可能。

import collections

Record = collections.namedtuple('Record', ('price', 'weight'))

class Item(object):
    def __init__(self):
        self._records = []

    def report_record(self, price, weight=1):
        self._records.append(Record(price, weight))

    def average_price(self):
        total, total_weight = 0, 0
        for record in self._records:
            total += record.price * record.weight
        return total / len(self._records)

class Shop_ID(object):
    def __init__(self):
        self._shop_items = {}

    def item(self, name):
        if name not in self._shop_items:
            self._shop_items[name] = Item()
        return self._shop_items[name]

    def average_price(self):
        total, count = 0, 0
        for shop_item in self._shop_items.values():
            total += shop_item.average_price()
            count += 1
        return total / count

class SalesRecord(object):
    def __init__(self):
        self._shop_id = {}

    def shop_id(self, name):
        if name not in self._shop_id:
            self._shop_id[name] = Shop_ID()
        return self._shop_id[name]

record = SalesRecord()
shop_111 = record.shop_id(111)
apple = shop_111.item('Apple')
apple.report_record(100, 0.8)
apple.report_record(100)
print(shop_111.average_price())

まとめ

• 値が他の辞書や長いタプルであるような辞書は作らない
• 辞書が複雑になったら、複数のヘルパークラスを使うように変更する

目的

複数の画像の各ピクセルの代表値を使って画像を生成する。
例えば10枚の画像の平均画像を作りたい。

準備

ライブラリのImport

import numpy as np
from scipy import stats

画像

10枚のRGB画像。
10枚の平均画像、中央値画像、最頻値画像をそれぞれ生成する。

imgs.shape    # (10, 128, 128, 3)

平均

x.shapeでaxisに設定したい軸を確認できる。

img_mean = np.mean(imgs, axis=0)
img_mean.shape    # (128, 128, 3)

中央値

img_median = np.median(imgs, axis=0)
img_median.shape    # (128, 128, 3)

最頻値

numpyに最頻値を求めるライブラリがないのでscipy。
他にいい方法あるかも。

# 返値がmodeとcountの2つあるのでmodeだけ拾う
img_mode = stats.mode(imgs, axis=0)[0]

# 形を他の代表値画像と合わせる
img_mode.shape    # (1, 128, 128, 3)
img_mode = img_mode.reshape(128, 128, 3)

img_mode.shape    # (128, 128, 3)

はじめに

・この記事では、家計簿アプリZaimからダウンロードしたデータを使います。もっと言えば、pythonを用いて、データから１年間の収入と支出をまとめたレポートをpdf形式で自動生成します。

・ソースコードと記事は、たぶん書き方が下手くそで美しくないところが多いと思いますが、許してください。

・筆者は「pythonを使いこなせば自動化できる！」って謳い文句につられてpythonを勉強し始めました。
　でも、初心者に優しい実践的な自動化の具体例があまりないと思っています。（大きな偏見あり）
　そこで、特に初心者の方に、pythonで自動化できるものには、こんなのがあるんだと知ってもらえると嬉しいです。

家計簿アプリZaimってなに？

↓ とりあえず公式のホームページ
https://zaim.co.jp/

筆者が愛用している家計簿アプリ。(2020年5月ぐらいから有料会員。）
月ごとの収入と支出等を、円グラフや積立棒グラフ等を使って分析できるのですごい便利。

そして、Zaimの有料会員になると、集めてきたデータをcsv形式でダウンロードすることが出来る。（ファイルをQiitaの記事に載せる方法が分からなかったので、写真だけ。）

作成した各ファイル，フォルダの概要

↓ 作業フォルダの構造

簡単に言うと、
・dataフォルダの中に、Zaimからダウンロードしたデータファイルが入ってます。（ファイル名：Zaim.2020.csv）

・font_dataフォルダは、pythonでpdfに書き込むためのフォントファイルが入ってます。（ファイル名：GenShinGothic-Monospace-Medium.ttf）

・analyze.pyでは、まず、Zaimからのデータファイル（Zaim.2020.csv）をpandas等で読み取ります。次に、pandas、numpyなどを使ってゴネゴネとデータを整形し、図を作ったり整形したデータを新しいファイルに出力したりします。

・format_dataフォルダでは、analyze.pyで整形したデータが、csv形式で入っています。（ファイル名：diff_money.csv）

・graphフォルダでは、analyze.pyで作成した画像データが、jpg形式で入っています。（ファイル名： in_money.jpg,　out_money.jpg,　money.jpg,　difference_money.jpg）

・report_pdf.pyでは、まず、作成するpdfレポートの細かな設定をします。次に、graphフォルダ内のmoney.jpg,　difference_money.jpgの画像をpdfに貼り付けます。そして、format_dataフォルダ内のdiff_money.csvから、１年間でどれだけの収支になったのかを読み取りpdfに書き込みます。その後、pdfファイルとして出力します。

・reportフォルダでは、report_pdf.pyで出力されたpdfデータが、pdfファイルとして入っています。（ファイル名：balance report.pdf）

analyze.pyのソースコード

↓ ソースコード

analyze.py　ライブラリのインポートとグラフの体裁

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

from matplotlib import rcParams
# グラフの体裁を整える
rcParams['font.family'] = 'sans-serif' #使用するフォント
rcParams['font.sans-serif'] = ['Hiragino Maru Gothic Pro', 'Yu Gothic', 'Meirio', 'Takao', 'IPAexGothic', 'IPAPGothic', 'VL PGothic', 'Noto Sans CJK JP']
rcParams['xtick.direction'] = 'in'#x軸の目盛線が内向き('in')か外向き('out')か双方向か('inout')
rcParams['ytick.direction'] = 'in'#y軸の目盛線が内向き('in')か外向き('out')か双方向か('inout')
rcParams['xtick.major.width'] = 1.0#x軸主目盛り線の線幅
rcParams['ytick.major.width'] = 1.0#y軸主目盛り線の線幅
rcParams['font.size'] = 15 #フォントの大きさ
rcParams['axes.linewidth'] = 1.0# 軸の線幅edge linewidth。囲みの太さ

グラフの体裁は良い記事がいっぱいあるから美味しいとこをもらおう。
https://qiita.com/qsnsr123/items/325d21621cfe9e553c17

analyze.py　関数の定義

 # 年月日を月ごとに使いやすい形へ変形
def devide_month(data):
    data = np.array(data)
    # 各月ごとのデータを入れるためのリストを用意
    month_01, month_02, month_03, month_04 = list(), list(), list(), list()
    month_05, month_06, month_07, month_08 = list(), list(), list(), list()
    month_09, month_10, month_11, month_12 = list(), list(), list(), list()
    # dataから月ごとの日付とkeyを取得
    for i in range(len(data.T[1])):
        # 1月のデータをmonth_01に格納
        if data.T[1][i] == "01":
            # 日付とdataframeのkeyをmonth_1に格納
            month_01.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "02":
            month_02.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "03":
            month_03.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "04":
            month_04.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "05":
            month_05.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "06":
            month_06.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "07":
            month_07.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "08":
            month_08.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "09":
            month_09.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "10":
            month_10.append([data.T[2][i], data.T[3][i]])
        elif data.T[1][i] == "11":
            month_11.append([data.T[2][i], data.T[3][i]])
        else:
            month_12.append([data.T[2][i], data.T[3][i]])
    # 月ごとのデータをまとめる
    date_month = [ month_01, month_02, month_03, month_04,
                   month_05, month_06, month_07, month_08,
                   month_09, month_10, month_11, month_12, ]
    return date_month

# dataframeから特定の種別のデータを引っ張って来る
def collect_data_from_dataframe(dataframe, name, date_month):
    monthly_data = list()
    for i in range(len(date_month)):
        # 月の中にデータがあれば、dataframeからデータを取得
        if date_month[i] != []:
            monthly_data.append(dataframe[name][int(date_month[i][0][1]):int(date_month[i][-1][1])+1].tolist())
        else:
            monthly_data.append(None)

    return monthly_data

# 月ごとの支出の合計金額をリストに格納する
def monthly_sum_money(money):
    sum_list = list()
    for i in range(len(money)):
        if money[i] == None:
            sum_list.append(0)
        else:
            sum_list.append(np.sum(money[i]))
    return sum_list

analyze.py　グラフのためのデータの準備

# Zaimからダウンロードしたデータのパス
read_data_path = "data/Zaim.2020.csv"
# データを読み込んでpandasに格納。
df = pd.read_csv(read_data_path, encoding="shift-jis")



key = 0
date_2020 = list()

# 日付のデータを取り出す
for date in df["日付"]:
    # 日付を年、月、日ごとに分割してリストに格納
    date = date.split("-")
    # 日付と元々のデータフレームを対応させるために、keyを付け加える
    date.append(str(key))
    # 日付とkeyを合わせたものをリストに加える
    date_2020.append(date)
    key += 1

# date_month.shape: [月][月のデータの番号][日,key]
date_month_2020 = devide_month(date_2020)

# 支出を月ごとにまとめて、リストで保持。
out_money = collect_data_from_dataframe(df, "支出", date_month_2020)
# 支出の各月の合計を計算し、リストで保持。
out_money_monthly = monthly_sum_money(out_money)

# 収入を月ごとにまとめて、リストで保持。
in_money = collect_data_from_dataframe(df, "収入", date_month_2020)
# 収入の各月の合計を計算し、リストで保持。
in_money_monthly = monthly_sum_money(in_money)

# 収支を計算、リスト形式で保持
difference_money = (np.array(in_money_monthly) - np.array(out_money_monthly)).tolist()

グラフの作成と保存

# グラフの保存先
save_file_path = "./graph/"
# 年間の支出のグラフ
# 去年の3月より前は完全にデータがないので省く（4月から12月まで）
plt.plot(range(4, 13), np.array(out_money_monthly[3:])/10000, color="blue", label="支出")
plt.xlabel('月')
plt.ylabel('支出 [万円]')
plt.grid(linestyle= '--')
plt.xlim(1,12)
plt.ylim(0,20)
# 軸の目盛間隔などの設定。この場合、x軸は1から12まで1ずつの間隔となる。
plt.xticks(np.arange(1, 13, 1))
plt.yticks(np.arange(0, 21, 1))
# グラフの保存
plt.savefig(save_file_path + "out_money.jpg")
plt.cla()

# 年間の収入のグラフ
plt.plot(range(4, 13), np.array(in_money_monthly[3:])/10000, color="red", label="収入")
plt.xlabel('月')
plt.ylabel('収入 [万円]')
plt.grid(linestyle= '--')
plt.xlim(1,12)
plt.ylim(0,20)
plt.xticks(np.arange(1, 13, 1))
plt.yticks(np.arange(0, 21, 1))

plt.savefig(save_file_path + "in_money.jpg")
plt.cla()


# 年間の支出と収入のグラフ
# 収入は赤線で、支出は青線で表示
plt.plot(range(4, 13), np.array(in_money_monthly[3:])/10000, color="red", label="収入")
plt.plot(range(4, 13), np.array(out_money_monthly[3:])/10000, color="blue", label="支出")
plt.legend(loc="lower right")
plt.xlabel('月')
plt.ylabel('金額 [万円]')
plt.grid(linestyle= '--')
plt.xlim(1,12)
plt.ylim(0,20)
plt.xticks(np.arange(1, 13, 1))
plt.yticks(np.arange(0, 21, 1))

plt.savefig(save_file_path + "money.jpg")
plt.cla()


# 年間の収支のグラフ
# 棒グラフで表現する際に、各月の収支がプラスかマイナスで色分けしたいため、データを二つに分ける。
positive_money = list()
negative_money = list()
positive_x = list()
negative_x = list()
# 収入と支出の差がプラスかマイナスでデータを分ける。
for i in range(len(difference_money)):
    if difference_money[i] >= 0:
        positive_money.append(difference_money[i])
        positive_x.append(i+1)
    else:
        negative_money.append(difference_money[i])
        negative_x.append(i+1)

# 収支がプラスは赤色、収支がマイナスの月は青色で表現。
plt.bar(positive_x, np.array(positive_money)/10000, color="red", label="収支＋")
plt.bar(negative_x, np.array(negative_money)/10000, color="blue", label="収支－")
plt.legend(loc="upper right")
plt.xlabel('月')
plt.ylabel('収支 [万円]')
plt.grid(linestyle= '--')
plt.xlim(1,12)
plt.ylim(-8,8)
plt.xticks(np.arange(1, 13, 1))
plt.yticks(np.arange(-8, 9, 1))

plt.savefig(save_file_path + "difference_money.jpg")
plt.cla()

# pdfレポートの自動作成のために、データを外部に出力
format_data_path = "format_data/"
np.savetxt(format_data_path + "diff_money.csv", difference_money)

report_pdf.pyのソースコード

この記事読んでないとpdfに出力する（書き込む？）やり方分からなくて出来ていなかったと思う。圧倒的感謝！
https://watlab-blog.com/2020/03/21/reportlab-pdf/

import numpy as np

from reportlab.pdfgen import canvas
from reportlab.lib.units import mm
from reportlab.pdfbase.pdfmetrics import registerFont
from reportlab.pdfbase.ttfonts import TTFont


# フォントを登録
registerFont(TTFont('GenShinGothic',
                    './font_data/GenShinGothic-Monospace-Medium.ttf'))

file_path = 'report/balance report.pdf'  # 出力ファイル名を設定
graph_path_transition = "graph/money.jpg"
graph_path_balance = "graph/difference_money.jpg"


# １年間の収支レポートを作成
paper = canvas.Canvas(file_path)  # 白紙のキャンバスを用意
paper.saveState()  # 初期化
paper.setFont('GenShinGothic', 20)  # フォントを設定

# 横wと縦hの用紙サイズを設定
w = 210 * mm
h = 297 * mm

paper.setPageSize((w, h))  # 用紙のサイズをセット
paper.drawString(w/2 - 90, h - 50,  # テキストの書き込み
                 '１年間の支出と収入')


# 画像を埋め込み(画像ファイルのパス, 横位置, 縦位置, 画像横サイズ, 画像縦サイズ)
paper.drawInlineImage(graph_path_transition, 31*mm, h-130*mm, 148*mm, 111*mm)
paper.setFont('GenShinGothic', 15)  # フォントを設定
paper.drawString(w/2 - 112.5, 160*mm, '図１　１年間の収入と支出の推移')

# ２つ目の画像を埋め込み(画像ファイルのパス, 横位置, 縦位置, 画像横サイズ, 画像縦サイズ)
paper.drawInlineImage(graph_path_balance, 31*mm, h-250*mm, 148*mm, 111*mm)
paper.setFont('GenShinGothic', 15)  # フォントを設定
paper.drawString(w/2 - 67.5, 40*mm, '図２　１年間の収支')

# 収支の結果を記入
difference_money = np.loadtxt('format_data/diff_money.csv', delimiter=',') # 収支のデータを読み込む
paper.setFont('GenShinGothic', 13)  # フォントを設定
paper.drawString(w/2-100, 23*mm,
                 f'１年間全体の収支は、{int(np.sum(difference_money))}円です。')
paper.drawString(w/2-100, 14 * mm,
                 f'つまり、１ヶ月あたり、{int(np.sum(difference_money)/len(difference_money))}円です。')

paper.save()  # PDFを保存

実行結果

analyze.pyの実行結果

analyze.pyを実行した結果グラフは、in_money.jpg,　out_money.jpg,　money.jpg,　difference_money.jpgの４つが作成されるが、内２つ（in_money.jpg,　out_money.jpg）は、money.jpgと内容が被っているので割愛する。

　　　　　　図１　１年間の収入と支出の推移（money.jpg）

　　　　　　図２　１年間の収支の推移（difference_money.jpg）

report_pdf.pyの実行結果

report_pdf.pyにより自動作成された家計簿レポートは次の通りになる。

経緯

ROSを使って色々イジイジしていたときのことです。カメラの入力ノードからRGBデータが送られてくるので、それを別のノードで受け取りたいと思い、cv_bridgeを使おうとしていました。下のような感じです。

from cv_bridge import CvBridge

def prediction(msg):
    bridge = CvBridge()
    img = bridge.imgmsg_to_cv2(msg, "bgr8")

しかし、ビルドして実行してみたところ、以下のようなエラーが発生しました。

File "/opt/ros/melodic/lib/python2.7/dist-packages/cv_bridge/core.py", line 91, in encoding_to_cvtype2
    from cv_bridge.boost.cv_bridge_boost import getCvType
ImportError: dynamic module does not define module export function (PyInit_cv_bridge_boost)

調べてみると、cv_bridgeがpython2でビルドされていることが原因のようでした。自分は実行環境がpython3(virtualenv)だったので、cv_bridgeをソースコードからビルドする必要がありました。

環境

• ROS melodic
• python3.6(virtualenv)
• ubuntu18.04
• Jetson AGX Xavier

最初にやった方法

最初はこちらの方法を参考にしてやってみました。ローカルのpython3が実行環境の場合は、上手く行くのかもしれませんが、virtualenvを使っていた自分の環境ではうまく行きませんでした。一応共有しておきます。

$ cd catkin_ws
$ catkin config -DPYTHON_EXECUTABLE=/usr/bin/python3 -DPYTHON_INCLUDE_DIR=/usr/include/python3.5m -DPYTHON_LIBRARY=/usr/lib/aarch64-linux-gnu/libpython3.5m.so
$ catkin config --install
$ git clone https://github.com/ros-perception/vision_opencv.git src/vision_opencv
$ apt-cache show ros-melodic-cv-bridge | grep Version
$ cd src/vision_opencv/
# たぶんの下のバージョンだった気がします。apt-cache showで出てきたものです。
$ git checkout 1.13.0
$ cd ../../
$ catkin build cv_bridge
$ source install/setup.bash --extend

virtualenvを使っていた自分の環境では、cv_bridgeは無事にビルドされるのですが、ビルド後に自前の他のノードを一緒のcatkin_wsでビルド実行すると下のようなエラーが発生してしまいました。

RLException: [sample.launch] is neither a launch file in package [sample_proc] nor is [sample_proc] a launch file name

ここから結構沼にハマってしまいました。犠牲者を減らすために、自分が解決した方法を共有します。

最終的な解決法

こちらのリポジトリの環境が、自分と同じようにvirtualenv上のpython3でcv_bridgeを使っていたので、そこからヒントを得ました。
大事なことは、2つあります。1つは、cv_bridgeのビルドを自前のワークスペースとは別のワークスペースで行うことです。もう1つは、自分のスクリプトのcv_bridgeをimportしている行の前で、python2.7のパスを削除することです。具体的な手順は以下の通りです。

$ mkdir -p cv_bridge_ws/src
$ git clone https://github.com/ros-perception/vision_opencv.git src/vision_opencv
$ apt-cache show ros-kinetic-cv-bridge | grep Version
$ cd src/vision_opencv/
# apt-cache showの結果でversionの値は変わります
$ git checkout 1.13.0
$ cd ../../
# 環境により、DPYTHON_LIBRARYのaarch64-linux-gnuの部分は変わります
$ catkin build
$ catkin config -DCMAKE_BUILD_TYPE=Release -DPYTHON_EXECUTABLE=/usr/bin/python3 -DPYTHON_INCLUDE_DIR=/usr/include/python3.6m -DPYTHON_LIBRARY=/usr/lib/aarch64-linux-gnu/libpython3.6m.so
$ source devel/setup.bash --extend
$ cd ../catkin_ws
# 自前のノードをビルド
$ catkin build
$ source devel/setup.bash

さらに自分のスクリプトも変更します。

import sys
sys.path.remove('/opt/ros/melodic/lib/python2.7/dist-packages')
from cv_bridge import CvBridge

def prediction(msg):
    bridge = CvBridge()
    img = bridge.imgmsg_to_cv2(msg, "bgr8")

これにより、python2のcv_bridgeではなく、python3のcv_bridgeが実行時に呼び出されるようになります。

まとめ

もうmelodicはいやだ(笑)。noeticかROS2使いたい。

はじめに

 こんにちは、麻菜結です。必要に迫られてしこしこ作ったプログラムが良い感じに作れたので紹介する記事です。作ったのはそこそこ実用的な連立一次方程式を解くプログラムです。バグがあったら優しく教えてください。なお、掃き出し方そのものの説明は行わないのでご了承ください。

環境

 環境はPython3系です。

> python --version
Python 3.9.0

プログラム

 プログラムは以下に示すものです。

solve_PE.py

import sys

def hakidashi(m, show_matrix):
    num_line = len(m)
    num_column = len(m[0])
    div = lambda a, b: 0 if b==0 else a / b
    for p in range(num_line):
        m[p] = [div(n, m[p][p]) for n in m[p]]
        for i in range(num_line):
            x = div(m[i][p], m[p][p])
            for j in range(num_column):
                if(i != p):
                    m[i][j] = m[i][j] - x * m[p][j]
    if(show_matrix):
        for e in m:
            print(e)
    return list(map(lambda m: m[-1], m))

def dec_to_frac(f, max_denominator, calc_accuracy):
    for i in range(1, max_denominator + 1):
        k1 = int(f * i)
        k2 = k1 + 1
        if(abs((k1 / i) - f) < calc_accuracy):
            return f"{k1}/{i}"
        elif(abs((k2 / i) - f) < calc_accuracy):
            return f"{k2}/{i}"
    else:
        return "not found"

def read_file(file_name, file_codec):
    c, m = [], []
    with open(file_name, encoding=file_codec) as f:
        lists = list(f.readlines())
        c = split_line(lists[0])
        for l in lists[1:]:
            m.append(split_line(l, num=True))
    return c, m

def split_line(line, num=False):
    line = line.strip()
    line += "#"
    data_line = []
    word = ""
    is_firstblank = True
    for e in line:
        if(" " == e and is_firstblank):
            data_line.append(word)
            is_firstblank = False
            word = ""
        elif("," == e):
            data_line.append(word)
            word = ""
        elif("#" == e):
            data_line.append(word)
            break
        else:
            is_firstblank = True
            word += e
    else:
        data_line.append(word)
    return [
        float(x.strip())
        if num else
        x.strip()
        for x in data_line 
        if x not in ["", " "]
    ]

if __name__ == "__main__":
    file_name = sys.argv[1]
    file_encode = "utf-8"
    max_denominator = 10000000
    calc_accuracy = 0.0000000001
    show_matrix = True

    c, m = read_file(file_name, file_encode)
    if(len(c) + 1 == len(m[0])):
        ans = hakidashi(m, show_matrix)
        for c, ans in zip(c, ans):
            f = dec_to_frac(ans, max_denominator, calc_accuracy)
            print(f"{c} = {ans} ({f})")
    else:
        print("data error", file=sys.stderr)

 以下はそれぞれの説明です。

掃き出し方

 hakidashi()が該当関数です。特に工夫はありませんが、ゼロで割っちゃうエラーを回避するためにdivという関数を内部で定義しており、ゼロで割りそうなときは0を返すようにしてあります。リターンの直前にprintしているのは、もし解が無いような方程式を渡されてしまったとき、定数項が正しく1にならないのでそこで見て判断できます。

分数表現

 dec_to_frac()が該当関数です。分母を変えつつ検証していって、許せる精度になったら分数表現にした文字列を返します。もし範囲に見当たらなかったときはnot foundを表示します。ここがforループで計算回数が多いのであんまりに重かったらmax_denominatorやcalc_accuracyの値を調整してください。

こちらのサイトを参考にさせていただきました。
FloatToFraction (JavaScript版)

ファイルの読み込み

 read_file()とsplit_line()が該当関数です。
 read_fileではコマンドラインからもらったファイル名からファイルデータを開き、split_lineに渡しています。cは文字ラベル(xとかyとか)を入れる配列、mは数値の配列データです。
 split_lineは、カンマセパレートでもスペースセパレートでも読み込めるように、またはコメントを書き込めるように丁寧に分けるやつです。

使い方

たとえば、以下のようなテキストを記述したファイルを使います。

mat1.txt

x y z
4 2 1 26
1 6 3 48
4 1 5 34

上のは以下の連立方程式を表しています。

$$
4x + 2y + z = 26 \\
x + 6y + 3z = 48 \\
4x + y + 5z = 34
$$

> python solve_PE.py mat1.txt
[1.0, 0.0, 0.0, 2.727272727272727]
[0.0, 1.0, 0.0, 5.818181818181818]
[0.0, 0.0, 1.0, 3.454545454545455]
x = 2.727272727272727 (30/11)
y = 5.818181818181818 (64/11)
z = 3.454545454545455 (38/11)

また、以下のように解が無い場合を考えます。

mat2.txt

x1, x2    # 下式が上式の
3, 6, 9   # ただの二倍になっているので
6, 12, 18 # 解はない

なお、#はコメントを表し、実際にファイルに記述することが可能です。

> python solve_PE.py mat2.txt
[1.0, 2.0, 3.0]
[0, 0, 0]
x1 = 3.0 (3/1)
x2 = 0 (0/1)

値は適当に入れられていますが、上の行列が三角っぽくないのでおかしいことがわかります。

おわりに

 お疲れさまでした。上手く動きましたか？百行も満たないのに結構使えるプログラムになったのでPythonすごいというのと、先人の考えたアルゴリズムすごいなという気持ちです（小学生並の感想）。やっぱ自分で組めると楽しいですね、こういうことに幸せを感じて生きていきたい。
 最後まで読んでくださってありがとうございます。役に立ったなら幸いです。どうでもいいですけど、私の次は何か書こうかなというのは本当にあてになりませんね。自分で過去の記事を見てびっくりします。手がかじかんできたのでこの辺にします。ありがとうございました。

機械学習の画像集めにicrawlerを利用したのでその紹介です。

icrawlerとは

pythonでwebクローリングを行い、画像を集めるためのフレームワークです。
非常に短いコードを記述するだけで画像を集めることができます。

インストール

pip

$ pip install icrawler

anaconda

$ conda install -c hellock icrawler

使い方

from icrawler.builtin import BingImageCrawler

crawler = BingImageCrawler(storage={"root_dir": './images'})
crawler.crawl(keyword='猫', max_num=100)

• root_dirに画像の保存先ディレクトリを指定します。
• keywordに集めたい画像のキーワードを指定します。
• max_numに集める画像の枚数を指定します。
• BingImageCrawlerの部分を他のImageCrawlerに変えることもでき、GoogleやFlickerも利用できます。
  • 利用可能なもの → https://icrawler.readthedocs.io/en/latest/builtin.html

Google利用時にjson.decoder.JSONDecodeErrorが出る際の対処法

1. google.pyを見つけます。
   • 例(anaconda利用): C:\Users\hoge\anaconda3\envs\env1\Lib\site-packages\icrawler\builtin\google.py
   • pip でインストールしている場合はパッケージの場所を検索できるのでそこから辿ってください
   • https://qiita.com/t-fuku/items/83c721ed7107ffe5d8ff
2. google.pyのparseメソッドを下記に変更します。
   • parseメソッドは144行目あたりにあります。

def parse(self, response):
        soup = BeautifulSoup(
            response.content.decode('utf-8', 'ignore'), 'lxml')
        #image_divs = soup.find_all('script')
        image_divs = soup.find_all(name='script')
        for div in image_divs:
            #txt = div.text
            txt = str(div)
            #if not txt.startswith('AF_initDataCallback'):
            if 'AF_initDataCallback' not in txt:
                continue
            if 'ds:0' in txt or 'ds:1' not in txt:
                continue
            #txt = re.sub(r"^AF_initDataCallback\({.*key: 'ds:(\d)'.+data:function\(\){return (.+)}}\);?$",
            #             "\\2", txt, 0, re.DOTALL)
            #meta = json.loads(txt)
            #data = meta[31][0][12][2]
            #uris = [img[1][3][0] for img in data if img[0] == 1]

            uris = re.findall(r'http.*?\.(?:jpg|png|bmp)', txt)
            return [{'file_url': uri} for uri in uris]

参考

https://github.com/hellock/icrawler
https://github.com/hellock/icrawler/issues/65

はじめに

日本分類学会の学会誌「データ分析の理論と応用」のVol.5 (2016)に載っていた、"対応分析を用いたアソシエーションルールによるアンケート結果の可視化" についてアンケート以外にも使えるんじゃないか？と思ってPythonで実施してみた、という話。
networkxの練習も兼ねている。
書きなぐったようなクソコードをさらしていくスタイル。

ざっくり内容説明

当該論文について、簡単に説明すると、コレスポンデンス分析の結果のプロット上に、アソシエーション分析の結果の組み合わせ同士を線で繋げてあげるイメージ。論文中では"属性特化型特徴抽出アソシエーションプロット"と呼ばれている。
論文中の例として、メディア層(属性)別の健康上の悩みアンケート結果に対して可視化を行っている。
可視化の結果、例えば「F3層は○○の悩みがある傾向で、一方F1層・F2層は××の悩みがある傾向だが、~~の悩みに関してはF2層F3層共通の悩みである」といったことがわかる。

POSデータへの適用

例えばID-POSデータを使って属性に当たる部分を地域や店舗、アンケートに当たる部分を購入商品カテゴリーやブランドの購入回数とすると、地域や店舗の購買傾向を可視化することができる。
実際に、計算機統計学 29巻 22号 の"売り上げ傾向による店舗の分類と購買傾向の分析と可視化"では店舗別の購買傾向について分析を行っている。
今回の試みも、POSデータに対して実施した。

使用データ

kaggleのsuperstore_dataを使う。
4年間のグローバルスーパーストアの小売データで、Customer IDやProduct ID、Cityなどがあるが、店舗IDのようなものは無い。
店舗IDが無いので、今回は国と、商品サブカテゴリーの購買傾向を可視化する。

<class 'pandas.core.frame.DataFrame'>
RangeIndex: 51290 entries, 0 to 51289
Data columns (total 24 columns):
 #   Column          Non-Null Count  Dtype         
---  ------          --------------  -----         
 0   Row ID          51290 non-null  int64         
 1   Order ID        51290 non-null  object        
 2   Order Date      51290 non-null  datetime64[ns]
 3   Ship Date       51290 non-null  datetime64[ns]
 4   Ship Mode       51290 non-null  object        
 5   Customer ID     51290 non-null  object        
 6   Customer Name   51290 non-null  object        
 7   Segment         51290 non-null  object        
 8   City            51290 non-null  object        
 9   State           51290 non-null  object        
 10  Country         51290 non-null  object        
 11  Postal Code     9994 non-null   float64       
 12  Market          51290 non-null  object        
 13  Region          51290 non-null  object        
 14  Product ID      51290 non-null  object        
 15  Category        51290 non-null  object        
 16  Sub-Category    51290 non-null  object        
 17  Product Name    51290 non-null  object        
 18  Sales           51290 non-null  float64       
 19  Quantity        51290 non-null  int64         
 20  Discount        51290 non-null  float64       
 21  Profit          51290 non-null  float64       
 22  Shipping Cost   51290 non-null  float64       
 23  Order Priority  51290 non-null  object        
dtypes: datetime64[ns](2), float64(5), int64(2), object(15)
memory usage: 9.4+ MB

コードを書く

準備

まず必要なパッケージをimport

# パッケージインポート
import numpy as np
import scipy
import matplotlib as mpl
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
from pandas.plotting import register_matplotlib_converters
import sklearn
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
import os
import mlxtend
from mlxtend.preprocessing import TransactionEncoder
from mlxtend.frequent_patterns import apriori, association_rules, fpgrowth
import networkx as nx
import mca
import codecs
sns.set()
# 日本語を使うときは必要になる(今回は不必要)
font_path = 'C:\\Users\\[YOUR_USERNAME]\\Anaconda3\\envs\\[ENV_NAME]\\Lib\\site-packages\\matplotlib\\mpl-data\\fonts\\ttf\\ipaexg.ttf'
font_prop = mpl.font_manager.FontProperties(fname=font_path)

# それぞれのパッケージのバージョン
"""
numpy 1.18.1
scipy 1.4.1
matplotlib 3.1.3
seaborn 0.10.0
pandas 1.0.3
sklearn 0.22.1
mlxtend 0.17.3
networkx 2.5
mca 1.0.3
"""

データを読み込む。今回は処理を軽くするために8割のデータは捨てる。

with codecs.open('superstore_dataset2011-2015.csv', "r", "shift-jis", "ignore") as f:
    df = pd.read_csv(f, parse_dates=['Order Date','Ship Date'], dayfirst=True)
# 後の処理のために一部のカラムを加工しておく
df['Order Date']=pd.to_datetime(df['Order Date'])
df['Ship Date']=pd.to_datetime(df['Ship Date'])
df['Country']='Country_'+df['Country']
df['Sub-Category']='SubC_'+df['Sub-Category']
# 処理が重かったので、8割くらいのデータは捨てる
df, gomi=train_test_split(df, test_size=0.8, random_state=0)
display(df)

# それぞれのカラムの値のユニークな数
"""
Customer ID 1493
Category 3 ;['Office Supplies' 'Technology' 'Furniture']
Sub-Category 17
Product Name 3038
Region 13
Market 7
Country 137
State 891
City 2431
"""

アソシエーション分析

まず、アソシエーション分析を行うためにエンコードなどデータ加工を行う。
※エンコードは昇順で番号が付けられていくので注意。今回はCountryがSub-Categoryよりも若い数字になるということを念頭にその後の処理を実施している。
※あれ？なんでわざわざエンコードなんかしたんだっけ…。しなくても全然問題ないんじゃ…？

# Customer IDごとのCountryのdfとCustomer IDごとのSub-Categoryのdfをconcatしてエンコードする
# エンコードは昇順で番号が付けられていく
ro='Country'
co='Sub-Category'
def create_city_product_matrix(df):
    df_store=df.groupby(['Customer ID',ro,'Order Date'])[[co]].count().reset_index()
    df_product=df.groupby(['Customer ID',co,'Order Date'])[[ro]].count().reset_index()
    df_concat=pd.concat([df_store[['Customer ID',ro,'Order Date']].rename(columns={ro:co}),df_product[['Customer ID',co,'Order Date']]])
    df_concat=df_concat.sort_values(by=['Customer ID','Order Date',co])
    le = LabelEncoder()
    encoded = le.fit_transform(df_concat[co].values)
    df_concat['encoded'] = encoded
    return df_concat[['Customer ID',co,'encoded']], le

# city_cnt:国のユニーク数(後の処理で使う);エンコード後の番号のCountryとSub-Categoryの分かれ目の数字
city_cnt=df[ro].unique().shape[0]

df_label,le=create_city_product_matrix(df)
df_label['flg']=1
display(df_label)

# Customer IDごとにエンコードした数字をlist化
df_list = df_label.groupby(["Customer ID"])["encoded"].apply(lambda x:list(x)).reset_index()
display(df_list)

# アソシエーション分析のためのマート作成
te = TransactionEncoder()
te_ary = te.fit(df_list['encoded']).transform(df_list['encoded'])
df_mart = pd.DataFrame(te_ary, columns=te.columns_)
#df_mart.columns=le.inverse_transform(df_mart.columns.values)
display(df_mart)

アソシエーション分析を実施。

# 支持度5%以上のitemに絞る
min_support=0.05
frequent_itemsets = fpgrowth(df_mart,min_support=min_support,use_colnames=True)
# アソシエーション分析実行
rules = association_rules(frequent_itemsets,metric='support',min_threshold=min_support)
display(rules)

条件部にはCountryが来てほしいので、条件部にCounty、結論部にSub-Categoryが来るようにデータを抽出。

# 条件部にCounty、結論部にSub-Categoryが来るようにrulesを抽出
labels_no_frozen=[i for i in rules['antecedents'].values]
labels_no=[list(i) for i in rules['antecedents'].values]
consequents_no_frozen=[i for i in rules['consequents'].values]
consequents_no=[list(i) for i in rules['consequents'].values]
city_labels=[]
product_labels=[]
for i,k,l,n in zip(labels_no,labels_no_frozen,consequents_no,consequents_no_frozen):
    for j in i:
        # city_cnt-1以下 = Countryのエンコード後の番号
        if j <= city_cnt-1:
            for m in l:
                # city_cnt-1より大きい = Sub-Categoryのエンコード後の番号
                if m > city_cnt-1:
                    # County番号を追加
                    city_labels.append(k)
                    # Sub-Category番号を追加
                    product_labels.append(n)
                    break
                break

rules1=rules[(rules['antecedents'].isin(city_labels))&(rules['consequents'].isin(product_labels))&(rules["antecedents"].apply(lambda x: len(x))==1)&(rules["consequents"].apply(lambda x: len(x))==1)]
max_support=rules1['support'].max()
print('max_support',max_support)
display(rules1)

後に、Sub-Category同士の組み合わせも可視化したいので、条件部にも結論部にもSub-Categoryが来るようにデータを抽出しておく。

# 条件部にも結論部にもSub-Categoryが来るようにrulesを抽出
labels_no_frozen=[i for i in rules['antecedents'].values]
labels_no=[list(i) for i in rules['antecedents'].values]
consequents_no_frozen=[i for i in rules['consequents'].values]
consequents_no=[list(i) for i in rules['consequents'].values]
city_labels=[]
product_labels=[]
for i,k,l,n in zip(labels_no,labels_no_frozen,consequents_no,consequents_no_frozen):
    for m in l:
        # city_cnt-1より大きい = Sub-Categoryのエンコード後の番号
        if m > city_cnt-1:
            city_labels.append(k)
            product_labels.append(n)
            break
        break

rules11=rules[(rules['antecedents'].isin(product_labels))&(rules['consequents'].isin(product_labels))&(rules["antecedents"].apply(lambda x: len(x))==1)&(rules["consequents"].apply(lambda x: len(x))==1)&(rules['support']<=max_support)]
display(rules11)

アソシエーション分析の結果の表をデコードする。

# アソシエーション分析の結果の表を作る
def create_association_matrix(rules1):
    # rules1のantecedentsをデコード
    antecedents_scale=pd.DataFrame(le.inverse_transform([list(i)[0] for i in rules1['antecedents'].unique()]),columns=[co])
    antecedents_scale['antecedents']=[i for i in rules1['antecedents'].unique()]

    # rules1のconsequentsをデコード
    # 一応複数の組み合わせがあっても大丈夫なようにしている
    consequents=[list(i) for i in rules1['consequents'].unique()]
    consequents=[([le.inverse_transform([i])[0] for i in j]) for j in consequents]
    consequents_scale=pd.DataFrame(consequents,columns=[co])
    consequents_scale['consequents']=[i for i in rules1['consequents'].unique()]

    rules3=pd.merge(rules1, antecedents_scale, on=['antecedents'], how='left').rename(columns={co:'antecedents_name'})
    rules3=pd.merge(rules3, consequents_scale, on=['consequents'], how='left').rename(columns={co:'consequents_name'})
    rules3=rules3.reindex(columns=['antecedents_name','consequents_name','support','confidence','lift','antecedents','consequents'])
    return rules3

rules3=create_association_matrix(rules1)
rules33=create_association_matrix(rules11)
display(rules3)
display(rules33)

アソシエーション分析の結果をネットワーク図で表現してみる。

# Countryごとの人数をカウント
def create_country_uu(df_label):
    count_UU=df_label.groupby([co,'Customer ID'])[['flg']].count().reset_index()
    count_UU['cnt']=1
    count_UU=count_UU.groupby([co])[['cnt']].sum().reset_index()
    return count_UU

# ノード、エッジを作る
GA=nx.from_pandas_edgelist(rules1[['antecedents','consequents','lift']],source='antecedents',target='consequents', edge_attr=True)

count_UU=create_country_uu(df_label)
count_UU_mst=pd.DataFrame(le.inverse_transform([list(i)[0] for i, j in GA.nodes(data=True) if i in rules1['antecedents'].values]),columns=[co])
count_UU_mst=pd.merge(count_UU_mst, count_UU, on=[co],how='left').rename(columns={co:ro})

# ノードのマスタを作る(デコード)
labels_no=[list(j) for j in [i for i in GA.nodes]]
labels=[]
for no in labels_no:
    labesl2=[]
    for label in no:
        labesl2.append(le.inverse_transform([label])[0])
    labels.append(labesl2)

new_labels2={}
for i, j in zip(GA.nodes, labels):
    new_labels2[i]=j

# アソシエーション分析の結果をネットワーク図で表現
def association_network(GA, rules1, count_UU_mst):
    fig, ax=plt.subplots(figsize=(30,15))
    pos = nx.kamada_kawai_layout(GA,scale=0.06)

    # エッジの太さはリフト値に比例させる
    edge_width = [d['lift']*8 for (u,v,d) in GA.edges(data=True)]

    # Countryを正方形でプロット
    # ノードサイズはCountryごとのUU数に比例させる
    nx.draw_networkx_nodes(GA, pos, alpha=0.5, node_shape="s", linewidths=5, node_color='red',
                           nodelist=[i for i, j in GA.nodes(data=True) if i in rules1['antecedents'].values],
                           node_size=[4.*v for v in count_UU_mst['cnt'].values])

    # Sub-Categoryを円形でプロット
    nx.draw_networkx_nodes(GA, pos, alpha=0.5, node_shape="o", linewidths=0, node_color='blue', node_size=600,
                           nodelist=[i for i, j in GA.nodes(data=True) if i in rules1['consequents'].values])

    # エッジをプロット
    nx.draw_networkx_edges(GA, pos, alpha=0.2, edge_color='grey', width=edge_width)
    #ラベルの設置をする
    datas = nx.draw_networkx_labels(GA,pos,new_labels2,font_size=20)

    # 日本語に対応できるようにするため、日本語が使えるフォントを設定
    #for t in datas.values():
    #    t.set_fontproperties(font_prop)

    plt.grid(False)
    plt.show()

association_network(GA, rules1, count_UU_mst)

これはまた見づらい図…。
赤い四角の大きさはその国のCustomer数に比例していて、灰色線の太さはリフト値に比例している。
先進国ではいろんなカテゴリーの商品が買われていて、発展途上国では一部のカテゴリーの商品が買われているような傾向にあると言えるかも。

ただし、まだこの図ではノード同士の位置関係は意味をなしていない。
コレスポンデンス分析を合わせることにより、ノードの位置関係も意味を持つ図を作るのが今回の目的だ。

コレスポンデンス分析

コレスポンデンス分析を行うためのマートを作成。

# CountryとSub-Categoryの売り上げのクロス表を作る
df_corre=df.copy()
df_corre=df_corre.pivot_table(index=ro, columns=co, values='Sales', aggfunc=lambda x:x.sum()).fillna(0)
display(df_corre)

コレスポンデンス分析を実施。

# コレスポンデンス分析をプロット
fig, ax=plt.subplots(figsize=(30,20))

mca_counts = mca.MCA(df_corre)
rows = mca_counts.fs_r(N=2)
cols = mca_counts.fs_c(N=2)

ax.scatter(rows[:,0], rows[:,1], c='red',marker='s', s=200)
labels = df_corre.index.values
for label,x,y in zip(labels,rows[:,0],rows[:,1]):
        ax.annotate(label,xy = (x, y),fontsize=20)

ax.scatter(cols[:,0], cols[:,1], c='blue',marker='x', s=400)
labels = df_corre.columns.values
for label,x,y in zip(labels,cols[:,0],cols[:,1]):
        ax.annotate(label,xy = (x, y),fontsize=20, color='c')

ax.tick_params(left=True, bottom=True, labelleft=True, labelbottom=True)
ax.axhline(0, color='gray')
ax.axvline(0, color='gray')
plt.show()

重なってほとんど見えないけど、各国や商品カテゴリーの位置関係に意味を持つ図ができる。(ザンビアとモーリタニアとアフガニスタンとシンガポールの購買傾向が似ているとか。本当か…？)

この図に、アソシエーション分析の結果のエッジを追加できれば、属性特化型特徴抽出アソシエーションプロットの完成である。

属性特化型特徴抽出アソシエーションプロット

まず、先ほどと同様にコレスポンデンス分析を実施。

mca_counts = mca.MCA(df_corre)
rows = mca_counts.fs_r(N=2)
cols = mca_counts.fs_c(N=2)

先ほどはコレスポンデンス分析の結果をplt.scatterでプロットしたが、ここではnetworkxを使う。

# コレスポンデンス分析とアソシエーション分析の結果をnetwork図で表現
def mca_association_plot(df_corre, df_label, rows, cols, new_labels2
                         , strong_node_row=None, strong_node_col=None
                         , xlim=[None, None], ylim=[None, None]):
    fig, ax=plt.subplots(figsize=(30,20))
    uu_list=create_country_uu(df_label)

    # Countryを赤い四角でプロット
    G = nx.Graph()
    node_weights=[]
    for node, pos in zip(df_corre.index, rows):
        if strong_node_row is None:
            G.add_node(node)
            G.nodes[node]["pos"] = (pos[0], pos[1])
            node_weights.append(uu_list[uu_list[co]==node]['cnt'].values[0])
        else:
            if node in strong_node_row:
                G.add_node(node)
                G.nodes[node]["pos"] = (pos[0], pos[1])
                node_weights.append(uu_list[uu_list[co]==node]['cnt'].values[0])
    position=np.array([v['pos'] for (u,v) in G.nodes(data=True)])
    pos = {n:(i[0], i[1]) for i, n in zip(position ,G.nodes)}
    nx.draw_networkx(G, pos=pos, node_color="red",ax=ax, linewidths=5, node_shape="s", node_size=[1.5*v for v in node_weights], alpha=0.5)
    new_labels2={}
    for i, j in zip(G.nodes, G.nodes):
        new_labels2[i]=j
    datas = nx.draw_networkx_labels(G,pos, new_labels2, font_size=20, font_color='k')
    #日本語に対応できるようにするため、日本語が使えるフォントを設定している
    #for t in datas.values():
    #    t.set_fontproperties(font_prop)

    # Sub-Categoryを青いバツ印でプロット    
    G2 = nx.Graph()
    node_weights2=[]
    for node, pos in zip(df_corre.columns, cols):
        if strong_node_col is None:
            G2.add_node(node)
            G2.nodes[node]["pos"] = (pos[0], pos[1])
            node_weights2.append(uu_list[uu_list[co]==node]['cnt'].values[0])
        else:
            if node in strong_node_col:
                G2.add_node(node)
                G2.nodes[node]["pos"] = (pos[0], pos[1])
                node_weights2.append(uu_list[uu_list[co]==node]['cnt'].values[0])
    position2=np.array([v['pos'] for (u,v) in G2.nodes(data=True)])
    pos2 = {n:(i[0], i[1]) for i, n in zip(position2 ,G2.nodes)}
    nx.draw_networkx(G2, pos=pos2, node_color="blue",ax=ax, node_shape="x", node_size=[5*v for v in node_weights2], alpha=0.5)
    new_labels2={}
    for i, j in zip(G2.nodes, G2.nodes):
        new_labels2[i]=j
    datas = nx.draw_networkx_labels(G2,pos2, new_labels2, font_size=20, font_color='b')
    #日本語に対応できるようにするため、日本語が使えるフォントを設定している
    #for t in datas.values():
    #    t.set_fontproperties(font_prop)

    # CountryとSub-Categoryの間のエッジをプロット（灰色でリフト値に比例して太くなる）
    U=nx.Graph()
    U.add_nodes_from(G.nodes(data=True)) #deals with isolated nodes
    U.add_nodes_from(G2.nodes(data=True))
    for edge1, edge2, lift in zip(rules3[rules3['lift']>=1.6]['antecedents_name'].values, rules3[rules3['lift']>=1.6]['consequents_name'].values, rules3[rules3['lift']>=1.6]['lift'].values):
        if strong_node_col is None:
            U.add_edge(edge1, edge2, lift=lift)
        else:
            if edge2 in strong_node_col:
                U.add_edge(edge1, edge2, lift=lift)
    pos_all = {n:(i[0], i[1]) for i, n in zip(np.vstack((position, position2)) ,U.nodes)}
    edge_width = [d['lift']*8. for (u,v,d) in U.edges(data=True)]
    nx.draw_networkx_edges(U, pos=pos_all, alpha=0.3, edge_color='grey', width=edge_width, ax=ax)

    # Sub-CategoryとSub-Categoryの間のエッジをプロット（水色でリフト値に比例して太くなる）
    V=nx.Graph()
    V.add_nodes_from(G.nodes(data=True)) #deals with isolated nodes
    V.add_nodes_from(G2.nodes(data=True))
    for edge1, edge2, lift in zip(rules33[rules33['lift']>=1.5]['antecedents_name'].values, rules33[rules33['lift']>=1.5]['consequents_name'].values, rules33[rules33['lift']>=1.5]['lift'].values):
        if strong_node_col is None:
            V.add_edge(edge1, edge2, lift=lift)
        else:
            if edge1 in strong_node_col and edge2 in strong_node_col:
                V.add_edge(edge1, edge2, lift=lift)
    pos_all2 = {n:(i[0], i[1]) for i, n in zip(np.vstack((position, position2)) ,V.nodes)}
    edge_width = [d['lift']*8. for (u,v,d) in V.edges(data=True)]
    nx.draw_networkx_edges(V, pos=pos_all2, alpha=0.2, edge_color='c', width=edge_width ,ax=ax)

    ax.tick_params(left=True, bottom=True, labelleft=True, labelbottom=True)
    ax.set_xlim(xlim[0], xlim[1])
    ax.set_ylim(ylim[0], ylim[1])
    ax.axhline(0, color='gray')
    ax.axvline(0, color='gray')
    plt.grid(True)
    plt.show()

いざ実施。

xlim=[None,None]
ylim=[None,None]
mca_association_plot(df_corre, df_label, rows, cols, new_labels2, strong_node_row=None, strong_node_col=None, xlim=xlim, ylim=ylim)

バーーーーン
コレスポンデンス分析の結果にアソシエーション分析の結果を反映させた図ができた。
灰色のエッジは国と商品サブカテゴリーとの関係、水色のエッジは商品サブカテゴリー同士の関係を表している。
エッジはリフト値が1.6以上(灰色)、1.5以上(水色)のものだけ引いている。
赤い四角と青いバツはCustomer IDのユニーク数に比例して大きくなる。
とはいっても重なってよく分からないので、真ん中をちょっと拡大して見る。

xlim=[-0.05,0.05]
ylim=[-0.05,0.05]
mca_association_plot(df_corre, df_label, rows, cols, new_labels2, strong_node_row=None, strong_node_col=None, xlim=xlim, ylim=ylim)

拡大しても、よくわかんない(笑)。
でもアメリカは購入する客が最も多くて、Bindersが売れる傾向にある位置にいるけどエッジはつながっていなくて、Tablesとエッジがつながっているとかはわかる。
売り上げ的に目立つのはBindersかもしれないけど、全ての国の中でTablesを購入した客の割合より、実はアメリカでTablesを購入した客の割合の方が1.6倍以上高いよって感じか。よくわからんね。

プロットされるノードを減らすため、支持度が0.05以上だった項目だけを対象に、再度プロットしてみる。

# 支持度0.05以上の項目だけ使う
strong_single_product=list(set([[j for j in i][0] for i in frequent_itemsets['itemsets']]))
strong_single_product=le.inverse_transform(strong_single_product)
row_word = 'Country_'
strong_node_row = [v for i, v in enumerate(strong_single_product) if row_word in v]
strong_node_col = [v for i, v in enumerate(strong_single_product) if row_word not in v]

xlim=[-0.05,0.05]
ylim=[-0.05,0.05]
mca_association_plot(df_corre, df_label, rows, cols, new_labels2, strong_node_row=strong_node_row, strong_node_col=strong_node_col, xlim=xlim, ylim=ylim)

結局エッジが飛び交いすぎて見にくい…。無理やり解釈していくと…。

・左のアメリカ、真ん中のドイツ・イギリス、右上のブラジル・メキシコ、下のスペイン・イタリヤなど、それぞれ傾向が異なると考えられる
・全ての国の中でTablesを購入した客の割合より、メキシコでTablesを購入した客の割合の方が1.6倍以上高い
・ブラジルはメキシコと同じ傾向の国なので、Tablesを推してもいいかもしれない
・TablesとChairs, Paperの組み合わせも買う人が多いと言えるので、アメリカはChairをもっと推してもいいかもしれない

とか？いや、解釈むずいわ。
とある国の店舗ごとの分析とかだったら解釈しやすいかもしれないな…。

おわりに

属性特化型特徴抽出アソシエーションプロットを実施した。
しかし解釈がかなり難しかった…。
計算機統計学 29巻 22号 の"売り上げ傾向による店舗の分類と購買傾向の分析と可視化"と同じようなアプローチだとまた意味のある分析ができるかもしれない。
個人的にはnetworkxの練習になったからよかった。

以上！

(オリジナルポスト)
https://qiita.com/suzu12/items/af0ccbca88f33dec2fda#実践
https://qiita.com/suzu12/items/af0ccbca88f33dec2fda

SymPy Liveで

Pythonでリストから重複した要素を抽出して、Pythonでlistのlistを引き算をしています。
もっと短くなるような気がします。教えて下さい。

https://live.sympy.org/

sum( [i for i in [1,2,3] if i not in [x for x in set([1,2,3]) if [1,2,3].count(x) > 1]])
6
sum( [i for i in [3,2,3] if i not in [x for x in set([3,2,3]) if [3,2,3].count(x) > 1]])
2
sum( [i for i in [3,3,3] if i not in [x for x in set([3,3,3]) if [3,3,3].count(x) > 1]])
0

はじめに

　Effective Python 第二版をよんで、感動した機能などを備忘録を兼ねてまとめておきたいと思います。チョイスは個人的な主観でやっているのでご承知おきください。。

代入式とは？

　簡単にいうと、if文で使う変数の宣言を短く記述できるものです。
具体的には、ifの条件式のなかで :=（セイウチ演算子）を使って式の評価と変数への代入をひといきに行います。

　例えば、フルーツの在庫を表す辞書があるとします。

fruits = {
  'apple': 10,
  'banana': 8,
  'lemon': 5,
}

　リンゴジュースをつくるのに少なくとも４つのリンゴが必要な場合、ジュースを実際につくれるかどうかは下のように条件分岐できます。

n_apple = fruits.get('apple', 0)
if n_apple >= 4:
  make_juice(n_apple)　# n_apple個のリンゴでジュースを作る
else:
  out_of_stock()　　　 # 在庫不足のためジュースを作らない

　これをセイウチ演算子を使って同じような動作を記述すると、

# リンゴの個数の評価と変数への代入を同時に行う
if (n_apple := fruits.get('apple', 0)) >= 4: 
  make_juice(n_apple)
else:
  out_of_stock()　　　　　　

　一見してコードが一行へっただけのように思えますが、このような書き方をすることで、n_appleがifの１つ目のブロックにしか使われないことを明確に表現することができます。

応用例

　これだけだと、いまいち良さが理解できないと思うので、よりうれしい例を取り上げてみます。
　バナナ→リンゴ→レモンの順番で優先順位をつけて在庫を消化したいとします。また、ジュースをつくるのに最低限必要なフルーツの個数はバナナは２本、リンゴは４個、レモンは１個とします。
　おそらく、以下のような入れ子になったif-elseブロックを作るのではないでしょうか。

n_banana = fruits.get('banana', 0)
if n_banana >= 2:　　　　
  make_juice(n_banana)　　 # バナナが２個以上ならバナナジュースを作る
else:
  n_apple = fruits.get('apple', 0)
  if n_apple >= 4:
    make_juice(n_apple)　　# リンゴが４個以上ならリンゴジュースを作る
  else:
    n_lemon = fruits.get('lemon', 0)
    if n_lemon:
      make_juice(n_lemon)　# レモンが１個以上ならレモンジュースを作る
    else:
      out_of_stock()　　　　# 在庫不足のためジュースを作らない

　これはかなり冗長な感じがしますよね。セイウチ演算子を使うと以下のようになります。

if (n_banana := fruits.get('banana', 0)) >= 2:　　　　
  make_juice(n_banana)　　 # バナナが２個以上ならバナナジュースを作る
elif (n_apple := fruits.get('apple', 0)) >= 4::
  make_juice(n_apple)　  　# リンゴが４個以上ならリンゴジュースを作る
elif n_lemon = fruits.get('lemon', 0):
  make_juice(n_lemon)　    # レモンが１個以上ならレモンジュースを作る
else:
  out_of_stock()　　　　    # 在庫不足のためジュースを作らない

　大分よみやすくなりました。
　どうですか？これなら使ってみたくなったのではないでしょうか(笑)。
　ただし、代入された変数を比較に使う場合には代入式を括弧でくくる必要があることに注意してください。

do-while文的な使い方

　もう少し応用的な使い方を紹介します。pythonにはdo/whileループがありませんが、whileブロックとbreak文で同じようなロジックを実装できます。
　例えば、在庫の補充とジュースの瓶詰を繰り返しおこなう処理を考えます。在庫の補充がなくなった時点で処理を終了させます。
make_juice()関数は、フルーツの種類と個数を受け取って適切な本数（batch）の瓶づめジュースを返すように変更したとします。

bottles = []　# 瓶を収納するケース
while True:
  fruits = pick_fruit() # フルーツを補充
  if not fruits:　　
    break　　　　　　　　# 補充がなければ終了
  for fruit, count in fruits.items():
    batch = make_juice(fruit, count)　# フルーツごとに適切な本数に瓶詰め
    bottles.extend(batch)　　　　　　　# 瓶をケースに追加

　このような処理も代入式で短く書くことができます。

bottles = []　# 瓶を収納するケース
while fruits := pick_fruit():　# 補充したフルーツをfruits変数に代入&空かどうか評価
  for fruit, count in fruits.items():
    batch = make_juice(fruit, count)　# フルーツごとに適切な本数に瓶詰め
    bottles.extend(batch)　　　　　　　# 瓶をケースに追加

　代入文とbreakのためのifブロックが不要になったのがうれしいですね。

おわりに

　セイウチ演算子、なかなか便利ですね。python3.8を仕事では使わないけど、この機能は3.7にもあってほしい。

参考文献

　Effective Python 第2版 ―Pythonプログラムを改良する90項目

言語処理100本ノック 2020 (Rev 2)ではファイルをダウンロードしてから処理することが多い

いろんなやり方があるので収集してみる。

ファイルは第3章: 正規表現のjawiki-country.json.gz

Google Colaboratoryにて実施

requests

requests.py

import requests

url='https://nlp100.github.io/data/jawiki-country.json.gz'
filename='jawiki-country.json.gz'

urlData = requests.get(url).content

with open(filename ,mode='wb') as f: # wb でバイト型を書き込める
  f.write(urlData)

自分のやり方。
Requetsはwgetとおんなじ感じで使えるのがいい。

ファイルの書き込みは定番のやり方

Python、Requestsを使ったダウンロードのように大きなファイルを扱うのもできるのでしょう。

requests2.py

import requests
import os

url='https://nlp100.github.io/data/jawiki-country.json.gz'
filename=os.path.basename(url)

r = requests.get(url, stream=True)
with open(filename, 'wb') as f:
  for chunk in r.iter_content(chunk_size=1024):
    if chunk:
      f.write(chunk)
      f.flush()

直で書き込むにはこちら。

urllib.request

urllib_request.py

import urllib.request

url='https://nlp100.github.io/data/jawiki-country.json.gz'
save_name='jawiki-country.json.gz'

urllib.request.urlretrieve(url, save_name)

今回で調べてみたら出てきました。
pythonでwebからのファイルのダウンロード
ファイルのセーブまでできるすぐれもの

pandas.read_X

read_X.py

import pandas as pd

url='https://nlp100.github.io/data/jawiki-country.json.gz'

df=pd.read_json(url, lines=True)

第２章で大活躍だったpandas

input/outputにある各コマンドはurlから直で読めて圧縮も自動で判別してくれる優れもの。

読み込むとデータフレームになってしまうので使いどころを選ぶけど、そのまま処理するのであればこちらでもいい。

まとめ

コマンドだとwgetでなにも考えなくていいけど、プログラムからやる時はそれなりに考えないといけない。

いずれかの方法でやっていけばいいと思います。