import os, stat, errno, fuse from pathlib import Path import re import dataclasses from collections.abc import Iterator from .favro_client import FavroClient, CardId, SeqId, Card fuse.fuse_python_api = (0, 2) hello_path = '/hello' hello_str = b'Hello World!\n' 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]: 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)) return # TODO 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): print('Trunca', path, new_size) # 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 = contents[0:new_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()