カッコ()で区切られたデータを連結したりして、正解と一致している区間を返す

Question

いつもお世話になっております。
Pythonのプログラミングでわからないところがあるので質問させていただきます。

前提

こんな実験をして、こんな状況があるとします

• ある照明器具があって、これが所望の時間帯に発行しているかを評価したい
• 照明器具はある計算規則によって発光する
• 発光を判断するセンサがあって、ある閾値を超えた輝度に関しては発光しているものと判断して、閾値以下のものを発光していないと判定してくれる

そしてこの実験から下記のCSVファイルが得られたします。

//data.csv
lamp,correct_time
  [(123 200) (200 400) (400 567)],[(268 574)]
  [(111 151) (412 567)],[(301 556)]
  [(168 199) (259 392)],[(174 580)]

のCSVファイルの詳細は次のとおりです

• 列名「lamp」というのは発光を判断するセンサが発光していると判断している時間帯を計測している。例えば左上の(123 200)は計測開始から123秒後から200秒後まで発光していると判断している
• このセンサは少し厄介なことがあって、計測開始から200秒毎に勝手に区切られてしまう。例えば一番上の行の(123 200) (200 400) (400 567)というのは実際には計測開始から123秒後から567秒後まで発光しているが200秒毎に区切られてしまって「123秒から200秒、200秒から400秒、400秒から567秒まで」というような表示になっている。
• 列名「correct_time」というのは、実際に光っていてほしい時間帯、すなわち正解の時間帯を示している。一番上の行の(268 574)は計測開始から268秒後から574秒後まで光っていてほしいという意味である

実現したいこと

やりたいことしては次のとおりです

• 列名「lamp」と「correct_time」がどれだけ一致しているかを返したいです。例えば、CSVの一番上の行であれば計測開始から268秒後から574秒後までが一致しているので、リストなどで[268 574]。
• 上から２番めであれば412秒後から556秒後までが一致しているので[412 556]
• 上から３番めであれば174秒後から199秒後と259秒後から392秒後までが一致しているので[174 199][259 392]

と返したいです。

試したこと

どのようにコードを書いたら良いのか分からず、全く試せていません・・
一応Pandasを使ってちょこっと書いてみましたが、ハテナです・・・
出来ればPandasを使って解決したいですが、その限りではございません。

全体的に分かりづらいかもしれませんが、是非ともご指南お願いいたします。

Answer 1

タスクとしては「文字列 "[(123 200) (200 400) (400 567)]" を扱える形のオブジェクトに変換すること」と「区間の共通部分を計算すること」の2つにわかれるかと思います。

前者は Python の文法に合うようにカンマを入れてあげて、組み込み関数 eval() でタプルのリストに変換するといいかと思います。

後者は自分で実装しようとすると少々面倒なので、ライブラリを使うといいと思います。
sympy の Sets で区間演算が行えるので、それを使うと楽です。

Sets — SymPy 1.4 documentation

手順

• 1. pandas の read_csv() で csv ファイルを読み込む。
• 1. 文字列をタプルのリストに変換する。

"[(123 200) (200 400) (400 567)]" -> [(123,200),(200,400),(400,567)]

• 1. タプルのリストを sympy.Interval のリストに変換する。
• 1. 列 lamp と列 correct_time の共通する区間のリストを計算する。(集合の積)
• 1. sympy.Interval のリストを文字列に戻す。

[Interval.Ropen(111, 151), Interval.Ropen(412, 567)] -> "[(111 151) (412 567)]"

1. 結果を csv に出力する。

サンプルコード

import itertools

import pandas as pd
import sympy as sy


def str_to_intervals(x):
    """文字列を sympy.Interval のリストに変換する。
    """
    # 空白をカンマに置換することで、python のリストのタプルとして、式を評価できるようにする。
    # 例: "[(123 200) (200 400) (400 567)]" -> [(123,200),(200,400),(400,567)]
    list_of_tuples = x.replace(" ", ",")
    # 文字列を sympy.Interval のリストに変換する。
    intervals = [sy.Interval(l, r, right_open=True) for l, r in eval(list_of_tuples)]

    return intervals


def intervals_to_str(intervals):
    """sympy.Interval を文字列に戻す。
    例: [Interval.Ropen(111, 151), Interval.Ropen(412, 567)] -> "[(111 151) (412 567)]"
    """
    str_intervals = [f"({i.left} {i.right})" for i in intervals]
    return "[" + " ".join(str_intervals) + "]"


def calc_overlap(row):
    """列 lamp と列 correct_time の共通する区間のリストを返す。
    """
    # 列 lamp の区間と列 correct_time の区間のすべての組み合わせにおいて、
    # 共通部分を計算し、その和集合をとる。
    overlaps = sy.EmptySet()
    for i1, i2 in itertools.product(row["lamp"], row["correct_time"]):
        overlaps += i1 & i2
    # Union の場合は、Interval のリストとして返す。
    # 区間が1つの場合でも Interval のリストとして返す。
    return overlaps.args if isinstance(overlaps, sy.Union) else [overlaps]


