AIGC - Stable Diffusion 搭建
1. Huggingface Token
url = ’‘
1. 注册获取账号
2. 生成获取用户项目token
2. Cloud GPU
url = '' # 可租用矩池云
1. 注册获取账号
2. 选择GPU档次及预装型号（建议huggingface,Pytorch）
3. huggingface-cli login # Cloud服务器通过token验证 | 下载模型
4. JupyterLab
pip3 install diffusers
pip3 install accelerate
pip3 install spacy ftfy==4.4.3
3. autocover.txt
import diffusers, torch, os
from time import strftime, localtime
model_id = "CompVis/stable-diffusion-v1-4"
device = torch.device("cuda")
pipe = diffusers.StableDiffusionPipeline.from_pretrained(model_id, use_auth_token=True)
pipe = pipe.to(device)
image = pipe(prompt).images[0]
timeprint = strftime("%Y%m%d%H%M%S", localtime())
fldpath = os.path.join('/mnt/', 'xxx/img')
if not os.path.exists(fldpath):
os.mkdir(fldpath)
img_filenm = f'{timeprint}.png'
image.save(os.path.join(fldpath, img_filenm))
print(f'{timeprint}.png')
4. userinput_write_pyfile.py:
def userinput_write_pyfile(user_input,local_put_path):
'''
- 将用户输入的文字信息写入远程py待执行文件中
'''
import os
f = open(local_put_path,'r')
index = 0
lines = []
for line in f.readlines():
index +=1
if index == 3:
prompt = f'prompt = "{user_input}"'
lines.append(prompt)
lines.append(line)
usr_path = os.path.join(os.path.expanduser('~'), 'Desktop/SSH/upload')
pyfilepath = os.path.join(usr_path,'autocover.py')
pf = open(pyfilepath,'w')
for line in lines:
pf.write(''.join(line))
pf.close()
return pyfilepath
5. SSH API
# -*- coding:utf-8 -*-
def get_SSH_account():
host = "" # 远程服务器ip
port = 0000 # 远程服务器端口
password = "" # 远程服务器密码
user = "" # 远程服务器用户名
account_info = [host,port,password,user] # 返回SSH登陆信息
return account_info
def SSH_connection(user_input,accounts,paths,write_py, specail_fname,is_DownloadALL,filter_fname):
import paramiko,os,time
(host, port, password, user) = accounts
(remote_get_path, remote_put_paths, local_get_path, local_put_paths, pyfilepath, python_path) = paths
class Pysftp(object):
"""
python3 自动上传、下载文件
"""
global local_put_paths, local_put_path, local_get_path, remote_put_paths,remote_put_path, remote_get_path, pyfilepath, python_path
def __init__(self, host, port, user, password):
self.host = host
self.port = port
self.user = user
self.password = password
def connect(self):
"""
建立ssh连接
"""
self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.ssh.connect(self.host, self.port, self.user, self.password)
print("[SSH] -> 连接成功")
# def cmd (self, cmd):
# """
# 需要执行的命令
# """
# cmd = "ls"
# stdin, stdout, stderr = self.ssh.exec_command(cmd)
# print(stdout.read().decode("utf8"))
# def mkdir(self):
# """
# 创建本地文件夹，存放上传、下载的文件
# """
# for lp in [local_put_path, local_get_path]:
# if not os.path.exists(lp):
# os.makedirs(lp, 666)
# print("创建本地文件夹:{}".format(lp))
# else:
# print("本地文件夹:{}已存在".format(lp))
def run_pyfile(self,pyfilepath):
'''
执行python文件
'''
cmd = f"export PATH={python_path}:$PATH;python3 -u {pyfilepath}"
stdin, stdout, stderr = self.ssh.exec_command(cmd)
for line in stderr :
print(line.rstrip())
flnm = stdout.readlines()[0].replace('\n', '')
print("[SSH] ->生成成功")
return flnm
def put(self,user_input,local_put_path,remote_put_path,path_i):
"""
上传文件
"""
sftp = paramiko.SFTPClient.from_transport(self.ssh.get_transport())
# sftp = self.ssh.open_sftp()
if path_i == 0:
pyfile_path = write_py(user_input,local_put_path) # ♻️
local_full_name = pyfile_path
fname = os.path.basename(pyfile_path)
else:
local_full_name = local_put_path
fname = specail_fname
# local_done_write(local_full_name) # 本地文件已写入完成，可以上传了
sftp.put(local_full_name, os.path.join(remote_put_path, fname))
print(f'[{fname}] -> 上传成功')
# os.remove(local_full_name)
# print("{}\n上传成功：本地文件:{}====>远程{}:{},已删除该本地文件\n".format(datetime.datetime.now(), local_full_name, self.host, remote_put_path))
def get(self,flnm,is_DownloadALL,filter_fname):
"""
下载文件
"""
sftp = paramiko.SFTPClient.from_transport(self.ssh.get_transport())
flidx = 0
for fname in sftp.listdir(remote_get_path):
flidx += 1
try:
if is_DownloadALL:
print(f'[{fname}] -> 正在下载')
remote_full_name = os.path.join(remote_get_path, fname)
self.remote_done_transffer(remote_full_name)
sftp.get(remote_full_name, os.path.join(local_get_path, fname))
# sftp.remove(remote_full_name)
print(f'[{fname}] -> 下载成功')
else:
if len(filter_fname) > 0:
if filter_fname in fname:
print(f'[{fname}] -> 正在下载')
remote_full_name = os.path.join(remote_get_path, fname)
self.remote_done_transffer(remote_full_name)
from time import strftime,localtime
timeprint = strftime("%Y%m%d%H%M%S", localtime())
nfname = f'{timeprint}{filter_fname}'
sftp.get(remote_full_name, os.path.join(local_get_path, nfname))
# sftp.remove(remote_full_name)
print(f'[{fname}] -> 下载成功')
flnm = nfname
break
else:
print(f'[{fname}] -> [{filter_fname}]')
if flidx == len(sftp.listdir(remote_get_path)) - 1:
print(f'[{fname}] -> [{filter_fname}] 无法找寻A')
else:
if flnm in fname:
print(f'[{fname}] -> 正在下载')
remote_full_name = os.path.join(remote_get_path, fname)
self.remote_done_transffer(remote_full_name)
sftp.get(remote_full_name, os.path.join(local_get_path, fname))
# sftp.remove(remote_full_name)
print(f'[{fname}] -> 下载成功')
break
else:
if flidx == len(sftp.listdir(remote_get_path))-1:
print(f'[{fname}] -> [{flnm}] 无法找寻B')
except Exception as e:
print(e)
return flnm
def stat(self, fpath):
"""
检查远程服务器文件状态
:param fpath:文件绝对路径
"""
sftp = paramiko.SFTPClient.from_transport(self.ssh.get_transport())
# sftp = self.ssh.open_sftp()
return sftp.stat(fpath)
def remote_done_transffer(self, fpath):
"""
检查文件是否传输完成
:param fpath:远程服务器上待下载文件绝对路径
"""
while True:
old_size = self.stat(fpath).st_size
time.sleep(3)
new_size = self.stat(fpath).st_size
if new_size <= old_size: # 传输已完成
return
def close(self):
"""
关闭ssh连接
"""
self.ssh.close()
print("[SSH] -> 连接关闭")
def local_done_write(self, fpath,old_size):
"""
检查本地文件是否已写入完成
:param fpath:本地待上传文件绝对路径
"""
new_size = os.stat(fpath).st_size
if new_size > old_size:
print('[SSH] -> 🌈 获取成功')
return
else:
print('[SSH] -> 没有新的写入')
def transffer(local_get_path):
"""
传输函数
"""
old_size = os.stat(local_get_path).st_size
obj = Pysftp(host, port, user, password)
obj.connect() # SSH连接
for path_i in range(len(local_put_paths)): # 上传py文件/其它附件
local_put_path = local_put_paths[path_i]
remote_put_path = remote_put_paths[path_i]
obj.put(user_input,local_put_path,remote_put_path,path_i)
flnm = obj.run_pyfile(pyfilepath) # 执行py文件
flnm = obj.get(flnm,is_DownloadALL,filter_fname)# 下载文件
obj.local_done_write(local_get_path,old_size) # 下载检查
obj.close() # 关闭连接
return flnm
flnm = transffer(local_get_path)
flpath = os.path.join(local_get_path,flnm)
return flpath
6. ffmpeg
解决音频内容GPU运算时遇到ffmpeg引起问题
# conda terminal install
conda config --add channels conda-forge
conda install ffmpeg
pip3 install ffmpy
# 找到ffmpeg安装的位置
which ffmpeg
# 寻找ffdec.py文件并修改33行 【/root/miniconda3/envs/myconda/lib/python3.8/site-packages/audioread/】
ori_COMMANDS = ('ffmpeg', 'avconv')
new_COMMANDS = ('/root/miniconda3/envs/myconda/bin/ffmpeg', 'avconv')
7. Piano Audio To Midi
# pip3 install
pip3 install piano_transcription_inference
# 将训练集放置在指定位置
8. Spleeter
write_py文章来源：https://www.toymoban.com/news/detail-489950.html
# pip3 install
# pip3 install Spleeter
# pip3 install pydub
# 将训练集放置在指定位置
def auto_spleeter_pywrite(userinput,local_put_path):
import os
f = open(local_put_path, 'r')
index = 0
lines = []
for line in f.readlines():
index +=1
if index == 1:
song_convert_path = f'song_convert_path = "/mnt/aigc_music/spleeter/source/song/"'
bounce_path = f'bounce_path = "/mnt/aigc_music/spleeter/bounce"'
lines.append(song_convert_path)
lines.append('\n')
lines.append(bounce_path)
lines.append('\n')
lines.append(line)
mainpath = os.path.join(os.path.expanduser('~'), 'Desktop/SSH')
usr_path = os.path.join(os.path.expanduser('~'), 'Desktop/SSH/upload')
downloadpath = os.path.join(os.path.expanduser('~'), 'Desktop/SSH/download')
paths = [mainpath, usr_path, downloadpath]
for path in paths:
if not os.path.exists(path):
os.mkdir(path)
pyfilepath = os.path.join(usr_path, 'audio_spleeter.py')
pf = open(pyfilepath, 'w')
for line in lines:
pf.write(''.join(line))
pf.close()
return pyfilepath
audio_spleeter.txt文章来源地址https://www.toymoban.com/news/detail-489950.html
import os,subprocess,spleeter
audio_path = os.path.dirname(song_convert_path) + '/'
result_path = bounce_path
# 执行spleeter
cmd = 'for f in ' + audio_path + '*.mp3; do spleeter separate -o ' + result_path + ' -p spleeter:5stems-16kHz $f -B tensorflow; done'
subprocess.call(cmd, shell=True)
# 压缩音频文件
def wav_to_mp3():
from pydub import AudioSegment
wav_foldepath = [os.path.join(bounce_path,i) for i in os.listdir(bounce_path) if not i.startswith('.')][0]
wav_flodepath = '/mnt/aigc_music/spleeter/bounce/temp_split' # os.path.dirname(wav_foldepath)
wav_files = [os.path.join(wav_foldepath,i) for i in os.listdir(wav_flodepath) if i.endswith('.wav')]
for w in wav_files:
sound = AudioSegment.from_wav(w)
mp3_flnm = w.replace('.wav', '.mp3')
sound.export(mp3_flnm, format='mp3', bitrate='320k')
wav_to_mp3()
# zip打包分轨
def get_zip_filepath():
zipfolders = [os.path.join(bounce_path,i) for i in os.listdir(bounce_path) if not i.startswith('.')][0]
zipfolders = '/mnt/aigc_music/spleeter/bounce/temp_split' # os.path.dirname(zipfolders) + '/'
zipsource_files = [os.path.join(zipfolders,i) for i in os.listdir(zipfolders) if i.endswith('.mp3')]
zip_file = os.path.join(bounce_path+'/temp_split/','zip_temp.zip')
def file2zip(zip_file, zipsource_files):
import zipfile, os
with zipfile.ZipFile(zip_file, mode='w', compression=zipfile.ZIP_DEFLATED) as zf:
for file in zipsource_files:
parent_path, name = os.path.split(file)
zf.write(file, arcname=name)
file2zip(
zip_file=zip_file,
zipsource_files=zipsource_files,
) # ♻️
zip_tmpfolderpath = zipfolders
if not os.path.exists(zip_tmpfolderpath):
os.mkdir(zip_tmpfolderpath)
zip_path = os.path.join(zip_tmpfolderpath,'zip_temp.zip')
return zip_path
zip_path = get_zip_filepath() # ♻️
print(zip_path)
到了这里，关于AIGC - Stable Diffusion 搭建【从零到一】的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
