Refactor data into dataclasses
This commit is contained in:
parent
c5b532b618
commit
76bedd42ee
7 changed files with 197 additions and 62 deletions
|
@ -1,3 +1,6 @@
|
|||
from habitmove.loopdata import Habit
|
||||
|
||||
|
||||
def migrate(db, trackers):
|
||||
c = db.cursor()
|
||||
habits = trackers_to_habits(trackers)
|
||||
|
@ -10,40 +13,33 @@ def migrate(db, trackers):
|
|||
return habits
|
||||
|
||||
|
||||
# By default set goal to half of max value
|
||||
NOMIE_MAX_TO_TARGET_VALUE_RATIO = 2
|
||||
|
||||
|
||||
def trackers_to_habits(trackers):
|
||||
habits = []
|
||||
for tracker_name in trackers.keys():
|
||||
t = trackers[tracker_name]
|
||||
for t in trackers:
|
||||
habits.append(
|
||||
{
|
||||
"archived": 0 if t["hidden"] == False else 1,
|
||||
"color": 11 if t.get("score", 0) != "-1" else 0,
|
||||
"description": t["tag"],
|
||||
"freq_den": 1,
|
||||
"freq_num": 1,
|
||||
"highlight": 0,
|
||||
"name": f"{t['emoji']} {t['label']}",
|
||||
"question": "",
|
||||
"position": 0,
|
||||
"unit": "" if t["uom"] == "num" else t["uom"],
|
||||
"uuid": t["id"],
|
||||
}
|
||||
Habit(
|
||||
archived=t.hidden,
|
||||
color=11 if t.score != "-1" else 0,
|
||||
description=t.tag,
|
||||
name=f"{t.emoji} {t.label}",
|
||||
unit="" if t.uom == "num" else t.uom,
|
||||
uuid=t.id,
|
||||
)
|
||||
)
|
||||
if t["type"] == "range" and len(habits) > 0:
|
||||
habits[-1]["type"] = 1
|
||||
if t.type == "range" and len(habits) > 0:
|
||||
habits[-1].type = 1
|
||||
# nomie only has concept of max value,
|
||||
# use a percentage of it for Loop range target
|
||||
habits[-1]["target_value"] = (
|
||||
int(t["max"]) // NOMIE_MAX_TO_TARGET_VALUE_RATIO
|
||||
)
|
||||
habits[-1].target_value = int(t.max) // NOMIE_MAX_TO_TARGET_VALUE_RATIO
|
||||
return habits
|
||||
|
||||
|
||||
def check_habit_duplicate(cursor, habit):
|
||||
cursor.execute("select name from Habits where uuid = ?", [habit["uuid"]])
|
||||
cursor.execute("select name from Habits where uuid = ?", [habit.uuid])
|
||||
name = cursor.fetchone()
|
||||
if name:
|
||||
return name[0]
|
||||
|
@ -54,15 +50,13 @@ def add_to_database(cursor, habit):
|
|||
"""Takes a habit in the form of a dictionary and inserts it into the Habits table.
|
||||
Parameters:
|
||||
cursor (db.cursor): SQL executing cursor
|
||||
habit (dict): A Loop habit to be added to the database. Must contain a minimum of
|
||||
'archived','color','description','freq_den','freq_num','highlight','name',
|
||||
'question','position','unit','uuid' as keys.
|
||||
habit (Habit): A Loop habit to be added to the database.
|
||||
"""
|
||||
|
||||
placeholder = ", ".join("?" * len(habit))
|
||||
columns = ", ".join(habit.keys())
|
||||
habit_data = habit.__dict__
|
||||
placeholder = ", ".join("?" * len(habit_data))
|
||||
columns = ", ".join(habit_data.keys())
|
||||
sql = "insert into `{table}` ({columns}) values ({values});".format(
|
||||
table="Habits", columns=columns, values=placeholder
|
||||
)
|
||||
values = list(habit.values())
|
||||
values = list(habit_data.values())
|
||||
cursor.execute(sql, values)
|
||||
|
|
25
habitmove/loopdata.py
Normal file
25
habitmove/loopdata.py
Normal file
|
@ -0,0 +1,25 @@
|
|||
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.
|
||||
@dataclass
|
||||
class Habit:
|
||||
name: str
|
||||
type: Optional[int] = 0 # 1 is range counter
|
||||
archived: Optional[int] = 0
|
||||
color: Optional[int] = 0
|
||||
highlight: Optional[int] = 0
|
||||
freq_den: Optional[int] = 1
|
||||
freq_num: Optional[int] = 1
|
||||
target_value: Optional[int] = 0
|
||||
description: Optional[str] = ""
|
||||
question: Optional[str] = ""
|
||||
unit: Optional[str] = ""
|
||||
position: Optional[int] = 0
|
||||
uuid: Optional[str] = None
|
||||
|
||||
# TODO test post init uuid setting
|
||||
def __post_init__(self):
|
||||
if not self.uuid:
|
||||
self.uuid = uuid4().hex
|
|
@ -1,6 +1,8 @@
|
|||
#!/usr/bin/env python
|
||||
|
||||
import json
|
||||
import re
|
||||
from habitmove.nomiedata import Tracker, Event, Activity, NomieImport
|
||||
|
||||
|
||||
def load_file(filename):
|
||||
|
@ -23,21 +25,77 @@ def confirmation_question(question, default_no=True):
|
|||
|
||||
|
||||
# display stats and ask user to confirm if they seem okay
|
||||
def verify_continue(data):
|
||||
def verify_continue(data: NomieImport):
|
||||
trackers = ""
|
||||
for t in data["trackers"]:
|
||||
trackers += t + ", "
|
||||
print(f"Exporting from nomie {data['nomie']['number']}:")
|
||||
for t in data.trackers:
|
||||
trackers += t.label + ", "
|
||||
trackers = trackers[:-2]
|
||||
|
||||
print(f"Exporting from nomie {data.version}:")
|
||||
print(f"Found trackers: {trackers}")
|
||||
print(f"Found events: {len(data['events'])} entries.")
|
||||
print(
|
||||
f"Found events: {len(data.events)} entries, containing {len(data.activities)} individual activities."
|
||||
)
|
||||
if not confirmation_question("Do you want to continue?", default_no=False):
|
||||
print("Aborted.")
|
||||
exit(0)
|
||||
|
||||
|
||||
def get_trackers(raw_trackers):
|
||||
tracker_list = list[Tracker]()
|
||||
for tracker_tuple in raw_trackers.items():
|
||||
tracker_list.append(Tracker(**tracker_tuple[1]))
|
||||
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 extract_tags_from_text(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
|
||||
|
||||
|
||||
def get_activities(tracker_list, note_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]))
|
||||
return activities
|
||||
|
||||
|
||||
# return the data belonging to nomie
|
||||
def get_data(file, interactive):
|
||||
nomie_data = load_file(file)
|
||||
def get_data(file, interactive=True):
|
||||
raw_data = load_file(file)
|
||||
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)
|
||||
|
||||
data = NomieImport(nomie_version, tracker_list, note_list, activity_list)
|
||||
if interactive:
|
||||
verify_continue(nomie_data)
|
||||
return nomie_data
|
||||
verify_continue(data)
|
||||
|
||||
return data
|
||||
|
|
58
habitmove/nomiedata.py
Normal file
58
habitmove/nomiedata.py
Normal file
|
@ -0,0 +1,58 @@
|
|||
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
|
||||
class Tracker:
|
||||
color: str
|
||||
emoji: str
|
||||
hidden: bool
|
||||
id: str
|
||||
ignore_zeros: bool
|
||||
label: str
|
||||
math: str
|
||||
one_tap: bool
|
||||
tag: str
|
||||
type: str
|
||||
uom: str
|
||||
# TODO no idea what include does
|
||||
include: str
|
||||
min: Optional[int] = 0
|
||||
max: Optional[int] = 0
|
||||
goal: Optional[int] = 0
|
||||
default: Optional[int] = 0
|
||||
# TODO score can be string (if custom) or int (if simple good/bad)
|
||||
score: Optional[str] = ""
|
||||
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
|
||||
class Activity:
|
||||
tracker: Tracker
|
||||
value: Optional[int] = 1
|
||||
|
||||
|
||||
@dataclass
|
||||
class NomieImport:
|
||||
version: str
|
||||
trackers: list[Tracker]
|
||||
events: list[Event]
|
||||
activities: list[Activity]
|
|
@ -26,8 +26,10 @@ def get_all_repetitions(habits, events):
|
|||
"""
|
||||
repetitions = []
|
||||
for event in events:
|
||||
for habit in habits:
|
||||
reps = tags_to_repetitions(habit, extract_tags(event["note"]), event["end"])
|
||||
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)
|
||||
|
||||
|
@ -51,7 +53,7 @@ def extract_tags(text, tagmarker="#"):
|
|||
|
||||
# does not do:
|
||||
# non-range habits but still #habit(3) number included, adding multiple?
|
||||
def tags_to_repetitions(habit, tags, timestamp):
|
||||
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.
|
||||
|
@ -64,10 +66,10 @@ def tags_to_repetitions(habit, tags, timestamp):
|
|||
"""
|
||||
reps = []
|
||||
for tag in tags:
|
||||
if habit["description"] in tag[0]:
|
||||
repetition = {"id": habit["id"], "timestamp": timestamp}
|
||||
if habit.description in tag[0]:
|
||||
repetition = {"id": habit_id, "timestamp": timestamp}
|
||||
if tag[1]:
|
||||
if "type" in habit and habit["type"] == 1:
|
||||
if habit.type == 1:
|
||||
repetition["value"] = tag[1] * LOOP_RANGE_VALUE_MULTIPLIER
|
||||
reps.append(repetition)
|
||||
return reps
|
||||
|
@ -78,15 +80,16 @@ def habit_list_add_ids(c, habitlist):
|
|||
"""Return the collection of habits with their sqlite id added.
|
||||
Parameters:
|
||||
c (sqlite.db.cursor): SQL cursor of database to query.
|
||||
habitlist (list): Full habit collection to return a simplified view of.
|
||||
habitlist (list[Habit]): Full habit collection to return a simplified view of.
|
||||
Returns:
|
||||
habitlist (list): The full habit collection passed in unchanged,
|
||||
except for an id added to each habit corresponding
|
||||
to their id in sqlite database and Loop repetitions.
|
||||
habit_id_dict (dict[Habit]): The habit collection as a dict with the keys
|
||||
consisting of the habit's sqlite database ID.
|
||||
"""
|
||||
with_id = habitlist.copy()
|
||||
for h in with_id:
|
||||
h["id"] = fetch_habit_id(c, h["uuid"])
|
||||
with_id = {}
|
||||
for h in habitlist:
|
||||
sql_id = fetch_habit_id(c, h.uuid)
|
||||
with_id[sql_id] = h
|
||||
|
||||
return with_id
|
||||
|
||||
|
||||
|
|
|
@ -5,7 +5,7 @@ def create_database(name):
|
|||
return sqlite3.connect(name)
|
||||
|
||||
|
||||
# better way to do the above
|
||||
# TODO better way to do the above
|
||||
# def create_connection(db_file):
|
||||
# """create a database connection to the SQLite database
|
||||
# specified by db_file
|
||||
|
|
19
run.py
19
run.py
|
@ -3,19 +3,20 @@ import habitmove.schema as schema
|
|||
import habitmove.habits as habits
|
||||
import habitmove.repetitions as rep
|
||||
import habitmove.nomie as nomie
|
||||
from habitmove.nomiedata import NomieImport
|
||||
|
||||
|
||||
import sys
|
||||
|
||||
|
||||
def migrate(trackers, events):
|
||||
def migrate(data: NomieImport):
|
||||
db = schema.migrate("output.db")
|
||||
if trackers is not None:
|
||||
if data.trackers is not None:
|
||||
|
||||
habitlist = habits.migrate(db, trackers)
|
||||
habitlist = habits.migrate(db, data.trackers)
|
||||
|
||||
if events is not None:
|
||||
rep.migrate(db, habitlist, events)
|
||||
if data.events is not None:
|
||||
rep.migrate(db, habitlist, data.events)
|
||||
|
||||
db.commit()
|
||||
db.close()
|
||||
|
@ -23,13 +24,9 @@ def migrate(trackers, events):
|
|||
|
||||
def main():
|
||||
file = sys.argv[1]
|
||||
data = nomie.get_data(file, True)
|
||||
migrate(data["trackers"], data["events"])
|
||||
data = nomie.get_data(file)
|
||||
migrate(data)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
# things to have as (data) classes:
|
||||
# tracker, event, tag?
|
||||
# habit, repetition
|
||||
|
|
Loading…
Reference in a new issue