Last active
December 21, 2023 22:53
-
-
Save MichaelTaylor3D/765a3069729851705bd6a1cfe502f8dc to your computer and use it in GitHub Desktop.
DataLayer Virtual FileSystem
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
import requests | |
import fuse | |
from fuse import FUSE, FuseOSError, Operations | |
class ChiaDataLayerFileSystem(Operations): | |
def __init__(self, rpc_base_url): | |
self.rpc_base_url = rpc_base_url | |
def get_subscriptions(self): | |
url = f"{self.rpc_base_url}/subscriptions" | |
headers = {"Content-Type": "application/json"} | |
response = requests.post(url, headers=headers, verify=False) | |
if response.status_code == 200 and response.json().get("success"): | |
return response.json()["store_ids"] | |
else: | |
raise FuseOSError(fuse.errno.ECOMM) | |
def get_keys(self, store_id): | |
url = f"{self.rpc_base_url}/get_keys" | |
headers = {"Content-Type": "application/json"} | |
body = {"id": store_id} | |
response = requests.post(url, json=body, headers=headers, verify=False) | |
if response.status_code == 200 and response.json().get("success"): | |
return [bytes.fromhex(key[2:]).decode('utf-8') for key in response.json()["keys"]] | |
else: | |
raise FuseOSError(fuse.errno.ECOMM) | |
def readdir(self, path, fh): | |
if path == "/": | |
try: | |
store_ids = self.get_subscriptions() | |
yield from ['.', '..'] + store_ids | |
except FuseOSError: | |
pass | |
else: | |
store_id = path.strip("/") | |
try: | |
files = self.get_keys(store_id) | |
yield from ['.', '..'] + files | |
except FuseOSError: | |
pass | |
def open(self, path, flags): | |
store_id, file_path = path.strip("/").split('/', 1) | |
files = self.get_keys(store_id) | |
if file_path in files: | |
return 0 # Dummy file handle | |
else: | |
raise FuseOSError(fuse.errno.ENOENT) # No such file | |
def read(self, path, size, offset, fh): | |
store_id, file_path = path.strip("/").split('/', 1) | |
url = f"{self.rpc_base_url}/get_value" | |
headers = {"Content-Type": "application/json"} | |
body = { | |
"id": store_id, | |
"key": file_path.encode().hex() | |
} | |
response = requests.post(url, json=body, headers=headers, verify=False) | |
if response.status_code == 200 and response.json().get("success"): | |
file_content = bytes.fromhex(response.json()["value"]).decode('utf-8') | |
return file_content[offset:offset + size] | |
else: | |
raise FuseOSError(fuse.errno.EIO) | |
def getattr(self, path, fh=None): | |
if path == '/': | |
st_mode = (stat.S_IFDIR | 0o755) | |
st_nlink = 2 | |
else: | |
st_mode = (stat.S_IFREG | 0o444) | |
st_nlink = 1 | |
st_size = 1024 # Arbitrary size; modify as needed | |
return { | |
'st_atime': 0, | |
'st_ctime': 0, | |
'st_gid': os.getgid(), | |
'st_mode': st_mode, | |
'st_mtime': 0, | |
'st_nlink': st_nlink, | |
'st_size': st_size, | |
'st_uid': os.getuid() | |
} | |
# Add other necessary methods like getattr, etc... | |
if __name__ == '__main__': | |
rpc_base_url = "https://localhost:8562" | |
fuse = FUSE(ChiaDataLayerFileSystem(rpc_base_url), '/mnt/chia', foreground=True, nothreads=True) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment