Make nomie dataclasses immutable

This commit is contained in:
Marty Oehme 2021-12-03 16:22:54 +01:00
parent 76bedd42ee
commit 76b2dd4408
Signed by: Marty
GPG key ID: B7538B8F50A1C800
4 changed files with 102 additions and 111 deletions

View file

@ -2,7 +2,8 @@ from typing import Optional
from dataclasses import dataclass
from uuid import uuid4
# A loop Habit representation. Tracks anything whose value can be encapsulated in a numerical value.
# A loop Habit representation.
# Tracks anything whose value can be encapsulated in a numerical value.
@dataclass
class Habit:
name: str
@ -23,3 +24,12 @@ class Habit:
def __post_init__(self):
if not self.uuid:
self.uuid = uuid4().hex
# A Loop repetition representation, containing only the bare minimum
# for its successful entry into the Loop Habit Tracker database.
@dataclass
class Repetition:
habit_uuid: str
timestamp: int
value: Optional[int]

View file

@ -31,10 +31,14 @@ def verify_continue(data: NomieImport):
trackers += t.label + ", "
trackers = trackers[:-2]
activity_count = 0
for e in data.events:
activity_count += len(e.activities) if e.activities else 0
print(f"Exporting from nomie {data.version}:")
print(f"Found trackers: {trackers}")
print(
f"Found events: {len(data.events)} entries, containing {len(data.activities)} individual activities."
f"Found events: {len(data.events)} entries, containing {activity_count} individual activities."
)
if not confirmation_question("Do you want to continue?", default_no=False):
print("Aborted.")
@ -48,15 +52,19 @@ def get_trackers(raw_trackers):
return tracker_list
def get_notes(raw_notes):
notes = list[Event]()
for note in raw_notes:
note["id"] = note["_id"]
note.pop("_id")
note["text"] = note["note"]
note.pop("note")
notes.append(Event(**note))
return notes
def get_events(raw_events, tracker_list):
events = list[Event]()
for event in raw_events:
event["id"] = event["_id"]
event.pop("_id")
event["text"] = event["note"]
event.pop("note")
activities = get_activities_for_event(event["text"], tracker_list)
events.append(Event(**event, activities=activities))
return events
def extract_tags_from_text(text, tagmarker="#"):
@ -74,14 +82,13 @@ def extract_tags_from_text(text, tagmarker="#"):
return tags_with_int_counters
def get_activities(tracker_list, note_list):
def get_activities_for_event(event_text, tracker_list):
activities = []
for note in note_list:
tag_list = extract_tags_from_text(note.text)
for tracker in tracker_list:
for tag in tag_list:
if tracker.tag in tag[0]:
activities.append(Activity(tracker=tracker, value=tag[1]))
tag_list = extract_tags_from_text(event_text)
for tracker in tracker_list:
for tag in tag_list:
if tracker.tag in tag[0]:
activities.append(Activity(tracker=tracker, value=tag[1]))
return activities
@ -91,10 +98,9 @@ def get_data(file, interactive=True):
nomie_version = raw_data["nomie"]["number"]
tracker_list = get_trackers(raw_data["trackers"])
note_list = get_notes(raw_data["events"])
activity_list = get_activities(tracker_list, note_list)
event_list = get_events(raw_data["events"], tracker_list)
data = NomieImport(nomie_version, tracker_list, note_list, activity_list)
data = NomieImport(nomie_version, tracker_list, event_list)
if interactive:
verify_continue(data)

View file

@ -2,7 +2,7 @@ from typing import Optional, Any
from dataclasses import dataclass, field
# A nomie habit tracker. Tracks anything whose value can be encapsulated in a numerical value.
@dataclass
@dataclass(frozen=True)
class Tracker:
color: str
emoji: str
@ -26,33 +26,33 @@ class Tracker:
score_calc: Optional[list[dict[str, Any]]] = field(default_factory=lambda: [])
# A nomie note. Records any circumstance of 'something happened' through prose.
# These are undigested events, whose changed trackers are still encapsulated
# in the 'note' field as continuous text.
@dataclass
class Event:
id: str
text: str
start: int
end: int
score: int
lat: float
lng: float
modified: Optional[bool] = False
source: Optional[str] = "n5"
offset: Optional[str] = ""
location: Optional[str] = ""
@dataclass
@dataclass(frozen=True)
class Activity:
tracker: Tracker
value: Optional[int] = 1
@dataclass
# A nomie note. Records any circumstance of 'something happened' through prose.
# These are undigested events, whose changed trackers are still encapsulated
# in the 'note' field as continuous text.
@dataclass(frozen=True)
class Event:
id: str
start: int
end: int
text: str
activities: Optional[list[Activity]] = field(default_factory=lambda: [])
score: Optional[int] = 0
lat: Optional[float] = 0.0
lng: Optional[float] = 0.0
location: Optional[str] = ""
modified: Optional[bool] = False
offset: Optional[str] = "" # local timezone offset?
source: Optional[str] = "n5" # nomie version
@dataclass(frozen=True)
class NomieImport:
version: str
trackers: list[Tracker]
events: list[Event]
activities: list[Activity]

