Obsidian import initial attempt

obsidian_import/__init__.py

@@ -0,0 +1,41 @@
+"""Obsidian Import.
+
+Sub-module for importing time-based data into Obsidian.
+"""
+
+from pathlib import Path
+from .obsidian import ObsidianVault
+from personal_data.util import load_csv_file
+import datetime
+from logging import getLogger
+logger = getLogger(__name__)
+
+def import_data(obsidian_path: Path, dry_run = True):
+    vault = ObsidianVault(obsidian_path, read_only = dry_run and 'silent' or None)
+
+    data_path = Path('/home/jmaa/Notes/workout.csv')
+    rows = load_csv_file(data_path)
+    logger.info('Loaded CSV with %d lines', len(rows))
+    num_updated = 0
+    for row in rows:
+        date = row['Date']
+        was_updated = False
+        mapping = {
+            'Cycling (mins)': ('Cycling (Duration)', 'minutes'),
+            'Cycling (kcals)': ('Cycling (kcals)', ''),
+            'Weight (Kg)': ('Weight (Kg)', ''),
+        }
+
+        for input_key, (output_key, unit) in mapping.items():
+            v = row.get(input_key)
+            if unit:
+                v = str(v) + ' ' + unit
+            if v:
+                was_updated |= vault.add_statistic(date, output_key, v)
+            del input_key, output_key, unit, v
+
+        if was_updated:
+            num_updated += 1
+        del row, date
+
+    logger.info('Updated %d files', num_updated)

obsidian_import/__main__.py

@@ -0,0 +1,30 @@
+import argparse
+import logging
+from pathlib import Path
+
+from . import import_data
+
+logger = logging.getLogger(__name__)
+
+def parse_arguments():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--vault', type=Path, required=True)
+    parser.add_argument('--yes', action='store_false', dest='dry_run')
+    return parser.parse_args()
+
+
+def main():
+    # Setup logging
+    logging.basicConfig()
+    logging.getLogger('obsidian_import').setLevel('INFO')
+
+    args = parse_arguments()
+    if args.dry_run:
+        logger.warning('Dry run')
+    import_data(args.vault, dry_run = args.dry_run)
+    if args.dry_run:
+        logger.warning('Dry run: Use --yes to execute')
+
+
+if __name__ == '__main__':
+    main()

obsidian_import/obsidian.py

@@ -0,0 +1,83 @@
+
+import datetime
+from typing import Any
+import json
+from pathlib import Path
+
+import frontmatter
+from decimal import Decimal
+from logging import getLogger
+logger = getLogger(__name__)
+
+StatisticKey = str
+
+class ObsidianVault:
+
+    def __init__(self, vault_path : Path, read_only: bool = 'silent'):
+        self.vault_path  = vault_path
+
+        assert (self.vault_path / '.obsidian').exists(), 'Not an Obsidian Vault'
+
+        with open(self.vault_path  / '.obsidian'  / 'daily-notes.json') as f:
+            daily_notes_config = json.load(f)
+        self.daily_folder = daily_notes_config['folder']
+        self.path_format = daily_notes_config['format']
+        self.template_file_path = daily_notes_config['template']
+        self.read_only = read_only
+
+    def get_statistic(self, date: datetime.date, statistic_key: StatisticKey) -> Any | None:
+        try:
+            with open(self._date_file_path(date)) as f:
+                data = frontmatter.load(f)
+        except FileNotFoundError:
+            return None
+
+        return data.metadata.get(statistic_key)
+
+    def add_statistic(self, date: datetime.date, statistic_key: StatisticKey, amount: Any) -> bool:
+        if self.read_only == 'silent':
+            logger.info('Real only ObsidianVault ignoring add_statistic(%s, "%s", ?)', date, statistic_key)
+            return False
+
+        self._create_date_if_not_present(date)
+
+        with open(self._date_file_path(date)) as f:
+            data = frontmatter.load(f)
+
+        if isinstance(amount, Decimal):
+            amount = float(amount)
+
+        if data.metadata.get(statistic_key) == amount:
+            return False
+
+        data.metadata[statistic_key] = amount
+
+        with open(self._date_file_path(date), 'wb') as f:
+            frontmatter.dump(data, f)
+
+        return True
+
+    def add_event(self, date: datetime.date, verb: str, subject: str) -> None:
+        if self.read_only == 'silent':
+            logger.info('Real only ObsidianVault ignoring add_event(%s, "%s", ?)', date, verb)
+            return
+
+        self._create_date_if_not_present(date)
+        # TODO
+
+    def _create_date_if_not_present(self, date: datetime.date):
+        date_file = self._date_file_path(date)
+        if date_file.exists():
+            return
+        logger.info('File "%s" doesn\'t exist, creating...', date)
+        with open(self._daily_template_path()) as f:
+            template_text = f.read()
+        with open(date_file, 'w') as f:
+            f.write(template_text)
+
+    def _date_file_path(self, date: datetime.date):
+        path = self.path_format.replace('YYYY', str(date.year)).replace('MM', '{:02d}'.format(date.month)).replace('DD', '{:02d}'.format(date.day))
+        return (self.vault_path / self.daily_folder / path).with_suffix('.md')
+
+    def _daily_template_path(self):
+        return (self.vault_path / self.template_file_path).with_suffix('.md')

personal_data/util.py

@@ -39,6 +39,8 @@ def csv_str_to_value(
     | bool
     | None
 ):
+    if s is None:
+        return None
     s = s.strip()
     if len(s) == 0:
         return None