Last modified 2026-07-05 01:14:03 UTC by Aibi#1673 — All History — Edit — Raw
need a pastebin or some shit
## ====== src/quest_engine/__init__.py
import json
import os
import argparse
import pathlib
import random
import rfc8785
import tempfile
from abc import ABC, abstractmethod
from pydantic import TypeAdapter
import paho.mqtt.client as mqtt
from .persistence import (
StorageEngine,
JsonStorageEngine,
iterate_jsonl_with_comments,
)
from .visitors import (
TaskIdSetRemoverVisitor,
TaskEmitRecordKeepAllVisitor,
)
from .model import (
TaskDefinitionRecord,
TaskDefinitionInstantiation,
TaskEmitRecord,
TaskEmitRecordVisitor,
)
MQTT_ADDRESS = os.environ.get("QUEST_ENGINE_MQTT_ADDRESS", "localhost")
MQTT_USERNAME = os.environ.get("QUEST_ENGINE_MQTT_USERNAME", "")
MQTT_PASSWORD = os.environ.get("QUEST_ENGINE_MQTT_PASSWORD", "")
# TODO get home directory dynamically.
DATA_PATH = os.environ.get("QUEST_ENGINE_DATA_PATH", "/home/sell/dev/life/todo-system/data.json")
ACTION_SCANNER_UNIQUE_NAME = "ssmisa" # "qs-1c440c97"
ACTION_SCANNER_TOPIC_STATE = f"{ACTION_SCANNER_UNIQUE_NAME}/profile/state"
QUEST_SCANNER_UNIQUE_NAME = "ssmis" # "qs-1c440c97"
QUEST_SCANNER_TOPIC_STATE = f"{QUEST_SCANNER_UNIQUE_NAME}/profile/state"
def load_config() -> dict:
with open(DATA_PATH, "r") as f:
return json.load(f)
parser = argparse.ArgumentParser('foo')
parser.add_argument('--seed', type=int, default=None, help='Set seed for reproducibility')
subparsers = parser.add_subparsers(dest='subcommand')
load_qd_parser = subparsers.add_parser('load-quest-definitions', help='')
load_qd_parser.add_argument('filename', type=pathlib.Path, help='')
def _remove_emitted(storage, docs):
task_id_candidates = []
for doc in docs:
for subtask in doc.subtasks():
task_id_candidates.append(subtask._jcs_task_id_hash())
visitor = TaskIdSetRemoverVisitor(task_id_candidates)
storage.visit_emitted_tasks(visitor)
out = []
for doc in docs:
this_task_id_set = set(subtask._jcs_task_id_hash() for subtask in doc.subtasks())
if this_task_id_set & visitor.candidates:
out.append(doc)
return out
def _main_load_quest_definitions(options):
assert options.subcommand == 'load-quest-definitions'
local_rng = random.Random(options.seed)
docs = []
with open(options.filename, 'r') as fh:
for doc in iterate_jsonl_with_comments(fh):
docs.append(TaskDefinitionRecord.model_validate(doc))
for idx, doc in enumerate(docs):
print(f"doc[{idx}] = {doc!r}")
with tempfile.TemporaryDirectory() as td:
tdp = pathlib.Path(td)
se = JsonStorageEngine(tdp)
assert docs
doc = local_rng.sample(docs, 1)[0]
for subtask in doc.subtasks():
se.emit_task(subtask)
visitor = TaskEmitRecordKeepAllVisitor()
se.visit_emitted_tasks(visitor)
print(f"seen: {visitor.items}")
# reordered = local_rng.sample(docs, len(docs))
# _remove_emitted(storage, reordered)
# for doc in docs:
# for subtask in doc.subtasks():
# taskhash = subtask._jcs_task_id_hash()
# print(f"subtask[#{taskhash}] = {subtask!r}")
def main():
options = parser.parse_args()
if options.subcommand == 'load-quest-definitions':
_main_load_quest_definitions(options)
else:
raise ValueError("bad subcommand")
def mqtt_loop():
config = load_config()
client = mqtt.Client()
engine = QuestEngine(config, client)
def on_connect(client, userdata, flags, rc):
print(f"Connected to MQTT broker (rc={rc})")
client.subscribe(QUEST_SCANNER_TOPIC_STATE)
client.subscribe(ACTION_SCANNER_UNIQUE_NAME)
def on_message(client, userdata, msg):
print(f"on_message(client={client!r}, userdata={userdata!r}, msg={msg!r})")
try:
payload = msg.payload.decode("utf-8").strip()
print(f"Received: {payload}")
engine.handle_input(msg.topic, payload)
except Exception as e:
print(f"Error handling message: {e}")
import traceback
traceback.print_exc()
client.on_connect = on_connect
client.on_message = on_message
print(f"Quest Engine starting...")
print(f" MQTT broker: {MQTT_ADDRESS}")
print(f" Config: {DATA_PATH}")
print(f" Database: {db.DB_PATH}")
if MQTT_PASSWORD or MQTT_USERNAME:
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
client.connect(MQTT_ADDRESS, 1883)
client.loop_forever()
## ====== src/quest_engine/model.py
import datetime
import hashlib
from dataclasses import dataclass
from typing import Self
from abc import ABC, abstractmethod
import rfc8785
from pydantic import BaseModel, ConfigDict, ValidationError
# TaskDefinitionRecord in a big json file, TaskEmitRecord's tracked in
# json blobs on file system, one per record? Then for completions for that date,
# we can log into a jsonl for point accumulation (and spending). each write can
# overwrite the previous rollup for that day, keeping a total given out and
# dispensed in seperate fields.
#
# track instantiations so we can look at what a TaskEmitRecord's true parent
# was at point-in-time.
@dataclass(frozen=True)
class KeyfunValue:
identifier: str
expires_at: datetime.datetime
@classmethod
def daily(cls, now: datetime.datetime) -> Self:
return cls(
now.strftime("%Y-%m-%d"),
now + datetime.timedelta(days=2),
)
@classmethod
def weekly(cls, now: datetime.datetime) -> Self:
year, week, _ = now.isocalendar()
return cls(
f"{year:04}-W{week:02}",
now + datetime.timedelta(days=14),
)
@classmethod
def monthly(cls, now: datetime.datetime) -> Self:
return cls(
now.strftime("%Y-%m"),
now + datetime.timedelta(days=62),
)
KeyfunValue.BY_NAME = {
"daily": KeyfunValue.daily,
"weekly": KeyfunValue.weekly,
"monthly": KeyfunValue.monthly,
}
class TaskDefinitionRecord(BaseModel):
model_config = ConfigDict(strict=True)
id: str
keyfun: str
tier: int
completion_award: int
charge_count: int
texts: list[str]
def _jcs_hash(self):
jcs_serialization = rfc8785.dumps(self.model_dump())
return hashlib.sha256(jcs_serialization).hexdigest()
def instantiate(self):
hash_id = self._jcs_hash()
return TaskDefinitionInstantiation(
hash_id=hash_id,
record=self,
)
def subtasks(self, now=None):
if now is None:
now = datetime.datetime.now()
for subtask_idx, text in enumerate(self.texts):
keyfun_value = KeyfunValue.BY_NAME[self.keyfun](now)
for charge_idx in range(self.charge_count):
yield TaskEmitRecord(
task_definition=self,
task_definition_hash=self._jcs_hash(),
expires_at=keyfun_value.expires_at,
keyfun_value=keyfun_value.identifier,
subtask_idx=subtask_idx,
charge_idx=charge_idx,
text=text,
)
class TaskDefinitionInstantiation(BaseModel):
model_config = ConfigDict(strict=True)
hash_id: str
record: TaskDefinitionRecord
class TaskEmitRecord(BaseModel):
model_config = ConfigDict(strict=True)
task_definition: TaskDefinitionRecord
task_definition_hash: str
expires_at: datetime.datetime
keyfun_value: str
subtask_idx: int
charge_idx: int
text: str
def _task_id(self):
return (
self.task_definition.id,
self.keyfun_value,
self.subtask_idx,
self.charge_idx,
)
def _jcs_task_id_hash(self):
jcs_serialization = rfc8785.dumps(self._task_id())
# 40 bits should be enough for a single-user quest system where all
# quests expire within a month
return hashlib.sha256(jcs_serialization).hexdigest()[:10]
class TaskEmitRecordVisitor(ABC):
@abstractmethod
def visit_emitted_task(self, emitted_task: TaskEmitRecord):
pass
def _not_used_hmac_payload(secret: str, audience: str, task_id: str) -> str:
encoded = secret.encode('utf8') + b"\0" + rfc8785.dumps({
'audience': audience,
'task_id': task_id
})
# 48 bits signature should be enough...
hashed_hmac = hashlib.sha256(encoded).hexdigest()[:12]
return f"{task_id}-{hashed_hmac}"
## ====== src/quest_engine/persistence.py
import json
import pathlib
from abc import ABC, abstractmethod
import rfc8785
from .model import TaskEmitRecord, TaskEmitRecordVisitor
def iterate_jsonl_with_comments(fh):
for line in fh:
if not line:
break
line = line.strip()
if line.startswith('#'):
continue
if not line:
continue
yield json.loads(line)
def serde_style_pydantic_object_inflater(class_map, /, ignore_unknown=False):
def inflate(doc):
doc_items = doc.items()
if len(doc_items) != 1:
raise ValueError("discriminator dictionary may only have one key")
(key, item) = list(doc_items)[0]
pydantic_cls = class_map.get(key, None)
if pydantic_cls is None and ignore_unknown:
return
elif pydantic_cls is None:
raise ValueError(f"unknown item type {key!r}")
# re-encode because pydantic's parser and validator are heavily intertwined..
return pydantic_cls.model_validate_json(json.dumps(item))
return inflate
class StorageEngine(ABC):
@abstractmethod
def emit_task(self, task: TaskEmitRecord):
pass
@abstractmethod
def visit_emitted_tasks(self, visitor: TaskEmitRecordVisitor):
pass
class JsonStorageEngine(StorageEngine):
def __init__(self, base_path: pathlib.Path):
self._base_path = base_path
def _activation_record_log(self):
return self._base_path / 'activation-records.jsonl'
def emit_task(self, task: TaskEmitRecord):
with self._activation_record_log().open('ab') as fh:
canonical_json = rfc8785.dumps({
'item': {
'task-emit-record': task.model_dump(mode='json'),
},
}) + b"\n"
fh.write(canonical_json)
def visit_emitted_tasks(self, visitor: TaskEmitRecordVisitor):
inflater = serde_style_pydantic_object_inflater({
'task-emit-record': TaskEmitRecord,
}, ignore_unknown=True)
with self._activation_record_log().open('r') as fh:
for doc in iterate_jsonl_with_comments(fh):
print(f"doc={doc!r}")
inflated_or_none = inflater(doc['item'])
if inflated_or_none is None:
continue
visitor.visit_emitted_task(inflated_or_none)
## ====== src/quest_engine/visitors.py
from abc import ABC, abstractmethod
from .model import (
TaskEmitRecord,
TaskEmitRecordVisitor,
)
class TaskIdSetRemoverVisitor(TaskEmitRecordVisitor):
def __init__(self, candidates):
self.candidates = set(candidates)
def visit_emitted_task(self, emitted_task: TaskEmitRecord):
self.candidates.discard(emitted_task._jcs_task_id_hash())
class TaskEmitRecordKeepAllVisitor(TaskEmitRecordVisitor):
def __init__(self):
self.items = []
def visit_emitted_task(self, emitted_task: TaskEmitRecord):
self.items.append(emitted_task)
## ====== src/quest_engine/test_general.py
import unittest
import pathlib
import random
import tempfile
import json
from quest_engine import (
JsonStorageEngine,
TaskDefinitionRecord,
TaskEmitRecordKeepAllVisitor,
)
class TestCalculator(unittest.TestCase):
def test_single_emit_of_simple_task(self):
local_rng = random.Random()
with tempfile.TemporaryDirectory() as td:
tdp = pathlib.Path(td)
doc_json = json.dumps({
"id": "example",
"keyfun": "daily",
"tier": 1,
"completion_award": 1000,
"charge_count": 1,
"texts": ["take 1 breath"]
})
docs = [
TaskDefinitionRecord.model_validate_json(doc_json)
]
se = JsonStorageEngine(tdp)
doc = local_rng.sample(docs, 1)[0]
for subtask in doc.subtasks():
se.emit_task(subtask)
visitor = TaskEmitRecordKeepAllVisitor()
se.visit_emitted_tasks(visitor)
items = list(visitor.items)
self.assertEqual(len(items), 1)
if __name__ == '__main__':
unittest.main()