v/fzfs
1
0
Fork 0
mirror of https://github.com/dakhnod/fzfs.git synced 2025-05-29 05:10:13 -07:00
Code Issues Packages Projects Releases Wiki Activity

fitted for pip install

Browse Source
This commit is contained in:
Daniel Dakhno 2024-02-16 02:14:50 +01:00
parent 5d756b54eb
commit a4736791f3
5 changed files with 45 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitmodules vendored
View File

@ -1,3 +0,0 @@
[submodule "flipperzero_protobuf_py"]
path = flipperzero_protobuf_py
url = https://github.com/flipperdevices/flipperzero_protobuf_py.git

37
flipper_api.py
View File

@ -6,9 +6,9 @@
import threading import threading
from flipperzero_protobuf_py.flipperzero_protobuf_compiled import flipper_pb2, storage_pb2 from flipperzero_protobuf.flipperzero_protobuf_compiled import flipper_pb2, storage_pb2
from flipperzero_protobuf_py.flipper_protobuf import ProtoFlipper from flipperzero_protobuf.flipper_proto import FlipperProto
from flipperzero_protobuf_py.cli_helpers import * from flipperzero_protobuf.cli_helpers import print_hex
class FlipperAPI(): class FlipperAPI():
@ -20,54 +20,54 @@ class FlipperAPI():
def connect(self): def connect(self):
with self.mutex: with self.mutex:
self.proto = ProtoFlipper(self.flipper) self.proto = FlipperProto(self.flipper)
self.proto._in_session = True
print("Ping result: ") print("Ping result: ")
print_hex(self.proto.cmd_system_ping()) print_hex(self.proto.rpc_system_ping())
def _cmd_storage_list_directory(self, path): def _cmd_storage_list_directory(self, path):
cmd_data = storage_pb2.ListRequest() cmd_data = storage_pb2.ListRequest()
cmd_data.path = path cmd_data.path = path
self.proto._cmd_send(cmd_data, 'storage_list_request') self.proto._rpc_send(cmd_data, 'storage_list_request')
def _cmd_storage_stat(self, path): def _cmd_storage_stat(self, path):
cmd_data = storage_pb2.StatRequest() cmd_data = storage_pb2.StatRequest()
cmd_data.path = path cmd_data.path = path
return self.proto._cmd_send_and_read_answer(cmd_data, 'storage_stat_request') return self.proto._rpc_send_and_read_answer(cmd_data, 'storage_stat_request')
def _cmd_storage_read(self, path): def _cmd_storage_read(self, path):
cmd_data = storage_pb2.ReadRequest() cmd_data = storage_pb2.ReadRequest()
cmd_data.path = path cmd_data.path = path
self.proto._cmd_send(cmd_data, 'storage_read_request') self.proto._rpc_send(cmd_data, 'storage_read_request')
def _cmd_storage_mkdir(self, path): def _cmd_storage_mkdir(self, path):
cmd_data = storage_pb2.MkdirRequest() cmd_data = storage_pb2.MkdirRequest()
cmd_data.path = path cmd_data.path = path
self.proto._cmd_send(cmd_data, 'storage_mkdir_request') self.proto._rpc_send(cmd_data, 'storage_mkdir_request')
def _cmd_storage_rmdir(self, path): def _cmd_storage_rmdir(self, path):
cmd_data = storage_pb2.RmdirRequest() cmd_data = storage_pb2.RmdirRequest()
cmd_data.path = path cmd_data.path = path
self.proto._cmd_send(cmd_data, 'storage_rmdir_request') self.proto._rpc_send(cmd_data, 'storage_rmdir_request')
def _cmd_storage_rename(self, old_path, new_path): def _cmd_storage_rename(self, old_path, new_path):
cmd_data = storage_pb2.RenameRequest() cmd_data = storage_pb2.RenameRequest()
cmd_data.old_path = old_path cmd_data.old_path = old_path
cmd_data.new_path = new_path cmd_data.new_path = new_path
self.proto._cmd_send(cmd_data, 'storage_rename_request') self.proto._rpc_send(cmd_data, 'storage_rename_request')
def _cmd_storage_delete(self, path, recursive): def _cmd_storage_delete(self, path, recursive):
cmd_data = storage_pb2.DeleteRequest() cmd_data = storage_pb2.DeleteRequest()
cmd_data.path = path cmd_data.path = path
cmd_data.recursive = recursive cmd_data.recursive = recursive
self.proto._cmd_send(cmd_data, 'storage_delete_request') self.proto._rpc_send(cmd_data, 'storage_delete_request')
def _cmd_storage_write(self, path, data): def _cmd_storage_write(self, path, data):
cmd_data = storage_pb2.WriteRequest() cmd_data = storage_pb2.WriteRequest()
cmd_data.path = path cmd_data.path = path
cmd_data.file.data = data cmd_data.file.data = data
self.proto._cmd_send(cmd_data, 'storage_write_request') self.proto._rpc_send(cmd_data, 'storage_write_request')
def check_response_status(self, response): def check_response_status(self, response):
if response.command_status == flipper_pb2.CommandStatus.ERROR_STORAGE_INVALID_NAME: if response.command_status == flipper_pb2.CommandStatus.ERROR_STORAGE_INVALID_NAME:
@ -81,7 +81,7 @@ class FlipperAPI():
files = [] files = []
while True: while True:
packet = self.proto._cmd_read_answer() packet = self.proto._rpc_read_answer()
self.check_response_status(packet) self.check_response_status(packet)
for file in packet.storage_list_response.file: for file in packet.storage_list_response.file:
files.append({**{ files.append({**{
@ -111,7 +111,7 @@ class FlipperAPI():
contents = [] contents = []
while True: while True:
packet = self.proto._cmd_read_answer() packet = self.proto._rpc_read_answer()
print(packet) print(packet)
self.check_response_status(packet) self.check_response_status(packet)
contents.extend(packet.storage_read_response.file.data) contents.extend(packet.storage_read_response.file.data)
@ -144,8 +144,9 @@ class FlipperAPI():
self._cmd_storage_write(path, data) self._cmd_storage_write(path, data)
def close(self): def close(self):
with self.mutex: pass
self.proto.cmd_flipper_stop_session() # with self.mutex:
# self.proto.cmd_flipper_stop_session()
class InvalidNameError(RuntimeError): class InvalidNameError(RuntimeError):
pass pass

1
flipperzero_protobuf_py

@ -1 +0,0 @@
Subproject commit 3a13eee93218e0954a04cc46d14c398bdf810d9f

26
pyproject.toml Normal file
View File

@ -0,0 +1,26 @@
[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"
[project]
name = "fzfs"
version = "0.0.1"
description = "Flipper Zero filesystem driver"
authors = [
{name = "Daniel Dakhno", email = "dakhnod@gmail.com"}
]
maintainers = [
{name = "Daniel Dakhno", email = "dakhnod@gmail.com"}
]
dependencies = [
"bleak==0.17.0",
"fusepy==3.0.1",
"pyserial==3.5",
"flipperzero_protobuf @ git+https://github.com/flipperdevices/flipperzero_protobuf_py.git@edd9d0da0d65f3911b66cbf8f899de6e7959d7e1"
]
[tool.setuptools]
py-modules = ["fzfs", "flipper_fs", "flipper_api", "flipper_serial", "serial_ble"]
[project.scripts]
fzfs = "fzfs:main"

7
requirements.txt
View File

@ -1,7 +0,0 @@
async-timeout==4.0.2
bleak==0.17.0
dbus-fast==1.4.0
fusepy==3.0.1
numpy==1.23.3
protobuf==4.21.6
pyserial==3.5