440 lines
15 KiB
Python
440 lines
15 KiB
Python
"""Python binding for the lls C API (ctypes, no compilation needed).
|
|
|
|
import lls
|
|
|
|
with lls.Database("tasks.db") as db:
|
|
db.add_account("Acrux")
|
|
db.add_task("Breeze A", account="Acrux", status="testing")
|
|
db.set_account_wait("Acrux", "4:30pm") # the account hit its limit
|
|
for task in db.tasks():
|
|
print(task.name, task.account, task.status_label)
|
|
db.save() # nothing is saved automatically
|
|
|
|
The shared library (liblls_api.so / liblls_api.dylib / lls_api.dll) is looked up in
|
|
the LLS_LIBRARY environment variable, then in ../build next to this file, then in
|
|
the current directory's build folder. Pass `library=` to Database to override.
|
|
"""
|
|
|
|
import ctypes
|
|
import os
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from enum import IntEnum
|
|
from pathlib import Path
|
|
from typing import List, Optional, Sequence, Tuple, Union
|
|
|
|
__all__ = [
|
|
"Database", "Task", "Account", "Status", "Level", "WaitState",
|
|
"LLSError", "InvalidArgument", "NotFound", "AlreadyExists", "DatabaseError",
|
|
"parse_time", "find_library",
|
|
]
|
|
|
|
# ---- Constants (mirror lls_api.h) --------------------------------------------------
|
|
|
|
OK, E_INVALID, E_NOT_FOUND, E_EXISTS, E_DATABASE, E_INTERNAL = range(6)
|
|
|
|
|
|
class Status(IntEnum):
|
|
PROMPT = 0
|
|
LIMIT_RESET = 1
|
|
TESTING = 2
|
|
WAIT = 3
|
|
UPLOADING = 4
|
|
FROZEN = 5
|
|
DONE = 6
|
|
|
|
|
|
class Level(IntEnum):
|
|
BUSY = 0
|
|
TAKEN = 1
|
|
FREE = 2
|
|
|
|
|
|
class WaitState(IntEnum):
|
|
NONE = 0
|
|
WAITING = 1
|
|
LIMIT_RESET = 2
|
|
|
|
|
|
# ---- Errors ------------------------------------------------------------------------
|
|
|
|
class LLSError(Exception):
|
|
"""Something the library refused to do; `code` is the LLS_E_* value."""
|
|
|
|
def __init__(self, code: int, message: str):
|
|
super().__init__(message)
|
|
self.code = code
|
|
|
|
|
|
class InvalidArgument(LLSError):
|
|
"""Bad name, unknown status, unreadable time..."""
|
|
|
|
|
|
class NotFound(LLSError):
|
|
"""No such task or account."""
|
|
|
|
|
|
class AlreadyExists(LLSError):
|
|
"""A task or account with that name exists."""
|
|
|
|
|
|
class DatabaseError(LLSError):
|
|
"""SQLite or file problem."""
|
|
|
|
|
|
_ERRORS = {E_INVALID: InvalidArgument, E_NOT_FOUND: NotFound, E_EXISTS: AlreadyExists, E_DATABASE: DatabaseError}
|
|
|
|
|
|
# ---- Results -----------------------------------------------------------------------
|
|
|
|
@dataclass(frozen=True)
|
|
class Task:
|
|
name: str
|
|
account: Optional[str]
|
|
status: Status # as shown: Wait / Limit reset come from the account
|
|
status_label: str # "Prompt", "Limit reset", "Wait 16:30", ...
|
|
stored_status: Status # what is saved underneath
|
|
until: Optional[datetime] # for Wait / Limit reset: when the wait ends (ended)
|
|
notes: str
|
|
notes_lines: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Account:
|
|
name: str
|
|
level: Level
|
|
level_label: str # "Busy", "Taken", "Free"
|
|
wait_state: WaitState
|
|
wait_until: Optional[datetime]
|
|
wait_listed: bool # the wait is not shown by any task (account is still Busy)
|
|
tasks: Tuple[str, ...] # tasks at the account's level
|
|
|
|
|
|
# ---- ctypes plumbing -----------------------------------------------------------------
|
|
|
|
class _TaskInfo(ctypes.Structure):
|
|
_fields_ = [
|
|
("name", ctypes.c_char_p),
|
|
("account", ctypes.c_char_p),
|
|
("status", ctypes.c_int),
|
|
("status_label", ctypes.c_char_p),
|
|
("stored_status", ctypes.c_int),
|
|
("until", ctypes.c_int64),
|
|
("notes", ctypes.c_char_p),
|
|
("notes_lines", ctypes.c_int),
|
|
]
|
|
|
|
|
|
class _AccountInfo(ctypes.Structure):
|
|
_fields_ = [
|
|
("name", ctypes.c_char_p),
|
|
("level", ctypes.c_int),
|
|
("level_label", ctypes.c_char_p),
|
|
("wait_state", ctypes.c_int),
|
|
("wait_until", ctypes.c_int64),
|
|
("wait_listed", ctypes.c_int),
|
|
("task_count", ctypes.c_int),
|
|
("tasks", ctypes.POINTER(ctypes.c_char_p)),
|
|
]
|
|
|
|
|
|
def _library_names() -> List[str]:
|
|
if sys.platform == "win32":
|
|
return ["lls_api.dll"]
|
|
if sys.platform == "darwin":
|
|
return ["liblls_api.dylib"]
|
|
return ["liblls_api.so"]
|
|
|
|
|
|
def find_library() -> str:
|
|
"""Path of the shared library, or raises FileNotFoundError."""
|
|
candidates: List[Path] = []
|
|
env = os.environ.get("LLS_LIBRARY")
|
|
if env:
|
|
candidates.append(Path(env))
|
|
here = Path(__file__).resolve().parent
|
|
for folder in (here.parent / "build", here.parent / "build" / "Release", here.parent / "build" / "Debug",
|
|
Path.cwd() / "build", Path.cwd(), here):
|
|
for name in _library_names():
|
|
candidates.append(folder / name)
|
|
for candidate in candidates:
|
|
if candidate.is_file():
|
|
return str(candidate)
|
|
raise FileNotFoundError("lls shared library not found; build the project or set LLS_LIBRARY. Tried: " +
|
|
", ".join(str(c) for c in candidates))
|
|
|
|
|
|
_loaded = {}
|
|
|
|
|
|
def _load(path: Optional[str]) -> ctypes.CDLL:
|
|
path = path or find_library()
|
|
if path in _loaded:
|
|
return _loaded[path]
|
|
|
|
lib = ctypes.CDLL(path)
|
|
c_int, c_str, c_p, c_i64 = ctypes.c_int, ctypes.c_char_p, ctypes.c_void_p, ctypes.c_int64
|
|
|
|
def declare(name, restype, *argtypes):
|
|
function = getattr(lib, name)
|
|
function.restype = restype
|
|
function.argtypes = list(argtypes)
|
|
|
|
declare("lls_api_version", c_int)
|
|
declare("lls_status_name", c_str, c_int)
|
|
declare("lls_parse_time", c_int, c_str, c_i64, ctypes.POINTER(c_i64))
|
|
declare("lls_open", c_int, c_str, ctypes.POINTER(c_p))
|
|
declare("lls_close", None, c_p)
|
|
declare("lls_last_error", c_str, c_p)
|
|
declare("lls_last_messages", c_str, c_p)
|
|
declare("lls_path", c_str, c_p)
|
|
declare("lls_is_dirty", c_int, c_p)
|
|
declare("lls_save", c_int, c_p)
|
|
declare("lls_backup", c_int, c_p, c_str, ctypes.c_char_p, ctypes.c_size_t)
|
|
declare("lls_restore", c_int, c_p, c_str)
|
|
declare("lls_set_clock", None, c_p, c_i64)
|
|
declare("lls_now", c_i64, c_p)
|
|
declare("lls_account_add", c_int, c_p, c_str)
|
|
declare("lls_account_delete", c_int, c_p, c_str)
|
|
declare("lls_account_rename", c_int, c_p, c_str, c_str)
|
|
declare("lls_account_set_wait", c_int, c_p, c_str, c_str)
|
|
declare("lls_account_set_wait_at", c_int, c_p, c_str, c_i64)
|
|
declare("lls_account_clear_wait", c_int, c_p, c_str, ctypes.POINTER(c_int))
|
|
declare("lls_task_add", c_int, c_p, c_str, c_str, c_str)
|
|
declare("lls_task_delete", c_int, c_p, c_str)
|
|
declare("lls_task_rename", c_int, c_p, c_str, c_str)
|
|
declare("lls_task_set_account", c_int, c_p, c_str, c_str)
|
|
declare("lls_task_set_status", c_int, c_p, c_str, c_str)
|
|
declare("lls_task_set_notes", c_int, c_p, c_str, c_str)
|
|
declare("lls_tasks", c_int, c_p, ctypes.POINTER(c_p))
|
|
declare("lls_task_list_count", c_int, c_p)
|
|
declare("lls_task_list_get", ctypes.POINTER(_TaskInfo), c_p, c_int)
|
|
declare("lls_task_list_free", None, c_p)
|
|
declare("lls_accounts", c_int, c_p, ctypes.POINTER(c_p))
|
|
declare("lls_account_list_count", c_int, c_p)
|
|
declare("lls_account_list_get", ctypes.POINTER(_AccountInfo), c_p, c_int)
|
|
declare("lls_account_list_free", None, c_p)
|
|
|
|
if lib.lls_api_version() != 1:
|
|
raise RuntimeError("lls library has API version %d, this module expects 1" % lib.lls_api_version())
|
|
_loaded[path] = lib
|
|
return lib
|
|
|
|
|
|
def _enc(text: Optional[str]) -> Optional[bytes]:
|
|
return None if text is None else text.encode("utf-8")
|
|
|
|
|
|
def _dec(raw: Optional[bytes]) -> Optional[str]:
|
|
return None if raw is None else raw.decode("utf-8", errors="replace")
|
|
|
|
|
|
def _when(seconds: int) -> Optional[datetime]:
|
|
return datetime.fromtimestamp(seconds).astimezone() if seconds else None
|
|
|
|
|
|
def _seconds(moment: Union[datetime, int, float]) -> int:
|
|
return int(moment.timestamp() if isinstance(moment, datetime) else moment)
|
|
|
|
|
|
def parse_time(text: str, now: Union[datetime, int, float, None] = None, library: Optional[str] = None) -> datetime:
|
|
"""Reads a clock time like "16:30" or "4:30pm" the way `acc wait` does (a time that has
|
|
passed today means tomorrow). Raises InvalidArgument."""
|
|
lib = _load(library)
|
|
out = ctypes.c_int64()
|
|
code = lib.lls_parse_time(_enc(text), 0 if now is None else _seconds(now), ctypes.byref(out))
|
|
if code != OK:
|
|
raise InvalidArgument(code, "Invalid time '%s'. Use e.g. 16:30, 4:30pm or 4pm." % text)
|
|
return _when(out.value)
|
|
|
|
|
|
# ---- The database --------------------------------------------------------------------
|
|
|
|
class Database:
|
|
"""A task / account database. Changes stay in memory until save()."""
|
|
|
|
def __init__(self, path: Union[str, os.PathLike], library: Optional[str] = None):
|
|
self._lib = _load(library)
|
|
self._handle = ctypes.c_void_p()
|
|
code = self._lib.lls_open(_enc(os.fspath(path)), ctypes.byref(self._handle))
|
|
if code != OK:
|
|
self._handle = None
|
|
self._raise(code, _dec(self._lib.lls_last_error(None)))
|
|
|
|
# -- lifecycle --
|
|
|
|
def close(self) -> None:
|
|
"""Closes the database WITHOUT saving."""
|
|
if self._handle:
|
|
self._lib.lls_close(self._handle)
|
|
self._handle = None
|
|
|
|
def __enter__(self) -> "Database":
|
|
return self
|
|
|
|
def __exit__(self, *exc) -> None:
|
|
self.close()
|
|
|
|
def __del__(self):
|
|
try:
|
|
self.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# -- helpers --
|
|
|
|
def _raise(self, code: int, message: str) -> None:
|
|
raise _ERRORS.get(code, LLSError)(code, message)
|
|
|
|
def _call(self, name: str, *args) -> None:
|
|
if not self._handle:
|
|
raise LLSError(E_INVALID, "The database is closed.")
|
|
code = getattr(self._lib, name)(self._handle, *args)
|
|
if code != OK:
|
|
self._raise(code, _dec(self._lib.lls_last_error(self._handle)))
|
|
|
|
# -- state --
|
|
|
|
@property
|
|
def path(self) -> str:
|
|
return _dec(self._lib.lls_path(self._handle))
|
|
|
|
@property
|
|
def dirty(self) -> bool:
|
|
"""True if there are unsaved changes."""
|
|
return bool(self._lib.lls_is_dirty(self._handle))
|
|
|
|
@property
|
|
def messages(self) -> List[str]:
|
|
"""What the last change did, as the interactive prompt would print it."""
|
|
text = _dec(self._lib.lls_last_messages(self._handle))
|
|
return text.split("\n") if text else []
|
|
|
|
def save(self) -> None:
|
|
self._call("lls_save")
|
|
|
|
def backup(self, file: Optional[str] = None) -> str:
|
|
"""Writes a copy and returns its path (a timestamped name if `file` is None)."""
|
|
buffer = ctypes.create_string_buffer(4096)
|
|
self._call("lls_backup", _enc(file), buffer, len(buffer))
|
|
return buffer.value.decode("utf-8")
|
|
|
|
def restore(self, file: str) -> None:
|
|
"""Replaces the data in memory with a file's contents (call save() to keep it)."""
|
|
self._call("lls_restore", _enc(file))
|
|
|
|
def set_clock(self, now: Union[datetime, int, float, None]) -> None:
|
|
"""Pretend it is `now` (for tests / simulations); None goes back to the real clock."""
|
|
self._lib.lls_set_clock(self._handle, 0 if now is None else _seconds(now))
|
|
|
|
@property
|
|
def now(self) -> datetime:
|
|
return _when(self._lib.lls_now(self._handle))
|
|
|
|
# -- accounts --
|
|
|
|
def add_account(self, name: str) -> None:
|
|
self._call("lls_account_add", _enc(name))
|
|
|
|
def delete_account(self, name: str) -> None:
|
|
"""Its tasks become unassigned."""
|
|
self._call("lls_account_delete", _enc(name))
|
|
|
|
def rename_account(self, name: str, new_name: str) -> None:
|
|
self._call("lls_account_rename", _enc(name), _enc(new_name))
|
|
|
|
def set_account_wait(self, name: str, until: Union[str, datetime, int, float]) -> None:
|
|
"""The account hit its limit until `until`: a clock time such as "16:30" / "4:30pm"
|
|
(a time that has passed today means tomorrow), a datetime, or Unix seconds."""
|
|
if isinstance(until, str):
|
|
self._call("lls_account_set_wait", _enc(name), _enc(until))
|
|
else:
|
|
self._call("lls_account_set_wait_at", _enc(name), _seconds(until))
|
|
|
|
def clear_account_wait(self, name: str) -> bool:
|
|
"""Clears its wait / limit reset. Returns False if it was not waiting."""
|
|
was_waiting = ctypes.c_int(0)
|
|
self._call("lls_account_clear_wait", _enc(name), ctypes.byref(was_waiting))
|
|
return bool(was_waiting.value)
|
|
|
|
# -- tasks --
|
|
|
|
def add_task(self, name: str, account: Optional[str] = None, status: Optional[str] = None) -> None:
|
|
"""`status` is what you would type at the prompt ("testing", "wait 4pm", "wait"...);
|
|
None means prompt."""
|
|
self._call("lls_task_add", _enc(name), _enc(account), _enc(status))
|
|
|
|
def delete_task(self, name: str) -> None:
|
|
self._call("lls_task_delete", _enc(name))
|
|
|
|
def rename_task(self, name: str, new_name: str) -> None:
|
|
self._call("lls_task_rename", _enc(name), _enc(new_name))
|
|
|
|
def set_task_account(self, name: str, account: Optional[str]) -> None:
|
|
self._call("lls_task_set_account", _enc(name), _enc(account))
|
|
|
|
def set_task_status(self, name: str, status: str) -> None:
|
|
self._call("lls_task_set_status", _enc(name), _enc(status))
|
|
|
|
def set_task_notes(self, name: str, notes: str) -> None:
|
|
"""Any length, any number of lines; "" clears them."""
|
|
self._call("lls_task_set_notes", _enc(name), _enc(notes))
|
|
|
|
# -- queries --
|
|
|
|
def tasks(self) -> List[Task]:
|
|
"""All tasks in display order."""
|
|
handle = ctypes.c_void_p()
|
|
self._call("lls_tasks", ctypes.byref(handle))
|
|
try:
|
|
result = []
|
|
for i in range(self._lib.lls_task_list_count(handle)):
|
|
info = self._lib.lls_task_list_get(handle, i).contents
|
|
result.append(Task(
|
|
name=_dec(info.name),
|
|
account=_dec(info.account),
|
|
status=Status(info.status),
|
|
status_label=_dec(info.status_label),
|
|
stored_status=Status(info.stored_status),
|
|
until=_when(info.until),
|
|
notes=_dec(info.notes),
|
|
notes_lines=info.notes_lines,
|
|
))
|
|
return result
|
|
finally:
|
|
self._lib.lls_task_list_free(handle)
|
|
|
|
def accounts(self) -> List[Account]:
|
|
"""All accounts, most free first."""
|
|
handle = ctypes.c_void_p()
|
|
self._call("lls_accounts", ctypes.byref(handle))
|
|
try:
|
|
result = []
|
|
for i in range(self._lib.lls_account_list_count(handle)):
|
|
info = self._lib.lls_account_list_get(handle, i).contents
|
|
result.append(Account(
|
|
name=_dec(info.name),
|
|
level=Level(info.level),
|
|
level_label=_dec(info.level_label),
|
|
wait_state=WaitState(info.wait_state),
|
|
wait_until=_when(info.wait_until),
|
|
wait_listed=bool(info.wait_listed),
|
|
tasks=tuple(_dec(info.tasks[k]) for k in range(info.task_count)),
|
|
))
|
|
return result
|
|
finally:
|
|
self._lib.lls_account_list_free(handle)
|
|
|
|
def task(self, name: str) -> Task:
|
|
for task in self.tasks():
|
|
if task.name.lower() == name.lower():
|
|
return task
|
|
raise NotFound(E_NOT_FOUND, "No task named '%s'." % name)
|
|
|
|
def account(self, name: str) -> Account:
|
|
for account in self.accounts():
|
|
if account.name.lower() == name.lower():
|
|
return account
|
|
raise NotFound(E_NOT_FOUND, "No account named '%s'." % name)
|