- Fix variable naming: TIME_COLUMN -> time_column, l -> components, COLUMNS -> columns - Extract exception string literals to variables (EM101) - Replace assert statements with proper error handling in obsidian_import - Use dict.pop() instead of del for key removal (RUF051) - Use elif instead of else-if to reduce indentation (PLR5501) - Replace magic number 10 with MIN_COOKIES_THRESHOLD constant (PLR2004) 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
from collections.abc import Iterator
|
|
from logging import getLogger
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from personal_data.activity import ActivitySample, Label
|
|
from personal_data.csv_import import determine_possible_keys, load_csv_file, start_end
|
|
|
|
logger = getLogger(__name__)
|
|
|
|
|
|
def iterate_samples_from_dicts(rows: list[dict[str, Any]]) -> Iterator[ActivitySample]:
|
|
if len(rows) == 0:
|
|
message = 'No rows provided'
|
|
raise ValueError(message)
|
|
|
|
if True:
|
|
event_data = rows[len(rows) // 2] # Hopefully select a useful representative.
|
|
possible_keys = determine_possible_keys(event_data)
|
|
logger.info('Found possible keys: %s', possible_keys)
|
|
del event_data
|
|
|
|
if len(possible_keys.time_start) + len(possible_keys.time_end) < 1:
|
|
message = 'No time columns found in data'
|
|
raise ValueError(message)
|
|
|
|
for event_data in rows:
|
|
"""
|
|
title = ': '.join(event_data[k] for k in possible_name_keys[:max_title_parts])
|
|
description = '\n\n'.join(
|
|
event_data[k] for k in possible_name_keys[max_title_parts:]
|
|
)
|
|
image = event_data[possible_keys.image[0]] if possible_keys.image else None
|
|
"""
|
|
|
|
(start_at, end_at) = start_end(event_data, possible_keys)
|
|
labels = [Label(k, event_data[k]) for k in possible_keys.misc]
|
|
|
|
# Create event
|
|
yield ActivitySample(
|
|
labels=tuple(labels),
|
|
start_at=start_at,
|
|
end_at=end_at,
|
|
)
|
|
|
|
del event_data
|
|
|
|
|
|
def iterate_samples_from_csv_file(file_path: Path) -> Iterator[ActivitySample]:
|
|
dicts = load_csv_file(file_path)
|
|
samples = list(iterate_samples_from_dicts(dicts))
|
|
if len(samples) == 0:
|
|
message = 'Did not find any samples'
|
|
raise ValueError(message)
|
|
yield from samples
|