100 lines
2.6 KiB
Python
100 lines
2.6 KiB
Python
'''
|
|
Rescuetime (phone activity tracking) data.
|
|
'''
|
|
REQUIRES = [
|
|
'git+https://github.com/karlicoss/rescuexport',
|
|
]
|
|
|
|
from collections.abc import Iterable, Sequence
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
from my.core import Stats, get_files, make_logger, stat
|
|
from my.core.cachew import mcachew
|
|
from my.core.error import Res, split_errors
|
|
|
|
from my.config import rescuetime as config # isort: skip
|
|
|
|
logger = make_logger(__name__)
|
|
|
|
|
|
def inputs() -> Sequence[Path]:
|
|
return get_files(config.export_path)
|
|
|
|
|
|
import rescuexport.dal as dal
|
|
|
|
DAL = dal.DAL
|
|
Entry = dal.Entry
|
|
|
|
|
|
@mcachew(depends_on=inputs)
|
|
def entries() -> Iterable[Res[Entry]]:
|
|
dal = DAL(inputs())
|
|
yield from dal.entries()
|
|
|
|
|
|
def groups(gap: timedelta=timedelta(hours=3)) -> Iterable[Res[Sequence[Entry]]]:
|
|
vit, eit = split_errors(entries(), ET=Exception)
|
|
yield from eit
|
|
from more_itertools import split_when
|
|
yield from split_when(vit, lambda a, b: (b.dt - a.dt) > gap)
|
|
|
|
|
|
# todo automatic dataframe interface?
|
|
from .core.pandas import DataFrameT, as_dataframe
|
|
|
|
|
|
def dataframe() -> DataFrameT:
|
|
return as_dataframe(entries())
|
|
|
|
|
|
def stats() -> Stats:
|
|
return {
|
|
**stat(groups),
|
|
**stat(entries),
|
|
}
|
|
|
|
|
|
# basically, hack config and populate it with fake data? fake data generated by DAL, but the rest is handled by this?
|
|
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
|
|
|
|
# todo take seed, or what?
|
|
@contextmanager
|
|
def fake_data(rows: int=1000) -> Iterator:
|
|
# todo also disable cachew automatically for such things?
|
|
import json
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from my.core.cachew import disabled_cachew
|
|
from my.core.cfg import tmp_config
|
|
with disabled_cachew(), TemporaryDirectory() as td:
|
|
tdir = Path(td)
|
|
f = tdir / 'rescuetime.json'
|
|
f.write_text(json.dumps(dal.fake_data_generator(rows=rows)))
|
|
|
|
class override:
|
|
class rescuetime:
|
|
export_path = tdir
|
|
|
|
with tmp_config(modules=__name__, config=override) as cfg:
|
|
yield cfg
|
|
# TODO ok, now it's something that actually could run on CI!
|
|
# todo would be kinda nice if doctor could run against the fake data, to have a basic health check of the module?
|
|
|
|
|
|
def fill_influxdb() -> None:
|
|
from my.core import influxdb
|
|
|
|
it = ({
|
|
'dt': e.dt,
|
|
'duration_d': e.duration_s,
|
|
'tags': {'activity': e.activity},
|
|
} for e in entries() if isinstance(e, Entry)) # TODO handle errors in core.influxdb
|
|
influxdb.fill(it, measurement=__name__)
|
|
|
|
|
|
# TODO lots of garbage in dir()? maybe need to del the imports...
|