3D-Druck/OctoPrint-BambuPrinter
1
0
Fork 0
Code Pull Requests Actions 1 Packages Releases 1 Activity

switch ftpclient class to support A1/P1 devices hopefully

Browse Source
This commit is contained in:
jneilliii 2024-02-10 11:30:38 -05:00
parent f910a6b03e
commit 15538a9d0d
5 changed files with 223 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
octoprint_bambu_printer/ftpsclient/__init__.py
View File

@ -1,2 +1 @@
from ._client import IoTFTPSClient
from ._version import __version__
from .ftpsclient import IoTFTPSClient

159
octoprint_bambu_printer/ftpsclient/_client.py
View File

@ -1,159 +0,0 @@
"""wrapper for FTPS server interactions"""
import ftplib
import ssl
from typing import List, Optional, Union
class ImplicitTLS(ftplib.FTP_TLS):
"""ftplib.FTP_TLS sub-class to support implicit SSL FTPS"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._sock = None
@property
def sock(self):
"""return socket"""
return self._sock
@sock.setter
def sock(self, value):
"""wrap and set SSL socket"""
if value is not None and not isinstance(value, ssl.SSLSocket):
value = self.context.wrap_socket(value)
self._sock = value
def ntransfercmd(self, cmd, rest=None):
conn, size = ftplib.FTP.ntransfercmd(self, cmd, rest)
if self._prot_p:
conn = self.context.wrap_socket(conn,
server_hostname=self.host,
session=self.sock.session) # this is the fix
return conn, size
class IoTFTPSClient:
"""iot ftps ftpsclient"""
ftps_host: str
ftps_port: int
ftps_user: str
ftps_pass: str
ssl_implicit: bool
ftps_session: Union[ftplib.FTP, ImplicitTLS]
def __init__(
self,
ftps_host: str,
ftps_port: Optional[int] = 21,
ftps_user: Optional[str] = "",
ftps_pass: Optional[str] = "",
ssl_implicit: Optional[bool] = False,
) -> None:
self.ftps_host = ftps_host
self.ftps_port = ftps_port
self.ftps_user = ftps_user
self.ftps_pass = ftps_pass
self.ssl_implicit = ssl_implicit
self.instantiate_ftps_session()
def __repr__(self) -> str:
return (
"IoT FTPS Client\n"
"--------------------\n"
f"host: {self.ftps_host}\n"
f"port: {self.ftps_port}\n"
f"user: {self.ftps_user}\n"
f"ssl: {self.ssl_implicit}"
)
def instantiate_ftps_session(self) -> None:
"""init ftps_session based on input params"""
try:
if self.ssl_implicit:
self.ftps_session = ImplicitTLS()
else:
self.ftps_session = ftplib.FTP()
self.ftps_session.connect(host=self.ftps_host, port=self.ftps_port)
if self.ftps_user != "" and self.ftps_pass != "":
self.ftps_session.login(user=self.ftps_user, passwd=self.ftps_pass)
else:
self.ftps_session.login()
if self.ssl_implicit:
self.ftps_session.prot_p()
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return
def disconnect(self) -> None:
"""disconnect the current session from the ftps server"""
try:
self.ftps_session.close()
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return
def download_file(self, source: str, dest: str) -> bool:
"""download a file to a path on the local filesystem"""
try:
with open(dest, "wb") as file:
self.ftps_session.retrbinary(f"RETR {source}", file.write)
return True
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return False
def upload_file(self, source: str, dest: str) -> bool:
"""upload a file to a path inside the FTPS server"""
try:
with open(source, "rb") as file:
self.ftps_session.storbinary(f"STOR {dest}", file)
return True
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return False
def delete_file(self, path: str) -> bool:
"""delete a file from under a path inside the FTPS server"""
try:
self.ftps_session.delete(path)
return True
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return False
def move_file(self, source: str, dest: str) -> bool:
"""move a file inside the FTPS server to another path inside the FTPS server"""
try:
self.ftps_session.rename(source, dest)
return True
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return False
def list_files(
self, path: str, file_pattern: Optional[str] = None
) -> Union[List[str], None]:
"""list files under a path inside the FTPS server"""
try:
files = self.ftps_session.nlst(path)
if not files:
return
if file_pattern:
return [f for f in files if file_pattern in f]
return files
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return

3
octoprint_bambu_printer/ftpsclient/_version.py
View File

@ -1,3 +0,0 @@
VERSION = "1.1.1"
__version__ = VERSION

221
octoprint_bambu_printer/ftpsclient/ftpsclient.py Normal file
View File

@ -0,0 +1,221 @@
"""
Based on: <https://github.com/dgonzo27/py-iot-utils>
MIT License
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
wrapper for FTPS server interactions
"""
import ftplib
import os
import socket
import ssl
from typing import Optional, Union, List
from contextlib import redirect_stdout
import io
import re
class ImplicitTLS(ftplib.FTP_TLS):
"""ftplib.FTP_TLS sub-class to support implicit SSL FTPS"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._sock = None
@property
def sock(self):
"""return socket"""
return self._sock
@sock.setter
def sock(self, value):
"""wrap and set SSL socket"""
if value is not None and not isinstance(value, ssl.SSLSocket):
value = self.context.wrap_socket(value)
self._sock = value
def ntransfercmd(self, cmd, rest=None):
conn, size = ftplib.FTP.ntransfercmd(self, cmd, rest)
if self._prot_p:
conn = self.context.wrap_socket(conn,
server_hostname=self.host,
session=self.sock.session) # this is the fix
return conn, size
class IoTFTPSClient:
"""iot ftps ftpsclient"""
ftps_host: str
ftps_port: int
ftps_user: str
ftps_pass: str
ssl_implicit: bool
ftps_session: Union[ftplib.FTP, ImplicitTLS]
last_error: Optional[str] = None
welcome: str
def __init__(
self,
ftps_host: str,
ftps_port: Optional[int] = 21,
ftps_user: Optional[str] = "",
ftps_pass: Optional[str] = "",
ssl_implicit: Optional[bool] = False,
) -> None:
self.ftps_host = ftps_host
self.ftps_port = ftps_port
self.ftps_user = ftps_user
self.ftps_pass = ftps_pass
self.ssl_implicit = ssl_implicit
self.instantiate_ftps_session()
def __repr__(self) -> str:
return (
"IoT FTPS Client\n"
"--------------------\n"
f"host: {self.ftps_host}\n"
f"port: {self.ftps_port}\n"
f"user: {self.ftps_user}\n"
f"ssl: {self.ssl_implicit}"
)
def instantiate_ftps_session(self) -> None:
"""init ftps_session based on input params"""
self.ftps_session = ImplicitTLS() if self.ssl_implicit else ftplib.FTP()
self.ftps_session.set_debuglevel(0)
self.welcome = self.ftps_session.connect(
host=self.ftps_host, port=self.ftps_port)
if self.ftps_user and self.ftps_pass:
self.ftps_session.login(user=self.ftps_user, passwd=self.ftps_pass)
else:
self.ftps_session.login()
if self.ssl_implicit:
self.ftps_session.prot_p()
def disconnect(self) -> None:
"""disconnect the current session from the ftps server"""
self.ftps_session.close()
def download_file(self, source: str, dest: str):
"""download a file to a path on the local filesystem"""
with open(dest, "wb") as file:
self.ftps_session.retrbinary(f"RETR {source}", file.write)
def upload_file(self, source: str, dest: str, callback=None):
"""upload a file to a path inside the FTPS server"""
file_size = os.path.getsize(source)
block_size = max(file_size // 100, 8192)
rest = None
# Taken from ftplib.storbinary but with custom ssl handling
# due to the shitty bambu p1p ftps server TODO fix properly.
with open(source, "rb") as fp:
self.ftps_session.voidcmd('TYPE I')
with self.ftps_session.transfercmd(f"STOR {dest}", rest) as conn:
while 1:
buf = fp.read(block_size)
if not buf:
break
conn.sendall(buf)
if callback:
callback(buf)
# shutdown ssl layer
if ftplib._SSLSocket is not None and isinstance(conn, ftplib._SSLSocket):
# Yeah this is suposed to be conn.unwrap
# But since we operate in prot p mode
# we can close the connection always.
# This is cursed but it works.
if "vsFTPd" in self.welcome:
conn.unwrap()
else:
conn.shutdown(socket.SHUT_RDWR)
return self.ftps_session.voidresp()
# Old api call.
# self.ftps_session.storbinary(
# f"STOR {dest}", file, blocksize=block_size, callback=callback)
def delete_file(self, path: str):
"""delete a file from under a path inside the FTPS server"""
self.ftps_session.delete(path)
def move_file(self, source: str, dest: str):
"""move a file inside the FTPS server to another path inside the FTPS server"""
self.ftps_session.rename(source, dest)
def mkdir(self, path: str) -> str:
return self.ftps_session.mkd(path)
def list_files(self, path: str, file_pattern: Optional[str] = None) -> Union[List[str], None]:
"""list files under a path inside the FTPS server"""
try:
files = self.ftps_session.nlst(path)
if not files:
return
if file_pattern:
return [f for f in files if file_pattern in f]
return files
except Exception as ex:
print(f"unexpected exception occurred: {ex}")
pass
return
def list_files_ex(self, path: str) -> Union[list[str], None]:
"""list files under a path inside the FTPS server"""
try:
f = io.StringIO()
with redirect_stdout(f):
self.ftps_session.dir(path)
s = f.getvalue()
files = []
for row in s.split("\n"):
if len(row) <= 0: continue
attribs = row.split(" ")
match = re.search(r".*\ (\d\d\:\d\d|\d\d\d\d)\ (.*)", row)
name = ""
if match:
name = match.groups(1)[1]
else:
name = attribs[len(attribs) - 1]
file = ( attribs[0], name )
files.append(file)
return files
except Exception as ex:
print(f"unexpected exception occurred: [{ex}]")
pass
return

2
setup.py
View File

@ -14,7 +14,7 @@ plugin_package = "octoprint_bambu_printer"
plugin_name = "OctoPrint-BambuPrinter"
# The plugin's version. Can be overwritten within OctoPrint's internal data via __plugin_version__ in the plugin module
plugin_version = "0.0.5"
plugin_version = "0.0.6"
# The plugin's description. Can be overwritten within OctoPrint's internal data via __plugin_description__ in the plugin
# module