init commit

Author: dclong

.gitignore

@@ -0,0 +1,10 @@
+__pycache__/
+.ipynb_checkpoints/
+venv/
+target/
+dist/
+.coverage
+.mypy
+.mypy_cache
+*.crc
+*.egg-info/

.style.yapf

@@ -0,0 +1,2 @@
+[style]
+based_on_style = facebook

dsutil/__init__.py

@@ -0,0 +1,12 @@
+from . import collections
+from . import const
+from . import dataframe
+from . import datetime
+from . import docker
+from . import git
+from . import misc
+from . import poetry
+from . import text
+from . import url
+
+__version__ = "0.8.1"

dsutil/collections.py

@@ -0,0 +1,51 @@
+from typing import Callable, Any
+
+
+def format_item_html(key: Any, value: Any) -> str:
+    """Format a (key, value) pair of a dictionary in HTML format.
+
+    :param key: An key of the dictionary.
+    :param value: The corresponding value of the key.
+    """
+    return ' ' * 4 + f'{key}: {value}'
+
+
+def format_item_plain(key: Any, value: Any) -> str:
+    """Format a (key, value) pair of a dictionary in HTML format.
+
+    :param key: An key of the dictionary.
+    :param value: The corresponding value of the key.
+    """
+    return ' ' * 4 + f'{key}: {value}'
+
+
+def format_dict_html(
+    dict_: dict,
+    fmt: Callable[[Any, Any], str] = format_item_html,
+    filter_: Callable[[Any, Any], bool] = lambda key, value: True
+):
+    """Format a dict in HTML format for pretty printing.
+
+    :param dict_: The dictionary to format.
+    :param fmt: A function to format a (key, value) pair.
+    :param filter_: A filtering function to select items from the dictionary.
+    :is_html: whether to format the dictionary in HTML format or in plain text format.
+    """
+    lines = (fmt(k, v) for k, v in dict_.items() if filter_(k, v))
+    return '{<br>' + '<br>'.join(lines) + '<br>}'
+
+
+def format_dict_plain(
+    dict_: dict,
+    fmt: Callable[[Any, Any], str] = format_item_plain,
+    filter_: Callable[[Any, Any], bool] = lambda key, value: True
+):
+    """Format a dict for pretty printing.
+
+    :param dict_: The dictionary to format.
+    :param fmt: A function to format a (key, value) pair.
+    :param filter_: A filtering function to select items from the dictionary.
+    :is_html: whether to format the dictionary in HTML format or in plain text format.
+    """
+    lines = (fmt(k, v) for k, v in dict_.items() if filter_(k, v))
+    return '{\n' + '\n'.join(lines) + '\n}'

dsutil/const.py

@@ -0,0 +1,38 @@
+import os
+import re
+import pandas as pd
+import subprocess as sp
+from typing import List, Sequence, Union
+
+
+def table_2w(frame: pd.DataFrame, columns: List[str], na_as=None):
+    """Create 2-way table from columns of a DataFrame.
+    """
+    if na_as is not None:
+        frame = frame.fillna(na_as)
+    if type(frame) == pd.Series:
+        df = frame.unstack()
+        df.index = pd.MultiIndex.from_product(
+            [[df.index.name], df.index.values]
+        )
+        df.columns = pd.MultiIndex.from_product(
+            [[df.columns.name], df.columns.values]
+        )
+        return df
+    if isinstance(frame, pd.DataFrame):
+        return table_2w(frame[columns].groupby(columns).size())  # pylint: disable=E1120
+    raise TypeError('"frame" must be pandas.Series or pandas.DataFrame.')
+
+
+def read_csv(path: str, **kwargs):
+    """Read many CSV files into a DataFrame at once.
+    """
+    if os.path.isfile(path):
+        return pd.read_csv(path, **kwargs)
+    frame_list = []
+    if os.path.isdir(path):
+        for file in os.listdir(path):
+            if os.path.splitext(file)[1].lower() == '.csv':
+                file = os.path.join(path, file)
+                frame_list.append(pd.read_csv(file, **kwargs))
+    return pd.concat(frame_list)

