From 743312a87bc2ba876f350b608fbdcf75f2c84ad8 Mon Sep 17 00:00:00 2001 From: Dima Gerasimov Date: Tue, 1 Sep 2020 18:43:37 +0100 Subject: [PATCH] my.body.blood: prettify, add stat() --- my/body/blood.py | 118 ++++++++++++++++++++++++------------------- my/core/orgmode.py | 21 ++++++++ my/emfit/__init__.py | 3 ++ 3 files changed, 91 insertions(+), 51 deletions(-) create mode 100644 my/core/orgmode.py diff --git a/my/body/blood.py b/my/body/blood.py index 9a614d3..ee3bf03 100755 --- a/my/body/blood.py +++ b/my/body/blood.py @@ -4,110 +4,126 @@ Blood tracking from datetime import datetime from typing import Iterable, NamedTuple, Optional -from itertools import chain -import porg -from ..common import listify -from ..error import Res, echain +from ..core.common import listify +from ..core.error import Res, echain +from ..core.orgmode import parse_org_datetime -from kython.org import parse_org_date - -from my.config import blood as config - import pandas as pd # type: ignore +import porg + + +from my.config import blood as config class Entry(NamedTuple): dt: datetime - ket: Optional[float]=None - glu: Optional[float]=None + ketones : Optional[float]=None + glucose : Optional[float]=None - vitd: Optional[float]=None - b12: Optional[float]=None + vitamin_d : Optional[float]=None + vitamin_b12 : Optional[float]=None - hdl: Optional[float]=None - ldl: Optional[float]=None - trig: Optional[float]=None + hdl : Optional[float]=None + ldl : Optional[float]=None + triglycerides: Optional[float]=None - extra: Optional[str]=None + source : Optional[str]=None + extra : Optional[str]=None Result = Res[Entry] -class ParseError(Exception): - pass - def try_float(s: str) -> Optional[float]: l = s.split() if len(l) == 0: return None + # meh. this is to strip away HI/LO? Maybe need extract_float instead or something x = l[0].strip() if len(x) == 0: return None return float(x) -def iter_gluc_keto_data() -> Iterable[Result]: +def glucose_ketones_data() -> Iterable[Result]: o = porg.Org.from_file(str(config.blood_log)) tbl = o.xpath('//table') + # todo some sort of sql-like interface for org tables might be ideal? for l in tbl.lines: kets = l['ket'].strip() glus = l['glu'].strip() extra = l['notes'] - dt = parse_org_date(l['datetime']) - assert isinstance(dt, datetime) - ket = try_float(kets) - glu = try_float(glus) - yield Entry( - dt=dt, - ket=ket, - glu=glu, - extra=extra, - ) + dt = parse_org_datetime(l['datetime']) + try: + assert isinstance(dt, datetime) + ket = try_float(kets) + glu = try_float(glus) + except Exception as e: + ex = RuntimeError(f'While parsing {l}') + ex.__cause__ = e + yield ex + else: + yield Entry( + dt=dt, + ketones=ket, + glucose=glu, + extra=extra, + ) -def iter_tests_data() -> Iterable[Result]: +def blood_tests_data() -> Iterable[Result]: o = porg.Org.from_file(str(config.blood_tests_log)) tbl = o.xpath('//table') for d in tbl.lines: try: - dt = parse_org_date(d['datetime']) - assert isinstance(dt, datetime) - # TODO rest + dt = parse_org_datetime(d['datetime']) + assert isinstance(dt, datetime), dt F = lambda n: try_float(d[n]) yield Entry( dt=dt, - vitd=F('VD nm/L'), - b12 =F('B12 pm/L'), + vitamin_d =F('VD nm/L'), + vitamin_b12 =F('B12 pm/L'), - hdl =F('HDL mm/L'), - ldl =F('LDL mm/L'), - trig=F('Trig mm/L'), + hdl =F('HDL mm/L'), + ldl =F('LDL mm/L'), + triglycerides=F('Trig mm/L'), - extra=d['misc'], + source =d['source'], + extra =d['notes'], ) except Exception as e: - print(e) - yield echain(ParseError(str(d)), e) + ex = RuntimeError(f'While parsing {d}') + ex.__cause__ = e + yield ex -def data(): - datas = list(chain(iter_gluc_keto_data(), iter_tests_data())) - return list(sorted(datas, key=lambda d: getattr(d, 'dt', datetime.min))) +def data() -> Iterable[Result]: + from itertools import chain + from ..core.error import sort_res_by + datas = chain(glucose_ketones_data(), blood_tests_data()) + return sort_res_by(datas, key=lambda e: e.dt) -@listify(wrapper=pd.DataFrame) -def dataframe(): - for d in data(): - if isinstance(d, Exception): - yield {'error': str(d)} +def dataframe() -> pd.DataFrame: + rows = [] + for x in data(): + if isinstance(x, Exception): + # todo use some core helper? this is a pretty common operation + d = {'error': str(x)} else: - yield d._asdict() + d = x._asdict() + rows.append(d) + return pd.DataFrame(rows) + + +def stats(): + from ..core import stat + return stat(data) def test(): diff --git a/my/core/orgmode.py b/my/core/orgmode.py new file mode 100644 index 0000000..3f4ac25 --- /dev/null +++ b/my/core/orgmode.py @@ -0,0 +1,21 @@ +""" +Various helpers for reading org-mode data +""" +from datetime import datetime + + +def parse_org_datetime(s: str) -> datetime: + s = s.strip('[]') + for fmt, cl in [ + ("%Y-%m-%d %a %H:%M", datetime), + ("%Y-%m-%d %H:%M" , datetime), + # todo not sure about these... fallback on 00:00? + # ("%Y-%m-%d %a" , date), + # ("%Y-%m-%d" , date), + ]: + try: + return datetime.strptime(s, fmt) + except ValueError: + continue + else: + raise RuntimeError(f"Bad datetime string {s}") diff --git a/my/emfit/__init__.py b/my/emfit/__init__.py index 09fc61b..0a1e1b3 100755 --- a/my/emfit/__init__.py +++ b/my/emfit/__init__.py @@ -100,7 +100,9 @@ def dataframe() -> DataFrameT: else: # todo it's change during the day?? dunno if reasonable metric hrv_change = s.hrv_evening - last.hrv_morning + # todo maybe changes need to be handled in a more generic way? + # todo ugh. get rid of hardcoding, just generate the schema automatically # TODO use 'workdays' provider.... d = { 'date' : dd, @@ -116,6 +118,7 @@ def dataframe() -> DataFrameT: 'hrv_morning': s.hrv_morning, 'recovery' : s.recovery, 'hrv_change' : hrv_change, + 'respiratory_rate_avg': s.respiratory_rate_avg, } last = s # meh dicts.append(d)