View file

@ -1,5 +1,6 @@
import sqlite3
import re
from datetime import datetime
from habitmove.loopdata import Habit, Repetition
def migrate(db, habitlist, events):
@ -7,9 +8,7 @@ def migrate(db, habitlist, events):
habits = habit_list_add_ids(c, habitlist)
repetitions = get_all_repetitions(habits, events)
for rep in repetitions:
add_to_database(
c, rep["id"], rep["timestamp"], 2 if "value" not in rep else rep["value"]
)
add_to_database(c, habits, rep)
LOOP_RANGE_VALUE_MULTIPLIER = 1000
@ -26,55 +25,20 @@ def get_all_repetitions(habits, events):
"""
repetitions = []
for event in events:
for habit_id in habits.keys():
reps = tags_to_repetitions(
habit_id, habits[habit_id], extract_tags(event.text), event.end
)
if reps:
repetitions.extend(reps)
for activity in event.activities:
for habit in habits.values():
# TODO Fix reaching a layer too far into activity -> tracker
if habit.uuid == activity.tracker.id:
rep = Repetition(
habit_uuid=habit.uuid, timestamp=event.end, value=2
)
if habit.type == 1 and activity.value:
rep.value = activity.value * LOOP_RANGE_VALUE_MULTIPLIER
repetitions.append(rep)
return repetitions
def extract_tags(text, tagmarker="#"):
"""Return lists of tuples of all event tags found in text.
Parameters:
text (str): The text to search through.
tagmarker (str): Optional character marking beginning of tag, defaults to '#'.
Returns:
tags (list): List of tuples in the form [('tag', '3'), ('anothertag', '')].
"""
string_tags = re.findall(rf"{tagmarker}(\w+)(?:\((\d+)\))?", text)
tags_with_int_counters = []
for tag in string_tags:
tags_with_int_counters.append((tag[0], None if tag[1] == "" else int(tag[1])))
return tags_with_int_counters
# does not do:
# non-range habits but still #habit(3) number included, adding multiple?
def tags_to_repetitions(habit_id, habit, tags, timestamp):
"""Return a list of all repetitions generated from the tags and habits passed in.
Parameters:
habits (list): Collection of habits, with minimum necessary fields description and id.
tags (list): Collection of tag tuples.
Returns:
repetitions (list): Collection of habits for which a corresponding tag has
been found. If they correspond that means at the timestamp
the habit has been checked in, and a repetition is created.
Contains fields id, timestamp, value (for ranges).
"""
reps = []
for tag in tags:
if habit.description in tag[0]:
repetition = {"id": habit_id, "timestamp": timestamp}
if tag[1]:
if habit.type == 1:
repetition["value"] = tag[1] * LOOP_RANGE_VALUE_MULTIPLIER
reps.append(repetition)
return reps
# TODO possibly just get rid of this entirely
def habit_list_add_ids(c, habitlist):
"""Return the collection of habits with their sqlite id added.
@ -93,30 +57,41 @@ def habit_list_add_ids(c, habitlist):
return with_id
def fetch_habit_id(c, uuid):
def fetch_habit_id(cursor: sqlite3.Cursor, uuid: str):
"""Return sqlite internal id for habit with uuid.
Parameters:
c (sqlite.db.cursor): SQL cursor of database to query.
uuid (str): Unique id of habit to query for.
:param c: SQL cursor of database to query.
:param uuid: Unique id of habit to query for.
Returns:
id (int): SQLite internal id for habit queried for.
:return id: SQLite internal id for habit queried for.
"""
c.execute("select id from Habits where uuid = ?", ([uuid]))
id = c.fetchone()
cursor.execute("select id from Habits where uuid = ?", ([uuid]))
id = cursor.fetchone()
if id is not None:
return id[0]
def add_to_database(cursor, habit_id, timestamp, value=2):
try:
cursor.execute(
"""
INSERT INTO
Repetitions(id, habit, timestamp, value)
VALUES (NULL, ?, ?, ?)
""",
(habit_id, timestamp, value),
)
except sqlite3.IntegrityError:
# TODO better error handling
print(f"fail to register {habit_id}: timestamp {timestamp} not unique")
def add_to_database(
cursor: sqlite3.Cursor, habits: dict[int, Habit], repetition: Repetition
):
"""Insert the repetition into a sqlite3 table suitable for Loop.
Parameters:
:param c: SQL cursor of database to query.
:sql_id: Internal sqlite database id of the habit the repetition belongs to.
"""
for sql_id, habit in habits.items():
if repetition.habit_uuid == habit.uuid:
try:
cursor.execute(
"""
INSERT INTO
Repetitions(id, habit, timestamp, value)
VALUES (NULL, ?, ?, ?)
""",
(sql_id, repetition.timestamp, repetition.value),
)
except sqlite3.IntegrityError:
# TODO better error handling
print(
f"{sql_id}, {habit.name}: timestamp {datetime.fromtimestamp(repetition.timestamp/1000)} not unique, moving timestamp slightly."
)