Calc weekly totals from daily totals.

calc_weekly_from_daily.py

#!/usr/bin/env python3
"""
Calc weekly totals from daily totals.

https://gist.github.com/trueroad/a3558410135051bad254eb257d45cbd9

Copyright (C) 2023 Masamichi Hosoda.
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:

* Redistributions of source code must retain the above copyright notice,
  this list of conditions and the following disclaimer.

* Redistributions in binary form must reproduce the above copyright notice,
  this list of conditions and the following disclaimer in the documentation
  and/or other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED.
IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
SUCH DAMAGE.

使用方法

以下のような日付毎のデータがある CSV を入力とする。
UTF-8 BOM 付きなどでよい。改行コードは自動判別される。

表側は利用者を識別する ID など。1 行 1 列にその名前を入れる。
表頭は ISO8601 年月日で YYYY-MM-DD のような表記。
表体は浮動小数点数で日毎の値を加算すると週毎の値になるもの、
例えば利用時間（秒など）、販売量（kg など）など。

```in_example.csv
ID,2023-12-01,2023-12-02,2023-12-03,2023-12-04,2023-12-05,2023-12-06
1,1.0,1.0,1.0,1.0,1.0,1.0
2,2.0,2.0,2.0,2.0,2.0,2.0
```

以下のように実行すると週毎に集計した CSV が得られる。

```
$ ./calc_weekly_from_daily.py in_example.csv out_example.csv
```

以下のような出力 CSV が得られる。
UTF-8 BOM 付き、改行コードは実行環境依存で出力される。

表側は入力と同じものが出力される。1 行 1 列の名前も同じ。
表頭は ISO8601 週番号を文字列表記にしたもので YYYY-Www のような表記。
表体は入力の日毎の値を加算して週毎の値にしたもの。

```out_example.csv
ID,2023-W48,2023-W49
1,3.0,3.0
2,6.0,6.0
```
"""

import os
import sys
from typing import Final, Union

import pandas as pd
import numpy as np
import numpy.typing as npt


def day_to_week(d: np.datetime64) -> np.datetime64:
    """日付からその日を含む ISO 週の初日（月曜）を返す."""
    year: int
    week: int
    year, week, _ = pd.Timestamp(d).isocalendar()
    return np.datetime64(pd.Timestamp.fromisocalendar(year, week, 1), 'D')


def week_str(d: np.datetime64) -> str:
    """日付からその日を含む ISO 週の文字列表記（YYYY-Www 表記）を返す."""
    year: int
    week: int
    year, week, _ = pd.Timestamp(d).isocalendar()
    if year < 0 or year > 9999:
        raise ValueError('Year is out of range.')
    return f'{year:04}-W{week:02}'


class calc_weekly_from_daily:
    """Calc class."""

    def calc(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        計算メイン.

        Args:
          df (pd.DataFrame): 日付毎のデータがあるデータフレーム
        Returns:
          pd.DataFrame: 週毎に集計したデータフレーム
        """
        # 列名から日付の一覧を作る
        dates: npt.NDArray[np.datetime64] = \
            np.sort(np.array(df.columns, dtype='datetime64[D]'))
        # 最初の日付
        date_begin: np.datetime64 = dates[0]
        # 最後の日付
        date_last: np.datetime64 = dates[-1]
        # 最初の日付を含む ISO 週（の初日）
        week_begin: np.datetime64 = day_to_week(date_begin)
        # 最後の日付を含む ISO 週（の初日）
        week_last: np.datetime64 = day_to_week(date_last)
        # ISO 週の一覧
        week_all: npt.NDArray[np.datetime64] = \
            np.arange(week_begin,
                      week_last + np.timedelta64(1, 'D'),
                      np.timedelta64(7, 'D'))
        # ISO 週の一覧の文字列版
        week_all_str: list[str] = \
            list(map(lambda x: week_str(x), week_all))

        print(f'Dates and ISO weeks: {date_begin} {week_all_str[0]} ~'
              f' {date_last} {week_all_str[-1]}')

        # 出力用データフレーム
        df_out: pd.DataFrame = pd.DataFrame(columns=week_all_str,
                                            dtype=np.float64)
        # インデックスのタイトルをコピー
        df_out.index.name = df.index.name

        row: pd.Series[float]
        for _, row in df.iterrows():
            # 入力用データフレームの行毎に処理
            print(f'Calculating row: {row.name}')

            # 出力用データフレームに行追加、表側は入力と同じ、表体は初期値 0.0
            df_out.loc[str(row.name)] = 0.0

            d: np.datetime64
            for d in dates:
                # 入力用データフレームの列（日付）毎に処理

                # 入力側の日付のデータを出力側の週のデータに加算
                df_out.at[str(row.name), week_str(d)] += \
                    df.at[row.name, str(d)]

        return df_out

    def process_csv(self,
                    filename_in: Union[str, os.PathLike[str]],
                    filename_out: Union[str, os.PathLike[str]]) -> bool:
        """
        CSV ファイルを処理する.

        Args:
          filename_in (Union[str, os.PathLike[str]]): 入力 CSV ファイル
          filename_out (Union[str, os.PathLike[str]]): 出力 CSV ファイル
        Returns:
          bool: True なら成功、False なら失敗
        """
        # 入力用データフレームに読み込む、最初の列（表側）は文字列
        print(f'Loading: {filename_in} ...')
        df: pd.DataFrame = pd.read_csv(filename_in,
                                       dtype={0: str})
        # 最初の列（表側）をインデックスにする
        df = df.set_index(df.columns[0])

        # 計算する
        df_out: pd.DataFrame = self.calc(df)

        # 集計結果を出力
        print(f'Writing: {filename_out} ...')
        # UTF-8 BOM 付き CSVで出力（Excel で開けるように）
        df_out.to_csv(filename_out, encoding='utf_8_sig')

        return True


def main() -> None:
    """Do main."""
    print('Calc weekly totals from daily totals.\n\n'
          'https://gist.github.com/trueroad/'
          'a3558410135051bad254eb257d45cbd9\n\n'
          'Copyright (C) 2023 Masamichi Hosoda.\n'
          'All rights reserved.\n')

    if len(sys.argv) != 3:
        print('Usage: ./calc_weekly_from_daily.py IN.csv OUT.csv')
        sys.exit(1)

    cwfd: calc_weekly_from_daily = calc_weekly_from_daily()
    cwfd.process_csv(sys.argv[1], sys.argv[2])

    print('Done.')


if __name__ == '__main__':
    main()