df = pd.read_csv(
    "test.csv", converters={"lamp": str_to_intervals, "correct_time": str_to_intervals}
)
# 共通部分を計算して、列 overlap に追加する。
df["overlap"] = df.apply(calc_overlap, axis="columns")
# sympy.Interval を文字列表現に戻す。
df = df.applymap(intervals_to_str)
# csv ファイルとして保存する。
df.to_csv("reuslt.csv")

入出力

入力

lamp,correct_time
[(123 200) (200 400) (400 567)],[(268 574)]
[(111 151) (412 567)],[(301 556)]
[(168 199) (259 392)],[(174 580)]

出力

,lamp,correct_time,overlap
0,[(123 200) (200 400) (400 567)],[(268 574)],[(268 567)]
1,[(111 151) (412 567)],[(301 556)],[(412 556)]
2,[(168 199) (259 392)],[(174 580)],[(174 199) (259 392)]

Comments

[pepasuke623]

ご丁寧な回答、感謝いたします。サンプルコードも添付していただいて非常に分かりやすいです。
走らせてみたところ
("'str' object has no attribute 'left'", 'occurred at index filename')
とエラーメッセージが出て色々ググってみましたが分からず・・・。初歩的なところを見落としているのかもしれませんが、よろしければアドバイス願います。

[tiitoi]

回答の「入力」の csv を用意すると、実行すると「出力」の csv のコードが出てくるというもので、こちらの環境では動作しています。
もし、入力の csv の形式が回答で想定しているものと異なる場合や質問者さんがコードを一部編集した場合は質問欄に追記していただけますか。

[pepasuke623]

ご返信ありがとうございます。

今回投稿させていただいた内容は質問用にデフォルメしてあります。
列名などは違うものの、CSVファイル内の該当部分のフォーマットは同じであるはずです。頂いたサンプルコードの列名などを変えていますが、それ以外は変更していないと思います。（只今出払っていて、エラーになった方のCSVは見れませんので、後日もう一度確認します）

また前回の私の説明が言葉足らずでしたが、
str_intervals = [f"({i.left} {i.right})" for i in intervals]
の行で
("'str' object has no attribute 'left'", 'occurred at index filename')
というエラーが出ていました。

後ほどもう一回見てみます。

[tiitoi]

エラーの直接的な原因は intervals は sympy.Interval のリストのはずなのですが、文字列 str が混ざってしまっているみたいです。
なぜそうなったのかはわからないのですが、元の読み込むデータに原因があるのではと推測しております。

お時間が取れるときでよろしいので、df = pd.read_csv() した直後の DataFrame を確認して、おかしなことになっていないか確認してみてください。

可能であれば、実際の csv ファイルの一部 (最初の10行ぐらい) を質問欄に記載いただくか、csv ファイルをどこかのアップローダーにアップしていただければこちらでも検証できます。

[pepasuke623]

コメントありがとうございます。解決しました。
おっしゃる通り、エラーが出た方のCSVは他に日付やその他の情報が入った列が存在するためにエラーになっていました。初歩的なところで、引っかかっていました。

Answer 2

以下のような方法ではどうでしょう。

https://www.gesource.jp/weblog/?p=8244を参考に、「正解の時間帯」の区間を定義して、ある時点が「正解の時間帯」に含まれているか否かを判別する関数を作る。

「実際に光っている時間帯」の開始時点と終了時点をそれぞれ上記関数で判別する。

「実際の開始時点＝正解の範囲内」かつ「実際の終了時点＝正解の範囲内」なら（実際の開始時点、実際の終了時点）を一致した区間として返す。
「実際の開始時点＝正解の範囲内」かつ「実際の終了時点＝正解の範囲外」なら（実際の開始時点、正解の終了時点）を一致した区間として返す。
以下同様に。

Comments

[pepasuke623]

回答ありがとうございます！！
なるほど、是非とも参考にさせていただきます。

ただ、私はどちらかというとカッコ入の文字列処理に苦戦しております。
力技でなんとか出来なくもないと思いますが、もっとスマートに出来ないかな、と思い質問させていただいた次第です。

[KojiDoi]

正規表現で（開始、終了）というパターンをキャプチャすればいいのではないでしょうか。

[pepasuke623]

回答有り難うございます。そうですね、正規表現は殆ど使ったことがないので、調べながら書いてみます。

Answer 3

データを違うフォーマットで吐けるのなら、それが一番スマートなんですが・・・（JSONにでもしてくれればいいのに）

正規表現使ってやるしかなさそうですね。とにかくpythonオブジェクトにしちゃえば、あとは力技でなんとかなるでしょう。

>>> import re
>>> s = "[(123 200) (200 400) (400 567)]"
>>> [list(map(int, data)) for data in re.findall(r"((\d+) (\d+))", s)]
[[123, 200], [200, 400], [400, 567]]

あるいは、「そのフォーマットを100%信じていいのなら」空白をカンマに置換してあげるとpythonの式として読める文字列になりそうなので、それで取り扱うか。

>>> from ast import literal_eval
>>> s = "[(123 200) (200 400) (400 567)]"
>>> literal_eval(s.replace(" ", ", "))
[(123, 200), (200, 400), (400, 567)]

Comments

[pepasuke623]

回答有り難うございます。勉強になりました。正規表現を殆ど使ったことがないので、調べてみます。このテクニックは他の問題にも適用できそうですね。