Skip to content

Instantly share code, notes, and snippets.

@MichaelTaylor3D
Last active December 21, 2023 22:53
Show Gist options
  • Save MichaelTaylor3D/765a3069729851705bd6a1cfe502f8dc to your computer and use it in GitHub Desktop.
Save MichaelTaylor3D/765a3069729851705bd6a1cfe502f8dc to your computer and use it in GitHub Desktop.
DataLayer Virtual FileSystem
#!/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