dsutil/dataframe.py

@@ -0,0 +1,123 @@
+import re
+import os
+from typing import Union, Iterable
+import datetime
+import pandas as pd
+#import pysnooper
+
+DATE_FORMAT_DASH = '%Y-%m-%d'
+DATE_FORMAT_DIGIT = '%Y%m%d'
+TIME_FORMAT_DASH = '%Y-%m-%d %H:%M:%S'
+
+
+def range_str(
+    start,
+    stop,
+    *,
+    step=datetime.timedelta(days=1),
+    format: str = TIME_FORMAT_DASH
+) -> Iterable[str]:
+    for ts in range(start=start, stop=stop, step=step):
+        yield ts.strftime(format)
+
+
+def range(start, stop,
+          step=datetime.timedelta(days=1)) -> Iterable[datetime.datetime]:
+    """Generate a range of datetime objects.
+    :param start: A datetime object or a string that can be parsed into a datetime.
+    :param stop: A datetime object or a string that can be parsed into a datetime.
+    :param step: A timedelta object specifying how much the values in the sequence increase at each step. 
+    """
+    start = pd.to_datetime(start)
+    stop = pd.to_datetime(stop)
+    curr_dt = start
+    while curr_dt < stop:
+        yield curr_dt
+        curr_dt += step
+
+
+def last_weekday(weekday) -> datetime.date:
+    """Get the date of latest occurrence of the specified weekday.
+    """
+    mapping = {
+        'Monday': 0,
+        'Mon': 0,
+        'Tuesday': 1,
+        'Tue': 1,
+        'Wednesday': 2,
+        'Wed': 2,
+        'Thursday': 3,
+        'Thu': 3,
+        'Friday': 4,
+        'Fri': 4,
+        'Saturday': 5,
+        'Sat': 5,
+        'Sunday': 6,
+    }
+    if type(weekday) == str:
+        weekday = mapping[weekday]
+    today = datetime.date.today()
+    diff = today.weekday() - weekday
+    if diff < 0:
+        diff += 7
+    return today - datetime.timedelta(days=diff)
+
+
+def last_monday() -> datetime.date:
+    """Get the date of latest occurrence of Monday.
+    """
+    return last_weekday('Mon')
+
+
+def last_tuesday() -> datetime.date:
+    """Get the date of latest occurrence of Monday.
+    """
+    return last_weekday('Tue')
+
+
+def last_wednesday() -> datetime.date:
+    """Get the date of latest occurrence of Monday.
+    """
+    return last_weekday('Wed')
+
+
+def last_thursday() -> datetime.date:
+    """Get the date of latest occurrence of Monday.
+    """
+    return last_weekday('Thu')
+
+
+def last_friday() -> datetime.date:
+    """Get the date of latest occurrence of Monday.
+    """
+    return last_weekday('Fri')
+
+
+def last_saturday() -> datetime.date:
+    """Get the date of latest occurrence of Monday.
+    """
+    return last_weekday('Sat')
+
+
+def last_sunday() -> datetime.date:
+    """Get the date of latest occurrence of Monday.
+    """
+    return last_weekday('Sun')
+
+
+def today(days: int = 0) -> datetime.date:
+    """Return the date of days after today as a date.
+    """
+    return datetime.date.today() + datetime.timedelta(days=days)
+
+
+def today_dash(days: int = 0) -> str:
+    """Return the date of days after today as a dash separated string.
+    """
+    return today(days).strftime(DATE_FORMAT_DASH)
+
+
+def today_digit(days: int = 0) -> str:
+    """Return the date of days after today as a 8-digit string.
+    """
+    return today(days).strftime(DATE_FORMAT_DIGIT)