use blood test data
This commit is contained in:
parent
23e6c91758
commit
3059fd85cb
1 changed files with 79 additions and 12 deletions
|
@ -1,25 +1,55 @@
|
||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Iterable, NamedTuple, Optional
|
from typing import Iterable, NamedTuple, Optional
|
||||||
|
from itertools import chain
|
||||||
|
|
||||||
import porg
|
import porg
|
||||||
from kython import listify
|
from kython import listify
|
||||||
from kython.org import parse_org_date
|
from kython.org import parse_org_date
|
||||||
|
from kython.kerror import Res, echain
|
||||||
|
|
||||||
from ..paths import LOGS
|
from ..paths import LOGS, MY
|
||||||
|
|
||||||
|
import pandas as pd # type: ignore
|
||||||
|
|
||||||
|
|
||||||
|
blood_tests_log = MY / 'blood.org'
|
||||||
blood_log = LOGS / 'blood.org'
|
blood_log = LOGS / 'blood.org'
|
||||||
|
|
||||||
class Entry(NamedTuple):
|
class Entry(NamedTuple):
|
||||||
dt: datetime
|
dt: datetime
|
||||||
ket: Optional[float]
|
|
||||||
glu: Optional[float]
|
ket: Optional[float]=None
|
||||||
extra: str
|
glu: Optional[float]=None
|
||||||
|
|
||||||
|
vitd: Optional[float]=None
|
||||||
|
b12: Optional[float]=None
|
||||||
|
|
||||||
|
hdl: Optional[float]=None
|
||||||
|
ldl: Optional[float]=None
|
||||||
|
trig: Optional[float]=None
|
||||||
|
|
||||||
|
extra: Optional[str]=None
|
||||||
|
|
||||||
|
|
||||||
def iter_data() -> Iterable[Entry]:
|
Result = Res[Entry]
|
||||||
o = porg.Org.from_file(blood_log)
|
|
||||||
|
class ParseError(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def try_float(s: str) -> Optional[float]:
|
||||||
|
l = s.split()
|
||||||
|
if len(l) == 0:
|
||||||
|
return None
|
||||||
|
x = l[0].strip()
|
||||||
|
if len(x) == 0:
|
||||||
|
return None
|
||||||
|
return float(x)
|
||||||
|
|
||||||
|
|
||||||
|
def iter_gluc_keto_data() -> Iterable[Result]:
|
||||||
|
o = porg.Org.from_file(str(blood_log))
|
||||||
tbl = o.xpath('//table')
|
tbl = o.xpath('//table')
|
||||||
for l in tbl.lines:
|
for l in tbl.lines:
|
||||||
kets = l['ket'].strip()
|
kets = l['ket'].strip()
|
||||||
|
@ -27,8 +57,8 @@ def iter_data() -> Iterable[Entry]:
|
||||||
extra = l['notes']
|
extra = l['notes']
|
||||||
dt = parse_org_date(l['datetime'])
|
dt = parse_org_date(l['datetime'])
|
||||||
assert isinstance(dt, datetime)
|
assert isinstance(dt, datetime)
|
||||||
ket = None if len(kets) == 0 else float(kets)
|
ket = try_float(kets)
|
||||||
glu = None if len(glus) == 0 else float(glus)
|
glu = try_float(glus)
|
||||||
yield Entry(
|
yield Entry(
|
||||||
dt=dt,
|
dt=dt,
|
||||||
ket=ket,
|
ket=ket,
|
||||||
|
@ -36,14 +66,51 @@ def iter_data() -> Iterable[Entry]:
|
||||||
extra=extra,
|
extra=extra,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def iter_tests_data() -> Iterable[Result]:
|
||||||
|
o = porg.Org.from_file(str(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
|
||||||
|
|
||||||
|
F = lambda n: try_float(d[n])
|
||||||
|
yield Entry(
|
||||||
|
dt=dt,
|
||||||
|
|
||||||
|
vitd=F('VD nm/L'),
|
||||||
|
b12 =F('B12 pm/L'),
|
||||||
|
|
||||||
|
hdl =F('HDL mm/L'),
|
||||||
|
ldl =F('LDL mm/L'),
|
||||||
|
trig=F('Trig mm/L'),
|
||||||
|
|
||||||
|
extra=d['misc'],
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
print(e)
|
||||||
|
yield echain(ParseError(str(d)), e)
|
||||||
|
|
||||||
|
|
||||||
def data():
|
def data():
|
||||||
datas = list(iter_data())
|
datas = list(chain(iter_gluc_keto_data(), iter_tests_data()))
|
||||||
return list(sorted(datas, key=lambda d: d.dt))
|
return list(sorted(datas, key=lambda d: getattr(d, 'dt', datetime.min)))
|
||||||
|
|
||||||
|
|
||||||
|
@listify(wrapper=pd.DataFrame)
|
||||||
def dataframe():
|
def dataframe():
|
||||||
import pandas as pd
|
for d in data():
|
||||||
return pd.DataFrame(map(lambda e: e._asdict(), data()))
|
if isinstance(d, Exception):
|
||||||
|
yield {'error': str(d)}
|
||||||
|
else:
|
||||||
|
yield d._asdict()
|
||||||
|
|
||||||
|
|
||||||
|
def test():
|
||||||
|
print(dataframe())
|
||||||
|
assert len(dataframe()) > 10
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
|
|
Loading…
Add table
Reference in a new issue