Last modified 2026-07-05 06:57:42 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
import subprocess
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,
TaskDefinitionRecordCollector,
TaskEmitRecordCompletedPreparerVisitor,
ScoreAccumulatorVisitor,
)
from .model import (
TaskDefinitionRecord,
TaskEmitRecord,
LogVisitor,
)
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())
remaining_candidates = TaskIdSetRemoverVisitor.apply(storage, task_id_candidates)
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 & remaining_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):
pass
# 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_logs(visitor)
subprocess.run(['cat', se._activation_record_log()], check=True)
# 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, ClassVar
from abc import ABC, abstractmethod
import rfc8785
from pydantic import BaseModel, ConfigDict, ValidationError
class MessageMetadata(BaseModel):
model_config = ConfigDict(strict=True)
record_id: str
supersedes: list[str] # `record_id` that this record replaces.
@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):
"""
this is loaded from user config, and within TaskDefinitionFirstUseRecord
"""
WIRE_NAME: ClassVar[str] = "task-definition-record"
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([
'TaskDefinitionRecord',
self.model_dump(),
])
return hashlib.sha256(jcs_serialization).hexdigest()
def metadata(self) -> MessageMetadata:
return MessageMetadata(record_id=self._jcs_hash(), supersedes=[])
# 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 TaskEmitRecordReturnToken(BaseModel):
"""
persists a "return token" value that is used to respond to an issued task,
keeps the string open for future consumption.
"""
WIRE_NAME: ClassVar[str] = "task-emit-record-return-token"
model_config = ConfigDict(strict=True)
hash_id_short: str
task_definition_hash: str
task_emit_record_hash: str
expires_at: datetime.datetime
def _jcs_hash(self):
jcs_serialization = rfc8785.dumps([
type(self).WIRE_NAME,
self.task_definition_hash,
self.task_emit_record_hash,
])
return hashlib.sha256(jcs_serialization).hexdigest()
def metadata(self) -> MessageMetadata:
return MessageMetadata(
record_id=self._jcs_hash(),
supersedes=[],
)
def mark_completed(self, task: TaskDefinitionRecord) -> 'TaskEmitRecordCompleted':
return TaskEmitRecordCompleted(
parent=self.metadata().record_id,
completion_award=task.completion_award,
)
class TaskEmitRecordCompleted(BaseModel):
"""
persists a "return token" value that is used to respond to an issued task,
keeps the string open for future consumption.
"""
WIRE_NAME: ClassVar[str] = "task-emit-record-completed"
model_config = ConfigDict(strict=True)
parent: str
completion_award: int
def _jcs_hash(self):
jcs_serialization = rfc8785.dumps([
type(self).WIRE_NAME,
self.parent,
self.completion_award,
])
return hashlib.sha256(jcs_serialization).hexdigest()
def metadata(self) -> MessageMetadata:
return MessageMetadata(
record_id=self._jcs_hash(),
supersedes=[self.parent],
)
class TaskEmitRecord(BaseModel):
WIRE_NAME: ClassVar[str] = "task-emit-record"
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_hash(self):
jcs_serialization = rfc8785.dumps([
type(self).WIRE_NAME,
self._task_id(),
])
return hashlib.sha256(jcs_serialization).hexdigest()
def metadata(self) -> MessageMetadata:
return MessageMetadata(record_id=self._jcs_hash(), supersedes=[])
def generate_short(self) -> TaskEmitRecordReturnToken:
hash_id = self._jcs_hash()
return TaskEmitRecordReturnToken(
hash_id_short=hash_id[:10],
task_definition_hash=self.task_definition_hash,
task_emit_record_hash=hash_id,
expires_at=self.expires_at,
)
class LogVisitor(ABC):
def visit_task_definition_record(self, meta: MessageMetadata, v: TaskDefinitionRecord):
pass
def visit_emitted_task(self, meta: MessageMetadata, v: TaskEmitRecord):
pass
def visit_task_emit_record_return_token(self, meta: MessageMetadata, v: TaskEmitRecordReturnToken):
pass
def visit_task_emit_record_completed(self, meta: MessageMetadata, v: TaskEmitRecordCompleted):
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,
LogVisitor,
TaskDefinitionRecord,
LogVisitor,
TaskEmitRecordReturnToken,
MessageMetadata,
TaskEmitRecordCompleted,
)
from .visitors import (
TaskEmitRecordCompletedPreparerVisitor,
)
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_logs(self, visitor: LogVisitor):
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 _write_record(self, fh, record):
item_key = type(record).WIRE_NAME
fh.write(rfc8785.dumps({
'metadata': record.metadata().model_dump(mode='json'),
'item': {
item_key: record.model_dump(mode='json'),
},
}) + b"\n")
def emit_task(self, task: TaskEmitRecord):
task_metadata = task.metadata()
first_use_searching_for = task.task_definition.metadata()
class Visitor(LogVisitor):
def __init__(self):
self.items = []
self.already_emitted = False
def visit_emitted_task(self, meta: MessageMetadata, v: TaskEmitRecord):
if task_metadata.record_id == meta.record_id:
self.already_emitted = True
def visit_task_definition_record(self, meta: MessageMetadata, v: TaskDefinitionRecord):
if first_use_searching_for.record_id == meta.record_id:
self.items.append(v)
visitor = Visitor()
self.visit_logs(visitor)
if visitor.already_emitted:
raise ValueError("already emitted")
with self._activation_record_log().open('ab') as fh:
if not visitor.items:
self._write_record(fh, task.task_definition)
self._write_record(fh, task)
self._write_record(fh, task.generate_short())
def complete_task(self, short_code: str):
return_visitor = TaskEmitRecordCompletedPreparerVisitor(short_code)
self.visit_logs(return_visitor)
completion_record = return_visitor.complete()
if completion_record:
with self._activation_record_log().open('ab') as fh:
self._write_record(fh, completion_record)
def visit_logs(self, visitor: LogVisitor):
if not self._activation_record_log().exists():
return
inflater = serde_style_pydantic_object_inflater({
'task-emit-record': TaskEmitRecord,
'task-definition-record': TaskDefinitionRecord,
'task-emit-record-return-token': TaskEmitRecordReturnToken,
'task-emit-record-completed': TaskEmitRecordCompleted
}, ignore_unknown=True)
with self._activation_record_log().open('r') as fh:
for doc in iterate_jsonl_with_comments(fh):
metadata = MessageMetadata.model_validate_json(json.dumps(doc['metadata']))
inflated_or_none = inflater(doc['item'])
if inflated_or_none is None:
continue
if isinstance(inflated_or_none, TaskEmitRecord):
visitor.visit_emitted_task(metadata, inflated_or_none)
elif isinstance(inflated_or_none, TaskDefinitionRecord):
visitor.visit_task_definition_record(metadata, inflated_or_none)
elif isinstance(inflated_or_none, TaskEmitRecordReturnToken):
visitor.visit_task_emit_record_return_token(metadata, inflated_or_none)
elif isinstance(inflated_or_none, TaskEmitRecordCompleted):
visitor.visit_task_emit_record_completed(metadata, inflated_or_none)
## ====== src/quest_engine/visitors.py
import datetime
from abc import ABC, abstractmethod
from .model import (
MessageMetadata,
TaskEmitRecord,
LogVisitor,
TaskDefinitionRecord,
TaskEmitRecordReturnToken,
TaskEmitRecordCompleted,
)
class TaskDefinitionRecordCollector(LogVisitor):
@classmethod
def apply(cls, storage):
v = cls()
storage.visit_logs(v)
return v.score
def __init__(self):
self.items = []
def visit_task_definition_record(self, meta: MessageMetadata, v: TaskDefinitionRecord):
self.items.append(meta.record_id)
class TaskIdSetRemoverVisitor(LogVisitor):
@classmethod
def apply(cls, storage, candidates):
v = cls(candidates)
storage.visit_logs(v)
return v.candidates
def __init__(self, candidates):
self.candidates = set(candidates)
def visit_emitted_task(self, meta: MessageMetadata, v: TaskEmitRecord):
self.candidates.discard(meta.record_id)
class TaskEmitRecordKeepAllVisitor(LogVisitor):
@classmethod
def apply(cls, storage):
v = cls()
storage.visit_logs(v)
return v.items
def __init__(self):
self.items = []
def visit_emitted_task(self, meta: MessageMetadata, v: TaskEmitRecord):
self.items.append(v)
class TaskEmitRecordCompletedPreparerVisitor(LogVisitor):
def __init__(
self,
short_code: str,
/,
now: datetime.datetime | None = None
):
self._return_token = None
self._task_definitions = dict()
self._is_completed = False
self._short_code = short_code
if now is None:
now = datetime.datetime.now()
self._now = now
def visit_task_definition_record(self, meta: MessageMetadata, v: TaskDefinitionRecord):
self._task_definitions[meta.record_id] = v
def visit_task_emit_record_return_token(self, meta: MessageMetadata, v: TaskEmitRecordReturnToken):
if v.expires_at < self._now:
return
if v.hash_id_short == self._short_code:
self._return_token = v
def visit_task_emit_record_completed(self, meta: MessageMetadata, v: TaskEmitRecordCompleted):
if self._return_token is None:
return
if self._return_token.metadata().record_id == v.parent:
self._is_completed = True
def complete(self):
if self._return_token is None:
raise ValueError("TaskEmitRecordReturnToken not found")
task_definition = self._task_definitions.get(self._return_token.task_definition_hash, None)
if task_definition is None:
raise ValueError("TaskEmitRecordReturnToken orphaned: no associated TaskDefinitionRecord")
if self._is_completed:
return
return self._return_token.mark_completed(task_definition)
class TaskEmitRecordFinder(LogVisitor):
@classmethod
def apply(cls, storage, short_code: str):
v = cls(short_code)
storage.visit_logs(v)
return v.items
def __init__(self, short_code: str):
self.items = []
self._short_code = short_code
def visit_task_emit_record_return_token(self, meta: MessageMetadata, v: TaskEmitRecordReturnToken):
if v.hash_id_short == self._short_code:
self.items.append(v)
class ScoreAccumulatorVisitor(LogVisitor):
@classmethod
def apply(cls, storage):
v = cls()
storage.visit_logs(v)
return v.score
def __init__(self):
self.score = 0
def visit_task_emit_record_completed(self, meta: MessageMetadata, v: TaskEmitRecordCompleted):
self.score += v.completion_award
## ====== src/quest_engine/test_general.py
import unittest
import pathlib
import random
import tempfile
import json
from quest_engine import (
JsonStorageEngine,
TaskDefinitionRecordCollector,
TaskDefinitionRecord,
TaskEmitRecordKeepAllVisitor,
ScoreAccumulatorVisitor,
TaskIdSetRemoverVisitor,
)
class TestSingleTaskEmit(unittest.TestCase):
def _emit_tasks(self, storage):
local_rng = random.Random()
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)
]
doc = local_rng.sample(docs, 1)[0]
subtasks = list(doc.subtasks())
remaining_candidates = TaskIdSetRemoverVisitor.apply(storage, [
s.metadata().record_id for s in subtasks
])
out = []
for subtask in subtasks:
if subtask.metadata().record_id in remaining_candidates:
storage.emit_task(subtask)
out.append(subtask)
return out
def test_find_emit_record(self):
with tempfile.TemporaryDirectory() as td:
tdp = pathlib.Path(td)
se = JsonStorageEngine(tdp)
self._emit_tasks(se)
visitor = TaskEmitRecordKeepAllVisitor()
se.visit_logs(visitor)
items = list(visitor.items)
self.assertEqual(len(items), 1)
def test_find_task_definition_first_use(self):
local_rng = random.Random()
with tempfile.TemporaryDirectory() as td:
tdp = pathlib.Path(td)
se = JsonStorageEngine(tdp)
self._emit_tasks(se)
visitor = TaskDefinitionRecordCollector()
se.visit_logs(visitor)
items = list(visitor.items)
self.assertEqual(len(items), 1)
def test_emit_and_complete(self):
local_rng = random.Random()
with tempfile.TemporaryDirectory() as td:
tdp = pathlib.Path(td)
se = JsonStorageEngine(tdp)
# get some tasks to do.
subtasks = self._emit_tasks(se)
first_subtask = subtasks[0]
# this represents the barcode that the user holds while doing the
# task
short_hash = first_subtask.generate_short().hash_id_short
# we scan it
se.complete_task(short_hash)
# and then we validate our score is accounted for
self.assertEqual(ScoreAccumulatorVisitor.apply(se), 1000)
# ensure double spend does not increment score, as if the user
# resubmitted a task completion code
se.complete_task(short_hash)
# it should remain the same value
self.assertEqual(ScoreAccumulatorVisitor.apply(se), 1000)
def test_task_emit_only_once(self):
with tempfile.TemporaryDirectory() as td:
tdp = pathlib.Path(td)
se = JsonStorageEngine(tdp)
# get some tasks to do, for this test-set, there is only one
subtasks = self._emit_tasks(se)
self.assertEqual(len(subtasks), 1)
# so we should run out if we try again.
subtasks = self._emit_tasks(se)
self.assertEqual(len(subtasks), 0)
if __name__ == '__main__':
unittest.main()