197 lines
5.3 KiB
Python
197 lines
5.3 KiB
Python
import dataclasses
|
|
import errno
|
|
import re
|
|
import stat
|
|
from collections.abc import Iterator
|
|
from logging import getLogger
|
|
|
|
import fuse
|
|
|
|
from .favro_client import FavroClient
|
|
from .favro_data_model import Card, SeqId
|
|
|
|
logger = getLogger(__name__)
|
|
|
|
fuse.fuse_python_api = (0, 2)
|
|
|
|
class MyStat(fuse.Stat):
|
|
def __init__(self):
|
|
self.st_mode = 0
|
|
self.st_ino = 0
|
|
self.st_dev = 0
|
|
self.st_nlink = 0
|
|
self.st_uid = 0
|
|
self.st_gid = 0
|
|
self.st_size = 0
|
|
self.st_atime = 0
|
|
self.st_mtime = 0
|
|
self.st_ctime = 0
|
|
|
|
|
|
CARD_FILENAME_FORMAT = 'PAR-{seq_id}.md'
|
|
CARD_FILENAME_REGEX = r'^\/PAR\-(\d+)\.md$'
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class Thing:
|
|
@staticmethod
|
|
def from_path(path_str: str) -> 'Thing | None':
|
|
if path_str == '/':
|
|
return RootThing()
|
|
if m := re.match(CARD_FILENAME_REGEX, path_str):
|
|
return CardThing(SeqId(int(m.group(1))))
|
|
return None
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class RootThing(Thing):
|
|
pass
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class CardThing(Thing):
|
|
seq_id: SeqId
|
|
|
|
|
|
def card_to_contents(card: Card) -> str:
|
|
ls = []
|
|
# ls.append('# ')
|
|
# ls.append(card.name)
|
|
# ls.append('\n\n')
|
|
ls.append(card.detailed_description or '')
|
|
return ''.join(ls)
|
|
|
|
|
|
class FavroFuse(fuse.Fuse):
|
|
"""Favro Filesystem in Userspace."""
|
|
|
|
def __init__(self, favro_client: FavroClient, **kwargs):
|
|
self.favro_client = favro_client
|
|
super().__init__(**kwargs)
|
|
|
|
def getattr(self, path: str) -> MyStat | int:
|
|
thing = Thing.from_path(path)
|
|
|
|
st = MyStat()
|
|
if isinstance(thing, RootThing):
|
|
st.st_mode = stat.S_IFDIR | 0o755
|
|
st.st_nlink = 2
|
|
elif isinstance(thing, CardThing):
|
|
card = self.favro_client.get_card(thing.seq_id)
|
|
|
|
st.st_mode = stat.S_IFREG | 0o666
|
|
st.st_nlink = 1
|
|
st.st_size = len(card_to_contents(card))
|
|
st.st_ctime = int(card.creation_date.timestamp())
|
|
st.st_mtime = st.st_ctime # TODO
|
|
else:
|
|
return -errno.ENOENT
|
|
return st
|
|
|
|
def readdir(self, path: str, offset: int) -> Iterator[fuse.Direntry]:
|
|
logger.warning('readdir(path=%s, offset=%s)',path, offset)
|
|
yield fuse.Direntry('.')
|
|
yield fuse.Direntry('..')
|
|
|
|
for card in self.favro_client.get_todo_list_cards():
|
|
yield fuse.Direntry(CARD_FILENAME_FORMAT.format(seq_id=card.seq_id.raw_id))
|
|
|
|
def open(self, path: str, flags) -> int | None:
|
|
thing = Thing.from_path(path)
|
|
if not isinstance(thing, CardThing):
|
|
return -errno.ENOENT
|
|
# accmode = os.O_RDONLY | os.O_WRONLY | os.O_RDWR
|
|
# if (flags & accmode) != os.O_RDONLY: return -errno.EACCES
|
|
return None
|
|
|
|
def read(self, path: str, size: int, offset: int) -> bytes | int:
|
|
# Check that this is a card thing.
|
|
thing = Thing.from_path(path)
|
|
if not isinstance(thing, CardThing):
|
|
return -errno.ENOENT
|
|
|
|
card = self.favro_client.get_card(thing.seq_id)
|
|
|
|
contents_str = card_to_contents(card)
|
|
contents = bytes(contents_str, 'utf8')
|
|
|
|
slen = len(contents)
|
|
if offset < slen:
|
|
if offset + size > slen:
|
|
size = slen - offset
|
|
buf = contents[offset : offset + size]
|
|
else:
|
|
buf = b''
|
|
return buf
|
|
|
|
def write(self, path: str, written_buffer: bytes, offset: int) -> int:
|
|
# Check that this is a card thing.
|
|
thing = Thing.from_path(path)
|
|
if not isinstance(thing, CardThing):
|
|
return -errno.ENOENT
|
|
|
|
card = self.favro_client.get_card(thing.seq_id)
|
|
|
|
# Splice contents
|
|
contents_str = card_to_contents(card)
|
|
contents = bytes(contents_str, 'utf8')
|
|
contents = splice(contents, written_buffer, offset)
|
|
contents_str = contents.decode('utf8')
|
|
|
|
# Write to favro
|
|
self.favro_client.update_card_description(card.card_id, contents_str)
|
|
|
|
# Return amount written
|
|
return len(written_buffer)
|
|
|
|
def truncate(self, path: str, new_size: int):
|
|
# Check that this is a card thing.
|
|
thing = Thing.from_path(path)
|
|
if not isinstance(thing, CardThing):
|
|
return -errno.ENOENT
|
|
|
|
card = self.favro_client.get_card(thing.seq_id)
|
|
|
|
# Splice contents
|
|
contents_str = card_to_contents(card)
|
|
contents = bytes(contents_str, 'utf8')
|
|
old_size = len(contents)
|
|
contents = contents[0:new_size] + b' ' * (old_size - new_size)
|
|
assert len(contents) == old_size
|
|
contents_str = contents.decode('utf8')
|
|
|
|
# Write to favro
|
|
self.favro_client.update_card_description(card.card_id, contents_str)
|
|
|
|
# Return amount written
|
|
return 0
|
|
|
|
|
|
def splice(original_buffer: bytes, input_buffer: bytes, offset: int) -> bytes:
|
|
return (
|
|
original_buffer[0 : offset - 1]
|
|
+ input_buffer
|
|
+ original_buffer[offset + len(input_buffer) + 1 : len(original_buffer)]
|
|
)
|
|
|
|
|
|
HELP = (
|
|
"""
|
|
Userspace hello example
|
|
|
|
"""
|
|
+ fuse.Fuse.fusage
|
|
)
|
|
|
|
|
|
def start_favro_fuse(favro_client: FavroClient):
|
|
# TODO:
|
|
server = FavroFuse(
|
|
favro_client=favro_client,
|
|
version='%prog ' + fuse.__version__,
|
|
usage=HELP,
|
|
dash_s_do='setsingle',
|
|
)
|
|
server.parse(errex=1)
|
|
server.main()
|