import glm
from OpenGL.GLUT import *
from OpenGL.GLU import *
from OpenGL.GL import *
import numpy

block_VAO = 0
draw = False
block_EBO_buffer_len = 0


def create_blocks(x: int, y: int, z: int):
    global draw, block_VAO, block_EBO_buffer_len
    if draw:
        return
    draw = True
    block_point_buffer = []
    block_color_buffer = []
    block_EBO_buffer = []
    block_point_buffer += [x - 0.5, y + 0.5, z - 0.5,  # V0
                           x + 0.5, y + 0.5, z - 0.5,  # V1
                           x + 0.5, y - 0.5, z - 0.5,  # V2
                           x - 0.5, y - 0.5, z - 0.5,  # V3
                           x - 0.5, y + 0.5, z + 0.5,  # V4
                           x + 0.5, y + 0.5, z + 0.5,  # V5
                           x + 0.5, y - 0.5, z + 0.5,  # V6
                           x - 0.5, y - 0.5, z + 0.5]  # V7
    block_color_buffer += [1.0, 1.0, 0.0, 1.0] * 8
    vertex_buffer = []
    for i in range(8):
        vertex_buffer += block_point_buffer[3 * i:3 * i + 3]
        vertex_buffer += block_color_buffer[4 * i:4 * i + 4]
    block_EBO_buffer += [0, 1, 5, 4,
                         3, 2, 6, 7,
                         0, 3, 7, 4,
                         1, 2, 6, 5,
                         0, 1, 2, 3,
                         4, 5, 6, 7]

    block_VAO = glGenVertexArrays(1)
    glBindVertexArray(block_VAO)

    vbo = glGenBuffers(1)
    glBindBuffer(GL_ARRAY_BUFFER, vbo)
    a = numpy.array(vertex_buffer, dtype='float32')
    print(sys.getsizeof(a), a.nbytes)
    glBufferData(GL_ARRAY_BUFFER, a.nbytes, a, GL_STATIC_DRAW)

    glEnableVertexAttribArray(0)
    glVertexAttribPointer(0, 3, GL_FLOAT, GL_FALSE, 7 * 4, None)
    glEnableVertexAttribArray(1)
    glVertexAttribPointer(1, 4, GL_FLOAT, GL_FALSE, 7 * 4, ctypes.c_void_p(3 * 4))

    ebo = glGenBuffers(1)
    glBindBuffer(GL_ELEMENT_ARRAY_BUFFER, ebo)
    a = numpy.array(block_EBO_buffer, dtype='uint32')
    glBufferData(GL_ELEMENT_ARRAY_BUFFER, a.nbytes, a, GL_STATIC_DRAW)
    block_EBO_buffer_len = len(block_EBO_buffer)

    glBindVertexArray(0)


rotate = [45, 0.0, 0]

def display(shader):
    glClear(GL_COLOR_BUFFER_BIT)

    glBindVertexArray(block_VAO)

    #projectionMatrix = Matrix.Perspective(45.0, 400 / 350, 0.1, 1000.0)
    projectionMatrix = glm.perspective(glm.radians(45.0), 400 / 350, 0.1, 1000.0)
    #viewMatrix = Matrix.LookAt([0.0, 0.0, 5], [0.0, 0.0, 0], Vector.up())
    viewMatrix = glm.lookAt(glm.vec3(0.0, 0.0, 5.0), glm.vec3(0.0, 0.0, 0.0), glm.vec3(0, 1, 0))
    #modelMatrix = Matrix.Transformation([0.0, 0.0, 0], rotate, [1.0, 1.0, 1.0])
    modelMatrix = glm.mat4(1.0)
    modelMatrix = glm.scale(modelMatrix, glm.vec3(1, 1, 1))

    modelMatrix = glm.rotate(modelMatrix, glm.radians(rotate[2]), glm.vec3(0, 0, 1)) # z
    modelMatrix = glm.rotate(modelMatrix, glm.radians(rotate[0]), glm.vec3(1, 0, 0)) # x
    modelMatrix = glm.rotate(modelMatrix, glm.radians(rotate[1]), glm.vec3(0, 1, 0)) # y

    modelMatrix = glm.translate(modelMatrix, glm.vec3(0.0, 0.0, 0))
    #shader.setUniform('MVP', projectionMatrix * viewMatrix * modelMatrix)
    mvp = projectionMatrix * viewMatrix * modelMatrix
    glUniformMatrix4fv(shader.getUniformID("MVP"), 1, GL_FALSE, glm.value_ptr(mvp))
    glDrawElements(GL_QUADS, block_EBO_buffer_len, GL_UNSIGNED_INT, None)

    rotate[0] += 0.2
    rotate[1] += 0.2
    rotate[2] += 0.2
    glutSwapBuffers()
    glutPostRedisplay()


def reshape(width, height):
    glViewport(0, 0, width, height)
    glMatrixMode(GL_PROJECTION)
    glLoadIdentity()
    gluPerspective(40.0, width / height, 0.5, 20.0)
    glMatrixMode(GL_MODELVIEW)


if __name__ == '__main__':
    glutInit(sys.argv)
    glutInitDisplayMode(GLUT_SINGLE | GLUT_RGBA | GLUT_DEPTH)
    glutInitWindowSize(400, 350)
    glutCreateWindow(b"OpenGL Window")
    create_blocks(0, 0, 0)
    glClearColor(0.0, 0.0, 0.0, 0.0)

    from static.shader import Shader
    shader = Shader('test', 'vShader.glsl', 'fShader.glsl')
    shader.useProgram()

    glPolygonMode(GL_FRONT_AND_BACK, GL_LINE)

    def _display():
        display(shader)


    glutDisplayFunc(_display)
    # glutReshapeFunc(reshape)
    glutMainLoop()

sample/opengl_example.py
92MING/Stable-Renderer

from OpenGL import GL as gl
from OpenGL import GLU as glu

def printOpenGLError():
    err = gl.glGetError() # pylint: disable=E1111
    if (err != gl.GL_NO_ERROR):
        print('GLERROR: ', glu.gluErrorString(err)) # pylint: disable=E1101

import os
# sys.path.append(os.path.abspath(os.path.dirname(__file__)))
class Shader(object):

    def initShaderFromGLSL(self, vertex_shader_paths, fragment_shader_paths):
        vertex_shader_source_list = []
        fragment_shader_source_list = []
        if(isinstance(vertex_shader_paths,list)):

            
            for GLSL in vertex_shader_paths:
                absDIR =  os.path.abspath(os.path.join(os.path.join(os.path.dirname(__file__),".."),GLSL))
                f = open(absDIR,'rb')
                vertex_shader_source_list.append(f.read())
                f.close()
            for GLSL in fragment_shader_paths:
                absDIR =  os.path.abspath(os.path.join(os.path.join(os.path.dirname(__file__),".."),GLSL))
                f = open(absDIR,'rb')
                fragment_shader_source_list.append(f.read())      
                f.close()    
            self.initShader(vertex_shader_source_list,fragment_shader_source_list)


    def initShader(self, vertex_shader_source_list, fragment_shader_source_list):
        # create program
        self.program= gl.glCreateProgram() # pylint: disable=E1111
        #print('create program ',self.program)
        printOpenGLError()

        # vertex shader
        #print('compile vertex shader...')
        self.vs = gl.glCreateShader(gl.GL_VERTEX_SHADER) # pylint: disable=E1111
        gl.glShaderSource(self.vs, vertex_shader_source_list)
        gl.glCompileShader(self.vs)
        if(gl.GL_TRUE!=gl.glGetShaderiv(self.vs, gl.GL_COMPILE_STATUS)):
            err =  gl.glGetShaderInfoLog(self.vs)
            raise Exception(err)
        gl.glAttachShader(self.program, self.vs)
        printOpenGLError()

        # fragment shader
        #print('compile fragment shader...')
        self.fs = gl.glCreateShader(gl.GL_FRAGMENT_SHADER) # pylint: disable=E1111
        gl.glShaderSource(self.fs, fragment_shader_source_list)
        gl.glCompileShader(self.fs)
        if(gl.GL_TRUE!=gl.glGetShaderiv(self.fs, gl.GL_COMPILE_STATUS)):
            err =  gl.glGetShaderInfoLog(self.fs)
            raise Exception(err)
        gl.glAttachShader(self.program, self.fs)
        printOpenGLError()

        #print('link...')
        gl.glLinkProgram(self.program)
        if(gl.GL_TRUE!=gl.glGetProgramiv(self.program, gl.GL_LINK_STATUS)):
            err =  gl.glGetShaderInfoLog(self.vs)
            raise Exception(err)
        printOpenGLError()

    def begin(self):
        if gl.glUseProgram(self.program):
            printOpenGLError()

    def end(self):
        gl.glUseProgram(0)
sample/tutorial/example_utils/shaderLoader.py
92MING/Stable-Renderer
# import os,sys
# sys.path.append(os.path.abspath(os.path.dirname(__file__)))

from OpenGL.GL import *  # pylint: disable=W0614
from static.data_types.matrix import Matrix
from static.data_types.vector import Vector
import glm
from sample.tutorial.example_utils.glutWindow import GlutWindow
from sample.tutorial.example_utils.shaderLoader import Shader
import numpy as np
np.set_printoptions(suppress=True)

g_vertex_buffer_data = [
		-1.0,-1.0,-1.0,
		-1.0,-1.0, 1.0,
		-1.0, 1.0, 1.0,
		 1.0, 1.0,-1.0,
		-1.0,-1.0,-1.0,
		-1.0, 1.0,-1.0,
		 1.0,-1.0, 1.0,
		-1.0,-1.0,-1.0,
		 1.0,-1.0,-1.0,
		 1.0, 1.0,-1.0,
		 1.0,-1.0,-1.0,
		-1.0,-1.0,-1.0,
		-1.0,-1.0,-1.0,
		-1.0, 1.0, 1.0,
		-1.0, 1.0,-1.0,
		 1.0,-1.0, 1.0,
		-1.0,-1.0, 1.0,
		-1.0,-1.0,-1.0,
		-1.0, 1.0, 1.0,
		-1.0,-1.0, 1.0,
		 1.0,-1.0, 1.0,
		 1.0, 1.0, 1.0,
		 1.0,-1.0,-1.0,
		 1.0, 1.0,-1.0,
		 1.0,-1.0,-1.0,
		 1.0, 1.0, 1.0,
		 1.0,-1.0, 1.0,
		 1.0, 1.0, 1.0,
		 1.0, 1.0,-1.0,
		-1.0, 1.0,-1.0,
		 1.0, 1.0, 1.0,
		-1.0, 1.0,-1.0,
		-1.0, 1.0, 1.0,
		 1.0, 1.0, 1.0,
		-1.0, 1.0, 1.0,
		 1.0,-1.0, 1.0]
g_color_buffer_data = [ 
		0.583,  0.771,  0.014,
		0.609,  0.115,  0.436,
		0.327,  0.483,  0.844,
		0.822,  0.569,  0.201,
		0.435,  0.602,  0.223,
		0.310,  0.747,  0.185,
		0.597,  0.770,  0.761,
		0.559,  0.436,  0.730,
		0.359,  0.583,  0.152,
		0.483,  0.596,  0.789,
		0.559,  0.861,  0.639,
		0.195,  0.548,  0.859,
		0.014,  0.184,  0.576,
		0.771,  0.328,  0.970,
		0.406,  0.615,  0.116,
		0.676,  0.977,  0.133,
		0.971,  0.572,  0.833,
		0.140,  0.616,  0.489,
		0.997,  0.513,  0.064,
		0.945,  0.719,  0.592,
		0.543,  0.021,  0.978,
		0.279,  0.317,  0.505,
		0.167,  0.620,  0.077,
		0.347,  0.857,  0.137,
		0.055,  0.953,  0.042,
		0.714,  0.505,  0.345,
		0.783,  0.290,  0.734,
		0.722,  0.645,  0.174,
		0.302,  0.455,  0.848,
		0.225,  0.587,  0.040,
		0.517,  0.713,  0.338,
		0.053,  0.959,  0.120,
		0.393,  0.621,  0.362,
		0.673,  0.211,  0.457,
		0.820,  0.883,  0.371,
		0.982,  0.099,  0.879
]        

class Tu01Win(GlutWindow):

	class GLContext(object):
		pass
	def init_opengl(self):
		glClearColor(0.0,0,0.4,0)
		glDepthFunc(GL_LESS)
		glEnable(GL_DEPTH_TEST)
		glEnable(GL_CULL_FACE)

	def init_context(self):
		self.context = self.GLContext()
		self.shader = shader = Shader()
		shader.initShaderFromGLSL(["glsl/tu01/vertex.glsl"],["glsl/tu01/fragment.glsl"])

		self.context.MVP_ID   = glGetUniformLocation(shader.program,"MVP")

		self.context.vertexbuffer  = glGenBuffers(1)
		glBindBuffer(GL_ARRAY_BUFFER,self.context.vertexbuffer)
		glBufferData(GL_ARRAY_BUFFER,len(g_vertex_buffer_data)*4,(GLfloat * len(g_vertex_buffer_data))(*g_vertex_buffer_data),GL_STATIC_DRAW)

		self.context.colorbuffer  = glGenBuffers(1)
		glBindBuffer(GL_ARRAY_BUFFER,self.context.colorbuffer)
		glBufferData(GL_ARRAY_BUFFER,len(g_color_buffer_data)*4,(GLfloat * len(g_color_buffer_data))(*g_color_buffer_data),GL_STATIC_DRAW)

	def calc_MVP(self,width=1920,height=1080):
		self.context.Projection = glm.perspective(glm.radians(45.0),float(width)/float(height),0.1,1000.0)
		self.context.View =  glm.lookAt(glm.vec3(4,3,-3), # Camera is at (4,3,-3), in World Space
						glm.vec3(0,0,0), #and looks at the (0.0.0))
						glm.vec3(0,1,0) ) #Head is up (set to 0,-1,0 to look upside-down)
		self.context.Model=  glm.mat4(1.0)

		self.context.MVP =  self.context.Projection * self.context.View * self.context.Model
		print(self.context.MVP)
		self.context.MVP = Matrix.Perspective(45, width/height, 0.1, 1000) * Matrix.LookAt([4,3,-3],[0,0,0],Vector.up()) * Matrix.Identity(4)
		#self.context.MVP = np.ascontiguousarray(self.context.MVP.flatten(), np.float32)
		#self.context.MVP = np.array(self.context.MVP.flatten(), np.float32)
		print(self.context.MVP)
	def resize(self,Width,Height):
		glViewport(0, 0, Width, Height)
		self.calc_MVP(Width,Height)

	def ogl_draw(self):

		print("draw++")
		#print self.context.MVP
		glClear(GL_COLOR_BUFFER_BIT | GL_DEPTH_BUFFER_BIT)

		self.shader.begin()
		#glUniformMatrix4fv(self.context.MVP_ID, 1, GL_FALSE, glm.value_ptr(self.context.MVP))
		glUniformMatrix4fv(self.context.MVP_ID, 1, GL_TRUE, self.context.MVP.data)
		#glUniformMatrix4fv(self.context.MVP_ID, 1, GL_FALSE,self.context.MVP)

		glEnableVertexAttribArray(0)
		glBindBuffer(GL_ARRAY_BUFFER, self.context.vertexbuffer)
		glVertexAttribPointer(0,3,GL_FLOAT,GL_FALSE,0,None)

		glEnableVertexAttribArray(1)
		glBindBuffer(GL_ARRAY_BUFFER, self.context.colorbuffer)
		glVertexAttribPointer(1,3,GL_FLOAT,GL_FALSE,0,None)


		glDrawArrays(GL_TRIANGLES, 0, 12*3) # 12*3 indices starting at 0 -> 12 triangles

		glDisableVertexAttribArray(0)
		glDisableVertexAttribArray(1)
		self.shader.end()

if __name__ == "__main__":
	win = Tu01Win()
	win.init_opengl()
	win.init_context()
	win.run()

sample/tutorial/tu_01_color_cube.py
92MING/Stable-Renderer
# import os,sys
# sys.path.append(os.path.abspath(os.path.dirname(__file__)))


from OpenGL.GL import *  # pylint: disable=W0614

import glm
from sample.tutorial.example_utils.textureLoader import textureLoader
from sample.tutorial.example_utils.glutWindow import GlutWindow
from sample.tutorial.example_utils.shaderLoader import Shader

g_vertex_buffer_data = [
		-1.0,-1.0,-1.0,
		-1.0,-1.0, 1.0,
		-1.0, 1.0, 1.0,
		 1.0, 1.0,-1.0,
		-1.0,-1.0,-1.0,
		-1.0, 1.0,-1.0,
		 1.0,-1.0, 1.0,
		-1.0,-1.0,-1.0,
		 1.0,-1.0,-1.0,
		 1.0, 1.0,-1.0,
		 1.0,-1.0,-1.0,
		-1.0,-1.0,-1.0,
		-1.0,-1.0,-1.0,
		-1.0, 1.0, 1.0,
		-1.0, 1.0,-1.0,
		 1.0,-1.0, 1.0,
		-1.0,-1.0, 1.0,
		-1.0,-1.0,-1.0,
		-1.0, 1.0, 1.0,
		-1.0,-1.0, 1.0,
		 1.0,-1.0, 1.0,
		 1.0, 1.0, 1.0,
		 1.0,-1.0,-1.0,
		 1.0, 1.0,-1.0,
		 1.0,-1.0,-1.0,
		 1.0, 1.0, 1.0,
		 1.0,-1.0, 1.0,
		 1.0, 1.0, 1.0,
		 1.0, 1.0,-1.0,
		-1.0, 1.0,-1.0,
		 1.0, 1.0, 1.0,
		-1.0, 1.0,-1.0,
		-1.0, 1.0, 1.0,
		 1.0, 1.0, 1.0,
		-1.0, 1.0, 1.0,
		 1.0,-1.0, 1.0]
g_uv_buffer_data = [
		0.000059, 1.0-0.000004, 
		0.000103, 1.0-0.336048, 
		0.335973, 1.0-0.335903, 
		1.000023, 1.0-0.000013, 
		0.667979, 1.0-0.335851, 
		0.999958, 1.0-0.336064, 
		0.667979, 1.0-0.335851, 
		0.336024, 1.0-0.671877, 
		0.667969, 1.0-0.671889, 
		1.000023, 1.0-0.000013, 
		0.668104, 1.0-0.000013, 
		0.667979, 1.0-0.335851, 
		0.000059, 1.0-0.000004, 
		0.335973, 1.0-0.335903, 
		0.336098, 1.0-0.000071, 
		0.667979, 1.0-0.335851, 
		0.335973, 1.0-0.335903, 
		0.336024, 1.0-0.671877, 
		1.000004, 1.0-0.671847, 
		0.999958, 1.0-0.336064, 
		0.667979, 1.0-0.335851, 
		0.668104, 1.0-0.000013, 
		0.335973, 1.0-0.335903, 
		0.667979, 1.0-0.335851, 
		0.335973, 1.0-0.335903, 
		0.668104, 1.0-0.000013, 
		0.336098, 1.0-0.000071, 
		0.000103, 1.0-0.336048, 
		0.000004, 1.0-0.671870, 
		0.336024, 1.0-0.671877, 
		0.000103, 1.0-0.336048, 
		0.336024, 1.0-0.671877, 
		0.335973, 1.0-0.335903, 
		0.667969, 1.0-0.671889, 
		1.000004, 1.0-0.671847, 
		0.667979, 1.0-0.335851
	]



class Tu01Win(GlutWindow):

	class GLContext(object):
		pass
	def init_opengl(self):
		glClearColor(0.0,0,0.4,0)
		glDepthFunc(GL_LESS)
		glEnable(GL_DEPTH_TEST)
		glEnable(GL_CULL_FACE)

	def init_context(self):
		self.context = self.GLContext()

		# vertex = glGenVertexArrays(1) # pylint: disable=W0612
		# glBindVertexArray(vertex)

		self.shader = shader = Shader()
		shader.initShaderFromGLSL(["glsl/tu02/vertex.glsl"],["glsl/tu02/fragment.glsl"])

		self.context.MVP_ID   = glGetUniformLocation(shader.program,"MVP")
		self.context.TextureID =  glGetUniformLocation(shader.program, "myTextureSampler")


		texture = textureLoader("resources/tu02/uvtemplate.tga")
		#texture = textureLoader("resources/tu02/uvtemplate.dds")

		self.context.textureGLID = texture.textureGLID

		self.context.vertexbuffer  = glGenBuffers(1)
		glBindBuffer(GL_ARRAY_BUFFER,self.context.vertexbuffer)
		glBufferData(GL_ARRAY_BUFFER,len(g_vertex_buffer_data)*4,(GLfloat * len(g_vertex_buffer_data))(*g_vertex_buffer_data),GL_STATIC_DRAW)

		if(texture.inversedVCoords):
			for index in range(0,len(g_uv_buffer_data)):
				if(index % 2):
					g_uv_buffer_data[index] = 1.0 - g_uv_buffer_data[index]

		self.context.uvbuffer  = glGenBuffers(1)
		glBindBuffer(GL_ARRAY_BUFFER,self.context.uvbuffer)
		glBufferData(GL_ARRAY_BUFFER,len(g_uv_buffer_data)*4,(GLfloat * len(g_uv_buffer_data))(*g_uv_buffer_data),GL_STATIC_DRAW)

	def calc_MVP(self,width=1920,height=1080):

		self.context.Projection = glm.perspective(glm.radians(45.0),float(width)/float(height),0.1,1000.0)
		self.context.View =  glm.lookAt(glm.vec3(4,3,-3), # Camera is at (4,3,-3), in World Space
						glm.vec3(0,0,0), #and looks at the (0.0.0))
						glm.vec3(0,1,0) ) #Head is up (set to 0,-1,0 to look upside-down)
		#fixed Cube Size
		self.context.Model=  glm.mat4(1.0)
		#print self.context.Model
		self.context.MVP =  self.context.Projection * self.context.View * self.context.Model	

	def resize(self,Width,Height):
		
		glViewport(0, 0, Width, Height)
		self.calc_MVP(Width,Height)

	def ogl_draw(self):

		print("draw++")
		#print self.context.MVP
		glClear(GL_COLOR_BUFFER_BIT | GL_DEPTH_BUFFER_BIT)

		self.shader.begin()
		glUniformMatrix4fv(self.context.MVP_ID,1,GL_FALSE,glm.value_ptr(self.context.MVP))


		glActiveTexture(GL_TEXTURE0)
		glBindTexture(GL_TEXTURE_2D, self.context.textureGLID)
		glUniform1i(self.context.TextureID, 0)


		glEnableVertexAttribArray(0)
		glBindBuffer(GL_ARRAY_BUFFER, self.context.vertexbuffer)
		glVertexAttribPointer(0,3,GL_FLOAT,GL_FALSE,0,None)

		glEnableVertexAttribArray(1)
		glBindBuffer(GL_ARRAY_BUFFER, self.context.uvbuffer)
		glVertexAttribPointer(1,2,GL_FLOAT,GL_FALSE,0,None)


		glDrawArrays(GL_TRIANGLES, 0, 12*3) # 12*3 indices starting at 0 -> 12 triangles

		glDisableVertexAttribArray(0)
		glDisableVertexAttribArray(1)
		self.shader.end()
        

if __name__ == "__main__":

    win = Tu01Win()
    win.init_opengl()
    win.init_context()
    win.run()

sample/tutorial/tu_02_texture_without_normal.py
92MING/Stable-Renderer
import os.path
import glm
from runtime.engine import Engine
from static.texture import Texture
from static.shader import Shader
from static.mesh import Mesh
from utils.path_utils import RESOURCES_DIR
import OpenGL.GL as gl

if __name__ == '__main__':
    class Sample(Engine):
        def beforePrepare(self):
            boatDir = os.path.join(RESOURCES_DIR, 'boat')
            self.boat = Mesh.Load(os.path.join(boatDir, 'boat.obj'))
            self.boatShader = Shader('boat_shader',os.path.join(boatDir, 'boat_vs.glsl'), os.path.join(boatDir, 'boat_fs.glsl'))
            self.boatDiffuseTex = Texture.Load(os.path.join(boatDir, 'boatColor.png'), 'boat_diffuse')
            proj = glm.perspective(glm.radians(45.0), self.WindowManager.AspectRatio, 0.1, 1000.0)
            view = glm.lookAt(glm.vec3(4, 4, -3), glm.vec3(0, 0, 0), glm.vec3(0, 1, 0))
            self.RenderManager.UpdateUBO_ProjMatrix(proj)
            self.RenderManager.UpdateUBO_ViewMatrix(view)

        def beforeFrameRun(self):
            modelM = self.RenderManager.UBO_ModelMatrix
            modelM = glm.rotate(modelM, glm.radians(0.05), glm.vec3(0.0, 1.0, 0.0))
            self.RenderManager.UpdateUBO_ModelMatrix(modelM)

            self.boatShader.useProgram()
            self.boatDiffuseTex.bind(0, self.boatShader.getUniformID('boatDiffuseTex'))
            self.boat.draw()

    Sample().run()
source/main.py
92MING/Stable-Renderer
from runtime.engineObj import EngineObj
from utils.global_utils import GetOrAddGlobalValue

_COMPONENT_CLSES = GetOrAddGlobalValue("_COMPONENT_CLSES", {})
class ComponentMeta(type):
    def __new__(cls, *args, **kwargs):
        clsname = args[0]
        if clsname in _COMPONENT_CLSES:
            return _COMPONENT_CLSES[clsname]
        else:
            cls = super().__new__(cls, *args, **kwargs)
            _COMPONENT_CLSES[clsname] = cls
            return cls
class Component(EngineObj, metaclass=ComponentMeta):
    '''Base cls of all components, e.g. Camera, Light, etc.'''

    # region cls properties
    Priority = 0
    '''
    Override this property to change the priority of this component.
    Lower value means higher priority.
    Note that the priority of a component is only for the components in the same GameObject.
    '''

    Unique = False
    '''
    Override this property to change the uniqueness of this component.
    If this property is True, there can be only one component of this type in a GameObject.
    '''

    RequireComponent = ()
    '''
    Override this property to change the required components of this component. You can input Component cls or str(cls name).
    e.g. Camera.RequireComponent = (Transform, )
    '''

    @classmethod
    def ComponentName(cls):
        return cls.__qualname__
    @classmethod
    def FindComponentCls(cls, cls_name):
        try:
            return _COMPONENT_CLSES[cls_name]
        except KeyError:
            raise KeyError(f'Component with class name: {cls_name} not found.')
    def __class_getitem__(cls, item):
        '''You can find a component cls by using this syntax: Component[cls_name]'''
        return cls.FindComponentCls(item)
    # endregion

    def __init__(self, gameObj:'GameObject', enable=True):
        if self.__class__.__qualname__ == 'Component':
            raise Exception('Component is an abstract class. You can not create an instance of it.')
        if gameObj is None:
            raise Exception('gameObj can not be None.')
        self._gameObj = gameObj
        self._enable = enable
        self._started = False
        self._awaked = False
    def __ge__(self, other):
        return self.Priority >= other.Priority
    def __gt__(self, other):
        return self.Priority > other.Priority
    def __le__(self, other):
        return self.Priority <= other.Priority
    def __lt__(self, other):
        return self.Priority < other.Priority

    def awake(self):
        '''Called when this component is added to a GameObject (in case it is enabled). Awake does not follow the priority rule & will be called immediately.'''
        pass
    def start(self):
        '''Called when this component runs for the first time'''
        pass

    def _tryAwake(self):
        if self.enable:
            if not self._awaked :
                self._awaked = True
                self.awake()
            self.onEnable()
    def _tryStart(self):
        if not self._started and self.enable and self._awaked:
            self._started = True
            self.start()
    def _checkAwakeAndStart(self):
        self._tryAwake()
        self._tryStart()
    def fixedUpdate(self):
        '''Called every fixed frame'''
        self._checkAwakeAndStart()
    def update(self):
        '''Called every frame'''
        self._checkAwakeAndStart()
    def lateUpdate(self):
        '''Called every frame after update'''
        self._checkAwakeAndStart()

    def onEnable(self):
        '''Called when this component is enabled. onEnable will not follow the priority rule.'''
        pass
    def onDisable(self):
        '''
        Called when this component is disabled. onDisable will not follow the priority rule.
        When destroy a GameObject, onDisable will be called before onDestroy.
        '''
        pass
    def onDestroy(self):
        '''
        Called when this component is destroyed.
        Component can only be destroyed by GameObject.removeComponent
        onDestroy will not follow the priority rule.
        onDisable will be called before onDestroy.
        '''
        pass

    @property
    def gameObj(self)->'GameObject':
        return self._gameObj
    @property
    def enable(self):
        return self._enable and self._gameObj.active
    @enable.setter
    def enable(self, value):
        self.setEnable(value)
    def setEnable(self, value):
        if self._enable != value:
            self._enable = value
            if self._gameObj.active:
                if value:
                    self._tryAwake()
                else:
                    self.onDisable()
source/runtime/component.py
92MING/Stable-Renderer
from runtime.components.transform import *
from static.color import Color
from static.enums import ProjectionType
from typing import Set
import glm

class Camera(Component):

    # region class properties / methods
    _Main_Camera = None # current main camera
    _All_Cameras:Set['Camera'] = set()
    _Has_Init_UBO = False
    _Camera_UBO_ID = None

    RequireComponent = (Transform, )

    @classmethod
    def ActiveCameras(cls):
        return [cam for cam in cls._All_Cameras if cam.enable]
    @classmethod
    def MainCamera(cls):
        return cls._Main_Camera
    # endregion

    def __init__(self,
                 gameObj,
                 enable=True,
                 fov=45.0,
                 near_plane=0.1,
                 far_plane=100.0,
                 ortho_size=1.0,
                 bgColor=Color.CLEAR,
                 projection_type=ProjectionType.PERSPECTIVE):
        super().__init__(gameObj, enable)
        self.fov = fov
        self.near_plane = near_plane
        self.far_plane = far_plane
        self.ortho_size = ortho_size
        self.bgColor = bgColor
        self.projection_type = projection_type

    def awake(self):
        Camera._All_Cameras.add(self)
    def onDestroy(self):
        Camera._All_Cameras.remove(self)
    def onDisable(self):
        if self.isMainCamera:
            Camera._Main_Camera = None
            for cam in Camera._All_Cameras: # try to set another camera as main camera
                if cam is not self:
                    if cam.set_as_main_camera():
                        break

    @property
    def isMainCamera(self):
        return Camera._Main_Camera == self
    def set_as_main_camera(self)->bool:
        '''
        Set this camera as the main camera. Return True if success, otherwise return False.
        :return: bool
        '''
        if not self.enable:
            return False
        Camera._Main_Camera = self
        return True

    @property
    def _pos(self):
        return self.gameObj.transform.globalPos
    @property
    def _forward(self):
        return self.gameObj.transform.forward
    @property
    def _up(self):
        return self.gameObj.transform.up
    @property
    def viewMatrix(self):
        return glm.lookAt(self._pos, self._pos + self._forward, self._up)
    @property
    def projectionMatrix(self):
        if self.projection_type == ProjectionType.PERSPECTIVE:
            return glm.perspective(
                self.fov,
                self.engine.WindowManager.AspectRatio,
                self.near_plane,
                self.far_plane,
            )
        elif self.projection_type == ProjectionType.ORTHOGRAPHIC:
            screen_scale = self.engine.WindowManager.AspectRatio * self.ortho_size / 2
            return glm.ortho(
                -screen_scale,
                screen_scale,
                -self.ortho_size / 2,
                self.ortho_size / 2,
                self.near_plane,
                self.far_plane,
            )

    def lateUpdate(self):
        if self.isMainCamera:
            if self.engine.bgColor != self.bgColor:
                self.engine.bgColor = self.bgColor
            viewMatrix, projectionMatrix = self.viewMatrix, self.projectionMatrix
            if self.engine.RenderManager.UBO_ViewMatrix != viewMatrix:
                self.engine.RenderManager.UpdateUBO_ViewMatrix(viewMatrix)
            if self.engine.RenderManager.UBO_ProjMatrix != projectionMatrix:
                self.engine.RenderManager.UpdateUBO_ProjMatrix(projectionMatrix)

source/runtime/components/camera.py
92MING/Stable-Renderer
import numpy as np
from typing import List
from OpenGL.GL import *
from OpenGL.GLUT import *
from static.engine import Engine
from static.data_types.vector import Vector
from static.data_types.matrix import Matrix
from static.color import Color
from runtime.components.camera import Camera
from static.enums import LightType
from static.data_structure import (
    SingleLightData,
    FBO_Texture,
    ShadowLightDataPointLight,
    ShadowLightDataOtherLight,
    LightData_SSBO
)


class Light:
    UNCHANGE = float('-inf')  # FLT_MIN

    has_init_SSBO = False
    light_SSBO_id = None
    ambient_light_color = None  # Set this appropriately
    ambient_light_intensity = None  # Set this appropriately
    lights_enable = set()  # set to ensure uniqueness
    lights_casting_shadow = set()

    # shadow
    cast_shadow = False
    has_init_frame_buffer_and_texture = False
    all_point_light_shadow_FBO_and_tex: List[FBO_Texture] = []  # static vector<FBO_Texture> allPointLightShadowFBOandTex;
    all_other_light_shadow_FBO_and_tex: List[FBO_Texture] = []  # static vector<FBO_Texture> allOtherLightShadowFBOandTex;
    frame_buffer_id = None
    shadow_map_id = None
    far_plane_point_light = 100.0

    def __init__(self, enable=True, cast_shadow=False, light_type=None, light_color=None, intensity=1.0):
        self.this_light_data = SingleLightData()  # Define the structure appropriately
        if enable:
            Light.lights_enable.add(self)
        self.this_light_data.light_type = light_type
        self.this_light_data.lightColor = light_color
        self.this_light_data.intensity = intensity
        if cast_shadow:
            self.set_cast_shadow(True)

    def __del__(self):
        Light.lights_enable.remove(self)
        if self.is_cast_shadow():
            self.set_cast_shadow(False)

    def set_light_enable(self, set_val):
        if set_val and self not in Light.lights_enable:
            Light.lights_enable.append(self)
        elif not set_val and self in Light.lights_enable:
            Light.lights_enable.remove(self)

    @staticmethod
    def init_SSBO():
        if Light.has_init_SSBO:
            return
        # Equivalent GL functions, assuming you're using PyOpenGL
        Light.light_SSBO_id = glGenBuffers(1)
        glBindBuffer(GL_SHADER_STORAGE_BUFFER, Light.light_SSBO_id)
        # You might need to adjust the size calculation for glBufferData
        glBufferData(GL_SHADER_STORAGE_BUFFER, 0, None, GL_DYNAMIC_DRAW)
        glBindBufferBase(GL_SHADER_STORAGE_BUFFER, 1, Light.light_SSBO_id)
        glBindBuffer(GL_SHADER_STORAGE_BUFFER, 0)
        Light.has_init_SSBO = True

    def pre_init_frame_buffer_and_texture():
        if not Light.has_init_frame_buffer_and_texture:
            # init point lights
            max_size = Engine.get_shadow_map_texture_size()
            for i in range(Engine.get_max_light_num_with_depth_map()):
                FBO_tex = FBO_Texture()
                FBO_tex.FBO_id = glGenFramebuffers(1)
                glBindFramebuffer(GL_FRAMEBUFFER, FBO_tex.FBO_id)
                FBO_tex.shadow_map_texture_id = glGenTextures(1)
                glBindTexture(GL_TEXTURE_CUBE_MAP, FBO_tex.shadow_map_texture_id)
                for j in range(6):
                    glTexImage2D(GL_TEXTURE_CUBE_MAP_POSITIVE_X + j, 0, GL_DEPTH_COMPONENT, max_size.x, max_size.y, 0, GL_DEPTH_COMPONENT, GL_FLOAT, None)
                    glTexParameteri(GL_TEXTURE_CUBE_MAP, GL_TEXTURE_MAG_FILTER, GL_NEAREST)
                    glTexParameteri(GL_TEXTURE_CUBE_MAP, GL_TEXTURE_MIN_FILTER, GL_NEAREST)
                    glTexParameteri(GL_TEXTURE_CUBE_MAP, GL_TEXTURE_WRAP_S, GL_CLAMP_TO_BORDER)
                    glTexParameteri(GL_TEXTURE_CUBE_MAP, GL_TEXTURE_WRAP_T, GL_CLAMP_TO_BORDER)
                    glTexParameteri(GL_TEXTURE_CUBE_MAP, GL_TEXTURE_WRAP_R, GL_CLAMP_TO_BORDER)
                    glTexParameterfv(GL_TEXTURE_CUBE_MAP, GL_TEXTURE_BORDER_COLOR, np.array([1.0, 1.0, 1.0, 1.0]))  # TODO: 不确定最后一个参数是否正确
                glFramebufferTexture(GL_FRAMEBUFFER, GL_DEPTH_ATTACHMENT, FBO_tex.shadow_map_texture_id, 0)
                if glCheckFramebufferStatus(GL_FRAMEBUFFER) != GL_FRAMEBUFFER_COMPLETE:
                    print("ERROR when generating light shadow map framebuffer: not complete.")
                    return
                glDrawBuffer(GL_NONE)
                glReadBuffer(GL_NONE)
                glBindTexture(GL_TEXTURE_CUBE_MAP, 0)
                glBindFramebuffer(GL_FRAMEBUFFER, 0)
                Light.all_point_light_shadow_FBO_and_tex.append(FBO_tex)

            # init other lights
            for i in range(Engine.get_max_light_num_with_depth_map()):
                FBO_tex = FBO_Texture()
                FBO_tex.FBO_id = glGenFramebuffers(1)
                glBindFramebuffer(GL_FRAMEBUFFER, FBO_tex.FBO_id)
                FBO_tex.shadow_map_texture_id = glGenTextures(1)
                glBindTexture(GL_TEXTURE_2D, FBO_tex.shadow_map_texture_id)
                glTexImage2D(GL_TEXTURE_2D, 0, GL_DEPTH_COMPONENT, max_size.x, max_size.y, 0, GL_DEPTH_COMPONENT, GL_FLOAT, None)
                glTexParameteri(GL_TEXTURE_2D, GL_TEXTURE_MAG_FILTER, GL_NEAREST)
                glTexParameteri(GL_TEXTURE_2D, GL_TEXTURE_MIN_FILTER, GL_NEAREST)
                glTexParameteri(GL_TEXTURE_2D, GL_TEXTURE_WRAP_S, GL_CLAMP_TO_BORDER)
                glTexParameteri(GL_TEXTURE_2D, GL_TEXTURE_WRAP_T, GL_CLAMP_TO_BORDER)
                glTexParameterfv(GL_TEXTURE_2D, GL_TEXTURE_BORDER_COLOR, np.array([1.0, 1.0, 1.0, 1.0]))
                glFramebufferTexture2D(GL_FRAMEBUFFER, GL_DEPTH_ATTACHMENT, GL_TEXTURE_2D, FBO_tex.shadow_map_texture_id, 0)
                glDrawBuffer(GL_NONE)
                glReadBuffer(GL_NONE)
                if glCheckFramebufferStatus(GL_FRAMEBUFFER) != GL_FRAMEBUFFER_COMPLETE:
                    print("ERROR when generating light shadow map framebuffer: not complete.")
                    return
                glBindTexture(GL_TEXTURE_2D, 0)
                glBindFramebuffer(GL_FRAMEBUFFER, 0)
                Light.all_other_light_shadow_FBO_and_tex.append(FBO_tex)

            print("Done generating all light shadow maps and textures.")
            Light.has_init_frame_buffer_and_texture = True

    def set_cast_shadow(self, castShadow):
        if self.cast_shadow == castShadow:
            return

        if castShadow:
            if self.this_light_data.light_type == LightType.POINT:
                for FBO_tex in Light.all_point_light_shadow_FBO_and_tex:
                    if not FBO_tex.occupied:
                        self.frame_buffer_id = FBO_tex.FBO_id
                        self.shadow_map_id = FBO_tex.shadow_map_texture_id
                        FBO_tex.occupied = True
                        Light.lights_casting_shadow.append(self)
                        self.cast_shadow = True
                        return

                print("no free point light shadow map for using")
                return

            else:
                for FBO_tex in Light.all_other_light_shadow_FBO_and_tex:
                    if not FBO_tex.occupied:
                        self.frame_buffer_id = FBO_tex.FBO_id
                        self.shadow_map_id = FBO_tex.shadow_map_texture_id
                        FBO_tex.occupied = True
                        Light.lights_casting_shadow.append(self)
                        self.cast_shadow = True
                        return

                print("no free other light shadow map for using")
                return

        else:
            Light.lights_casting_shadow.remove(self)
            if self.this_light_data.light_type == LightType.POINT:
                for FBO_tex in Light.all_point_light_shadow_FBO_and_tex:
                    if FBO_tex.FBO_id == self.frame_buffer_id:
                        FBO_tex.occupied = False
                        self.frame_buffer_id = None
                        self.shadow_map_id = None
                        return
            else:
                for FBO_tex in Light.all_other_light_shadow_FBO_and_tex:
                    if FBO_tex.FBO_id == self.frame_buffer_id:
                        FBO_tex.occupied = False
                        self.frame_buffer_id = None
                        self.shadow_map_id = None
                        return
            glDeleteTextures(1, self.shadow_map_id)

    def set_light_type(self, light_type: LightType):
        if self.this_light_data.light_type == light_type:
            return
        if self.cast_shadow:
            self.set_cast_shadow(False)
            self.this_light_data.light_type = light_type
            self.set_cast_shadow(True)
        else:
            self.this_light_data.light_type = light_type

    def set_spot_light_cutoff(self, new_cutoff=-1, new_outer_cutoff=-1):
        if self.this_light_data.light_type != LightType.SPOT:
            print("This light is not a spot light. No need to set spot light cutoff.")
            return
        if new_cutoff >= 0:
            self.this_light_data.cutOff = new_cutoff
        if new_outer_cutoff >= 0:
            self.this_light_data.outer_cutoff = new_outer_cutoff

    def set_distance_formula(self, quadratic=UNCHANGE, linear=UNCHANGE, constant=UNCHANGE):
        if self.this_light_data.light_type not in (LightType.POINT, LightType.SPOT):
            print("This light is not a point light or spot light. No need to set distance formula.")
            return
        if quadratic != self.UNCHANGE:
            self.this_light_data.quadratic = quadratic
        if linear != self.UNCHANGE:
            self.this_light_data.linear = linear
        if constant != self.UNCHANGE:
            self.this_light_data.constant = constant

    def set_light_color(self, color: Color):
        self.this_light_data.lightColor = color

    def set_light_intensity(self, intensity):
        if intensity < 0:
            print("Light intensity cannot be negative.")
            return
        self.this_light_data.intensity = intensity

    def set_point_light_shadow_map_far_plane(self, farPlane):
        if self.this_light_data.light_type != "POINT_LIGHT":
            print("This light is not a point light. No need to set point light shadow map far plane.")
            return
        elif farPlane < 0:
            print("Point light shadow map far plane cannot be negative.")
            return
        self.far_plane_point_light = farPlane

    def is_shadow_light(self):
        return self.cast_shadow

    @ staticmethod
    def set_ambient_light(color, intensity=UNCHANGE):
        global ambient_light_color, ambient_light_intensity  # Assuming these are global variables in your code
        ambient_light_color = color
        if intensity != Light.UNCHANGE:
            ambient_light_intensity = intensity

    def get_this_shadow_light_data_other_light(self):
        if not Light.cast_shadow or self.this_light_data.light_type == LightType.POINT:
            return ShadowLightDataOtherLight()

        shadow_light_data = ShadowLightDataOtherLight()
        shadow_light_data.frame_buffer_id = self.frame_buffer_id
        shadow_light_data.shadow_map_id = self.shadow_map_id
        shadow_light_data.light_VP = Matrix.Identity(4)
        shadow_light_data.light_type = self.this_light_data.light_type

        if self.this_light_data.light_type == LightType.SPOT:
            shadow_light_data.light_pos_or_light_dir = self.this_light_data.position
        elif self.this_light_data.light_type == LightType.DIRECTIONAL:
            shadow_light_data.light_pos_or_light_dir = self.this_light_data.direction

        if not Camera.has_main_cam():
            return shadow_light_data

        cam: Camera = Camera.get_main_cam()
        v = Matrix.Identity(4)
        p = Matrix.Identity(4)

        if self.this_light_data.light_type == LightType.DIRECTIONAL:
            scale = (Engine.get_shadow_map_texture_size().x / float(Engine.get_shadow_map_texture_size().y)) / 2
            p = Matrix.Orthographic(-scale * Engine.get_ortho_distance(), scale * Engine.get_ortho_distance(), -scale *
                                    Engine.get_ortho_distance(), scale * Engine.get_ortho_distance(), cam.nearPlane, cam.farPlane)
            Matrix.LookAt(-10.0 * self.this_light_data.direction, Vector(0.0, 0.0, 0.0), cam.get_up())

        elif self.this_light_data.light_type == LightType.SPOT:
            p = Matrix.Perspective(
                np.radians(self.this_light_data.outer_cutoff * 2),
                Engine.get_shadow_map_texture_size().x / float(Engine.get_shadow_map_texture_size().y),
                cam.near_plane,
                cam.far_plane
            )
            v = Matrix.LookAt(
                self.this_light_data.position,
                self.this_light_data.position + self.this_light_data.direction,
                cam.get_up()
            )

        shadow_light_data.light_VP = p * v
        return shadow_light_data

    def get_this_shadow_light_data_point_light(self) -> ShadowLightDataPointLight:
        if not self.cast_shadow or self.this_light_data.light_type != LightType.POINT:
            return ShadowLightDataPointLight()

        shadow_light_data = ShadowLightDataPointLight()
        shadow_light_data.frame_buffer_id = self.frame_buffer_id
        shadow_light_data.shadow_map_id = self.shadow_map_id
        shadow_light_data.light_pos = self.this_light_data.position
        shadow_light_data.far_plane = self.far_plane_point_light
        shadow_light_data.light_proj = Matrix.Perspective(
            90.0,
            Engine.get_shadow_map_texture_size()[0] / float(Engine.get_shadow_map_texture_size()[1]),
            0.0,
            self.far_plane_point_light
        )
        position = self.this_light_data.position
        shadow_light_data.light_views = [
            Matrix.LookAt(position, position + Vector(1.0, 0.0, 0.0), Vector(0.0, -1.0, 0.0)),
            Matrix.LookAt(position, position + Vector(-1.0, 0.0, 0.0), Vector(0.0, -1.0, 0.0)),
            Matrix.LookAt(position, position + Vector(0.0, 1.0, 0.0), Vector(0.0, 0.0, 1.0)),
            Matrix.LookAt(position, position + Vector(0.0, -1.0, 0.0), Vector(0.0, 0.0, -1.0)),
            Matrix.LookAt(position, position + Vector(0.0, 0.0, 1.0), Vector(0.0, -1.0, 0.0)),
            Matrix.LookAt(position, position + Vector(0.0, 0.0, -1.0), Vector(0.0, -1.0, 0.0))
        ]
        return shadow_light_data

    # these methods may be unnecessary
    @ classmethod
    def get_light_casting_shadow(cls):
        return cls.lights_casting_shadow

    @ classmethod
    def get_lights_enable(cls):
        return cls.lights_enable

    @ classmethod
    def current_enable_light_num(cls):
        return len(cls.lights_enable)

    @ classmethod
    def current_light_using_depth_map_num(cls):
        return len(cls.lights_casting_shadow)

    @ classmethod
    def get_ambient_light_power(cls):
        return cls.ambient_light_intensity

    @ classmethod
    def get_ambient_light_color(cls):
        return cls.ambient_light_color

    @ classmethod
    def get_light_SSBO_id(cls):
        if not cls.has_init_SSBO:
            cls.init_SSBO()  # You'll need to define InitSSBO method in Python
        return cls.light_SSBO_id

    @ classmethod
    def set_all_SSBO_data(cls, all_lights, ambient_light, ambient_power):
        light_data_SSBO = LightData_SSBO()  # This should be a dict or another class
        light_data_SSBO.all_single_light_data = all_lights
        light_data_SSBO.single_light_data_length = len(all_lights)
        light_data_SSBO.ambient_light = ambient_light
        light_data_SSBO.ambient_power = ambient_power

        if not cls.has_init_SSBO:
            cls.init_SSBO()

        glBindBuffer(GL_SHADER_STORAGE_BUFFER, cls.light_SSBO_id)

        glBufferSubData(GL_SHADER_STORAGE_BUFFER,
                        0,
                        sys.getsizeof(float) + sys.getsizeof(Color) + sys.getsizeof(int) + sys.getsizeof(all_lights),
                        light_data_SSBO
                        )

        glBindBuffer(GL_SHADER_STORAGE_BUFFER, 0)

    @ classmethod
    def delete_light_SSBO(cls):
        if cls.has_init_SSBO:
            glDeleteBuffers(1, cls.light_SSBO_id)
            cls.has_init_SSBO = False

source/runtime/components/light.py
92MING/Stable-Renderer
from runtime.component import Component
from glm import *
from typing import Optional, Union, Sequence

class Transform(Component):
    '''記載物件的位置、旋轉、縮放等資訊、操作'''

    Unique = True # 一個物件只能有一個Transform

    def __init__(self, gameObj:'GameObject', enable=True):
        if not enable:
            raise Exception('Transform can not be disabled.')
        super().__init__(gameObj)
        self.localPos = vec3(0, 0, 0)
        self._localRot = quat() # rotation cannot be set directly since it is quaternion
        self.localScale = vec3(1, 1, 1)
    @property
    def enable(self):
        return True
    @enable.setter
    def enable(self, value):
        if not value:
            raise Exception('Transform can not be disabled.')
    @property
    def parent(self)->Optional['Transform']:
        '''return the transform on parent gameObject. If this transform has no parent or parent has no transform, return None'''
        if self.gameObj.parent is not None and self.gameObj.parent.hasComponent(Transform):
            return self.gameObj.parent.transform
        return None

    # region rotation
    def localRot(self, radian=False)->vec3:
        if not radian:
            return degrees(eulerAngles(self._localRot))
        else:
            return eulerAngles(self._localRot)
    def setLocalRot(self, x, y, z, radian=False):
        if not radian:
            x, y, z = radians(x), radians(y), radians(z)
        self._localRot = quat(vec3(x, y, z))
    def localRotX(self, radian=False)->float:
        return self.localRot(radian)[0]
    def localRotY(self, radian=False)->float:
        return self.localRot(radian)[1]
    def localRotZ(self, radian=False)->float:
        return self.localRot(radian)[2]
    def setLocalRotX(self, x, radian=False):
        localRot = self.localRot(radian)
        self.setLocalRot(x, localRot[1], localRot[2], radian)
    def setLocalRotY(self, y, radian=False):
        localRot = self.localRot(radian)
        self.setLocalRot(localRot[0], y, localRot[2], radian)
    def setLocalRotZ(self, z, radian=False):
        localRot = self.localRot(radian)
        self.setLocalRot(localRot[0], localRot[1], z, radian)

    def globalRot(self, radian=False)->vec3:
        gloablForward = self.forward
        globalUp = self.up
        if not radian:
            return degrees(eulerAngles(quatLookAt(gloablForward, globalUp)))
        else:
            return eulerAngles(quatLookAt(gloablForward, globalUp))

    def rotateLocalAxis(self, axis:vec3, angle:float, radian=False):
        if not radian:
            angle = radians(angle)
        angle = -angle
        self._localRot = rotate(self._localRot, angle, axis)
    def rotateLocalX(self, angle:float, radian=False):
        '''rotate around local x axis'''
        self.rotateLocalAxis(self.localRight, angle, radian)
    def rotateLocalY(self, angle:float, radian=False):
        '''rotate around local y axis'''
        self.rotateLocalAxis(self.localUp, angle, radian)
    def rotateLocalZ(self, angle:float, radian=False):
        '''rotate around local z axis'''
        self.rotateLocalAxis(self.localForward, angle, radian)
    def rotateGlobalAxis(self, globalAxis:vec3, angle:float, radian=False):
        if not radian:
            angle = radians(angle)
        newGlobalPos = rotate(self.globalPos, angle, globalAxis)
        newGlobalForward = rotate(self.forward, angle, globalAxis)
        self.globalPos = newGlobalPos
        self.forward = newGlobalForward

    @property
    def localForward(self)->vec3:
        '''return vec3(0, 0, -1)'''
        return vec3(0, 0, -1)
    @property
    def localUp(self)->vec3:
        '''return vec3(0, 1, 0)'''
        return vec3(0, 1, 0)
    @property
    def localRight(self)->vec3:
        '''return vec3(1, 0, 0)'''
        return vec3(1, 0, 0)

    @property
    def forward(self)->vec3:
        '''return z- direction in local space transform to global space'''
        return self.transformDirection(self.localForward)
    @forward.setter
    def forward(self, value):
        '''set z- direction in local space transform to global space. Keep up direction unchanged'''
        newRight = normalize(cross(self.up, value))
        newUp = normalize(cross(value, newRight))
        self._localRot = quatLookAt(value, newUp)
    @property
    def up(self)->vec3:
        '''return y+ direction in local space transform to global space'''
        return self.transformDirection(self.localUp)
    @up.setter
    def up(self, value):
        '''set y+ direction in local space transform to global space. Keep forward direction unchanged'''
        newRight = normalize(cross(value, self.forward))
        newForward = normalize(cross(newRight, value))
        self._localRot = quatLookAt(newForward, value)
    @property
    def right(self)->vec3:
        '''return x+ direction in local space transform to global space'''
        return self.transformDirection(self.localRight)
    @right.setter
    def right(self, value):
        '''set x+ direction in local space transform to global space. Keep up direction unchanged'''
        newForward = normalize(cross(value, self.up))
        newUp = normalize(cross(newForward, value))
        self._localRot = quatLookAt(newForward, newUp)
    def lookAt(self, globalTarget:Union['Transform','GameObject', vec3, Sequence]):
        '''rotate transform to make forward direction point to globalTarget, and up direction point to up'''
        from runtime.gameObj import GameObject
        if isinstance(globalTarget, GameObject):
            globalTarget = globalTarget.transform.globalPos
        elif isinstance(globalTarget, Transform):
            globalTarget = globalTarget.globalPos
        target = self.inverseTransformPoint(globalTarget)
        oldForward = self.localForward
        newForward = normalize(target - self.localPos)
        rotationAxis = normalize(cross(oldForward, newForward))
        rotationAngle = acos(dot(oldForward, newForward))
        self.rotateLocalAxis(rotationAxis, rotationAngle, True)
    # endregion

    # region position
    @property
    def globalPos(self)->vec3:
        '''return the position of this transform in global space'''
        return self.transformPoint(self.localPos)
    @globalPos.setter
    def globalPos(self, value):
        self.setGlobalPos(value.x, value.y, value.z)
    def setGlobalPos(self, x,y,z):
        '''set the position of this transform in global space'''
        self.localPos = self.inverseTransformPoint(vec3(x,y,z))
    def setGlobalPosX(self, x):
        globalPos = self.globalPos
        self.setGlobalPos(x, globalPos.y, globalPos.z)
    def setGlobalPosY(self, y):
        globalPos = self.globalPos
        self.setGlobalPos(globalPos.x, y, globalPos.z)
    def setGlobalPosZ(self, z):
        globalPos = self.globalPos
        self.setGlobalPos(globalPos.x, globalPos.y, z)
    # endregion

    # region scale
    @property
    def globalScale(self)->vec3:
        '''return the scale of this transform in global space'''
        if self.parent is None:
            return self.localScale
        return (self.parent.globalScaleMatrix * vec4(self.localScale, 1)).xyz
    @property
    def globalScaleX(self)->float:
        return self.globalScale.x
    @globalScaleX.setter
    def globalScaleX(self, value):
        self.setGlobalScaleX(value)
    @property
    def globalScaleY(self)->float:
        return self.globalScale.y
    @globalScaleY.setter
    def globalScaleY(self, value):
        self.setGlobalScaleY(value)
    @property
    def globalScaleZ(self)->float:
        return self.globalScale.z
    @globalScaleZ.setter
    def globalScaleZ(self, value):
        self.setGlobalScaleZ(value)
    @globalScale.setter
    def globalScale(self, value):
        self.setGlobalScale(value.x, value.y, value.z)
    def setGlobalScale(self, x,y,z):
        '''set the scale of this transform in global space'''
        scale = vec3(x, y, z)
        if self.parent is None:
            self.localScale = scale
        else:
            self.localScale = (self.parent.inverseGlobalScaleMatrix * vec4(scale, 1)).xyz
    def setGlobalScaleX(self, x):
        globalScale = self.globalScale
        self.setGlobalScale(x, globalScale.y, globalScale.z)
    def setGlobalScaleY(self, y):
        globalScale = self.globalScale
        self.setGlobalScale(globalScale.x, y, globalScale.z)
    def setGlobalScaleZ(self, z):
        globalScale = self.globalScale
        self.setGlobalScale(globalScale.x, globalScale.y, z)
    # endregion

    # region transform
    @property
    def translationMatrix(self)->mat4:
        '''return the matrix that can transform a vector to local position'''
        return translate(self.localPos)
    @property
    def rotationMatrix(self)->mat4:
        '''return the matrix that can transform a vector to local rotation'''
        return mat4_cast(self._localRot)
    @property
    def scaleMatrix(self)->mat4:
        '''return the matrix that can transform a vector to local scale'''
        return scale(self.localScale)
    @property
    def transformMatrix(self)->mat4:
        '''return the matrix that can transform a vector to local space'''
        return self.translationMatrix * self.rotationMatrix * self.scaleMatrix
    @property
    def globalTransformMatrix(self)->mat4:
        '''return the matrix that can transform a vector to global space'''
        if self.parent is None:
            return self.transformMatrix
        return self.parent.globalTransformMatrix * self.transformMatrix
    @property
    def globalScaleMatrix(self)->mat4:
        '''return the matrix that can transform a vector to global scale'''
        if self.parent is None:
            return self.scaleMatrix
        return self.parent.globalScaleMatrix * self.scaleMatrix
    @property
    def inverseGlobalTransformMatrix(self)->mat4:
        '''return the matrix that can transform a vector from global space to local space'''
        return inverse(self.globalTransformMatrix)
    @property
    def inverseGlobalScaleMatrix(self)->mat4:
        '''return the matrix that can transform a vector from global scale to local scale'''
        return inverse(self.globalScaleMatrix)

    def transformPoint(self, localPos)->vec3:
        '''transform a point from local space to global space'''
        if self.parent is None:
            return localPos
        return (self.parent.globalTransformMatrix * vec4(localPos, 1)).xyz
    def inverseTransformPoint(self, globalPos)->vec3:
        '''transform a point from global space to local space'''
        if self.parent is None:
            return globalPos
        return (self.parent.inverseGlobalTransformMatrix * vec4(globalPos, 1)).xyz

    def transformDirection(self, localDir)->vec3:
        '''transform a direction from local space to global space (return a normalized vector)'''
        return normalize(self.globalTransformMatrix * vec4(localDir, 0)).xyz
    def inverseTransformDirection(self, globalDir)->vec3:
        '''transform a direction from global space to local space (return a normalized vector)'''
        return normalize(self.inverseGlobalTransformMatrix * vec4(globalDir, 0)).xyz
    # endregion
source/runtime/components/transform.py
92MING/Stable-Renderer
import glfw, time
from utils.global_utils import *
from static.scene import *
import numpy as np
np.set_printoptions(suppress=True)
from runtime.managers import *

_engine_singleton = GetOrAddGlobalValue("_ENGINE_SINGLETON", None)
class Engine:

    def __new__(cls, *args, **kwargs):
        global _engine_singleton
        if _engine_singleton is not None:
            clsName = _engine_singleton.__class__.__qualname__
            if clsName != cls.__qualname__:
                raise RuntimeError(f'Engine is running as {clsName}. As a singleton, it can not be created again.')
            _engine_singleton.__init__ = lambda *a, **kw: None  # 防止再次初始化
            return _engine_singleton
        else:
            e = super().__new__(cls)
            SetGlobalValue("_ENGINE_SINGLETON", e)
            _engine_singleton = e
            return e
    def __init__(self, scene: Scene = None, winTitle=None, winSize=(800, 480),):
        self._scene = scene
        if winTitle is not None:
            title = winTitle
        elif self._scene is not None:
            title = self._scene.name
        else:
            title = 'Stable Renderer'

        self._windowManager = WindowManager(title, winSize)
        self._inputManager = InputManager(self._windowManager.Window)
        self._runtimeManager = RuntimeManager()
        self._renderManager = RenderManager()
        self._sceneManager = SceneManager(self._scene)
        self._resourceManager = ResourcesManager()
        # endregion

    @property
    def WindowManager(self)->WindowManager:
        '''Window Manager. GLFW related.'''
        return self._windowManager
    @property
    def InputManager(self)->InputManager:
        '''Input info/ events...'''
        return self._inputManager
    @property
    def RuntimeManager(self)->RuntimeManager:
        '''Manager of GameObjs, Components, etc.'''
        return self._runtimeManager
    @property
    def RenderManager(self)->RenderManager:
        '''Manager of Renderers, Shaders, UBO, ... etc.'''
        return self._renderManager
    @property
    def SceneManager(self)->SceneManager:
        return self._sceneManager
    @property
    def ResourcesManager(self)->ResourcesManager:
        return self._resourceManager

    # region overridable methods
    def beforePrepare(self):...
    def afterPrepare(self):...
    def beforeFrameBegin(self):...
    def beforeFrameRun(self):...
    def beforeFrameEnd(self):...
    def beforeRelease(self):...
    def afterRelease(self):...
    # endregion

    def run(self):

        self.beforePrepare()
        Manager._RunPrepare()  # prepare work, mainly for sceneManager to build scene, load resources, etc.
        self.afterPrepare()

        while not glfw.window_should_close(self.WindowManager.Window):

            self.beforeFrameBegin()
            Manager._RunFrameBegin()  # input events / clear buffers / etc.

            self.beforeFrameRun()
            Manager._RunFrameRun() # update gameobjs, components, etc.

            self.beforeFrameEnd()
            Manager._RunFrameEnd() # render / swap buffers / time count / etc.

        self.beforeRelease()
        Manager._RunRelease()
        self.afterRelease()

    @classmethod
    def Run(cls):
        global _engine_singleton
        if _engine_singleton is None:
            _engine_singleton = cls()
        _engine_singleton.run()
source/runtime/engine.py
92MING/Stable-Renderer
from utils.global_utils import GetOrAddGlobalValue

class EngineObj:
    @property
    def engine(self)->'Engine':
        _engine = GetOrAddGlobalValue("_ENGINE_SINGLETON", None)
        if _engine is None:
            raise RuntimeError("Engine is not initialized yet.")
        return _engine
source/runtime/engineObj.py
92MING/Stable-Renderer
'''GameObject is the class contains components and behaviors.'''
from utils.base_clses import NamedObj
from utils.global_utils import GetOrAddGlobalValue
from runtime.engineObj import EngineObj
from typing import Union, List, Optional
import heapq
from runtime.component import Component, ComponentMeta
from runtime.components.transform import Transform

class AutoSortList(list):
    def append(self, obj):
        heapq.heappush(self, obj)
    def extend(self, objs):
        for obj in objs:
            self.append(obj)

_gameObj_tags = GetOrAddGlobalValue("_GAMEOBJ_TAGS", {}) # tag: set of gameObj
_root_gameObjs = GetOrAddGlobalValue("_ROOT_GAMEOBJS", []) # list of root gameObj(no parent)
class GameObject(EngineObj, NamedObj):

    # region class methods
    @staticmethod
    def RootObjs()->list:
        return _root_gameObjs
    @staticmethod
    def FindObjs_ByTag(tag)->set:
        return _gameObj_tags.get(tag, set())
    @classmethod
    def FindObj_ByName(cls, name)->'GameObject':
        '''Return None if not found'''
        return cls.GetInstance(name)
    # endregion

    def __init__(self, name, active=True, parent=None, tags:Union[str, list, tuple, set]=None, needTransform=True):
        super().__init__(name)

        self._active = active
        self._tags = set()
        self._parent = parent
        self._children = []
        self._components:List[Component] = AutoSortList()

        if parent is None:
            _root_gameObjs.append(self)
        else:
            self.parent.children.append(self) if self not in self.parent.children else None

        if tags is not None:
            if isinstance(tags, str):
                tags = [tags]
            for tag in tags:
                self.addTag(tag)

        if needTransform:
            self.addComponent(Transform)

    # region child & parent
    def hasChild(self, child):
        return child in self.children
    def getChild(self, index):
        return self.children[index]
    @property
    def children(self)->List['GameObject']:
        return self._children
    @property
    def childCount(self):
        return len(self.children)
    @property
    def parent(self):
        return self._parent
    @parent.setter
    def parent(self, new_parent):
        """
        Set parent object. If `new_parent` is None, this object will be moved to root object list.
        """
        if new_parent == self.parent:
            return
        if self.parent is not None:
            self.parent.children.remove(self)
        else:
            GameObject.RootObjs().remove(self)
        self._parent = new_parent
        if self.parent is not None:
            self.parent.children.append(self)
        else:
            GameObject.RootObjs().append(self)
    @property
    def siblingIndex(self):
        if self.parent is not None:
            return self.parent.children.index(self)
        else:
            return GameObject.RootObjs().index(self)
    @siblingIndex.setter
    def siblingIndex(self, new_index):
        if new_index == self.siblingIndex:
            return
        if self.parent is not None:
            self.parent.children.remove(self)
            self.parent.children.insert(new_index, self)
        else:
            GameObject.RootObjs().remove(self)
            GameObject.RootObjs().insert(new_index, self)
    # endregion

    # region active & destroy
    @property
    def active(self):
        active = self._active
        par = self.parent
        while par is not None and active:
            active = par.active
            par = par.parent
        return active
    @active.setter
    def active(self, value:bool):
        if self._active == value:
            return
        self._active = value
        for comp in self.allComponents(enableOnly=True):
            if not value:
                comp.onDisable()
            else:
                comp.onEnable()
    def destroy(self):
        if self.parent is not None:
            self.parent.children.remove(self)
        else:
            GameObject.root_game_objects.remove(self)
        for att_list in self.attributes.values():
            for att in att_list:
                att.destroy()
        for child in list(self.children):
            child.destroy()
    # endregion

    # region tags
    @property
    def tags(self):
        return self._tags
    def hasTag(self, tag):
        return tag in self.tags
    def addTag(self, tag):
        if self.hasTag(tag):
            return
        self.tags.add(tag)
        if tag not in _gameObj_tags:
            _gameObj_tags[tag] = set()
        _gameObj_tags[tag].add(self)
    def removeTag(self, tag):
        if not self.hasTag(tag):
            return
        self.tags.remove(tag)
        _gameObj_tags[tag].remove(self)
    # endregion

    # region components
    def components(self, enableOnly=True)->List[Component]:
        '''
        return components it has
        :param enableOnly: only return enabled components
        :return:
        '''
        if enableOnly:
            return [comp for comp in self._components if comp.enable]
        else:
            return self._components
    def allComponents(self, enableOnly=True)->List[Component]:
        '''
        return all components it and its children have
        :param enableOnly: only return enabled components
        '''
        result = AutoSortList()
        for child in self.children:
            result.extend(child.childComponents(enableOnly))
        result.extend(self.components(enableOnly))
        return result
    def hasComponent(self, comp: Union[ComponentMeta, str, Component], enableOnly=False):
        '''
        whether it has component of type `comp`. If there exist a subclass of `comp`, also return True
        :param comp: Subclass of Component / component instance / component name
        :param enableOnly: whether only search enabled components
        :return: bool
        '''
        if isinstance(comp, ComponentMeta):
            return any(isinstance(c, comp) for c in self.components(enableOnly=enableOnly))
        elif isinstance(comp, str):
            return any(c.ComponentName == comp for c in self.components(enableOnly=enableOnly))
        elif isinstance(comp, Component):
            return comp in self.components(enableOnly=enableOnly)
        else:
            raise TypeError("comp must be a subclass of Component, or a component instance, or component name")
    def getComponent(self, comp: Union[ComponentMeta, str, Component], enableOnly=False)->Optional[Component]:
        '''
        return the first component of type `comp`
        :param comp: Subclass of Component / component instance / component name
        :param enableOnly: whether only search enabled components
        :return: component or None if not found
        '''
        if not isinstance(comp, (ComponentMeta, str, Component)):
            raise TypeError("comp must be a subclass of Component, or a component instance, or component name")
        if isinstance(comp, str):
            for c in self.components(enableOnly=enableOnly):
                if c.ComponentName == comp:
                    return c
        elif isinstance(comp, Component):
            for c in self.components(enableOnly=enableOnly):
                if c == comp:
                    return c
        else:
            for c in self.components(enableOnly=enableOnly):
                if isinstance(c, comp):
                    return c
        return None
    def getComponents(self, comp: Union[ComponentMeta, str, Component], enableOnly=False):
        '''
        return all components of type `comp`
        :param comp: Subclass of Component / component instance / component name
        :param enableOnly: whether only search enabled components
        :return: list of components
        '''
        if not isinstance(comp, (ComponentMeta, str, Component)):
            raise TypeError("comp must be a subclass of Component, or a component instance, or component name")
        if isinstance(comp, ComponentMeta):
            return [c for c in self.components(enableOnly=enableOnly) if isinstance(c, comp)]
        elif isinstance(comp, str):
            return [c for c in self.components(enableOnly=enableOnly) if c.ComponentName == comp]
        elif isinstance(comp, Component):
            return [c for c in self.components(enableOnly=enableOnly) if c == comp]
    def removeComponent(self, comp: Union[ComponentMeta, str, Component], enableOnly=False):
        '''
        remove the first component of type `comp`
        :param comp: Subclass of Component / component instance / component name
        '''
        if not isinstance(comp, (ComponentMeta, str, Component)):
            raise TypeError("comp must be a subclass of Component, or a component instance, or component name")
        if isinstance(comp, str):
            for c in self.components(enableOnly=enableOnly):
                if c.ComponentName == comp:
                    c.onDisable()
                    c.onDestory()
                    return
        elif isinstance(comp, Component):
            for c in self.components(enableOnly=enableOnly):
                if c == comp:
                    c.onDisable()
                    c.onDestory()
                    return
        else:
            for c in self.components(enableOnly=enableOnly):
                if isinstance(c, comp):
                    c.onDisable() # call onDisable before destroy
                    c.onDestory()
                    return
    def removeComponents(self, comp: Union[ComponentMeta, str, Component], enableOnly=False):
        '''
        remove all components of type `comp`
        :param comp: Subclass of Component / component instance / component name
        '''
        if not isinstance(comp, (ComponentMeta, str, Component)):
            raise TypeError("comp must be a subclass of Component")
        if isinstance(comp, str):
            for c in self.components(enableOnly=enableOnly):
                if c.ComponentName == comp:
                    c.onDisable()
                    c.onDestory()
        elif isinstance(comp, Component):
            for c in self.components(enableOnly=enableOnly):
                if c == comp:
                    c.onDisable()
                    c.onDestory()
        else:
            for c in self.components(enableOnly=enableOnly):
                if isinstance(c, comp):
                    c.onDisable() # call onDisable before destroy
                    c.onDestory()
    def addComponent(self, comp: Union[ComponentMeta, str, Component], enable=True, *args, **kwargs)->Optional[Component]:
        '''
        add a component of type `comp`
        :param comp: Subclass of Component / component instance / component name
        :param enable: whether enable the component
        :param args: args for the constructor of `comp`
        :param kwargs: kwargs for the constructor of `comp`
        :return: the component added
        '''
        if not isinstance(comp, (ComponentMeta, str, Component)):
            raise TypeError("comp must be a subclass of Component")
        if isinstance(comp, (ComponentMeta, str)):
            comp = Component.FindComponentCls(comp) if isinstance(comp, str) else comp
            if comp.Unique and self.hasComponent(comp):
                return None # already has a unique component
            for require in comp.RequireComponent:
                if not self.hasComponent(require):
                    newComp = self.addComponent(require, enable=enable, )
                    if newComp is None:
                        raise RuntimeError(f"GameObject: {self.name} has no component: {require}. Auto add failed")
            c = comp(self, enable, *args, **kwargs)
            self._components.append(c)
            c._tryAwake() # call awake if the gameobject is active and the component is enabled
            return c
        else: # comp is a component instance
            if comp.gameObj != self:
                raise RuntimeError(f"Component: {comp} is not belong to GameObject: {self.name}. Can't add")
            if comp.Unique and self.hasComponent(comp):
                return None
            for require in comp.RequireComponent:
                if not self.hasComponent(require):
                    newComp = self.addComponent(require, enable=enable, )
                    if newComp is None:
                        raise RuntimeError(f"GameObject: {self.name} has no component: {require}. Auto add failed")
            self._components.append(comp)
            comp._tryAwake()  # call awake if the gameobject is active and the component is enabled
    @property
    def transform(self) -> Transform:
        t = self.getComponent(Transform)
        if t is None:
            raise RuntimeError(f"GameObject: {self.name} has no Transform")
        return t
    # endregion

    # region internal use
    def _fixedUpdate(self):
        '''internal use for fixedUpdate and components'''
        comps = self.components(enableOnly=True)
        for comp in comps:
            comp.fixedUpdate()
        for child in self.children:
            child._fixedUpdate()
    def _lateUpdate(self):
        '''internal use for lateUpdate and components'''
        comps = self.components(enableOnly=True)
        for comp in comps:
            comp.lateUpdate()
        for child in self.children:
            child._lateUpdate()
    def _update(self):
        '''internal use for update and components'''
        comps = self.components(enableOnly=True)
        for comp in comps:
            comp.update()
        for child in self.children:
            child._update()
    @staticmethod
    def _RunFixedUpdate():
        '''internal use for fixedUpdate and components'''
        for obj in _root_gameObjs:
            obj._fixedUpdate()
    @staticmethod
    def _RunLateUpdate():
        '''internal use for lateUpdate and components'''
        for obj in _root_gameObjs:
            obj._lateUpdate()
    @staticmethod
    def _RunUpdate():
        '''internal use for update and components'''
        for obj in _root_gameObjs:
            obj._update()
    # endregion
source/runtime/gameObj.py
92MING/Stable-Renderer
'''
Managers are used to manage the different parts of the engine.
All of them are singletons.
'''

from .manager import *
from .windowManager import *
from .inputManager import *
from .runtimeManager import *
from .renderManager import *
from .sceneManager import *
from .resourcesManager import *
source/runtime/managers/__init__.py
92MING/Stable-Renderer
from .manager import Manager
from static.enums import *
import glfw

class InputManager(Manager):
    def __init__(self, window):
        super().__init__()
        self._mousePos = (0, 0)  # (x, y)
        self._mouseDelta = (0, 0)  # (dx, dy)
        self._mouseScroll = (0, 0)  # (dx, dy)
        self._mouseKeys = {}  # key: (action, modifier)
        self._keyboardKeys = {}  # key: (action, modifier)
        self._glfw_init(window)

    def _cursorCallback(self, window, x, y):
        # for glfw callback
        self._mouseDelta = (x - self._mousePos[0], y - self._mousePos[1])
        self._mousePos = (x, y)

    def _scrollCallback(self, window, x, y):
        # for glfw callback
        self._mouseScroll = (x, y)

    def _mouseButtonCallback(self, window, btn, action, mod):
        # for glfw callback
        btn = MouseButton.GetEnum(btn)
        action = InputAction.GetEnum(action)
        mod = InputModifier.GetEnum(mod)
        if btn not in self._mouseKeys:
            self._mouseKeys[btn] = [None, None]
        current = self._mouseKeys[btn]
        if action == InputAction.PRESS:
            if current[0] != InputAction.HOLD:
                current = [action, mod]
        elif action == InputAction.RELEASE:
            current = [action, mod]
        self._mouseKeys[btn] = current

    def _keyCallback(self, window, key, scancode, action, mod):
        # for glfw callback
        key = Key.GetEnum(key)
        if key not in self._keyboardKeys:
            self._keyboardKeys[key] = [None, None]  # (action, modifier)
        action = InputAction.GetEnum(action)
        mod = InputModifier.GetEnum(mod)
        current = self._keyboardKeys[key]
        if action == InputAction.PRESS:
            if current[0] != InputAction.HOLD:
                current = [action, mod]
        elif action == InputAction.RELEASE:
            current = [action, mod]
        self._keyboardKeys[key] = current

    def _glfw_init(self, window):
        glfw.set_cursor_pos_callback(window, self._cursorCallback)
        glfw.set_scroll_callback(window, self._scrollCallback)
        glfw.set_mouse_button_callback(window, self._mouseButtonCallback)
        glfw.set_key_callback(window, self._keyCallback)

    def _onFrameBegin(self):
        glfw.poll_events()  # update input info

    def _onFrameEnd(self):
        self._mouseDelta = (0, 0)
        self._mouseScroll = (0, 0)
        for key in self._mouseKeys:
            cur = self._mouseKeys[key]
            if cur[0] == InputAction.RELEASE:
                self._mouseKeys.pop(key)
            elif cur[0] == InputAction.PRESS:
                self._mouseKeys[key][0] = InputAction.HOLD
        for key in self._keyboardKeys:
            cur = self._keyboardKeys[key]
            if cur[0] == InputAction.RELEASE:
                self._keyboardKeys.pop(key)
            elif cur[0] == InputAction.PRESS:
                self._keyboardKeys[key][0] = InputAction.HOLD

    @property
    def mousePos(self):
        return self._mousePos

    @property
    def mouseDelta(self):
        return self._mouseDelta

    @property
    def mouseScroll(self):
        return self._mouseScroll

    def GetMouseBtn(self, btn: MouseButton, mod: InputModifier = None):
        '''Check if mouse btn is pressed. If mod is not None, any modifier will be accepted.'''
        if btn not in self._mouseKeys:
            return False
        else:
            cur = self._mouseKeys[btn]
            if mod is None:
                return cur[0] != InputAction.RELEASE
            else:
                return cur[0] != InputAction.RELEASE and cur[1] == mod

    def GetMouseBtnDown(self, btn: MouseButton, mod: InputModifier = None):
        '''Check if mouse btn is pressed down. If mod is not None, any modifier will be accepted.'''
        if btn not in self._mouseKeys:
            return False
        else:
            cur = self._mouseKeys[btn]
            if mod is None:
                return cur[0] == InputAction.PRESS
            else:
                return cur[0] == InputAction.PRESS and cur[1] == mod

    def GetMouseBtnUp(self, btn: MouseButton):
        '''Check if mouse btn is released.'''
        if btn not in self._mouseKeys:
            return False
        else:
            return self._mouseKeys[btn][0] == InputAction.RELEASE

    def GetMouseBtnHold(self, btn: MouseButton, mod: InputModifier = None):
        '''Check if mouse btn is hold. If mod is not None, any modifier will be accepted.'''
        if btn not in self._mouseKeys:
            return False
        else:
            cur = self._mouseKeys[btn]
            if mod is None:
                return cur[0] == InputAction.HOLD
            else:
                return cur[0] == InputAction.HOLD and cur[1] == mod

    def GetKey(self, key: Key, mod: InputModifier = None):
        '''Check if key is pressed. If mod is not None, any modifier will be accepted.'''
        if key not in self._keyboardKeys:
            return False
        else:
            cur = self._keyboardKeys[key]
            if mod is None:
                return cur[0] != InputAction.RELEASE
            else:
                return cur[0] != InputAction.RELEASE and cur[1] == mod

    def GetKeyDown(self, key: Key, mod: InputModifier = None):
        '''Check if key is pressed down. If mod is not None, any modifier will be accepted.'''
        if key not in self._keyboardKeys:
            return False
        else:
            cur = self._keyboardKeys[key]
            if mod is None:
                return cur[0] == InputAction.PRESS
            else:
                return cur[0] == InputAction.PRESS and cur[1] == mod

    def GetKeyUp(self, key: Key):
        '''Check if key is released.'''
        if key not in self._keyboardKeys:
            return False
        else:
            return self._keyboardKeys[key][0] == InputAction.RELEASE

    def GetKeyHold(self, key: Key, mod: InputModifier = None):
        '''Check if key is hold. If mod is not None, any modifier will be accepted.'''
        if key not in self._keyboardKeys:
            return False
        else:
            cur = self._keyboardKeys[key]
            if mod is None:
                return cur[0] == InputAction.HOLD
            else:
                return cur[0] == InputAction.HOLD and cur[1] == mod


__all__ = ['InputManager']
source/runtime/managers/inputManager.py
92MING/Stable-Renderer
from utils.global_utils import GetOrAddGlobalValue, GetGlobalValue
import time

_MANAGERS = GetOrAddGlobalValue('_ENGINE_MANAGERS', set())
_MANAGER_FUNCS = GetOrAddGlobalValue('_ENGINE_MANAGER_FUNCS', dict())
class Manager:

    _PrepareFuncOrder = 0
    _ReleaseFuncOrder = 0
    _FrameBeginFuncOrder = 0
    _FrameRunFuncOrder = 0
    _FrameEndFuncOrder = 0

    def __new__(cls, *args, **kwargs):
        for manager in _MANAGERS:
            if manager.__class__.__qualname__ == cls.__qualname__:
                manager.__init__ = lambda *a, **kw: None
                return manager
        obj = super().__new__(cls)
        _MANAGERS.add(obj)
        return obj
    def __init__(self):
        print('Initializing manager: ', self.__class__.__qualname__)

    def _onFrameBegin(self):
        '''The first function will be called in each frame.'''
        pass
    def _onFrameRun(self):
        '''Will be called after "_onFrameBegin"'''
        pass
    def _onFrameEnd(self):
        '''Will be called after "_onFrameRun"'''
        pass
    def _prepare(self):
        '''Prepare will be call before the loop begins.'''
        pass
    def _release(self):
        '''Release will be called when the program is going to exit.'''
        pass

    @property
    def engine(self)->'Engine':
        return GetGlobalValue('_ENGINE_SINGLETON')

    @staticmethod
    def _RunPrepare():
        if 'prepare' not in _MANAGER_FUNCS:
            _MANAGER_FUNCS['prepare'] = sorted(_MANAGERS, key=lambda m: m._PrepareFuncOrder)
        for manager in _MANAGER_FUNCS['prepare']:
            try:
                manager._prepare()
            except Exception as e:
                print(f'Warning: Error when running "_prepare" of {manager.__class__.__qualname__}. Err msg: {e}')
    @staticmethod
    def _RunFrameBegin():
        if 'begin' not in _MANAGER_FUNCS:
            _MANAGER_FUNCS['begin'] = sorted(_MANAGERS, key=lambda m: m._FrameBeginFuncOrder)
        for manager in _MANAGER_FUNCS['begin']:
            try:
                manager._onFrameBegin()
            except Exception as e:
                print(f'Warning: Error when running "_onFrameBegin" of {manager.__class__.__qualname__}. Err msg: {e}')
    @staticmethod
    def _RunFrameRun():
        if 'run' not in _MANAGER_FUNCS:
            _MANAGER_FUNCS['run'] = sorted(_MANAGERS, key=lambda m: m._FrameRunFuncOrder)
        for manager in _MANAGER_FUNCS['run']:
            try:
                manager._onFrameRun()
            except Exception as e:
                print(f'Warning: Error when running "_onFrameRun" of {manager.__class__.__qualname__}. Err msg: {e}')
    @staticmethod
    def _RunFrameEnd():
        if 'end' not in _MANAGER_FUNCS:
            _MANAGER_FUNCS['end'] = sorted(_MANAGERS, key=lambda m: m._FrameEndFuncOrder)
        for manager in _MANAGER_FUNCS['end']:
            try:
                manager._onFrameEnd()
            except Exception as e:
                print(f'Warning: Error when running "_onFrameEnd" of {manager.__class__.__qualname__}.Err msg: {e}')
    @staticmethod
    def _RunRelease():
        if 'release' not in _MANAGER_FUNCS:
            _MANAGER_FUNCS['release'] = sorted(_MANAGERS, key=lambda m: m._ReleaseFuncOrder)
        for manager in _MANAGERS:
            try:
                manager._release()
            except Exception as e:
                print(f'Warning: Error when running "_release" of {manager.__class__.__qualname__}. Err msg: {e}')

__all__ = ['Manager']
source/runtime/managers/manager.py
92MING/Stable-Renderer
from .manager import Manager
from .runtimeManager import RuntimeManager
from utils.data_struct.event import AutoSortTask
import glm
import OpenGL.GL as gl
import OpenGL.GLU as glu

class RenderManager(Manager):
    '''Manager of all rendering stuffs'''

    _FrameRunFuncOrder = RuntimeManager._FrameRunFuncOrder + 1  # always run after runtimeManager

    def __init__(self):
        super().__init__()
        self._renderTasks = AutoSortTask()
        self._init_opengl()
    def _init_opengl(self):
        gl.glClearColor(0, 0, 0, 0)
        gl.glEnable(gl.GL_DEPTH_TEST)
        gl.glEnable(gl.GL_CULL_FACE)

        self._UBO_modelMatrix = glm.mat4(1.0)
        self._UBO_viewMatrix = glm.mat4(1.0)
        self._UBO_projectionMatrix = glm.mat4(1.0)
        self._UBO_MVP = self._UBO_projectionMatrix * self._UBO_viewMatrix * self._UBO_modelMatrix

        self._matrixUBO = gl.glGenBuffers(1)
        gl.glBindBuffer(gl.GL_UNIFORM_BUFFER, self._matrixUBO)
        gl.glBufferData(gl.GL_UNIFORM_BUFFER, 4 * glm.sizeof(glm.mat4), None, gl.GL_DYNAMIC_DRAW)
        gl.glBindBufferBase(gl.GL_UNIFORM_BUFFER, self.MatrixUBO_BindingPoint, self._matrixUBO)
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, 0, glm.sizeof(glm.mat4), glm.value_ptr(self._UBO_modelMatrix))
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, glm.sizeof(glm.mat4), glm.sizeof(glm.mat4),
                        glm.value_ptr(self._UBO_viewMatrix))
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, 2 * glm.sizeof(glm.mat4), glm.sizeof(glm.mat4),
                        glm.value_ptr(self._UBO_projectionMatrix))
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, 3 * glm.sizeof(glm.mat4), glm.sizeof(glm.mat4),
                        glm.value_ptr(self._UBO_MVP))  # MVP

    # region UBO
    @property
    def MatrixUBO(self):
        return self._matrixUBO
    @property
    def MatrixUBO_BindingPoint(self):
        return 0
    @property
    def UBO_ModelMatrix(self):
        return self._UBO_modelMatrix
    @property
    def UBO_ViewMatrix(self):
        return self._UBO_viewMatrix
    @property
    def UBO_ProjMatrix(self):
        return self._UBO_projectionMatrix
    def UpdateUBO_ModelMatrix(self, modelMatrix: glm.mat4):
        self._UBO_modelMatrix = modelMatrix
        self._UBO_MVP = self._UBO_projectionMatrix * self._UBO_viewMatrix * self._UBO_modelMatrix
        gl.glBindBuffer(gl.GL_UNIFORM_BUFFER, self.MatrixUBO)
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, 0, glm.sizeof(glm.mat4), glm.value_ptr(self._UBO_modelMatrix))
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, 3 * glm.sizeof(glm.mat4), glm.sizeof(glm.mat4), glm.value_ptr(self._UBO_MVP))
    def UpdateUBO_ViewMatrix(self, viewMatrix: glm.mat4):
        self._UBO_viewMatrix = viewMatrix
        self._UBO_MVP = self._UBO_projectionMatrix * self._UBO_viewMatrix * self._UBO_modelMatrix
        gl.glBindBuffer(gl.GL_UNIFORM_BUFFER, self.MatrixUBO)
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, glm.sizeof(glm.mat4), glm.sizeof(glm.mat4),  glm.value_ptr(self._UBO_viewMatrix))
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, 3 * glm.sizeof(glm.mat4), glm.sizeof(glm.mat4), glm.value_ptr(self._UBO_MVP))
    def UpdateUBO_ProjMatrix(self, projectionMatrix: glm.mat4):
        self._UBO_projectionMatrix = projectionMatrix
        self._UBO_MVP = self._UBO_projectionMatrix * self._UBO_viewMatrix * self._UBO_modelMatrix
        gl.glBindBuffer(gl.GL_UNIFORM_BUFFER, self.MatrixUBO)
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, 2 * glm.sizeof(glm.mat4), glm.sizeof(glm.mat4),  glm.value_ptr(self._UBO_projectionMatrix))
        gl.glBufferSubData(gl.GL_UNIFORM_BUFFER, 3 * glm.sizeof(glm.mat4), glm.sizeof(glm.mat4), glm.value_ptr(self._UBO_MVP))
    def printOpenGLError(self):
        try:
            gl.glGetError() # nothing to do with error, just clear error flag
        except Exception as e:
            print('GL ERROR: ', e)
    # endregion

    @property
    def RenderTasks(self):
        return self._renderTasks

    def _onFrameBegin(self):
        gl.glClear(gl.GL_COLOR_BUFFER_BIT | gl.GL_DEPTH_BUFFER_BIT)
    def _onFrameRun(self):
        self._renderTasks.execute()


__all__ = ['RenderManager']
source/runtime/managers/renderManager.py
92MING/Stable-Renderer
from .manager import Manager
from .sceneManager import SceneManager
from utils.global_utils import GetOrAddGlobalValue

class ResourcesManager(Manager):

    _PrepareFuncOrder = SceneManager._PrepareFuncOrder + 1  # after SceneManager (after all gameobj/components are created)
    _ReleaseFuncOrder = -1
    def _prepare(self):
        resources_clses = GetOrAddGlobalValue('_RESOURCES_CLSES', dict()).values()
        base_clses = [cls for cls in resources_clses if cls._BaseName == cls.ClsName()]
        for cls in sorted(base_clses, key=lambda cls: cls._LoadOrder):
            for instance in cls.AllInstances():
                try:
                    instance.sendToGPU()
                except Exception:
                    raise Exception(f'Error when sending {instance.__class__.__qualname__}:{instance.name} to GPU')

    def _release(self):
        resources_clses = GetOrAddGlobalValue('_RESOURCES_CLSES', dict()).values()
        base_clses = [cls for cls in resources_clses if cls._BaseName == cls.ClsName()]
        for cls in sorted(base_clses, key=lambda cls: cls._LoadOrder, reverse=True):
            for instance in cls.AllInstances():
                try:
                    instance.clear()
                except Exception:
                    raise Exception(f'Error when clearing {instance.__class__.__qualname__}:{instance.name}')
source/runtime/managers/resourcesManager.py
92MING/Stable-Renderer
from .manager import Manager
from runtime.gameObj import GameObject
import glfw

class RuntimeManager(Manager):
    '''Manager of GameObjs, Components, etc.'''

    _FrameEndFuncOrder = 999  # run at the end of each frame, since it need to count the frame time

    def __init__(self, fixedUpdateMaxFPS=60):
        super().__init__()
        self._fixedUpdateMaxFPS = fixedUpdateMaxFPS
        self._minFixedUpdateDeltaTime = 1.0 / fixedUpdateMaxFPS
        self._deltaTime = 0.0
        self._startTime = 0.0
        self._firstFrame = True

    @property
    def FixedUpdateMaxFPS(self):
        return self._fixedUpdateMaxFPS
    @FixedUpdateMaxFPS.setter
    def FixedUpdateMaxFPS(self, value):
        self._fixedUpdateMaxFPS = value
        self._minFixedUpdateDeltaTime = 1.0 / value
    @property
    def DeltaTime(self):
        return self._deltaTime

    def _onFrameBegin(self):
        self._startTime = glfw.get_time()
    def _onFrameRun(self):
        if self._firstFrame or self.DeltaTime >= self._minFixedUpdateDeltaTime:
            GameObject._RunFixedUpdate()
        GameObject._RunUpdate()
        GameObject._RunLateUpdate()
        if self.DeltaTime >= self._minFixedUpdateDeltaTime:
            self._deltaTime = 0.0
        self._firstFrame = False
    def _onFrameEnd(self):
        self._deltaTime += (glfw.get_time() - self._startTime)

__all__ = ['RuntimeManager']
source/runtime/managers/runtimeManager.py
92MING/Stable-Renderer
from .manager import Manager

class SceneManager(Manager):
    def __init__(self, mainScene=None, scenes=()):
        super().__init__()
        self._mainScene = mainScene
        self._scenes = list(scenes)
        if mainScene is not None and mainScene not in self._scenes:
            self._scenes.append(mainScene)
        if mainScene is None and len(self._scenes)!=0:
            self._mainScene = self._scenes[0] # the first scene is the main scene if mainScene is not specified

    @property
    def MainScene(self):
        return self._mainScene
    def SwitchToScene(self, scene):
        if scene == self.MainScene:
            return
        # TODO: switch to scene
        pass

    def _prepare(self):
        if self.MainScene is not None:
            self.MainScene.prepare()

__all__ = ['SceneManager']
source/runtime/managers/sceneManager.py
92MING/Stable-Renderer
from .manager import Manager
from .renderManager import RenderManager
from utils.data_struct import DelayEvent, Event
import glfw
import OpenGL.GL as gl

class WindowManager(Manager):

    _FrameEndFuncOrder = RenderManager._FrameEndFuncOrder + 1 # swap buffer should be called after render
    _ReleaseFuncOrder = 999 # terminate glfw should be called at the end

    def __init__(self, title, size):
        super().__init__()
        self._init_glfw(title, size)
        self._onWindowResize = DelayEvent(int, int)
        self._onWindowResize.addListener(lambda width, height: gl.glViewport(0, 0, width, height))

    def _init_glfw(self, winTitle, winSize):
        self._title = winTitle
        self._size = winSize
        glfw.init()
        glfw.window_hint(glfw.CONTEXT_VERSION_MAJOR, 3)
        glfw.window_hint(glfw.CONTEXT_VERSION_MINOR, 3)
        glfw.window_hint(glfw.OPENGL_PROFILE, glfw.OPENGL_CORE_PROFILE)
        glfw.window_hint(glfw.OPENGL_FORWARD_COMPAT, gl.GL_TRUE)
        glfw.window_hint(glfw.RESIZABLE, gl.GL_TRUE)
        glfw.window_hint(glfw.DOUBLEBUFFER, gl.GL_TRUE)
        glfw.window_hint(glfw.DEPTH_BITS, 24)
        glfw.window_hint(glfw.SAMPLES, 4)  # MSAA
        glfw.window_hint(glfw.STENCIL_BITS, 8)
        self._window = glfw.create_window(winSize[0], winSize[1], winTitle, None, None)
        if not self._window:
            glfw.terminate()
            exit()
        glfw.make_context_current(self._window)
        glfw.set_window_size_callback(self._window, self._resizeCallback)
    def _resizeCallback(self, window, width, height):
        self._onWindowResize.invoke(width, height)

    def _onFrameBegin(self):
        self._onWindowResize.release()
    def _onFrameEnd(self):
        glfw.swap_buffers(self._window)
    def _release(self):
        glfw.terminate()

    @property
    def OnWindowResize(self)->Event:
        return self._onWindowResize
    @property
    def Window(self):
        return self._window
    @property
    def Title(self):
        return self._title
    @Title.setter
    def Title(self, value):
        self.SetWindowTitle(value)
    def SetWindowTitle(self, title):
        self._title = title
        glfw.set_window_title(self._window, title)
    @property
    def WindowSize(self):
        return self._size
    @WindowSize.setter
    def WindowSize(self, value):
        self.SetWindowSize(*value)
    def SetWindowSize(self, width, height):
        self._size = (width, height)
        glfw.set_window_size(self._window, *self._size)
    @property
    def AspectRatio(self):
        return self._size[0] / self._size[1]
__all__ = ['WindowManager']
source/runtime/managers/windowManager.py
92MING/Stable-Renderer
import inspect
import re
from typing import Any, Callable, Dict, List, Optional, Union

import numpy as np
import PIL
import torch
from packaging import version
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.configuration_utils import FrozenDict
from diffusers.image_processor import VaeImageProcessor
from diffusers.loaders import FromSingleFileMixin, LoraLoaderMixin, TextualInversionLoaderMixin
from diffusers.models import AutoencoderKL, UNet2DConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput, StableDiffusionSafetyChecker
from diffusers.schedulers import KarrasDiffusionSchedulers
from diffusers.utils import (
    PIL_INTERPOLATION,
    deprecate,
    is_accelerate_available,
    is_accelerate_version,
    logging,
)
from diffusers.utils.torch_utils import randn_tensor


# ------------------------------------------------------------------------------

logger = logging.get_logger(__name__)  # pylint: disable=invalid-name

re_attention = re.compile(
    r"""
\\\(|
\\\)|
\\\[|
\\]|
\\\\|
\\|
\(|
\[|
:([+-]?[.\d]+)\)|
\)|
]|
[^\\()\[\]:]+|
:
""",
    re.X,
)


def parse_prompt_attention(text):
    """
    Parses a string with attention tokens and returns a list of pairs: text and its associated weight.
    Accepted tokens are:
      (abc) - increases attention to abc by a multiplier of 1.1
      (abc:3.12) - increases attention to abc by a multiplier of 3.12
      [abc] - decreases attention to abc by a multiplier of 1.1
      \( - literal character '('
      \[ - literal character '['
      \) - literal character ')'
      \] - literal character ']'
      \\ - literal character '\'
      anything else - just text
    >>> parse_prompt_attention('normal text')
    [['normal text', 1.0]]
    >>> parse_prompt_attention('an (important) word')
    [['an ', 1.0], ['important', 1.1], [' word', 1.0]]
    >>> parse_prompt_attention('(unbalanced')
    [['unbalanced', 1.1]]
    >>> parse_prompt_attention('\(literal\]')
    [['(literal]', 1.0]]
    >>> parse_prompt_attention('(unnecessary)(parens)')
    [['unnecessaryparens', 1.1]]
    >>> parse_prompt_attention('a (((house:1.3)) [on] a (hill:0.5), sun, (((sky))).')
    [['a ', 1.0],
     ['house', 1.5730000000000004],
     [' ', 1.1],
     ['on', 1.0],
     [' a ', 1.1],
     ['hill', 0.55],
     [', sun, ', 1.1],
     ['sky', 1.4641000000000006],
     ['.', 1.1]]
    """

    res = []
    round_brackets = []
    square_brackets = []

    round_bracket_multiplier = 1.1
    square_bracket_multiplier = 1 / 1.1

    def multiply_range(start_position, multiplier):
        for p in range(start_position, len(res)):
            res[p][1] *= multiplier

    for m in re_attention.finditer(text):
        text = m.group(0)
        weight = m.group(1)

        if text.startswith("\\"):
            res.append([text[1:], 1.0])
        elif text == "(":
            round_brackets.append(len(res))
        elif text == "[":
            square_brackets.append(len(res))
        elif weight is not None and len(round_brackets) > 0:
            multiply_range(round_brackets.pop(), float(weight))
        elif text == ")" and len(round_brackets) > 0:
            multiply_range(round_brackets.pop(), round_bracket_multiplier)
        elif text == "]" and len(square_brackets) > 0:
            multiply_range(square_brackets.pop(), square_bracket_multiplier)
        else:
            res.append([text, 1.0])

    for pos in round_brackets:
        multiply_range(pos, round_bracket_multiplier)

    for pos in square_brackets:
        multiply_range(pos, square_bracket_multiplier)

    if len(res) == 0:
        res = [["", 1.0]]

    # merge runs of identical weights
    i = 0
    while i + 1 < len(res):
        if res[i][1] == res[i + 1][1]:
            res[i][0] += res[i + 1][0]
            res.pop(i + 1)
        else:
            i += 1

    return res


def get_prompts_with_weights(pipe: DiffusionPipeline, prompt: List[str], max_length: int):
    r"""
    Tokenize a list of prompts and return its tokens with weights of each token.

    No padding, starting or ending token is included.
    """
    tokens = []
    weights = []
    truncated = False
    for text in prompt:
        texts_and_weights = parse_prompt_attention(text)
        text_token = []
        text_weight = []
        for word, weight in texts_and_weights:
            # tokenize and discard the starting and the ending token
            token = pipe.tokenizer(word).input_ids[1:-1]
            text_token += token
            # copy the weight by length of token
            text_weight += [weight] * len(token)
            # stop if the text is too long (longer than truncation limit)
            if len(text_token) > max_length:
                truncated = True
                break
        # truncate
        if len(text_token) > max_length:
            truncated = True
            text_token = text_token[:max_length]
            text_weight = text_weight[:max_length]
        tokens.append(text_token)
        weights.append(text_weight)
    if truncated:
        logger.warning("Prompt was truncated. Try to shorten the prompt or increase max_embeddings_multiples")
    return tokens, weights


def pad_tokens_and_weights(tokens, weights, max_length, bos, eos, pad, no_boseos_middle=True, chunk_length=77):
    r"""
    Pad the tokens (with starting and ending tokens) and weights (with 1.0) to max_length.
    """
    max_embeddings_multiples = (max_length - 2) // (chunk_length - 2)
    weights_length = max_length if no_boseos_middle else max_embeddings_multiples * chunk_length
    for i in range(len(tokens)):
        tokens[i] = [bos] + tokens[i] + [pad] * (max_length - 1 - len(tokens[i]) - 1) + [eos]
        if no_boseos_middle:
            weights[i] = [1.0] + weights[i] + [1.0] * (max_length - 1 - len(weights[i]))
        else:
            w = []
            if len(weights[i]) == 0:
                w = [1.0] * weights_length
            else:
                for j in range(max_embeddings_multiples):
                    w.append(1.0)  # weight for starting token in this chunk
                    w += weights[i][j * (chunk_length - 2): min(len(weights[i]), (j + 1) * (chunk_length - 2))]
                    w.append(1.0)  # weight for ending token in this chunk
                w += [1.0] * (weights_length - len(w))
            weights[i] = w[:]

    return tokens, weights


def get_unweighted_text_embeddings(
    pipe: DiffusionPipeline,
    text_input: torch.Tensor,
    chunk_length: int,
    no_boseos_middle: Optional[bool] = True,
):
    """
    When the length of tokens is a multiple of the capacity of the text encoder,
    it should be split into chunks and sent to the text encoder individually.
    """
    max_embeddings_multiples = (text_input.shape[1] - 2) // (chunk_length - 2)
    if max_embeddings_multiples > 1:
        text_embeddings = []
        for i in range(max_embeddings_multiples):
            # extract the i-th chunk
            text_input_chunk = text_input[:, i * (chunk_length - 2): (i + 1) * (chunk_length - 2) + 2].clone()

            # cover the head and the tail by the starting and the ending tokens
            text_input_chunk[:, 0] = text_input[0, 0]
            text_input_chunk[:, -1] = text_input[0, -1]
            text_embedding = pipe.text_encoder(text_input_chunk)[0]

            if no_boseos_middle:
                if i == 0:
                    # discard the ending token
                    text_embedding = text_embedding[:, :-1]
                elif i == max_embeddings_multiples - 1:
                    # discard the starting token
                    text_embedding = text_embedding[:, 1:]
                else:
                    # discard both starting and ending tokens
                    text_embedding = text_embedding[:, 1:-1]

            text_embeddings.append(text_embedding)
        text_embeddings = torch.concat(text_embeddings, axis=1)
    else:
        text_embeddings = pipe.text_encoder(text_input)[0]
    return text_embeddings


def get_weighted_text_embeddings(
    pipe: DiffusionPipeline,
    prompt: Union[str, List[str]],
    uncond_prompt: Optional[Union[str, List[str]]] = None,
    max_embeddings_multiples: Optional[int] = 3,
    no_boseos_middle: Optional[bool] = False,
    skip_parsing: Optional[bool] = False,
    skip_weighting: Optional[bool] = False,
):
    r"""
    Prompts can be assigned with local weights using brackets. For example,
    prompt 'A (very beautiful) masterpiece' highlights the words 'very beautiful',
    and the embedding tokens corresponding to the words get multiplied by a constant, 1.1.

    Also, to regularize of the embedding, the weighted embedding would be scaled to preserve the original mean.

    Args:
        pipe (`DiffusionPipeline`):
            Pipe to provide access to the tokenizer and the text encoder.
        prompt (`str` or `List[str]`):
            The prompt or prompts to guide the image generation.
        uncond_prompt (`str` or `List[str]`):
            The unconditional prompt or prompts for guide the image generation. If unconditional prompt
            is provided, the embeddings of prompt and uncond_prompt are concatenated.
        max_embeddings_multiples (`int`, *optional*, defaults to `3`):
            The max multiple length of prompt embeddings compared to the max output length of text encoder.
        no_boseos_middle (`bool`, *optional*, defaults to `False`):
            If the length of text token is multiples of the capacity of text encoder, whether reserve the starting and
            ending token in each of the chunk in the middle.
        skip_parsing (`bool`, *optional*, defaults to `False`):
            Skip the parsing of brackets.
        skip_weighting (`bool`, *optional*, defaults to `False`):
            Skip the weighting. When the parsing is skipped, it is forced True.
    """
    max_length = (pipe.tokenizer.model_max_length - 2) * max_embeddings_multiples + 2
    if isinstance(prompt, str):
        prompt = [prompt]

    if not skip_parsing:
        prompt_tokens, prompt_weights = get_prompts_with_weights(pipe, prompt, max_length - 2)
        if uncond_prompt is not None:
            if isinstance(uncond_prompt, str):
                uncond_prompt = [uncond_prompt]
            uncond_tokens, uncond_weights = get_prompts_with_weights(pipe, uncond_prompt, max_length - 2)
    else:
        prompt_tokens = [
            token[1:-1] for token in pipe.tokenizer(prompt, max_length=max_length, truncation=True).input_ids
        ]
        prompt_weights = [[1.0] * len(token) for token in prompt_tokens]
        if uncond_prompt is not None:
            if isinstance(uncond_prompt, str):
                uncond_prompt = [uncond_prompt]
            uncond_tokens = [
                token[1:-1]
                for token in pipe.tokenizer(uncond_prompt, max_length=max_length, truncation=True).input_ids
            ]
            uncond_weights = [[1.0] * len(token) for token in uncond_tokens]

    # round up the longest length of tokens to a multiple of (model_max_length - 2)
    max_length = max([len(token) for token in prompt_tokens])
    if uncond_prompt is not None:
        max_length = max(max_length, max([len(token) for token in uncond_tokens]))

    max_embeddings_multiples = min(
        max_embeddings_multiples,
        (max_length - 1) // (pipe.tokenizer.model_max_length - 2) + 1,
    )
    max_embeddings_multiples = max(1, max_embeddings_multiples)
    max_length = (pipe.tokenizer.model_max_length - 2) * max_embeddings_multiples + 2

    # pad the length of tokens and weights
    bos = pipe.tokenizer.bos_token_id
    eos = pipe.tokenizer.eos_token_id
    pad = getattr(pipe.tokenizer, "pad_token_id", eos)
    prompt_tokens, prompt_weights = pad_tokens_and_weights(
        prompt_tokens,
        prompt_weights,
        max_length,
        bos,
        eos,
        pad,
        no_boseos_middle=no_boseos_middle,
        chunk_length=pipe.tokenizer.model_max_length,
    )
    prompt_tokens = torch.tensor(prompt_tokens, dtype=torch.long, device=pipe.device)
    if uncond_prompt is not None:
        uncond_tokens, uncond_weights = pad_tokens_and_weights(
            uncond_tokens,
            uncond_weights,
            max_length,
            bos,
            eos,
            pad,
            no_boseos_middle=no_boseos_middle,
            chunk_length=pipe.tokenizer.model_max_length,
        )
        uncond_tokens = torch.tensor(uncond_tokens, dtype=torch.long, device=pipe.device)

    # get the embeddings
    text_embeddings = get_unweighted_text_embeddings(
        pipe,
        prompt_tokens,
        pipe.tokenizer.model_max_length,
        no_boseos_middle=no_boseos_middle,
    )
    prompt_weights = torch.tensor(prompt_weights, dtype=text_embeddings.dtype, device=text_embeddings.device)
    if uncond_prompt is not None:
        uncond_embeddings = get_unweighted_text_embeddings(
            pipe,
            uncond_tokens,
            pipe.tokenizer.model_max_length,
            no_boseos_middle=no_boseos_middle,
        )
        uncond_weights = torch.tensor(uncond_weights, dtype=uncond_embeddings.dtype, device=uncond_embeddings.device)

    # assign weights to the prompts and normalize in the sense of mean
    # TODO: should we normalize by chunk or in a whole (current implementation)?
    if (not skip_parsing) and (not skip_weighting):
        previous_mean = text_embeddings.float().mean(axis=[-2, -1]).to(text_embeddings.dtype)
        text_embeddings *= prompt_weights.unsqueeze(-1)
        current_mean = text_embeddings.float().mean(axis=[-2, -1]).to(text_embeddings.dtype)
        text_embeddings *= (previous_mean / current_mean).unsqueeze(-1).unsqueeze(-1)
        if uncond_prompt is not None:
            previous_mean = uncond_embeddings.float().mean(axis=[-2, -1]).to(uncond_embeddings.dtype)
            uncond_embeddings *= uncond_weights.unsqueeze(-1)
            current_mean = uncond_embeddings.float().mean(axis=[-2, -1]).to(uncond_embeddings.dtype)
            uncond_embeddings *= (previous_mean / current_mean).unsqueeze(-1).unsqueeze(-1)

    if uncond_prompt is not None:
        return text_embeddings, uncond_embeddings
    return text_embeddings, None


def preprocess_image(image, batch_size):
    w, h = image.size
    w, h = (x - x % 8 for x in (w, h))  # resize to integer multiple of 8
    image = image.resize((w, h), resample=PIL_INTERPOLATION["lanczos"])
    image = np.array(image).astype(np.float32) / 255.0
    image = np.vstack([image[None].transpose(0, 3, 1, 2)] * batch_size)
    image = torch.from_numpy(image)
    return 2.0 * image - 1.0


def preprocess_mask(mask, batch_size, scale_factor=8):
    if not isinstance(mask, torch.FloatTensor):
        mask = mask.convert("L")
        w, h = mask.size
        w, h = (x - x % 8 for x in (w, h))  # resize to integer multiple of 8
        mask = mask.resize((w // scale_factor, h // scale_factor), resample=PIL_INTERPOLATION["nearest"])
        mask = np.array(mask).astype(np.float32) / 255.0
        mask = np.tile(mask, (4, 1, 1))
        mask = np.vstack([mask[None]] * batch_size)
        mask = 1 - mask  # repaint white, keep black
        mask = torch.from_numpy(mask)
        return mask

    else:
        valid_mask_channel_sizes = [1, 3]
        # if mask channel is fourth tensor dimension, permute dimensions to pytorch standard (B, C, H, W)
        if mask.shape[3] in valid_mask_channel_sizes:
            mask = mask.permute(0, 3, 1, 2)
        elif mask.shape[1] not in valid_mask_channel_sizes:
            raise ValueError(
                f"Mask channel dimension of size in {valid_mask_channel_sizes} should be second or fourth dimension,"
                f" but received mask of shape {tuple(mask.shape)}"
            )
        # (potentially) reduce mask channel dimension from 3 to 1 for broadcasting to latent shape
        mask = mask.mean(dim=1, keepdim=True)
        h, w = mask.shape[-2:]
        h, w = (x - x % 8 for x in (h, w))  # resize to integer multiple of 8
        mask = torch.nn.functional.interpolate(mask, (h // scale_factor, w // scale_factor))
        return mask


class StableDiffusionLongPromptWeightingPipeline(
    DiffusionPipeline, TextualInversionLoaderMixin, LoraLoaderMixin, FromSingleFileMixin
):
    r"""
    Pipeline for text-to-image generation using Stable Diffusion without tokens length limit, and support parsing
    weighting in prompt.

    This model inherits from [`DiffusionPipeline`]. Check the superclass documentation for the generic methods the
    library implements for all the pipelines (such as downloading or saving, running on a particular device, etc.)

    Args:
        vae ([`AutoencoderKL`]):
            Variational Auto-Encoder (VAE) Model to encode and decode images to and from latent representations.
        text_encoder ([`CLIPTextModel`]):
            Frozen text-encoder. Stable Diffusion uses the text portion of
            [CLIP](https://huggingface.co/docs/transformers/model_doc/clip#transformers.CLIPTextModel), specifically
            the [clip-vit-large-patch14](https://huggingface.co/openai/clip-vit-large-patch14) variant.
        tokenizer (`CLIPTokenizer`):
            Tokenizer of class
            [CLIPTokenizer](https://huggingface.co/docs/transformers/v4.21.0/en/model_doc/clip#transformers.CLIPTokenizer).
        unet ([`UNet2DConditionModel`]): Conditional U-Net architecture to denoise the encoded image latents.
        scheduler ([`SchedulerMixin`]):
            A scheduler to be used in combination with `unet` to denoise the encoded image latents. Can be one of
            [`DDIMScheduler`], [`LMSDiscreteScheduler`], or [`PNDMScheduler`].
        safety_checker ([`StableDiffusionSafetyChecker`]):
            Classification module that estimates whether generated images could be considered offensive or harmful.
            Please, refer to the [model card](https://huggingface.co/CompVis/stable-diffusion-v1-4) for details.
        feature_extractor ([`CLIPImageProcessor`]):
            Model that extracts features from generated images to be used as inputs for the `safety_checker`.
    """

    _optional_components = ["safety_checker", "feature_extractor"]

    def __init__(
        self,
        vae: AutoencoderKL,
        text_encoder: CLIPTextModel,
        tokenizer: CLIPTokenizer,
        unet: UNet2DConditionModel,
        scheduler: KarrasDiffusionSchedulers,
        safety_checker: StableDiffusionSafetyChecker,
        feature_extractor: CLIPImageProcessor,
        requires_safety_checker: bool = True,
    ):
        super().__init__()

        if hasattr(scheduler.config, "steps_offset") and scheduler.config.steps_offset != 1:
            deprecation_message = (
                f"The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"
                f" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure "
                "to update the config accordingly as leaving `steps_offset` might led to incorrect results"
                " in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,"
                " it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`"
                " file"
            )
            deprecate("steps_offset!=1", "1.0.0", deprecation_message, standard_warn=False)
            new_config = dict(scheduler.config)
            new_config["steps_offset"] = 1
            scheduler._internal_dict = FrozenDict(new_config)

        if hasattr(scheduler.config, "clip_sample") and scheduler.config.clip_sample is True:
            deprecation_message = (
                f"The configuration file of this scheduler: {scheduler} has not set the configuration `clip_sample`."
                " `clip_sample` should be set to False in the configuration file. Please make sure to update the"
                " config accordingly as not setting `clip_sample` in the config might lead to incorrect results in"
                " future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it would be very"
                " nice if you could open a Pull request for the `scheduler/scheduler_config.json` file"
            )
            deprecate("clip_sample not set", "1.0.0", deprecation_message, standard_warn=False)
            new_config = dict(scheduler.config)
            new_config["clip_sample"] = False
            scheduler._internal_dict = FrozenDict(new_config)

        if safety_checker is None and requires_safety_checker:
            logger.warning(
                f"You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"
                " that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered"
                " results in services or applications open to the public. Both the diffusers team and Hugging Face"
                " strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling"
                " it only for use-cases that involve analyzing network behavior or auditing its results. For more"
                " information, please have a look at https://github.com/huggingface/diffusers/pull/254 ."
            )

        if safety_checker is not None and feature_extractor is None:
            raise ValueError(
                "Make sure to define a feature extractor when loading {self.__class__} if you want to use the safety"
                " checker. If you do not want to use the safety checker, you can pass `'safety_checker=None'` instead."
            )

        is_unet_version_less_0_9_0 = hasattr(unet.config, "_diffusers_version") and version.parse(
            version.parse(unet.config._diffusers_version).base_version
        ) < version.parse("0.9.0.dev0")
        is_unet_sample_size_less_64 = hasattr(unet.config, "sample_size") and unet.config.sample_size < 64
        if is_unet_version_less_0_9_0 and is_unet_sample_size_less_64:
            deprecation_message = (
                "The configuration file of the unet has set the default `sample_size` to smaller than"
                " 64 which seems highly unlikely. If your checkpoint is a fine-tuned version of any of the"
                " following: \n- CompVis/stable-diffusion-v1-4 \n- CompVis/stable-diffusion-v1-3 \n-"
                " CompVis/stable-diffusion-v1-2 \n- CompVis/stable-diffusion-v1-1 \n- runwayml/stable-diffusion-v1-5"
                " \n- runwayml/stable-diffusion-inpainting \n you should change 'sample_size' to 64 in the"
                " configuration file. Please make sure to update the config accordingly as leaving `sample_size=32`"
                " in the config might lead to incorrect results in future versions. If you have downloaded this"
                " checkpoint from the Hugging Face Hub, it would be very nice if you could open a Pull request for"
                " the `unet/config.json` file"
            )
            deprecate("sample_size<64", "1.0.0", deprecation_message, standard_warn=False)
            new_config = dict(unet.config)
            new_config["sample_size"] = 64
            unet._internal_dict = FrozenDict(new_config)
        self.register_modules(
            vae=vae,
            text_encoder=text_encoder,
            tokenizer=tokenizer,
            unet=unet,
            scheduler=scheduler,
            safety_checker=safety_checker,
            feature_extractor=feature_extractor,
        )
        self.vae_scale_factor = 2 ** (len(self.vae.config.block_out_channels) - 1)

        self.image_processor = VaeImageProcessor(vae_scale_factor=self.vae_scale_factor)
        self.register_to_config(
            requires_safety_checker=requires_safety_checker,
        )

    def enable_vae_slicing(self):
        r"""
        Enable sliced VAE decoding.

        When this option is enabled, the VAE will split the input tensor in slices to compute decoding in several
        steps. This is useful to save some memory and allow larger batch sizes.
        """
        self.vae.enable_slicing()

    def disable_vae_slicing(self):
        r"""
        Disable sliced VAE decoding. If `enable_vae_slicing` was previously invoked, this method will go back to
        computing decoding in one step.
        """
        self.vae.disable_slicing()

    def enable_vae_tiling(self):
        r"""
        Enable tiled VAE decoding.

        When this option is enabled, the VAE will split the input tensor into tiles to compute decoding and encoding in
        several steps. This is useful to save a large amount of memory and to allow the processing of larger images.
        """
        self.vae.enable_tiling()

    def disable_vae_tiling(self):
        r"""
        Disable tiled VAE decoding. If `enable_vae_tiling` was previously invoked, this method will go back to
        computing decoding in one step.
        """
        self.vae.disable_tiling()

    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline.enable_sequential_cpu_offload
    def enable_sequential_cpu_offload(self, gpu_id=0):
        r"""
        Offloads all models to CPU using accelerate, significantly reducing memory usage. When called, unet,
        text_encoder, vae and safety checker have their state dicts saved to CPU and then are moved to a
        `torch.device('meta') and loaded to GPU only when their specific submodule has its `forward` method called.
        Note that offloading happens on a submodule basis. Memory savings are higher than with
        `enable_model_cpu_offload`, but performance is lower.
        """
        if is_accelerate_available() and is_accelerate_version(">=", "0.14.0"):
            from accelerate import cpu_offload
        else:
            raise ImportError("`enable_sequential_cpu_offload` requires `accelerate v0.14.0` or higher")

        device = torch.device(f"cuda:{gpu_id}")

        if self.device.type != "cpu":
            self.to("cpu", silence_dtype_warnings=True)
            torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

        for cpu_offloaded_model in [self.unet, self.text_encoder, self.vae]:
            cpu_offload(cpu_offloaded_model, device)

        if self.safety_checker is not None:
            cpu_offload(self.safety_checker, execution_device=device, offload_buffers=True)

    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline.enable_model_cpu_offload
    def enable_model_cpu_offload(self, gpu_id=0):
        r"""
        Offloads all models to CPU using accelerate, reducing memory usage with a low impact on performance. Compared
        to `enable_sequential_cpu_offload`, this method moves one whole model at a time to the GPU when its `forward`
        method is called, and the model remains in GPU until the next model runs. Memory savings are lower than with
        `enable_sequential_cpu_offload`, but performance is much better due to the iterative execution of the `unet`.
        """
        if is_accelerate_available() and is_accelerate_version(">=", "0.17.0.dev0"):
            from accelerate import cpu_offload_with_hook
        else:
            raise ImportError("`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.")

        device = torch.device(f"cuda:{gpu_id}")

        if self.device.type != "cpu":
            self.to("cpu", silence_dtype_warnings=True)
            torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

        hook = None
        for cpu_offloaded_model in [self.text_encoder, self.unet, self.vae]:
            _, hook = cpu_offload_with_hook(cpu_offloaded_model, device, prev_module_hook=hook)

        if self.safety_checker is not None:
            _, hook = cpu_offload_with_hook(self.safety_checker, device, prev_module_hook=hook)

        # We'll offload the last model manually.
        self.final_offload_hook = hook

    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def _execution_device(self):
        r"""
        Returns the device on which the pipeline's models will be executed. After calling
        `pipeline.enable_sequential_cpu_offload()` the execution device can only be inferred from Accelerate's module
        hooks.
        """
        if not hasattr(self.unet, "_hf_hook"):
            return self.device
        for module in self.unet.modules():
            if (
                hasattr(module, "_hf_hook")
                and hasattr(module._hf_hook, "execution_device")
                and module._hf_hook.execution_device is not None
            ):
                return torch.device(module._hf_hook.execution_device)
        return self.device

    def _encode_prompt(
        self,
        prompt,
        device,
        num_images_per_prompt,
        do_classifier_free_guidance,
        negative_prompt=None,
        max_embeddings_multiples=3,
        prompt_embeds: Optional[torch.FloatTensor] = None,
        negative_prompt_embeds: Optional[torch.FloatTensor] = None,
    ):
        r"""
        Encodes the prompt into text encoder hidden states.

        Args:
            prompt (`str` or `list(int)`):
                prompt to be encoded
            device: (`torch.device`):
                torch device
            num_images_per_prompt (`int`):
                number of images that should be generated per prompt
            do_classifier_free_guidance (`bool`):
                whether to use classifier free guidance or not
            negative_prompt (`str` or `List[str]`):
                The prompt or prompts not to guide the image generation. Ignored when not using guidance (i.e., ignored
                if `guidance_scale` is less than `1`).
            max_embeddings_multiples (`int`, *optional*, defaults to `3`):
                The max multiple length of prompt embeddings compared to the max output length of text encoder.
        """
        if prompt is not None and isinstance(prompt, str):
            batch_size = 1
        elif prompt is not None and isinstance(prompt, list):
            batch_size = len(prompt)
        else:
            batch_size = prompt_embeds.shape[0]

        if negative_prompt_embeds is None:
            if negative_prompt is None:
                negative_prompt = [""] * batch_size
            elif isinstance(negative_prompt, str):
                negative_prompt = [negative_prompt] * batch_size
            if batch_size != len(negative_prompt):
                raise ValueError(
                    f"`negative_prompt`: {negative_prompt} has batch size {len(negative_prompt)}, but `prompt`:"
                    f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"
                    " the batch size of `prompt`."
                )
        if prompt_embeds is None or negative_prompt_embeds is None:
            if isinstance(self, TextualInversionLoaderMixin):
                prompt = self.maybe_convert_prompt(prompt, self.tokenizer)
                if do_classifier_free_guidance and negative_prompt_embeds is None:
                    negative_prompt = self.maybe_convert_prompt(negative_prompt, self.tokenizer)

            prompt_embeds1, negative_prompt_embeds1 = get_weighted_text_embeddings(
                pipe=self,
                prompt=prompt,
                uncond_prompt=negative_prompt if do_classifier_free_guidance else None,
                max_embeddings_multiples=max_embeddings_multiples,
            )
            if prompt_embeds is None:
                prompt_embeds = prompt_embeds1
            if negative_prompt_embeds is None:
                negative_prompt_embeds = negative_prompt_embeds1

        bs_embed, seq_len, _ = prompt_embeds.shape
        # duplicate text embeddings for each generation per prompt, using mps friendly method
        prompt_embeds = prompt_embeds.repeat(1, num_images_per_prompt, 1)
        prompt_embeds = prompt_embeds.view(bs_embed * num_images_per_prompt, seq_len, -1)

        if do_classifier_free_guidance:
            bs_embed, seq_len, _ = negative_prompt_embeds.shape
            negative_prompt_embeds = negative_prompt_embeds.repeat(1, num_images_per_prompt, 1)
            negative_prompt_embeds = negative_prompt_embeds.view(bs_embed * num_images_per_prompt, seq_len, -1)
            prompt_embeds = torch.cat([negative_prompt_embeds, prompt_embeds])

        return prompt_embeds

    def check_inputs(
        self,
        prompt,
        height,
        width,
        strength,
        callback_steps,
        negative_prompt=None,
        prompt_embeds=None,
        negative_prompt_embeds=None,
    ):
        if height % 8 != 0 or width % 8 != 0:
            raise ValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}.")

        if strength < 0 or strength > 1:
            raise ValueError(f"The value of strength should in [0.0, 1.0] but is {strength}")

        if (callback_steps is None) or (
            callback_steps is not None and (not isinstance(callback_steps, int) or callback_steps <= 0)
        ):
            raise ValueError(
                f"`callback_steps` has to be a positive integer but is {callback_steps} of type"
                f" {type(callback_steps)}."
            )

        if prompt is not None and prompt_embeds is not None:
            raise ValueError(
                f"Cannot forward both `prompt`: {prompt} and `prompt_embeds`: {prompt_embeds}. Please make sure to"
                " only forward one of the two."
            )
        elif prompt is None and prompt_embeds is None:
            raise ValueError(
                "Provide either `prompt` or `prompt_embeds`. Cannot leave both `prompt` and `prompt_embeds` undefined."
            )
        elif prompt is not None and (not isinstance(prompt, str) and not isinstance(prompt, list)):
            raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(prompt)}")

        if negative_prompt is not None and negative_prompt_embeds is not None:
            raise ValueError(
                f"Cannot forward both `negative_prompt`: {negative_prompt} and `negative_prompt_embeds`:"
                f" {negative_prompt_embeds}. Please make sure to only forward one of the two."
            )

        if prompt_embeds is not None and negative_prompt_embeds is not None:
            if prompt_embeds.shape != negative_prompt_embeds.shape:
                raise ValueError(
                    "`prompt_embeds` and `negative_prompt_embeds` must have the same shape when passed directly, but"
                    f" got: `prompt_embeds` {prompt_embeds.shape} != `negative_prompt_embeds`"
                    f" {negative_prompt_embeds.shape}."
                )

    def get_timesteps(self, num_inference_steps, strength, device, is_text2img):
        if is_text2img:
            return self.scheduler.timesteps.to(device), num_inference_steps
        else:
            # get the original timestep using init_timestep
            init_timestep = min(int(num_inference_steps * strength), num_inference_steps)

            t_start = max(num_inference_steps - init_timestep, 0)
            timesteps = self.scheduler.timesteps[t_start * self.scheduler.order:]

            return timesteps, num_inference_steps - t_start

    def run_safety_checker(self, image, device, dtype):
        if self.safety_checker is not None:
            safety_checker_input = self.feature_extractor(self.numpy_to_pil(image), return_tensors="pt").to(device)
            image, has_nsfw_concept = self.safety_checker(
                images=image, clip_input=safety_checker_input.pixel_values.to(dtype)
            )
        else:
            has_nsfw_concept = None
        return image, has_nsfw_concept

    def decode_latents(self, latents):
        latents = 1 / self.vae.config.scaling_factor * latents
        image = self.vae.decode(latents).sample
        image = (image / 2 + 0.5).clamp(0, 1)
        # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
        image = image.cpu().permute(0, 2, 3, 1).float().numpy()
        return image

    def prepare_extra_step_kwargs(self, generator, eta):
        # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
        # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
        # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
        # and should be between [0, 1]

        accepts_eta = "eta" in set(inspect.signature(self.scheduler.step).parameters.keys())
        extra_step_kwargs = {}
        if accepts_eta:
            extra_step_kwargs["eta"] = eta

        # check if the scheduler accepts generator
        accepts_generator = "generator" in set(inspect.signature(self.scheduler.step).parameters.keys())
        if accepts_generator:
            extra_step_kwargs["generator"] = generator
        return extra_step_kwargs

    def prepare_latents(
        self,
        image,
        timestep,
        num_images_per_prompt,
        batch_size,
        num_channels_latents,
        height,
        width,
        dtype,
        device,
        generator,
        latents=None,
    ):
        if image is None:
            batch_size = batch_size * num_images_per_prompt
            shape = (batch_size, num_channels_latents, height // self.vae_scale_factor, width // self.vae_scale_factor)
            if isinstance(generator, list) and len(generator) != batch_size:
                raise ValueError(
                    f"You have passed a list of generators of length {len(generator)}, but requested an effective batch"
                    f" size of {batch_size}. Make sure the batch size matches the length of the generators."
                )

            if latents is None:
                latents = randn_tensor(shape, generator=generator, device=device, dtype=dtype)
            else:
                latents = latents.to(device)

            # scale the initial noise by the standard deviation required by the scheduler
            latents = latents * self.scheduler.init_noise_sigma
            return latents, None, None
        else:
            image = image.to(device=self.device, dtype=dtype)
            init_latent_dist = self.vae.encode(image).latent_dist
            init_latents = init_latent_dist.sample(generator=generator)
            init_latents = self.vae.config.scaling_factor * init_latents

            # Expand init_latents for batch_size and num_images_per_prompt
            init_latents = torch.cat([init_latents] * num_images_per_prompt, dim=0)
            init_latents_orig = init_latents

            # add noise to latents using the timesteps
            noise = randn_tensor(init_latents.shape, generator=generator, device=self.device, dtype=dtype)
            init_latents = self.scheduler.add_noise(init_latents, noise, timestep)
            latents = init_latents
            return latents, init_latents_orig, noise

    @torch.no_grad()
    def __call__(
        self,
        prompt: Union[str, List[str]],
        negative_prompt: Optional[Union[str, List[str]]] = None,
        image: Union[torch.FloatTensor, PIL.Image.Image] = None,
        mask_image: Union[torch.FloatTensor, PIL.Image.Image] = None,
        height: int = 512,
        width: int = 512,
        num_inference_steps: int = 50,
        guidance_scale: float = 7.5,
        strength: float = 0.8,
        num_images_per_prompt: Optional[int] = 1,
        add_predicted_noise: Optional[bool] = False,
        eta: float = 0.0,
        generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None,
        latents: Optional[torch.FloatTensor] = None,
        prompt_embeds: Optional[torch.FloatTensor] = None,
        negative_prompt_embeds: Optional[torch.FloatTensor] = None,
        max_embeddings_multiples: Optional[int] = 3,
        output_type: Optional[str] = "pil",
        return_dict: bool = True,
        callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,
        is_cancelled_callback: Optional[Callable[[], bool]] = None,
        callback_steps: int = 1,
        cross_attention_kwargs: Optional[Dict[str, Any]] = None,
    ):
        r"""
        Function invoked when calling the pipeline for generation.

        Args:
            prompt (`str` or `List[str]`):
                The prompt or prompts to guide the image generation.
            negative_prompt (`str` or `List[str]`, *optional*):
                The prompt or prompts not to guide the image generation. Ignored when not using guidance (i.e., ignored
                if `guidance_scale` is less than `1`).
            image (`torch.FloatTensor` or `PIL.Image.Image`):
                `Image`, or tensor representing an image batch, that will be used as the starting point for the
                process.
            mask_image (`torch.FloatTensor` or `PIL.Image.Image`):
                `Image`, or tensor representing an image batch, to mask `image`. White pixels in the mask will be
                replaced by noise and therefore repainted, while black pixels will be preserved. If `mask_image` is a
                PIL image, it will be converted to a single channel (luminance) before use. If it's a tensor, it should
                contain one color channel (L) instead of 3, so the expected shape would be `(B, H, W, 1)`.
            height (`int`, *optional*, defaults to 512):
                The height in pixels of the generated image.
            width (`int`, *optional*, defaults to 512):
                The width in pixels of the generated image.
            num_inference_steps (`int`, *optional*, defaults to 50):
                The number of denoising steps. More denoising steps usually lead to a higher quality image at the
                expense of slower inference.
            guidance_scale (`float`, *optional*, defaults to 7.5):
                Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
                `guidance_scale` is defined as `w` of equation 2. of [Imagen
                Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale >
                1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`,
                usually at the expense of lower image quality.
            strength (`float`, *optional*, defaults to 0.8):
                Conceptually, indicates how much to transform the reference `image`. Must be between 0 and 1.
                `image` will be used as a starting point, adding more noise to it the larger the `strength`. The
                number of denoising steps depends on the amount of noise initially added. When `strength` is 1, added
                noise will be maximum and the denoising process will run for the full number of iterations specified in
                `num_inference_steps`. A value of 1, therefore, essentially ignores `image`.
            num_images_per_prompt (`int`, *optional*, defaults to 1):
                The number of images to generate per prompt.
            add_predicted_noise (`bool`, *optional*, defaults to True):
                Use predicted noise instead of random noise when constructing noisy versions of the original image in
                the reverse diffusion process
            eta (`float`, *optional*, defaults to 0.0):
                Corresponds to parameter eta (η) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
                [`schedulers.DDIMScheduler`], will be ignored for others.
            generator (`torch.Generator` or `List[torch.Generator]`, *optional*):
                One or a list of [torch generator(s)](https://pytorch.org/docs/stable/generated/torch.Generator.html)
                to make generation deterministic.
            latents (`torch.FloatTensor`, *optional*):
                Pre-generated noisy latents, sampled from a Gaussian distribution, to be used as inputs for image
                generation. Can be used to tweak the same generation with different prompts. If not provided, a latents
                tensor will ge generated by sampling using the supplied random `generator`.
            prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt weighting. If not
                provided, text embeddings will be generated from `prompt` input argument.
            negative_prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated negative text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt
                weighting. If not provided, negative_prompt_embeds will be generated from `negative_prompt` input
                argument.
            max_embeddings_multiples (`int`, *optional*, defaults to `3`):
                The max multiple length of prompt embeddings compared to the max output length of text encoder.
            output_type (`str`, *optional*, defaults to `"pil"`):
                The output format of the generate image. Choose between
                [PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`.
            return_dict (`bool`, *optional*, defaults to `True`):
                Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
                plain tuple.
            callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. The function will be
                called with the following arguments: `callback(step: int, timestep: int, latents: torch.FloatTensor)`.
            is_cancelled_callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. If the function returns
                `True`, the inference will be cancelled.
            callback_steps (`int`, *optional*, defaults to 1):
                The frequency at which the `callback` function will be called. If not specified, the callback will be
                called at every step.
            cross_attention_kwargs (`dict`, *optional*):
                A kwargs dictionary that if specified is passed along to the `AttentionProcessor` as defined under
                `self.processor` in
                [diffusers.cross_attention](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/cross_attention.py).

        Returns:
            `None` if cancelled by `is_cancelled_callback`,
            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`:
            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] if `return_dict` is True, otherwise a `tuple.
            When returning a tuple, the first element is a list with the generated images, and the second element is a
            list of `bool`s denoting whether the corresponding generated image likely represents "not-safe-for-work"
            (nsfw) content, according to the `safety_checker`.
        """
        # 0. Default height and width to unet
        height = height or self.unet.config.sample_size * self.vae_scale_factor
        width = width or self.unet.config.sample_size * self.vae_scale_factor

        # 1. Check inputs. Raise error if not correct
        self.check_inputs(
            prompt, height, width, strength, callback_steps, negative_prompt, prompt_embeds, negative_prompt_embeds
        )

        # 2. Define call parameters
        if prompt is not None and isinstance(prompt, str):
            batch_size = 1
        elif prompt is not None and isinstance(prompt, list):
            batch_size = len(prompt)
        else:
            batch_size = prompt_embeds.shape[0]

        device = self._execution_device
        # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
        # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
        # corresponds to doing no classifier free guidance.
        do_classifier_free_guidance = guidance_scale > 1.0

        # 3. Encode input prompt
        prompt_embeds = self._encode_prompt(
            prompt,
            device,
            num_images_per_prompt,
            do_classifier_free_guidance,
            negative_prompt,
            max_embeddings_multiples,
            prompt_embeds=prompt_embeds,
            negative_prompt_embeds=negative_prompt_embeds,
        )
        dtype = prompt_embeds.dtype

        # 4. Preprocess image and mask
        if isinstance(image, PIL.Image.Image):
            image = preprocess_image(image, batch_size)
        if image is not None:
            image = image.to(device=self.device, dtype=dtype)
        if isinstance(mask_image, PIL.Image.Image):
            mask_image = preprocess_mask(mask_image, batch_size, self.vae_scale_factor)
        if mask_image is not None:
            mask = mask_image.to(device=self.device, dtype=dtype)
            mask = torch.cat([mask] * num_images_per_prompt)
        else:
            mask = None

        # 5. set timesteps
        self.scheduler.set_timesteps(num_inference_steps, device=device)
        timesteps, num_inference_steps = self.get_timesteps(num_inference_steps, strength, device, image is None)
        latent_timestep = timesteps[:1].repeat(batch_size * num_images_per_prompt)

        # 6. Prepare latent variables
        latents, init_latents_orig, noise = self.prepare_latents(
            image,
            latent_timestep,
            num_images_per_prompt,
            batch_size,
            self.unet.config.in_channels,
            height,
            width,
            dtype,
            device,
            generator,
            latents,
        )

        # 7. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
        extra_step_kwargs = self.prepare_extra_step_kwargs(generator, eta)

        # 8. Denoising loop
        num_warmup_steps = len(timesteps) - num_inference_steps * self.scheduler.order
        with self.progress_bar(total=num_inference_steps) as progress_bar:
            for i, t in enumerate(timesteps):
                # expand the latents if we are doing classifier free guidance
                latent_model_input = torch.cat([latents] * 2) if do_classifier_free_guidance else latents
                latent_model_input = self.scheduler.scale_model_input(latent_model_input, t)

                # predict the noise residual
                noise_pred = self.unet(
                    latent_model_input,
                    t,
                    encoder_hidden_states=prompt_embeds,
                    cross_attention_kwargs=cross_attention_kwargs,
                ).sample

                # perform guidance
                if do_classifier_free_guidance:
                    noise_pred_uncond, noise_pred_text = noise_pred.chunk(2)
                    noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                # compute the previous noisy sample x_t -> x_t-1
                latents = self.scheduler.step(noise_pred, t, latents, **extra_step_kwargs).prev_sample

                if mask is not None:
                    # masking
                    if add_predicted_noise:
                        init_latents_proper = self.scheduler.add_noise(
                            init_latents_orig, noise_pred_uncond, torch.tensor([t])
                        )
                    else:
                        init_latents_proper = self.scheduler.add_noise(init_latents_orig, noise, torch.tensor([t]))
                    latents = (init_latents_proper * mask) + (latents * (1 - mask))

                # call the callback, if provided
                if i == len(timesteps) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0):
                    progress_bar.update()
                    if i % callback_steps == 0:
                        if callback is not None:
                            callback(i, t, latents)
                        if is_cancelled_callback is not None and is_cancelled_callback():
                            return None

        if output_type == "latent":
            image = latents
            has_nsfw_concept = None
        elif output_type == "pil":
            # 9. Post-processing
            image = self.decode_latents(latents)

            # 10. Run safety checker
            image, has_nsfw_concept = self.run_safety_checker(image, device, prompt_embeds.dtype)

            # 11. Convert to PIL
            image = self.numpy_to_pil(image)
        else:
            # 9. Post-processing
            image = self.decode_latents(latents)

            # 10. Run safety checker
            image, has_nsfw_concept = self.run_safety_checker(image, device, prompt_embeds.dtype)

        # Offload last model to CPU
        if hasattr(self, "final_offload_hook") and self.final_offload_hook is not None:
            self.final_offload_hook.offload()

        if not return_dict:
            return image, has_nsfw_concept

        return StableDiffusionPipelineOutput(images=image, nsfw_content_detected=has_nsfw_concept)

    def text2img(
        self,
        prompt: Union[str, List[str]],
        negative_prompt: Optional[Union[str, List[str]]] = None,
        height: int = 512,
        width: int = 512,
        num_inference_steps: int = 50,
        guidance_scale: float = 7.5,
        num_images_per_prompt: Optional[int] = 1,
        eta: float = 0.0,
        generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None,
        latents: Optional[torch.FloatTensor] = None,
        prompt_embeds: Optional[torch.FloatTensor] = None,
        negative_prompt_embeds: Optional[torch.FloatTensor] = None,
        max_embeddings_multiples: Optional[int] = 3,
        output_type: Optional[str] = "pil",
        return_dict: bool = True,
        callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,
        is_cancelled_callback: Optional[Callable[[], bool]] = None,
        callback_steps: int = 1,
        cross_attention_kwargs: Optional[Dict[str, Any]] = None,
    ):
        r"""
        Function for text-to-image generation.
        Args:
            prompt (`str` or `List[str]`):
                The prompt or prompts to guide the image generation.
            negative_prompt (`str` or `List[str]`, *optional*):
                The prompt or prompts not to guide the image generation. Ignored when not using guidance (i.e., ignored
                if `guidance_scale` is less than `1`).
            height (`int`, *optional*, defaults to 512):
                The height in pixels of the generated image.
            width (`int`, *optional*, defaults to 512):
                The width in pixels of the generated image.
            num_inference_steps (`int`, *optional*, defaults to 50):
                The number of denoising steps. More denoising steps usually lead to a higher quality image at the
                expense of slower inference.
            guidance_scale (`float`, *optional*, defaults to 7.5):
                Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
                `guidance_scale` is defined as `w` of equation 2. of [Imagen
                Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale >
                1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`,
                usually at the expense of lower image quality.
            num_images_per_prompt (`int`, *optional*, defaults to 1):
                The number of images to generate per prompt.
            eta (`float`, *optional*, defaults to 0.0):
                Corresponds to parameter eta (η) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
                [`schedulers.DDIMScheduler`], will be ignored for others.
            generator (`torch.Generator` or `List[torch.Generator]`, *optional*):
                One or a list of [torch generator(s)](https://pytorch.org/docs/stable/generated/torch.Generator.html)
                to make generation deterministic.
            latents (`torch.FloatTensor`, *optional*):
                Pre-generated noisy latents, sampled from a Gaussian distribution, to be used as inputs for image
                generation. Can be used to tweak the same generation with different prompts. If not provided, a latents
                tensor will ge generated by sampling using the supplied random `generator`.
            prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt weighting. If not
                provided, text embeddings will be generated from `prompt` input argument.
            negative_prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated negative text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt
                weighting. If not provided, negative_prompt_embeds will be generated from `negative_prompt` input
                argument.
            max_embeddings_multiples (`int`, *optional*, defaults to `3`):
                The max multiple length of prompt embeddings compared to the max output length of text encoder.
            output_type (`str`, *optional*, defaults to `"pil"`):
                The output format of the generate image. Choose between
                [PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`.
            return_dict (`bool`, *optional*, defaults to `True`):
                Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
                plain tuple.
            callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. The function will be
                called with the following arguments: `callback(step: int, timestep: int, latents: torch.FloatTensor)`.
            is_cancelled_callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. If the function returns
                `True`, the inference will be cancelled.
            callback_steps (`int`, *optional*, defaults to 1):
                The frequency at which the `callback` function will be called. If not specified, the callback will be
                called at every step.
            cross_attention_kwargs (`dict`, *optional*):
                A kwargs dictionary that if specified is passed along to the `AttentionProcessor` as defined under
                `self.processor` in
                [diffusers.cross_attention](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/cross_attention.py).

        Returns:
            `None` if cancelled by `is_cancelled_callback`,
            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`:
            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] if `return_dict` is True, otherwise a `tuple.
            When returning a tuple, the first element is a list with the generated images, and the second element is a
            list of `bool`s denoting whether the corresponding generated image likely represents "not-safe-for-work"
            (nsfw) content, according to the `safety_checker`.
        """
        return self.__call__(
            prompt=prompt,
            negative_prompt=negative_prompt,
            height=height,
            width=width,
            num_inference_steps=num_inference_steps,
            guidance_scale=guidance_scale,
            num_images_per_prompt=num_images_per_prompt,
            eta=eta,
            generator=generator,
            latents=latents,
            prompt_embeds=prompt_embeds,
            negative_prompt_embeds=negative_prompt_embeds,
            max_embeddings_multiples=max_embeddings_multiples,
            output_type=output_type,
            return_dict=return_dict,
            callback=callback,
            is_cancelled_callback=is_cancelled_callback,
            callback_steps=callback_steps,
            cross_attention_kwargs=cross_attention_kwargs,
        )

    def img2img(
        self,
        image: Union[torch.FloatTensor, PIL.Image.Image],
        prompt: Union[str, List[str]],
        negative_prompt: Optional[Union[str, List[str]]] = None,
        strength: float = 0.8,
        num_inference_steps: Optional[int] = 50,
        guidance_scale: Optional[float] = 7.5,
        num_images_per_prompt: Optional[int] = 1,
        eta: Optional[float] = 0.0,
        generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None,
        prompt_embeds: Optional[torch.FloatTensor] = None,
        negative_prompt_embeds: Optional[torch.FloatTensor] = None,
        max_embeddings_multiples: Optional[int] = 3,
        output_type: Optional[str] = "pil",
        return_dict: bool = True,
        callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,
        is_cancelled_callback: Optional[Callable[[], bool]] = None,
        callback_steps: int = 1,
        cross_attention_kwargs: Optional[Dict[str, Any]] = None,
    ):
        r"""
        Function for image-to-image generation.
        Args:
            image (`torch.FloatTensor` or `PIL.Image.Image`):
                `Image`, or tensor representing an image batch, that will be used as the starting point for the
                process.
            prompt (`str` or `List[str]`):
                The prompt or prompts to guide the image generation.
            negative_prompt (`str` or `List[str]`, *optional*):
                The prompt or prompts not to guide the image generation. Ignored when not using guidance (i.e., ignored
                if `guidance_scale` is less than `1`).
            strength (`float`, *optional*, defaults to 0.8):
                Conceptually, indicates how much to transform the reference `image`. Must be between 0 and 1.
                `image` will be used as a starting point, adding more noise to it the larger the `strength`. The
                number of denoising steps depends on the amount of noise initially added. When `strength` is 1, added
                noise will be maximum and the denoising process will run for the full number of iterations specified in
                `num_inference_steps`. A value of 1, therefore, essentially ignores `image`.
            num_inference_steps (`int`, *optional*, defaults to 50):
                The number of denoising steps. More denoising steps usually lead to a higher quality image at the
                expense of slower inference. This parameter will be modulated by `strength`.
            guidance_scale (`float`, *optional*, defaults to 7.5):
                Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
                `guidance_scale` is defined as `w` of equation 2. of [Imagen
                Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale >
                1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`,
                usually at the expense of lower image quality.
            num_images_per_prompt (`int`, *optional*, defaults to 1):
                The number of images to generate per prompt.
            eta (`float`, *optional*, defaults to 0.0):
                Corresponds to parameter eta (η) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
                [`schedulers.DDIMScheduler`], will be ignored for others.
            generator (`torch.Generator` or `List[torch.Generator]`, *optional*):
                One or a list of [torch generator(s)](https://pytorch.org/docs/stable/generated/torch.Generator.html)
                to make generation deterministic.
            prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt weighting. If not
                provided, text embeddings will be generated from `prompt` input argument.
            negative_prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated negative text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt
                weighting. If not provided, negative_prompt_embeds will be generated from `negative_prompt` input
                argument.
            max_embeddings_multiples (`int`, *optional*, defaults to `3`):
                The max multiple length of prompt embeddings compared to the max output length of text encoder.
            output_type (`str`, *optional*, defaults to `"pil"`):
                The output format of the generate image. Choose between
                [PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`.
            return_dict (`bool`, *optional*, defaults to `True`):
                Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
                plain tuple.
            callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. The function will be
                called with the following arguments: `callback(step: int, timestep: int, latents: torch.FloatTensor)`.
            is_cancelled_callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. If the function returns
                `True`, the inference will be cancelled.
            callback_steps (`int`, *optional*, defaults to 1):
                The frequency at which the `callback` function will be called. If not specified, the callback will be
                called at every step.
            cross_attention_kwargs (`dict`, *optional*):
                A kwargs dictionary that if specified is passed along to the `AttentionProcessor` as defined under
                `self.processor` in
                [diffusers.cross_attention](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/cross_attention.py).

        Returns:
            `None` if cancelled by `is_cancelled_callback`,
            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] if `return_dict` is True, otherwise a `tuple.
            When returning a tuple, the first element is a list with the generated images, and the second element is a
            list of `bool`s denoting whether the corresponding generated image likely represents "not-safe-for-work"
            (nsfw) content, according to the `safety_checker`.
        """
        return self.__call__(
            prompt=prompt,
            negative_prompt=negative_prompt,
            image=image,
            num_inference_steps=num_inference_steps,
            guidance_scale=guidance_scale,
            strength=strength,
            num_images_per_prompt=num_images_per_prompt,
            eta=eta,
            generator=generator,
            prompt_embeds=prompt_embeds,
            negative_prompt_embeds=negative_prompt_embeds,
            max_embeddings_multiples=max_embeddings_multiples,
            output_type=output_type,
            return_dict=return_dict,
            callback=callback,
            is_cancelled_callback=is_cancelled_callback,
            callback_steps=callback_steps,
            cross_attention_kwargs=cross_attention_kwargs,
        )

    def inpaint(
        self,
        image: Union[torch.FloatTensor, PIL.Image.Image],
        mask_image: Union[torch.FloatTensor, PIL.Image.Image],
        prompt: Union[str, List[str]],
        negative_prompt: Optional[Union[str, List[str]]] = None,
        strength: float = 0.8,
        num_inference_steps: Optional[int] = 50,
        guidance_scale: Optional[float] = 7.5,
        num_images_per_prompt: Optional[int] = 1,
        add_predicted_noise: Optional[bool] = False,
        eta: Optional[float] = 0.0,
        generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None,
        prompt_embeds: Optional[torch.FloatTensor] = None,
        negative_prompt_embeds: Optional[torch.FloatTensor] = None,
        max_embeddings_multiples: Optional[int] = 3,
        output_type: Optional[str] = "pil",
        return_dict: bool = True,
        callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,
        is_cancelled_callback: Optional[Callable[[], bool]] = None,
        callback_steps: int = 1,
        cross_attention_kwargs: Optional[Dict[str, Any]] = None,
    ):
        r"""
        Function for inpaint.
        Args:
            image (`torch.FloatTensor` or `PIL.Image.Image`):
                `Image`, or tensor representing an image batch, that will be used as the starting point for the
                process. This is the image whose masked region will be inpainted.
            mask_image (`torch.FloatTensor` or `PIL.Image.Image`):
                `Image`, or tensor representing an image batch, to mask `image`. White pixels in the mask will be
                replaced by noise and therefore repainted, while black pixels will be preserved. If `mask_image` is a
                PIL image, it will be converted to a single channel (luminance) before use. If it's a tensor, it should
                contain one color channel (L) instead of 3, so the expected shape would be `(B, H, W, 1)`.
            prompt (`str` or `List[str]`):
                The prompt or prompts to guide the image generation.
            negative_prompt (`str` or `List[str]`, *optional*):
                The prompt or prompts not to guide the image generation. Ignored when not using guidance (i.e., ignored
                if `guidance_scale` is less than `1`).
            strength (`float`, *optional*, defaults to 0.8):
                Conceptually, indicates how much to inpaint the masked area. Must be between 0 and 1. When `strength`
                is 1, the denoising process will be run on the masked area for the full number of iterations specified
                in `num_inference_steps`. `image` will be used as a reference for the masked area, adding more
                noise to that region the larger the `strength`. If `strength` is 0, no inpainting will occur.
            num_inference_steps (`int`, *optional*, defaults to 50):
                The reference number of denoising steps. More denoising steps usually lead to a higher quality image at
                the expense of slower inference. This parameter will be modulated by `strength`, as explained above.
            guidance_scale (`float`, *optional*, defaults to 7.5):
                Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
                `guidance_scale` is defined as `w` of equation 2. of [Imagen
                Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale >
                1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`,
                usually at the expense of lower image quality.
            num_images_per_prompt (`int`, *optional*, defaults to 1):
                The number of images to generate per prompt.
            add_predicted_noise (`bool`, *optional*, defaults to True):
                Use predicted noise instead of random noise when constructing noisy versions of the original image in
                the reverse diffusion process
            eta (`float`, *optional*, defaults to 0.0):
                Corresponds to parameter eta (η) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
                [`schedulers.DDIMScheduler`], will be ignored for others.
            generator (`torch.Generator` or `List[torch.Generator]`, *optional*):
                One or a list of [torch generator(s)](https://pytorch.org/docs/stable/generated/torch.Generator.html)
                to make generation deterministic.
            prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt weighting. If not
                provided, text embeddings will be generated from `prompt` input argument.
            negative_prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated negative text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt
                weighting. If not provided, negative_prompt_embeds will be generated from `negative_prompt` input
                argument.
            max_embeddings_multiples (`int`, *optional*, defaults to `3`):
                The max multiple length of prompt embeddings compared to the max output length of text encoder.
            output_type (`str`, *optional*, defaults to `"pil"`):
                The output format of the generate image. Choose between
                [PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`.
            return_dict (`bool`, *optional*, defaults to `True`):
                Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
                plain tuple.
            callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. The function will be
                called with the following arguments: `callback(step: int, timestep: int, latents: torch.FloatTensor)`.
            is_cancelled_callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. If the function returns
                `True`, the inference will be cancelled.
            callback_steps (`int`, *optional*, defaults to 1):
                The frequency at which the `callback` function will be called. If not specified, the callback will be
                called at every step.
            cross_attention_kwargs (`dict`, *optional*):
                A kwargs dictionary that if specified is passed along to the `AttentionProcessor` as defined under
                `self.processor` in
                [diffusers.cross_attention](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/cross_attention.py).

        Returns:
            `None` if cancelled by `is_cancelled_callback`,
            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] if `return_dict` is True, otherwise a `tuple.
            When returning a tuple, the first element is a list with the generated images, and the second element is a
            list of `bool`s denoting whether the corresponding generated image likely represents "not-safe-for-work"
            (nsfw) content, according to the `safety_checker`.
        """
        return self.__call__(
            prompt=prompt,
            negative_prompt=negative_prompt,
            image=image,
            mask_image=mask_image,
            num_inference_steps=num_inference_steps,
            guidance_scale=guidance_scale,
            strength=strength,
            num_images_per_prompt=num_images_per_prompt,
            add_predicted_noise=add_predicted_noise,
            eta=eta,
            generator=generator,
            prompt_embeds=prompt_embeds,
            negative_prompt_embeds=negative_prompt_embeds,
            max_embeddings_multiples=max_embeddings_multiples,
            output_type=output_type,
            return_dict=return_dict,
            callback=callback,
            is_cancelled_callback=is_cancelled_callback,
            callback_steps=callback_steps,
            cross_attention_kwargs=cross_attention_kwargs,
        )

source/sd/diffuser_pipeline/lpw_stable_diffusion.py
92MING/Stable-Renderer
import math
import torch
import PIL
import numpy
import cv2
import torch.nn.functional as F
from PIL import Image
from packaging import version
from typing import List, Dict, Callable, Union, Optional, Any, Tuple
from diffusers.image_processor import PipelineImageInput, VaeImageProcessor
from diffusers.utils.torch_utils import is_compiled_module
from diffusers.pipelines.controlnet.multicontrolnet import MultiControlNetModel
from diffusers.models.controlnet import ControlNetModel
from diffusers import (
    StableDiffusionControlNetInpaintPipeline,
    AutoencoderKL,
    UNet2DConditionModel,
    DiffusionPipeline
)
from diffusers.configuration_utils import FrozenDict
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import deprecate, logging
from transformers import CLIPTextModel, CLIPTokenizer
from .diffuser_pipeline.lpw_stable_diffusion import StableDiffusionLongPromptWeightingPipeline, preprocess_image, preprocess_mask


logger = logging.get_logger(__name__)  # pylint: disable=invalid-name


class StableDiffusionImg2VideoPipeline(StableDiffusionLongPromptWeightingPipeline, StableDiffusionControlNetInpaintPipeline, DiffusionPipeline):
    def __init__(
        self,
        vae: AutoencoderKL,
        text_encoder: CLIPTextModel,
        tokenizer: CLIPTokenizer,
        unet: UNet2DConditionModel,
        controlnet: Union[ControlNetModel, List[ControlNetModel], Tuple[ControlNetModel], MultiControlNetModel],
        scheduler,
        safety_checker,
        feature_extractor,
        requires_safety_checker: bool = True
    ):
        DiffusionPipeline.__init__(self)

        if hasattr(scheduler.config, "steps_offset") and scheduler.config.steps_offset != 1:
            deprecation_message = (
                f"The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"
                f" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure "
                "to update the config accordingly as leaving `steps_offset` might led to incorrect results"
                " in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,"
                " it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`"
                " file"
            )
            deprecate("steps_offset!=1", "1.0.0", deprecation_message, standard_warn=False)
            new_config = dict(scheduler.config)
            new_config["steps_offset"] = 1
            scheduler._internal_dict = FrozenDict(new_config)

        if hasattr(scheduler.config, "clip_sample") and scheduler.config.clip_sample is True:
            deprecation_message = (
                f"The configuration file of this scheduler: {scheduler} has not set the configuration `clip_sample`."
                " `clip_sample` should be set to False in the configuration file. Please make sure to update the"
                " config accordingly as not setting `clip_sample` in the config might lead to incorrect results in"
                " future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it would be very"
                " nice if you could open a Pull request for the `scheduler/scheduler_config.json` file"
            )
            deprecate("clip_sample not set", "1.0.0", deprecation_message, standard_warn=False)
            new_config = dict(scheduler.config)
            new_config["clip_sample"] = False
            scheduler._internal_dict = FrozenDict(new_config)

        if safety_checker is None and requires_safety_checker:
            logger.warning(
                f"You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"
                " that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered"
                " results in services or applications open to the public. Both the diffusers team and Hugging Face"
                " strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling"
                " it only for use-cases that involve analyzing network behavior or auditing its results. For more"
                " information, please have a look at https://github.com/huggingface/diffusers/pull/254 ."
            )

        if safety_checker is not None and feature_extractor is None:
            raise ValueError(
                "Make sure to define a feature extractor when loading {self.__class__} if you want to use the safety"
                " checker. If you do not want to use the safety checker, you can pass `'safety_checker=None'` instead."
            )

        is_unet_version_less_0_9_0 = hasattr(unet.config, "_diffusers_version") and version.parse(
            version.parse(unet.config._diffusers_version).base_version
        ) < version.parse("0.9.0.dev0")
        is_unet_sample_size_less_64 = hasattr(unet.config, "sample_size") and unet.config.sample_size < 64
        if is_unet_version_less_0_9_0 and is_unet_sample_size_less_64:
            deprecation_message = (
                "The configuration file of the unet has set the default `sample_size` to smaller than"
                " 64 which seems highly unlikely. If your checkpoint is a fine-tuned version of any of the"
                " following: \n- CompVis/stable-diffusion-v1-4 \n- CompVis/stable-diffusion-v1-3 \n-"
                " CompVis/stable-diffusion-v1-2 \n- CompVis/stable-diffusion-v1-1 \n- runwayml/stable-diffusion-v1-5"
                " \n- runwayml/stable-diffusion-inpainting \n you should change 'sample_size' to 64 in the"
                " configuration file. Please make sure to update the config accordingly as leaving `sample_size=32`"
                " in the config might lead to incorrect results in future versions. If you have downloaded this"
                " checkpoint from the Hugging Face Hub, it would be very nice if you could open a Pull request for"
                " the `unet/config.json` file"
            )
            deprecate("sample_size<64", "1.0.0", deprecation_message, standard_warn=False)
            new_config = dict(unet.config)
            new_config["sample_size"] = 64
            unet._internal_dict = FrozenDict(new_config)

        if isinstance(controlnet, (list, tuple)):
            controlnet = MultiControlNetModel(controlnet)

        self.register_modules(
            vae=vae,
            text_encoder=text_encoder,
            tokenizer=tokenizer,
            unet=unet,
            controlnet=controlnet,
            scheduler=scheduler,
            safety_checker=safety_checker,
            feature_extractor=feature_extractor,
        )
        self.vae_scale_factor = 2 ** (len(self.vae.config.block_out_channels) - 1)

        self.image_processor = VaeImageProcessor(vae_scale_factor=self.vae_scale_factor)
        self.control_image_processor = VaeImageProcessor(
            vae_scale_factor=self.vae_scale_factor, do_convert_rgb=True, do_normalize=False
        )
        self.register_to_config(
            requires_safety_checker=requires_safety_checker,
        )

    def check_inputs(
        self,
        prompt,
        images,
        height,
        width,
        callback_steps,
        negative_prompt=None,
        prompt_embeds=None,
        negative_prompt_embeds=None,
        controlnet_conditioning_scale=1.0,
        control_guidance_start=0.0,
        control_guidance_end=1.0,
    ):
        if height is not None and height % 8 != 0 or width is not None and width % 8 != 0:
            raise ValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}.")

        if (callback_steps is None) or (
            callback_steps is not None and (not isinstance(callback_steps, int) or callback_steps <= 0)
        ):
            raise ValueError(
                f"`callback_steps` has to be a positive integer but is {callback_steps} of type"
                f" {type(callback_steps)}."
            )

        if prompt is not None and prompt_embeds is not None:
            raise ValueError(
                f"Cannot forward both `prompt`: {prompt} and `prompt_embeds`: {prompt_embeds}. Please make sure to"
                " only forward one of the two."
            )
        elif prompt is None and prompt_embeds is None:
            raise ValueError(
                "Provide either `prompt` or `prompt_embeds`. Cannot leave both `prompt` and `prompt_embeds` undefined."
            )
        elif prompt is not None and (not isinstance(prompt, str) and not isinstance(prompt, list)):
            raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(prompt)}")

        if negative_prompt is not None and negative_prompt_embeds is not None:
            raise ValueError(
                f"Cannot forward both `negative_prompt`: {negative_prompt} and `negative_prompt_embeds`:"
                f" {negative_prompt_embeds}. Please make sure to only forward one of the two."
            )

        if prompt_embeds is not None and negative_prompt_embeds is not None:
            if prompt_embeds.shape != negative_prompt_embeds.shape:
                raise ValueError(
                    "`prompt_embeds` and `negative_prompt_embeds` must have the same shape when passed directly, but"
                    f" got: `prompt_embeds` {prompt_embeds.shape} != `negative_prompt_embeds`"
                    f" {negative_prompt_embeds.shape}."
                )

        # `prompt` needs more sophisticated handling when there are multiple
        # conditionings.
        if isinstance(self.controlnet, MultiControlNetModel):
            if isinstance(prompt, list):
                logger.warning(
                    f"You have {len(self.controlnet.nets)} ControlNets and you have passed {len(prompt)}"
                    " prompts. The conditionings will be fixed across the prompts."
                )

        # Check `image`
        is_compiled = hasattr(F, "scaled_dot_product_attention") and isinstance(
            self.controlnet, torch._dynamo.eval_frame.OptimizedModule
        )
        if (
            isinstance(self.controlnet, ControlNetModel)
            or is_compiled
            and isinstance(self.controlnet._orig_mod, ControlNetModel)
        ):
            [self.check_image(image, prompt, prompt_embeds) for image in images]
        elif (
            isinstance(self.controlnet, MultiControlNetModel)
            or is_compiled
            and isinstance(self.controlnet._orig_mod, MultiControlNetModel)
        ):
            for image in images:
                if not isinstance(image, list):
                    raise TypeError("For multiple controlnets: `images` must be type `list` of `list`")

                # When `image` is a nested list:
                # (e.g. [[canny_image_1, pose_image_1], [canny_image_2, pose_image_2]])
                elif any(isinstance(i, list) for i in image):
                    raise ValueError("A single batch of multiple conditionings are supported at the moment.")
                elif len(image) != len(self.controlnet.nets):
                    raise ValueError(
                        f"For multiple controlnets: `image` must have the same length as the number of controlnets, but got {len(images)} images and {len(self.controlnet.nets)} ControlNets."
                    )

                for image_ in image:
                    self.check_image(image_, prompt, prompt_embeds)
        else:
            assert False

        # Check `controlnet_conditioning_scale`
        if (
            isinstance(self.controlnet, ControlNetModel)
            or is_compiled
            and isinstance(self.controlnet._orig_mod, ControlNetModel)
        ):
            if not isinstance(controlnet_conditioning_scale, float):
                raise TypeError("For single controlnet: `controlnet_conditioning_scale` must be type `float`.")
        elif (
            isinstance(self.controlnet, MultiControlNetModel)
            or is_compiled
            and isinstance(self.controlnet._orig_mod, MultiControlNetModel)
        ):
            if isinstance(controlnet_conditioning_scale, list):
                if any(isinstance(i, list) for i in controlnet_conditioning_scale):
                    raise ValueError("A single batch of multiple conditionings are supported at the moment.")
            elif isinstance(controlnet_conditioning_scale, list) and len(controlnet_conditioning_scale) != len(
                self.controlnet.nets
            ):
                raise ValueError(
                    "For multiple controlnets: When `controlnet_conditioning_scale` is specified as `list`, it must have"
                    " the same length as the number of controlnets"
                )
        else:
            assert False

        if len(control_guidance_start) != len(control_guidance_end):
            raise ValueError(
                f"`control_guidance_start` has {len(control_guidance_start)} elements, but `control_guidance_end` has {len(control_guidance_end)} elements. Make sure to provide the same number of elements to each list."
            )

        if isinstance(self.controlnet, MultiControlNetModel):
            if len(control_guidance_start) != len(self.controlnet.nets):
                raise ValueError(
                    f"`control_guidance_start`: {control_guidance_start} has {len(control_guidance_start)} elements but there are {len(self.controlnet.nets)} controlnets available. Make sure to provide {len(self.controlnet.nets)}."
                )

        for start, end in zip(control_guidance_start, control_guidance_end):
            if start >= end:
                raise ValueError(
                    f"control guidance start: {start} cannot be larger or equal to control guidance end: {end}."
                )
            if start < 0.0:
                raise ValueError(f"control guidance start: {start} can't be smaller than 0.")
            if end > 1.0:
                raise ValueError(f"control guidance end: {end} can't be larger than 1.0.")

    # Copied from diffusers.pipelines.controlnet.pipeline_controlnet.StableDiffusionControlNetPipeline.prepare_image
    def prepare_control_image(
        self,
        image,
        width,
        height,
        batch_size,
        num_images_per_prompt,
        device,
        dtype,
        do_classifier_free_guidance=False,
        guess_mode=False,
    ):
        image = self.control_image_processor.preprocess(image, height=height, width=width).to(dtype=torch.float32)
        image_batch_size = image.shape[0]

        if image_batch_size == 1:
            repeat_by = batch_size
        else:
            # image batch size is the same as prompt batch size
            repeat_by = num_images_per_prompt

        image = image.repeat_interleave(repeat_by, dim=0)

        image = image.to(device=device, dtype=dtype)

        if do_classifier_free_guidance and not guess_mode:
            image = torch.cat([image] * 2)

        return image

    @torch.no_grad()
    def __call__(
        self,
        prompt: Union[str, List[str]],
        negative_prompt: Optional[Union[str, List[str]]] = None,
        images: List[PipelineImageInput] = None,
        mask_images: List[PipelineImageInput] = None,
        control_images: Union[List[PipelineImageInput], List[List[PipelineImageInput]]] = None,
        blur_radius: int = 0,
        height: int = 512,
        width: int = 512,
        num_inference_steps: int = 50,
        guidance_scale: float = 7.5,
        strength: float = 0.8,
        num_images_per_prompt: Optional[int] = 1,
        add_predicted_noise: Optional[bool] = False,
        eta: float = 0.0,
        generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None,
        latents: Optional[torch.FloatTensor] = None,
        prompt_embeds: Optional[torch.FloatTensor] = None,
        negative_prompt_embeds: Optional[torch.FloatTensor] = None,
        max_embeddings_multiples: Optional[int] = 3,
        output_type: Optional[str] = "pil",
        return_dict: bool = True,
        callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,
        is_cancelled_callback: Optional[Callable[[], bool]] = None,
        callback_steps: int = 1,
        cross_attention_kwargs: Optional[Dict[str, Any]] = None,
        guidance_rescale: float = 0.0,
        controlnet_conditioning_scale: Union[float, List[float]] = 0.5,
        guess_mode: bool = False,
        control_guidance_start: Union[float, List[float]] = 0.0,
        control_guidance_end: Union[float, List[float]] = 1.0,
    ):
        r"""
        Function invoked when calling the pipeline for generation.

        Args:
            prompt (`str` or `List[str]`):
                The prompt or prompts to guide the image generation.
            negative_prompt (`str` or `List[str]`, *optional*):
                The prompt or prompts not to guide the image generation. Ignored when not using guidance (i.e., ignored
                if `guidance_scale` is less than `1`).
            images (`List[torch.FloatTensor]` or `List[PIL.Image.Image]`):
                List of frames. Every element, `Image`, or tensor representing an image batch, that will be used as the
                starting point for the process.
            mask_images (`List[torch.FloatTensor]` or `List[PIL.Image.Image]`):
                `Image`, or tensor representing an image batch, to mask `image`. White pixels in the mask will be
                replaced by noise and therefore repainted, while black pixels will be preserved. If `mask_image` is a
                PIL image, it will be converted to a single channel (luminance) before use. If it's a tensor, it should
                contain one color channel (L) instead of 3, so the expected shape would be `(B, H, W, 1)`.
            control_images (`List[torch.FloatTensor]`, `List[PIL.Image.Image]`, `List[List[torch.FloatTensor]]`, or `List[List[PIL.Image.Image]]`):
                List of control images. Every element, `Image`, or tensor representing an image batch, that will be used as the
                control image for the process. If `control_images` is a list of lists, then every element of the outer list will
                be used as a set of control images for the corresponding element of `images`. This is for MultiControlNet.
            blur_radius (`int`, *optional*, defaults to 0):
                The radius of the blur filter applied to the mask image.
            height (`int`, *optional*, defaults to 512):
                The height in pixels of the generated image.
            width (`int`, *optional*, defaults to 512):
                The width in pixels of the generated image.
            num_inference_steps (`int`, *optional*, defaults to 50):
                The number of denoising steps. More denoising steps usually lead to a higher quality image at the
                expense of slower inference.
            guidance_scale (`float`, *optional*, defaults to 7.5):
                Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
                `guidance_scale` is defined as `w` of equation 2. of [Imagen
                Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale >
                1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`,
                usually at the expense of lower image quality.
            strength (`float`, *optional*, defaults to 0.8):
                Conceptually, indicates how much to transform the reference `image`. Must be between 0 and 1.
                `image` will be used as a starting point, adding more noise to it the larger the `strength`. The
                number of denoising steps depends on the amount of noise initially added. When `strength` is 1, added
                noise will be maximum and the denoising process will run for the full number of iterations specified in
                `num_inference_steps`. A value of 1, therefore, essentially ignores `image`.
            num_images_per_prompt (`int`, *optional*, defaults to 1):
                The number of images to generate per prompt.
            add_predicted_noise (`bool`, *optional*, defaults to True):
                Use predicted noise instead of random noise when constructing noisy versions of the original image in
                the reverse diffusion process
            eta (`float`, *optional*, defaults to 0.0):
                Corresponds to parameter eta (η) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
                [`schedulers.DDIMScheduler`], will be ignored for others.
            generator (`torch.Generator` or `List[torch.Generator]`, *optional*):
                One or a list of [torch generator(s)](https://pytorch.org/docs/stable/generated/torch.Generator.html)
                to make generation deterministic.
            latents (`torch.FloatTensor`, *optional*):
                Pre-generated noisy latents, sampled from a Gaussian distribution, to be used as inputs for image
                generation. Can be used to tweak the same generation with different prompts. If not provided, a latents
                tensor will ge generated by sampling using the supplied random `generator`.
            prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt weighting. If not
                provided, text embeddings will be generated from `prompt` input argument.
            negative_prompt_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated negative text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt
                weighting. If not provided, negative_prompt_embeds will be generated from `negative_prompt` input
                argument.
            max_embeddings_multiples (`int`, *optional*, defaults to `3`):
                The max multiple length of prompt embeddings compared to the max output length of text encoder.
            output_type (`str`, *optional*, defaults to `"pil"`):
                The output format of the generate image. Choose between
                [PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`.
            return_dict (`bool`, *optional*, defaults to `True`):
                Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
                plain tuple.
            callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. The function will be
                called with the following arguments: `callback(step: int, timestep: int, latents: torch.FloatTensor)`.
            is_cancelled_callback (`Callable`, *optional*):
                A function that will be called every `callback_steps` steps during inference. If the function returns
                `True`, the inference will be cancelled.
            callback_steps (`int`, *optional*, defaults to 1):
                The frequency at which the `callback` function will be called. If not specified, the callback will be
                called at every step.
            cross_attention_kwargs (`dict`, *optional*):
                A kwargs dictionary that if specified is passed along to the `AttentionProcessor` as defined under
                `self.processor` in
                [diffusers.cross_attention](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/cross_attention.py).
            guidance_rescale (`float`, *optional*, defaults to 0.7):
                Guidance rescale factor proposed by [Common Diffusion Noise Schedules and Sample Steps are
                Flawed](https://arxiv.org/pdf/2305.08891.pdf) `guidance_scale` is defined as `φ` in equation 16. of
                [Common Diffusion Noise Schedules and Sample Steps are Flawed](https://arxiv.org/pdf/2305.08891.pdf).
                Guidance rescale factor should fix overexposure when using zero terminal SNR.
            controlnet_conditioning_scale (`float` or `List[float]`, *optional*, defaults to 0.5):
                The outputs of the ControlNet are multiplied by `controlnet_conditioning_scale` before they are added
                to the residual in the original `unet`. If multiple ControlNets are specified in `init`, you can set
                the corresponding scale as a list.
            guess_mode (`bool`, *optional*, defaults to `False`):
                The ControlNet encoder tries to recognize the content of the input image even if you remove all
                prompts. A `guidance_scale` value between 3.0 and 5.0 is recommended.
            control_guidance_start (`float` or `List[float]`, *optional*, defaults to 0.0):
                The percentage of total steps at which the ControlNet starts applying.
            control_guidance_end (`float` or `List[float]`, *optional*, defaults to 1.0):
                The percentage of total steps at which the ControlNet stops applying.

        Returns:
            `None` if cancelled by `is_cancelled_callback`,
            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`:
            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] if `return_dict` is True, otherwise a `tuple.
            When returning a tuple, the first element is a list with the generated images, and the second element is a
            list of `bool`s denoting whether the corresponding generated image likely represents "not-safe-for-work"
            (nsfw) content, according to the `safety_checker`.
        """

        # 1. Align format for control guidance
        controlnet = self.controlnet._orig_mod if is_compiled_module(self.controlnet) else self.controlnet

        if not isinstance(control_guidance_start, list) and isinstance(control_guidance_end, list):
            control_guidance_start = len(control_guidance_end) * [control_guidance_start]
        elif not isinstance(control_guidance_end, list) and isinstance(control_guidance_start, list):
            control_guidance_end = len(control_guidance_start) * [control_guidance_end]
        elif not isinstance(control_guidance_start, list) and not isinstance(control_guidance_end, list):
            mult = len(controlnet.nets) if isinstance(controlnet, MultiControlNetModel) else 1
            control_guidance_start, control_guidance_end = mult * [control_guidance_start], mult * [
                control_guidance_end
            ]

        # 2. Default height and width to unet
        height = height or self.unet.config.sample_size * self.vae_scale_factor
        width = width or self.unet.config.sample_size * self.vae_scale_factor

        # 3. Check inputs. Raise error if not correct
        self.check_inputs(
            prompt=prompt,
            images=images,
            height=height,
            width=width,
            callback_steps=callback_steps,
            negative_prompt=negative_prompt,
            prompt_embeds=prompt_embeds,
            negative_prompt_embeds=negative_prompt_embeds,
            controlnet_conditioning_scale=controlnet_conditioning_scale,
            control_guidance_start=control_guidance_start,
            control_guidance_end=control_guidance_end,
        )

        # 4. Define call parameters
        if prompt is not None and isinstance(prompt, str):
            batch_size = 1
        elif prompt is not None and isinstance(prompt, list):
            batch_size = len(prompt)
        else:
            batch_size = prompt_embeds.shape[0]

        device = self._execution_device
        # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
        # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
        # corresponds to doing no classifier free guidance.
        do_classifier_free_guidance = guidance_scale > 1.0

        if isinstance(controlnet, MultiControlNetModel) and isinstance(controlnet_conditioning_scale, float):
            controlnet_conditioning_scale = [controlnet_conditioning_scale] * len(controlnet.nets)

        global_pool_conditions = (
            controlnet.config.global_pool_conditions
            if isinstance(controlnet, ControlNetModel)
            else controlnet.nets[0].config.global_pool_conditions
        )
        guess_mode = guess_mode or global_pool_conditions

        # 5. Encode input prompt
        prompt_embeds = self._encode_prompt(
            prompt,
            device,
            num_images_per_prompt,
            do_classifier_free_guidance,
            negative_prompt,
            max_embeddings_multiples,
            prompt_embeds=prompt_embeds,
            negative_prompt_embeds=negative_prompt_embeds,
        )
        dtype = prompt_embeds.dtype

        # 6. Preprocess image and mask
        if images is not None:
            for i, image in enumerate(images):
                if isinstance(image, PIL.Image.Image):
                    image = preprocess_image(image, batch_size)
                    image = image.to(device=self.device, dtype=dtype)
                else:
                    image = None
                images[i] = image
        # 6.1 Blur mask edges
        if blur_radius:
            mask_images = [cv2.GaussianBlur(mask_image, (blur_radius, blur_radius), sigmaX=0) for mask_image in mask_images]

        if mask_images is not None:
            masks = []
            for i, mask_image in enumerate(mask_images):
                if isinstance(mask_image, PIL.Image.Image):
                    mask = preprocess_mask(mask_image, batch_size, self.vae_scale_factor)
                    mask = mask.to(device=self.device, dtype=dtype)
                    mask = torch.cat([mask] * num_images_per_prompt)
                else:
                    mask_image = None
                masks.append(mask_image)

        # 7. Prepare control image
        if isinstance(controlnet, ControlNetModel):
            control_images: List = [self.prepare_control_image(
                image=control_image,
                width=width,
                height=height,
                batch_size=batch_size * num_images_per_prompt,
                num_images_per_prompt=num_images_per_prompt,
                device=device,
                dtype=controlnet.dtype,
                do_classifier_free_guidance=do_classifier_free_guidance,
                guess_mode=guess_mode,
            ) for control_image in control_images]

        elif isinstance(controlnet, MultiControlNetModel):
            all_control_images = []
            for control_image in control_images:

                control_images_ = []

                for control_image_ in control_image:
                    control_image_ = self.prepare_control_image(
                        image=control_image_,
                        width=width,
                        height=height,
                        batch_size=batch_size * num_images_per_prompt,
                        num_images_per_prompt=num_images_per_prompt,
                        device=device,
                        dtype=controlnet.dtype,
                        do_classifier_free_guidance=do_classifier_free_guidance,
                        guess_mode=guess_mode,
                    )

                    control_images_.append(control_image_)

                all_control_images.append(control_images_)

            control_images: List[List] = all_control_images
        else:
            assert False

        # 8. Set timesteps
        self.scheduler.set_timesteps(num_inference_steps, device=device)
        timesteps, num_inference_steps = self.get_timesteps(num_inference_steps, strength, device, images is None)
        latent_timestep = timesteps[:1].repeat(batch_size * num_images_per_prompt)

        # 9. Prepare latent variables
        if images is not None:
            num_images = len(images)
            latents_list = []
            init_latents_orig_list = []
            noise_list = []
            for image in images:
                latents, init_latents_orig, noise = self.prepare_latents(
                    image,
                    latent_timestep,
                    num_images_per_prompt,
                    batch_size,
                    self.unet.config.in_channels,
                    height,
                    width,
                    dtype,
                    device,
                    generator,
                    latents,
                )
                latents_list.append(latents)
                init_latents_orig_list.append(init_latents_orig)
                noise_list.append(noise)
        else:
            num_images = 0
            latents, init_latents_orig, noise = self.prepare_latents(
                None,
                latent_timestep,
                num_images_per_prompt,
                batch_size,
                self.unet.config.in_channels,
                height,
                width,
                dtype,
                device,
                generator,
                latents,
            )
            latents_list = [latents]
            init_latents_orig_list = [init_latents_orig]
            noise_list = [noise]

        # 10. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
        extra_step_kwargs = self.prepare_extra_step_kwargs(generator, eta)

        # 10.1 Create tensor stating which controlnets to keep
        controlnet_keep = []
        for i in range(len(timesteps)):
            keeps = [
                1.0 - float(i / len(timesteps) < i or (i + 1) / len(timesteps) > e)
                for i, e in zip(control_guidance_start, control_guidance_end)
            ]
            controlnet_keep.append(keeps[0] if isinstance(controlnet, ControlNetModel) else keeps)

        # 11. Denoising loop
        num_warmup_steps = len(timesteps) - num_inference_steps * self.scheduler.order
        with self.progress_bar(total=num_inference_steps) as progress_bar:
            for s, t in enumerate(timesteps):
                # zero value
                denoised_latents_frame_list = []
                for j in range(num_images):
                    latents_frame = latents_list[j]
                    control_image = control_images[j]

                    # expand the latents if we are doing classifier-free guidance to avoid doing two forward passes.
                    latent_model_input = torch.cat([latents_frame] * 2) if do_classifier_free_guidance else latents_frame
                    latent_model_input = self.scheduler.scale_model_input(latent_model_input, t)

                    # controlnet(s) inference
                    if guess_mode and do_classifier_free_guidance:
                        # Infer ControlNet only for the conditional batch.
                        control_model_input = latents
                        control_model_input = self.scheduler.scale_model_input(control_model_input, t)
                        controlnet_prompt_embeds = prompt_embeds.chunk(2)[1]
                    else:
                        control_model_input = latent_model_input
                        controlnet_prompt_embeds = prompt_embeds

                    if isinstance(controlnet_keep[j], list):
                        cond_scale = [c * s for c, s in zip(controlnet_conditioning_scale, controlnet_keep[s])]
                    else:
                        controlnet_cond_scale = controlnet_conditioning_scale
                        if isinstance(controlnet_cond_scale, list):
                            controlnet_cond_scale = controlnet_cond_scale[0]
                        cond_scale = controlnet_cond_scale * controlnet_keep[s]

                    down_block_res_samples, mid_block_res_sample = self.controlnet(
                        control_model_input,
                        t,
                        encoder_hidden_states=controlnet_prompt_embeds,
                        controlnet_cond=control_image,
                        conditioning_scale=cond_scale,
                        guess_mode=guess_mode,
                        return_dict=False,
                    )

                    if guess_mode and do_classifier_free_guidance:
                        # Infered ControlNet only for the conditional batch.
                        # To apply the output of ControlNet to both the unconditional and conditional batches,
                        # add 0 to the unconditional batch to keep it unchanged.
                        down_block_res_samples = [torch.cat([torch.zeros_like(d), d]) for d in down_block_res_samples]
                        mid_block_res_sample = torch.cat([torch.zeros_like(mid_block_res_sample), mid_block_res_sample])

                    # predict the noise residual
                    noise_pred = self.unet(
                        latent_model_input,
                        t,
                        down_block_additional_residuals=down_block_res_samples,
                        mid_block_additional_residual=mid_block_res_sample,
                        encoder_hidden_states=prompt_embeds,
                        cross_attention_kwargs=cross_attention_kwargs,
                    ).sample

                    del latent_model_input

                    # perform guidance
                    if do_classifier_free_guidance:
                        noise_pred_uncond, noise_pred_cond = noise_pred.chunk(2)
                        noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_cond - noise_pred_uncond)
                        noise_pred = rescale_noise_cfg(noise_pred, noise_pred_cond, guidance_rescale=guidance_rescale)

                        del noise_pred_uncond, noise_pred_cond

                    # compute the previous noisy sample x_t -> x_t-1
                    latents_frame = self.scheduler.step(noise_pred, t, latents_frame, **extra_step_kwargs, return_dict=False)[0]
                    self.scheduler._step_index -= 1

                    # handle inpainting
                    if mask_images is not None:
                        mask = masks[j]
                        init_latents_orig = init_latents_orig[j]
                        noise = noise_list[j]
                        # masking
                        if add_predicted_noise:
                            init_latents_proper = self.scheduler.add_noise(
                                init_latents_orig, noise_pred_uncond, torch.tensor([t])
                            )
                        else:
                            init_latents_proper = self.scheduler.add_noise(init_latents_orig, noise, torch.tensor([t]))
                        latents = (init_latents_proper * mask) + (latents * (1 - mask))

                    denoised_latents_frame_list.append(latents_frame)

                self.scheduler._step_index += 1

                # Overlapping
                for i, latents_frame in enumerate(denoised_latents_frame_list):
                    latents_list[i].zero_()
                    count = 0
                    for j in range(num_images):
                        distance = abs(i - j)
                        weight = math.exp(-distance)
                        latents_list[i] += denoised_latents_frame_list[j] * weight
                        count += weight
                    latents_list[i] /= count

                # call the callback, if provided
                if s == len(timesteps) - 1 or ((s + 1) > num_warmup_steps and (s + 1) % self.scheduler.order == 0):
                    progress_bar.update()
                    if s % callback_steps == 0:
                        if callback is not None:
                            callback(s, t, latents_list)
                        if is_cancelled_callback is not None and is_cancelled_callback():
                            return None

        if output_type == "latent":
            images = latents_list
            # has_nsfw_concept = None
        elif output_type == "pil":
            # 12. Post-processing
            images = [self.decode_latents(latents) for latents in latents_list]

            # 13. Run safety checker
            # image, has_nsfw_concept = self.run_safety_checker(image, device, prompt_embeds.dtype)

            # 14. Convert to PIL
            images = [self.numpy_to_pil(image) for image in images]
        else:
            # 12. Post-processing
            images = [self.decode_latents(latents) for latents in latents_list]

            # 13. Run safety checker
            # image, has_nsfw_concept = self.run_safety_checker(image, device, prompt_embeds.dtype)

        # Offload last model to CPU
        if hasattr(self, "final_offload_hook") and self.final_offload_hook is not None:
            self.final_offload_hook.offload()

        if not return_dict:
            return images, None

        return StableDiffusionPipelineOutput(images=images, nsfw_content_detected=None)


def make_canny_images(images: List[Image.Image], threshold1=100, threshold2=200) -> List[Image.Image]:
    """
    Make canny images from a list of PIL images.
    """
    canny_images = []
    for image in images:
        image = numpy.array(image)
        image = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
        image = cv2.Canny(image, threshold1, threshold2)
        image = Image.fromarray(image)
        canny_images.append(image)
    return canny_images


def rescale_noise_cfg(noise_cfg, noise_pred_text, guidance_rescale=0.0):
    """
    Rescale `noise_cfg` according to `guidance_rescale`. Based on findings of [Common Diffusion Noise Schedules and
    Sample Steps are Flawed](https://arxiv.org/pdf/2305.08891.pdf). See Section 3.4
    """
    std_text = noise_pred_text.std(dim=list(range(1, noise_pred_text.ndim)), keepdim=True)
    std_cfg = noise_cfg.std(dim=list(range(1, noise_cfg.ndim)), keepdim=True)
    # rescale the results from guidance (fixes overexposure)
    noise_pred_rescaled = noise_cfg * (std_text / std_cfg)
    # mix with the original results from guidance by factor guidance_rescale to avoid "plain looking" images
    noise_cfg = guidance_rescale * noise_pred_rescaled + (1 - guidance_rescale) * noise_cfg
    return noise_cfg


def load_pipe(model_path, control_net_model_paths, use_safetensors=True, scheduler_type="euler-ancestral") -> StableDiffusionImg2VideoPipeline:
    # load control net and stable diffusion v1-5 model into one pipeline.

    if isinstance(control_net_model_paths, str):
        controlnet = ControlNetModel.from_pretrained(
            control_net_model_paths,
            torch_dtype=torch.float16
        )
    elif isinstance(control_net_model_paths, list):
        controlnets = []
        for control_net_model_path in control_net_model_paths:
            controlnet = ControlNetModel.from_pretrained(
                control_net_model_path,
                torch_dtype=torch.float16
            )
            controlnets.append(controlnet)
        controlnet = MultiControlNetModel(controlnets)

    pipe: StableDiffusionImg2VideoPipeline = StableDiffusionImg2VideoPipeline.from_single_file(
        model_path,
        controlnet=controlnet,
        torch_dtype=torch.float16,
        use_safetensors=use_safetensors,
        local_files_only=True,
        scheduler_type=scheduler_type,
    ).to("cuda")

    pipe.safety_checker = None

    # Enable XFormers
    pipe.enable_xformers_memory_efficient_attention(attention_op=None)
    pipe.enable_model_cpu_offload()
    return pipe


def make_inpaint_condition(image, image_mask):
    image = numpy.array(image.convert("RGB")).astype(numpy.float32) / 255.0
    image_mask = numpy.array(image_mask.convert("L")).astype(numpy.float32) / 255.0

    assert image.shape[0:1] == image_mask.shape[0:1], "image and image_mask must have the same image size"
    image[image_mask > 0.5] = -1.0  # set as masked pixel
    image = numpy.expand_dims(image, 0).transpose(0, 3, 1, 2)
    image = torch.from_numpy(image)
    return image


if __name__ == "__main__":
    import random
    from pathlib import Path
    pipe: StableDiffusionImg2VideoPipeline = load_pipe(
        model_path="models/Stable-Diffusion/AIDv2.10.safetensors",
        use_safetensors=True,
        scheduler_type="euler-ancestral"
    )

    neg_emb_path, neg_emb_token = "models/embeddings/aid210.pt", 'aid210'
    pipe.load_textual_inversion(neg_emb_path, token=neg_emb_token)

    prompt = "best quality, masterpiece, by ask, 1boy, solo, male focus, handsome"
    neg_prompt = neg_emb_token

    width = 848
    height = 480

    image_dir = Path("test/multidiffusion/frames")
    images = [Image.open(impath) for impath in image_dir.iterdir()]
    images = [image.resize((width, height), resample=Image.Resampling.LANCZOS) for image in images]
    canny_images = make_canny_images(images)

    seed = random.randint(0, 9999999999)
    generator = torch.Generator(device="cuda").manual_seed(seed)

    images = pipe.__call__(
        prompt=prompt,
        negative_prompt=neg_prompt,
        images=images,
        mask_images=None,
        control_images=canny_images,
        width=width,
        height=height,
        num_inference_steps=32,
        strength=0.75,
        generator=generator,
        guidance_scale=9
    ).images

    output_dir = Path("test/multidiffusion/outputs/frames")
    output_dir.mkdir(parents=True, exist_ok=True)

    for i, image in enumerate(images):
        image[0].save(output_dir.joinpath(f"frame_{i}.png"))

source/sd/multi_frame_stable_diffusion.py
92MING/Stable-Renderer
import glm
from typing import cast

def _getVal(val):
    if isinstance(val, int):
        val = val / 255.0
    if isinstance(val, float):
        val = max(0.0, min(1.0, val))
    else:
        raise TypeError('val must be int or float')
    return val

class _classproperty:
    def __init__(self, func):
        self._func = func
    def __get__(self, cls, owner):
        return self._func(owner)

class Color(glm.vec4):

    def __new__(cls, r, g, b, a=1.0):
        r = _getVal(r)
        g = _getVal(g)
        b = _getVal(b)
        a = _getVal(a)
        return cast(Color, glm.vec4(r, g, b, a))

    # region constants
    @_classproperty
    def BLACK(cls):
        if not hasattr(cls, '_BLACK'):
            cls._BLACK = cls(0, 0, 0)
        return cls._BLACK
    @_classproperty
    def WHITE(cls):
        if not hasattr(cls, '_WHITE'):
            cls._WHITE = cls(255, 255, 255)
        return cls._WHITE
    @_classproperty
    def RED(cls):
        if not hasattr(cls, '_RED'):
            cls._RED = cls(255, 0, 0)
        return cls._RED
    @_classproperty
    def GREEN(cls):
        if not hasattr(cls, '_GREEN'):
            cls._GREEN = cls(0, 255, 0)
        return cls._GREEN
    @_classproperty
    def BLUE(cls):
        if not hasattr(cls, '_BLUE'):
            cls._BLUE = cls(0, 0, 255)
        return cls._BLUE
    @_classproperty
    def YELLOW(cls):
        if not hasattr(cls, '_YELLOW'):
            cls._YELLOW = cls(255, 255, 0)
        return cls._YELLOW
    @_classproperty
    def ORANGE(cls):
        if not hasattr(cls, '_ORANGE'):
            cls._ORANGE = cls(255, 165, 0)
        return cls._ORANGE
    @_classproperty
    def PURPLE(cls):
        if not hasattr(cls, '_PURPLE'):
            cls._PURPLE = cls(128, 0, 128)
        return cls._PURPLE
    @_classproperty
    def CYAN(cls):
        if not hasattr(cls, '_CYAN'):
            cls._CYAN = cls(0, 255, 255)
        return cls._CYAN
    @_classproperty
    def MAGENTA(cls):
        if not hasattr(cls, '_MAGENTA'):
            cls._MAGENTA = cls(255, 0, 255)
        return cls._MAGENTA
    @_classproperty
    def GRAY(cls):
        if not hasattr(cls, '_GRAY'):
            cls._GRAY = cls(128, 128, 128)
        return cls._GRAY
    @_classproperty
    def CLEAR(cls):
        if not hasattr(cls, '_CLEAR'):
            cls._CLEAR = cls(0, 0, 0, 0)
        return cls._CLEAR
    # endregion

    def _tidy_rgba(self):
        self.r = _getVal(self.r)
        self.g = _getVal(self.g)
        self.b = _getVal(self.b)
        self.a = _getVal(self.a)

    @property
    def r(self):
        return self[0]
    @r.setter
    def r(self, val):
        self[0] = _getVal(val)
    @property
    def g(self):
        return self[1]
    @g.setter
    def g(self, val):
        self[1] = _getVal(val)
    @property
    def b(self):
        return self[2]
    @b.setter
    def b(self, val):
        self[2] = _getVal(val)
    @property
    def a(self):
        return self[3]
    @a.setter
    def a(self, val):
        self[3] = _getVal(val)
    @property
    def rgb(self):
        return self.xyz
    def hsv(self):
        '''return hsv color in glm.vec4(h, s, v, a)'''
        h, s, v = 0.0, 0.0, 0.0
        max_val = max(self.r, self.g, self.b)
        min_val = min(self.r, self.g, self.b)

        if max_val == min_val:
            h = 0
        elif max_val == self.r:
            h = 60 * (self.g - self.b) / (max_val - min_val)
        elif max_val == self.g:
            h = 60 * (self.b - self.r) / (max_val - min_val) + 120
        elif max_val == self.b:
            h = 60 * (self.r - self.g) / (max_val - min_val) + 240
        if h < 0:
            h += 360
        s = 0 if max_val == 0 else (max_val - min_val) / max_val
        v = max_val
        return glm.vec4(h, s, v, self.a)
    def set_from_hsv(self, hsv: glm.vec3, alpha: float = 1.0):
        h, s, v = hsv.x, hsv.y, hsv.z
        if s == 0:
            self.r, self.g, self.b = v, v, v
        else:
            i = int(h / 60)
            f = h / 60 - i
            p = v * (1 - s)
            q = v * (1 - s * f)
            t = v * (1 - s * (1 - f))

            if i == 0:
                self.r, self.g, self.b = v, t, p
            elif i == 1:
                self.r, self.g, self.b = q, v, p
            elif i == 2:
                self.r, self.g, self.b = p, v, t
            elif i == 3:
                self.r, self.g, self.b = p, q, v
            elif i == 4:
                self.r, self.g, self.b = t, p, v
            elif i == 5:
                self.r, self.g, self.b = v, p, q
        self.a = alpha
        self._tidy_rgba()

__all__ = ['Color']
source/static/color.py
92MING/Stable-Renderer
# -*- coding: utf-8 -*-
'''this module contains enums, constants, etc.'''
from enum import Enum
import OpenGL.GL as gl

# region opengl
class DepthFunc(Enum):
    NEVER = gl.GL_NEVER
    LESS = gl.GL_LESS
    EQUAL = gl.GL_EQUAL
    LEQUAL = gl.GL_LEQUAL
    GREATER = gl.GL_GREATER
    NOTEQUAL = gl.GL_NOTEQUAL
    GEQUAL = gl.GL_GEQUAL
    ALWAYS = gl.GL_ALWAYS
# endregion

# region render
class RenderOrder:
    """
    Render order of GameObjects.
    Note that this is not a enum class.
    """
    OPAQUE = 1000
    TRANSPARENT = 2000
    OVERLAY = 3000

class ProjectionType(Enum):
    PERSPECTIVE = 0
    ORTHOGRAPHIC = 1

class LightType(Enum):
    DIRECTIONAL_LIGHT = 0
    POINT_LIGHT = 1
    SPOT_LIGHT = 2
# endregion

# region input
class _FindableEnum(Enum):
    @classmethod
    def GetEnum(cls, value):
        for e in cls:
            if e.value == value:
                return e
        return None
class MouseButton(_FindableEnum):
    LEFT = 0
    RIGHT = 1
    MIDDLE = 2
    MOUSE_BTN_3 = 3
    MOUSE_BTN_4 = 4
    MOUSE_BTN_5 = 5
    MOUSE_BTN_6 = 6
    MOUSE_BTN_7 = 7
    MOUSE_BTN_8 = 8
class InputAction(_FindableEnum):
    RELEASE = 0
    PRESS = 1
    HOLD = 999
class InputModifier(_FindableEnum):
    NONE = 0
    SHIFT = 1
    CTRL = 2
    ALT = 4
    SUPER = 8
class Key(_FindableEnum):
    UNKNOWN = -1
    SPACE = 32
    APOSTROPHE = 39
    COMMA = 44
    MINUS = 45
    PERIOD = 46
    SLASH = 47
    KEY_0 = 48
    KEY_1 = 49
    KEY_2 = 50
    KEY_3 = 51
    KEY_4 = 52
    KEY_5 = 53
    KEY_6 = 54
    KEY_7 = 55
    KEY_8 = 56
    KEY_9 = 57
    SEMICOLON = 59
    EQUAL = 61
    A = 65
    B = 66
    C = 67
    D = 68
    E = 69
    F = 70
    G = 71
    H = 72
    I = 73
    J = 74
    K = 75
    L = 76
    M = 77
    N = 78
    O = 79
    P = 80
    Q = 81
    R = 82
    S = 83
    T = 84
    U = 85
    V = 86
    W = 87
    X = 88
    Y = 89
    Z = 90
    LEFT_BRACKET = 91
    BACKSLASH = 92
    RIGHT_BRACKET = 93
    GRAVE_ACCENT = 96
    WORLD_1 = 161
    WORLD_2 = 162
    ESCAPE = 256
    ENTER = 257
    TAB = 258
    BACKSPACE = 259
    INSERT = 260
    DELETE = 261
    RIGHT = 262
    LEFT = 263
    DOWN = 264
    UP = 265
    PAGE_UP = 266
    PAGE_DOWN = 267
    HOME = 268
    END = 269
    CAPS_LOCK = 280
    SCROLL_LOCK = 281
    NUM_LOCK = 282
    PRINT_SCREEN = 283
    PAUSE = 284
    F1 = 290
    F2 = 291
    F3 = 292
    F4 = 293
    F5 = 294
    F6 = 295
    F7 = 296
    F8 = 297
    F9 = 298
    F10 = 299
    F11 = 300
    F12 = 301
# endregion

# region resource
class TextureWrap(Enum):
    REPEAT = gl.GL_REPEAT
    MIRRORED_REPEAT = gl.GL_MIRRORED_REPEAT
    MIRROR_CLAMP_TO_EDGE = gl.GL_MIRROR_CLAMP_TO_EDGE
    CLAMP_TO_EDGE = gl.GL_CLAMP_TO_EDGE
    CLAMP_TO_BORDER = gl.GL_CLAMP_TO_BORDER
class TextureFilter(Enum):
    NEAREST = gl.GL_NEAREST
    LINEAR = gl.GL_LINEAR
    NEAREST_MIPMAP_NEAREST = gl.GL_NEAREST_MIPMAP_NEAREST
    LINEAR_MIPMAP_NEAREST = gl.GL_LINEAR_MIPMAP_NEAREST
    NEAREST_MIPMAP_LINEAR = gl.GL_NEAREST_MIPMAP_LINEAR
    LINEAR_MIPMAP_LINEAR = gl.GL_LINEAR_MIPMAP_LINEAR
class TextureFormat(Enum):
    RGB = gl.GL_RGB
    RGBA = gl.GL_RGBA
    DEPTH_COMPONENT = gl.GL_DEPTH_COMPONENT
    DEPTH_STENCIL = gl.GL_DEPTH_STENCIL
    def get_PIL_convert_mode(self):
        if self == TextureFormat.RGB:
            return 'RGB'
        elif self == TextureFormat.RGBA:
            return 'RGBA'
        elif self == TextureFormat.DEPTH_COMPONENT:
            return 'L'
        elif self == TextureFormat.DEPTH_STENCIL:
            return 'L'
    def get_default_internal_format(self):
        if self == TextureFormat.RGB:
            return gl.GL_RGB
        elif self == TextureFormat.RGBA:
            return gl.GL_RGBA
        elif self == TextureFormat.DEPTH_COMPONENT:
            return gl.GL_DEPTH_COMPONENT16
        elif self == TextureFormat.DEPTH_STENCIL:
            return gl.GL_DEPTH24_STENCIL8

class PrimitiveType(Enum):
    """Primitive drawing types of OpenGL"""
    POINTS = gl.GL_POINTS
    LINES = gl.GL_LINES
    LINE_STRIP = gl.GL_LINE_STRIP
    LINE_LOOP = gl.GL_LINE_LOOP
    TRIANGLES = gl.GL_TRIANGLES
    TRIANGLE_STRIP = gl.GL_TRIANGLE_STRIP
    TRIANGLE_FAN = gl.GL_TRIANGLE_FAN
    QUADS = gl.GL_QUADS
    QUAD_STRIP = gl.GL_QUAD_STRIP
    POLYGON = gl.GL_POLYGON
class ShaderType(Enum):
    """Shader types of OpenGL"""
    VERTEX = gl.GL_VERTEX_SHADER
    FRAGMENT = gl.GL_FRAGMENT_SHADER
    GEOMETRY = gl.GL_GEOMETRY_SHADER
    TESS_CONTROL = gl.GL_TESS_CONTROL_SHADER
    TESS_EVALUATION = gl.GL_TESS_EVALUATION_SHADER
    COMPUTE = gl.GL_COMPUTE_SHADER
# endregion
source/static/enums.py
92MING/Stable-Renderer
from .mesh import Mesh
from .mesh_OBJ import Mesh_OBJ
source/static/mesh/__init__.py
92MING/Stable-Renderer
import os
from static.resourcesObj import ResourcesObj

class Mesh(ResourcesObj):
    _BaseName = 'Mesh'

    def load(self, path:str):
        '''Load data from file. Override this function to implement loading data from file'''
        raise NotImplementedError
    def draw(self):
        '''Draw the mesh'''
        raise NotImplementedError

    @classmethod
    def Load(cls, path: str, name=None)->'Mesh':
        '''
        load a mesh from file
        :param path: path of the mesh file
        :param name: name is used to identify the mesh. If not specified, the name will be the path of the mesh file
        :return: Mesh object
        '''
        path, name = cls._GetPathAndName(path, name)
        if '.' not in os.path.basename(path):
            raise ValueError('path must contain file extension')
        ext = path.split('.')[-1]
        formatCls = cls.FindFormat(ext)
        if formatCls is None:
            raise ValueError(f'unsupported mesh format: {ext}')
        mesh = formatCls(name)
        mesh.load(path)
        return mesh
source/static/mesh/mesh.py
92MING/Stable-Renderer
from static.enums import PrimitiveType
import OpenGL.GL as gl
import numpy as np
import ctypes
from .mesh import Mesh

class Mesh_OBJ(Mesh):

    _Format = 'obj'

    def __init__(self, name):
        super().__init__(name)
        self.drawMode : PrimitiveType = None
        self.vao = None
        self.vbo = None
        self.has_normals = False
        self.has_uvs = False
        self.vertexCount = 0 # number of vertices per face
        self.vertices = []
    def clear(self):
        super().clear()
        if self.vbo is not None:
            gl.glDeleteBuffers(1, self.vbo)
            self.vbo = None
        if self.vao is not None:
            gl.glDeleteVertexArrays(1, self.vao)
            self.vao = None
        self.has_normals = False
        self.has_uvs = False
        self.vertices.clear()
    def load(self, path:str):
        temp_points = []
        temp_normals = []
        temp_uvs = []
        for line in open(path, 'r').readlines():
            if line.startswith('#'): continue
            if line.startswith('v'): temp_points.extend(map(float, line.split(' ')[1:4]))
            if line.startswith('vn'): temp_normals.extend(map(float, line.split(' ')[1:4]))
            if line.startswith('vt'): temp_uvs.extend(map(float, line.split(' ')[1:3]))
            if line.startswith('f'):
                faces = line.split(' ')[1:]
                if self.drawMode is None:
                    if len(faces) == 3:
                        self.drawMode = PrimitiveType.TRIANGLES
                    elif len(faces) == 4:
                        self.drawMode = PrimitiveType.QUADS
                    elif len(faces) > 4:
                        self.drawMode = PrimitiveType.POLYGON
                    else:
                        raise ValueError(f'invalid faces: {line}')
                    self.vertexCount = len(faces)
                for face in faces:
                    face = face.split('/')
                    if len(face) == 1: # only point index
                        index = int(face[0]) - 1
                        self.vertices.extend(temp_points[index * 3 : index * 3 + 3])
                    elif (len(face)==2): # point index and uv index
                        ptIndex = int(face[0]) - 1
                        uvIndex = int(face[1]) - 1
                        self.vertices.extend(temp_points[ptIndex * 3 : ptIndex * 3 + 3])
                        self.vertices.extend(temp_uvs[uvIndex * 2 : uvIndex * 2 + 2])
                    elif len(face)==3:
                        if face[1] == '': # point index and normal index
                            ptIndex = int(face[0]) - 1
                            normalIndex = int(face[2]) - 1
                            self.vertices.extend(temp_points[ptIndex * 3 : ptIndex * 3 + 3])
                            self.vertices.extend(temp_normals[normalIndex * 3 : normalIndex * 3 + 3])
                        else: # point index, uv index and normal index
                            ptIndex = int(face[0]) - 1
                            uvIndex = int(face[1]) - 1
                            normalIndex = int(face[2]) - 1
                            self.vertices.extend(temp_points[ptIndex * 3 : ptIndex * 3 + 3])
                            self.vertices.extend(temp_normals[normalIndex * 3: normalIndex * 3 + 3])
                            self.vertices.extend(temp_uvs[uvIndex * 2 : uvIndex * 2 + 2])
                    else:
                        raise ValueError(f'invalid faces: {line}')
        self.has_normals = len(temp_normals) > 0
        self.has_uvs = len(temp_uvs) > 0
    def sendToGPU(self):
        super().sendToGPU()
        if self.vao is not None:
            return # already sent to GPU
        if len(self.vertices) == 0:
            raise Exception('No data to send to GPU')
        self.vao = gl.glGenVertexArrays(1)
        gl.glBindVertexArray(self.vao)

        self.vbo = gl.glGenBuffers(1)
        gl.glBindBuffer(gl.GL_ARRAY_BUFFER, self.vbo)
        vertices_data = np.array(self.vertices, dtype=np.float32)
        gl.glBufferData(gl.GL_ARRAY_BUFFER, vertices_data.nbytes, vertices_data.data, gl.GL_STATIC_DRAW)

        if self.has_normals and self.has_uvs:
            stride = 8 * 4
        elif self.has_normals:
            stride = 6 * 4
        elif self.has_uvs:
            stride = 5 * 4
        else:
            stride = 3 * 4

        # position
        gl.glEnableVertexAttribArray(0)
        gl.glVertexAttribPointer(0, 3, gl.GL_FLOAT, gl.GL_FALSE, stride, None)

        # normal6
        if self.has_normals: # if there is no normal data, then no need to send
            gl.glEnableVertexAttribArray(1)
            gl.glVertexAttribPointer(1, 3, gl.GL_FLOAT, gl.GL_FALSE, stride, ctypes.c_void_p(3 * 4))
        # uv
        if self.has_uvs: # if there is no uv data, then no need to send
            gl.glEnableVertexAttribArray(2)
            gl.glVertexAttribPointer(2, 2, gl.GL_FLOAT, gl.GL_FALSE, stride, ctypes.c_void_p((3 + self.has_normals * 3) * 4))

    def draw(self):
        if len(self.vertices) == 0:
            raise Exception('No data to draw')
        if self.vao is None:
            raise Exception('Data is not sent to GPU')
        gl.glBindVertexArray(self.vao)
        gl.glDrawArrays(self.drawMode.value, 0, len(self.vertices) // self.vertexCount)

source/static/mesh/mesh_OBJ.py
92MING/Stable-Renderer
from utils.global_utils import GetOrAddGlobalValue, GetGlobalValue
import os

_RESOURCES_CLSES = GetOrAddGlobalValue('_RESOURCES_CLSES', dict()) #cls_name: cls
class ResourcesObjMeta(type):
    _Format_Clses:dict = None # format: sub cls
    def __new__(cls, *args, **kwargs):
        cls_name = args[0]
        if cls_name in _RESOURCES_CLSES:
            return _RESOURCES_CLSES[cls_name]
        cls = super().__new__(cls, *args, **kwargs)
        if cls_name == 'ResourcesObj':
            return cls
        if cls._Format_Clses is None:
            cls._Format_Clses = {}
        else:
            if cls.Format() is not None and cls.Format() not in cls._Format_Clses:
                cls._Format_Clses[cls.Format()] = cls
        _RESOURCES_CLSES[cls_name] = cls
        return cls

class ResourcesObj(metaclass=ResourcesObjMeta):
    '''Class for resources that need to load to GPU before get into the main loop'''

    # region cls properties
    _Format:str = None # format of the data, e.g. "obj" for "Mesh_OBJ"
    '''override this property to specify the format of the mesh data, e.g. "obj"'''
    _LoadOrder = 0
    '''override this property to specify the order of loading to GPU by ResourcesManager, the smaller the earlier'''
    _BaseName = None
    '''Override this to set the base name of the class, e.g. "Mesh" for "Mesh_OBJ"'''

    _Instances: dict = None  # {name: Obj, ...}

    @classmethod
    def ClsName(cls):
        return cls.__qualname__
    @classmethod
    def BaseClsName(cls):
        baseClsName = cls._BaseName
        if baseClsName is None:
            baseClsName = cls.ClsName()
            if '_' in baseClsName:
                baseClsName = baseClsName.split('_')[0]
        return baseClsName
    @classmethod
    def FindFormat(cls, format):
        return cls._Format_Clses.get(format, None)
    @classmethod
    def Format(cls):
        if cls._Format is None:
            return None
        format = cls._Format.lower().strip()
        if format.startswith('.'):
            format = format[1:]
        return format
    @classmethod
    def SupportedFormats(cls):
        return cls._Format_Clses.keys()
    def __class_getitem__(cls, item):
        return cls.FindFormat(item)
    @classmethod
    def Find(cls, name):
        if cls._Instances is None:
            cls._Instances = {}
        return cls._Instances.get(name, None)
    @classmethod
    def AllInstances(cls):
        if cls._Instances is None:
            cls._Instances = {}
        return cls._Instances.values()

    @classmethod
    def _GetPathAndName(cls, path, name=None):
        '''Get proper path & name for cls.Load()'''
        if name is None:
            name = os.path.basename(path).split('.')[0]
        if cls.Find(name) is not None:
            raise ValueError(f'Resources type {cls.BaseClsName()} with name "{name}" already exists')
        return path, name
    @classmethod
    def Load(cls, path, name=None)->'ResourcesObj':
        '''
        Override this method to load the data from file.
        e.g.:
            def Load(cls, path, name=None):
                path, name = cls._GetNameAndPath(path, name)
                ... # your code here
        '''
        raise NotImplementedError
    # endregion

    def __new__(cls, name, *args, **kwargs):
        if cls._Instances is None:
            cls._Instances = {}
        if name in cls._Instances:
            obj = cls._Instances[name]
            obj.__init__ = lambda *a, **kw: None
            return obj
        else:
            obj = super().__new__(cls)
            obj._name = name
            cls._Instances[name] = obj
            return obj
    def __init__(self, name, *args, **kwargs):
        pass
    @property
    def name(self):
        return self._name
    @property
    def engine(self)->'Engine':
        return GetGlobalValue('_ENGINE_SINGLETON')

    # region abstract methods
    def sendToGPU(self):
        '''Override this to send the data to GPU'''
        print(f'sending {self.BaseClsName()}: {self.name} to GPU...')
    def clear(self):
        '''Override this to clear the data from GPU'''
        print(f'clearing {self.BaseClsName()}: {self.name} from GPU...')
    # endregion

__all__ = ['ResourcesObj', 'ResourcesObjMeta']
source/static/resourcesObj.py
92MING/Stable-Renderer
import os.path

from utils.base_clses import NamedObj
import inspect
from typing import Literal

class Scene(NamedObj):

    def prepare(self):
        '''Override this method to prepare your scene.'''
        raise NotImplementedError

    @classmethod
    def Load(cls, path):
        '''Load a scene from a file.'''
        basename = os.path.basename(path)
        name = os.path.splitext(basename)[0]
        with open(path, 'r') as f:
            source = f.read()
        scene = cls(name)
        scene.deserialize(source)
        return scene

    def serialize(self, format:Literal['python','json']='python')->str:
        '''Serialize this scene to a file.'''
        if format == 'python':
            code = inspect.getsource(self.prepare)
            codeLines = [line.strip() for line in code.split('\n')][2:]
            codeLines.insert(0, '# Python code')
            code = '\n'.join(codeLines)
            return code
        else:
            # TODO: implement this
            raise NotImplementedError
    def deserialize(self, source:str):
        '''Deserialize this scene from source str.'''
        lines = source.split('\n')
        if 'Python' in lines[0]:
            self.prepare = lambda : exec(source)
        elif 'JSON' in lines[0]:
            # TODO: implement this
            raise NotImplementedError
        else:
            raise Exception('Unknown format')
    def save(self, path, overwrite=False):
        '''Save this scene to a file.'''
        if os.path.exists(path):
            if os.path.isdir(path):
                path = os.path.join(path, f'{self.name}.scene')
                if os.path.exists(path) and not overwrite:
                    raise Exception(f'File {path} already exists.')
                with open(path, 'w') as f:
                    f.write(self.serialize())
            elif not overwrite:
                raise Exception(f'File {path} already exists.')
            else:
                with open(path, 'w') as f:
                    f.write(self.serialize())
        else:
            with open(path, 'w') as f:
                f.write(self.serialize())

    def __repr__(self):
        return f"<Scene {self.name}>"
    def __str__(self):
        return self.__repr__()

source/static/scene.py
92MING/Stable-Renderer
from OpenGL.GL import *
from utils.base_clses import NamedObj
from .enums import ShaderType
import glm
from runtime.engineObj import EngineObj

class Shader(NamedObj, EngineObj):

    def __init__(self, name, vertex_source_path:str, fragment_source_path:str):
        super().__init__(name)
        self._vertex_source = open(vertex_source_path, 'r').read()
        self._fragment_source = open(fragment_source_path, 'r').read()
        self._v_shaderID = self._init_shader(self._vertex_source, ShaderType.VERTEX)
        self._f_shaderID = self._init_shader(self._fragment_source, ShaderType.FRAGMENT)
        self._programID = self._init_program(self._v_shaderID, self._f_shaderID)
        print('Loaded shader: ', name)

    def _init_shader(self, source:str, type:ShaderType):
        shaderID = glCreateShader(type.value)
        if shaderID == 0:
            self.engine.RenderManager.printOpenGLError()
            raise RuntimeError(f'Failed to create shader {self.name}')
        glShaderSource(shaderID, source)
        glCompileShader(shaderID)
        # check shader compile status
        if glGetShaderiv(shaderID, GL_COMPILE_STATUS, None) == GL_FALSE:
            info_log = glGetShaderInfoLog(shaderID)
            raise Exception(f'Failed to compile shader {self.name}. Error msg: {info_log}')
        return shaderID
    def _init_program(self, v_shaderID, f_shaderID):
        program = glCreateProgram()
        if program == 0:
            self.engine.RenderManager.printOpenGLError()
            raise RuntimeError(f'Failed to create program when initializing shader: {self.name}')
        glAttachShader(program, v_shaderID)
        glAttachShader(program, f_shaderID)
        glLinkProgram(program)
        if glGetProgramiv(program, GL_LINK_STATUS, None) == GL_FALSE:
            raise RuntimeError(f'Failed to link program when initializing shader: {self.name}')

        # try to bind uniform blocks
        matrixBlockIndex = glGetUniformBlockIndex(program, "Matrices")
        if matrixBlockIndex != GL_INVALID_INDEX:
            glUniformBlockBinding(program, matrixBlockIndex, self.engine.RenderManager.MatrixUBO_BindingPoint)

        return program

    # region properties
    @property
    def programID(self):
        return self._programID
    @property
    def vertexSource(self):
        return self._vertex_source
    @property
    def fragmentSource(self):
        return self._fragment_source
    @property
    def vertexShaderID(self):
        return self._v_shaderID
    @property
    def fragmentShaderID(self):
        return self._f_shaderID
    # endregion

    def useProgram(self):
        if glUseProgram(self._programID):
            self.engine.RenderManager.printOpenGLError()

    def getUniformID(self, name:str):
        return glGetUniformLocation(self._programID, name)
    def setUniform(self, name, value):
        self.useProgram()
        val_id = self.getUniformID(name)
        if isinstance(value, (tuple, list)):
            if len(value) == 1:
                value = value[0]
            elif len(value) == 2:
                value = glm.vec2(value)
            elif len(value) == 3:
                value = glm.vec3(value)
            elif len(value) == 4:
                value = glm.vec4(value)
            else:
                raise Exception(f'Invalid uniform value: {value}')
        if isinstance(value, glm.uint32):
            return glUniform1ui(val_id, value)
        if isinstance(value, (int, glm.int32)):
            return glUniform1i(val_id, value)
        elif isinstance(value, float):
            return glUniform1f(val_id, value)
        elif isinstance(value, bool):
            return glUniform1i(val_id, int(value))
        elif isinstance(value, glm.vec1):
            if isinstance(value, glm.dvec1):
                return glUniform1d(val_id, value.x)
            elif isinstance(value, glm.ivec1):
                return glUniform1i(val_id, value.x)
            elif isinstance(value, glm.uvec1):
                return glUniform1ui(val_id, value.x)
            else:
                return glUniform1f(val_id, value.x)
        elif isinstance(value, glm.vec2):
            if isinstance(value, glm.dvec2):
                return glUniform2d(val_id, value.x, value.y)
            elif isinstance(value, glm.ivec2):
                return glUniform2i(val_id, value.x, value.y)
            elif isinstance(value, glm.uvec2):
                return glUniform2ui(val_id, value.x, value.y)
            else:
                return glUniform2f(val_id, value.x, value.y)
        elif isinstance(value, glm.vec3):
            if isinstance(value, glm.dvec3):
                return glUniform3d(val_id, value.x, value.y, value.z)
            elif isinstance(value, glm.ivec3):
                return glUniform3i(val_id, value.x, value.y, value.z)
            elif isinstance(value, glm.uvec3):
                return glUniform3ui(val_id, value.x, value.y, value.z)
            else:
                return glUniform3f(val_id, value.x, value.y, value.z)
        elif isinstance(value, glm.vec4):
            if isinstance(value, glm.dvec4):
                return glUniform4d(val_id, value.x, value.y, value.z, value.w)
            elif isinstance(value, glm.ivec4):
                return glUniform4i(val_id, value.x, value.y, value.z, value.w)
            elif isinstance(value, glm.uvec4):
                return glUniform4ui(val_id, value.x, value.y, value.z, value.w)
            else:
                return glUniform4f(val_id, value.x, value.y, value.z, value.w)
        elif isinstance(value, glm.mat2):
            if isinstance(value, glm.dmat2):
                return glUniformMatrix2dv(val_id, 1, GL_FALSE, glm.value_ptr(value))
            else:
                return glUniformMatrix2fv(val_id, 1, GL_FALSE, glm.value_ptr(value))
        elif isinstance(value, glm.mat3):
            if isinstance(value, glm.dmat3):
                return glUniformMatrix3dv(val_id, 1, GL_FALSE, glm.value_ptr(value))
            else:
                return glUniformMatrix3fv(val_id, 1, GL_FALSE, glm.value_ptr(value))
        elif isinstance(value, glm.mat4):
            if isinstance(value, glm.dmat4):
                return glUniformMatrix4dv(val_id, 1, GL_FALSE, glm.value_ptr(value))
            else:
                return glUniformMatrix4fv(val_id, 1, GL_FALSE, glm.value_ptr(value))
        elif isinstance(value, glm.mat2x3):
            if isinstance(value, glm.dmat2x3):
                return glUniformMatrix2x3dv(val_id, 1, GL_FALSE, glm.value_ptr(value))
            else:
                return glUniformMatrix2x3fv(val_id, 1, GL_FALSE, glm.value_ptr(value))
        elif isinstance(value, glm.mat2x4):
            if isinstance(value, glm.dmat2x4):
                return glUniformMatrix2x4dv(val_id, 1, GL_FALSE, glm.value_ptr(value))
            else:
                return glUniformMatrix2x4fv(val_id, 1, GL_FALSE, glm.value_ptr(value))
        elif isinstance(value, glm.mat3x2):
            if isinstance(value, glm.dmat3x2):
                return glUniformMatrix3x2dv(val_id, 1, GL_FALSE, glm.value_ptr(value))
            else:
                return glUniformMatrix3x2fv(val_id, 1, GL_FALSE, glm.value_ptr(value))
        elif isinstance(value, glm.mat3x4):
            if isinstance(value, glm.dmat3x4):
                return glUniformMatrix3x4dv(val_id, 1, GL_FALSE, glm.value_ptr(value))
            else:
                return glUniformMatrix3x4fv(val_id, 1, GL_FALSE, glm.value_ptr(value))
        elif isinstance(value, glm.mat4x2):
            if isinstance(value, glm.dmat4x2):
                return glUniformMatrix4x2dv(val_id, 1, GL_FALSE, glm.value_ptr(value))
            else:
                return glUniformMatrix4x2fv(val_id, 1, GL_FALSE, glm.value_ptr(value))
        elif isinstance(value, glm.mat4x3):
            if isinstance(value, glm.dmat4x3):
                return glUniformMatrix4x3dv(val_id, 1, GL_FALSE, glm.value_ptr(value))
            else:
                return glUniformMatrix4x3fv(val_id, 1, GL_FALSE, glm.value_ptr(value))
        else:
            raise TypeError("Invalid uniform type: {}".format(type(value)))

source/static/shader.py
92MING/Stable-Renderer
from .texture import Texture
from .texture_DDS import Texture_DDS
source/static/texture/__init__.py
92MING/Stable-Renderer
'''
A modified module from https://github.com/robertkist/py_dds by Robert Kist
'''
__author__ = "Robert Kist"
from enum import Enum
import struct
import math
from dataclasses import dataclass
from typing import List, Dict, Optional, Callable, Any, Union, Type, Tuple, Literal
from OpenGL.GL.EXT.texture_compression_s3tc import *
from OpenGL.GL.EXT.texture_sRGB import *
import OpenGL.GL as gl

class DDSException(Exception):
    """a class for throwing DDSImage specific exceptions"""

class DDSFormat(Enum):
    """supported DDS formats, including DX9 and DX10 formats"""
    DXT1 = "DXT1"  # includes DXT1a, DXT1c
    DXT3 = "DXT3"  # includes DXT2, DXT3
    DXT5 = "DXT5"  # includes DXT3, DXT4
    UNCOMPRESSED32 = "UNCOMPRESSED32"
    UNCOMPRESSED24 = "UNCOMPRESSED24"
    DXGI_FORMAT_BC1_UNORM_SRGB = "DXGI_FORMAT_BC1_UNORM_SRGB"
    DXGI_FORMAT_BC2_UNORM_SRGB = "DXGI_FORMAT_BC2_UNORM_SRGB"
    DXGI_FORMAT_BC3_UNORM_SRGB = "DXGI_FORMAT_BC3_UNORM_SRGB"
    def getOpenGLformat(self):
        """returns the OpenGL texture format for the DDSFormat"""
        if self == DDSFormat.DXT1:
            return GL_COMPRESSED_RGBA_S3TC_DXT1_EXT
        elif self == DDSFormat.DXT3:
            return GL_COMPRESSED_RGBA_S3TC_DXT3_EXT
        elif self == DDSFormat.DXT5:
            return GL_COMPRESSED_RGBA_S3TC_DXT5_EXT
        elif self == DDSFormat.UNCOMPRESSED32:
            return gl.GL_RGBA
        elif self == DDSFormat.UNCOMPRESSED24:
            return gl.GL_RGB
        elif self == DDSFormat.DXGI_FORMAT_BC1_UNORM_SRGB:
            return GL_COMPRESSED_SRGB_ALPHA_S3TC_DXT1_EXT
        elif self == DDSFormat.DXGI_FORMAT_BC2_UNORM_SRGB:
            return GL_COMPRESSED_SRGB_ALPHA_S3TC_DXT3_EXT
        elif self == DDSFormat.DXGI_FORMAT_BC3_UNORM_SRGB:
            return GL_COMPRESSED_SRGB_ALPHA_S3TC_DXT5_EXT
    def rgbFormat(self)->Literal['RGB', 'RGBA']:
        """returns the OpenGL texture format for the DDSFormat"""
        if self == DDSFormat.UNCOMPRESSED24:
            return 'RGB'
        return 'RGBA'
    def bytes_per_pixel(self):
        """returns the number of bytes per pixel for the DDSFormat"""
        if self == DDSFormat.DXT1:
            return 0.5
        elif self == DDSFormat.DXT3:
            return 1
        elif self == DDSFormat.DXT5:
            return 1
        elif self == DDSFormat.UNCOMPRESSED32:
            return 4
        elif self == DDSFormat.UNCOMPRESSED24:
            return 3
        elif self == DDSFormat.DXGI_FORMAT_BC1_UNORM_SRGB:
            return 0.5
        elif self == DDSFormat.DXGI_FORMAT_BC2_UNORM_SRGB:
            return 1
        elif self == DDSFormat.DXGI_FORMAT_BC3_UNORM_SRGB:
            return 1

DDS_HEADER_SIZE = 128
DXT10_HEADER_SIZE = 20
SetPixelCallback = Callable[[int, int, int, int, int, int], None]
BCAlphaClass = Union['_BC1Alpha', '_BC2Alpha', '_BC3Alpha']


# helper dict for mapping DX10 header file formats to the unified DDSFormat enum; for internal use
_DxgiFormat: Dict[int, DDSFormat] = {
    72: DDSFormat.DXGI_FORMAT_BC1_UNORM_SRGB,
    75: DDSFormat.DXGI_FORMAT_BC2_UNORM_SRGB,
    78: DDSFormat.DXGI_FORMAT_BC3_UNORM_SRGB,
}


class _DXTCompression(Enum):
    """enum with supported dxt compression types; for internal use"""
    BC1 = "BC1"  # used in DXT1a/c
    BC2 = "BC2"  # used in DXT3/2
    BC3 = "BC3"  # used in DXT5/4


@dataclass(frozen=True)
class _DDSMip:
    """a container for storing mip map information; for internal use"""
    width: int
    height: int
    offset: int  # data start offset in the dds file in bytes
    size: int  # data size in bytes


@dataclass(frozen=True)
class _Color:
    """a container for a single RGBA color value; for internal use by _DDSPalette"""
    r: int = 0
    g: int = 0
    b: int = 0
    a: int = 255


class _DDSPalette:
    """a class for generating color palette information for a 4x4 BC1/2/3 compressed pixel chunk"""
    def __init__(self, data: bytes, ofs: int) -> None:
        color0 = struct.unpack_from("<H", data, ofs)[0]
        color1 = struct.unpack_from("<H", data, ofs + 2)[0]
        self.__palette: List[_Color] = []
        self.__load_palette(color0, color1)

    def get(self, index: int) -> _Color:
        """returns the RGBA color for the given palette index"""
        return self.__palette[index]

    def __lerp(self, vmin: int, vmax: int, value: float) -> int:
        """linearly interpolates value between given min and max"""
        return int(round((vmax - vmin) * value + vmin))

    def __lerp_rgb(self, color1: _Color, color2: _Color, value: float) -> _Color:
        """blends two colors together based on a ratio"""
        r: int = self.__lerp(color1.r, color2.r, value)
        g: int = self.__lerp(color1.g, color2.g, value)
        b: int = self.__lerp(color1.b, color2.b, value)
        return _Color(r, g, b)

    def __get_dxt_color(self, color: int) -> _Color:
        """Converts a 16bit DXT RGB (R(5 bits), G(6 bits), B(5 bits)) value into a regular 8bpp RGB value."""
        r = int(round(255.0 / 31.0 * (color >> 11)))
        g = int(round(255.0 / 63.0 * ((color >> 5) & int('111111', 2))))
        b = int(round(255.0 / 31.0 * (color & int('11111', 2))))
        return _Color(r, g, b)

    def __load_palette(self, color0: int, color1: int) -> None:
        """Constructs all 4 palette entries for a chunk's palette from 2 input colors for DXT1/3/5 compression"""
        c_color0: _Color = self.__get_dxt_color(color0)  # read color 1
        c_color1: _Color = self.__get_dxt_color(color1)  # read color 2
        self.__palette = [c_color0, c_color1]
        if color0 > color1:
            self.__palette.append(self.__lerp_rgb(c_color0, c_color1, 0.3333))  # interpolate color 3
            self.__palette.append(self.__lerp_rgb(c_color0, c_color1, 0.6666))  # interpolate color 4
        else:
            self.__palette.append(self.__lerp_rgb(c_color0, c_color1, 0.5))  # interpolate color 3
            self.__palette.append(_Color(0, 0, 0, 0))  # transparent color


class _BC1Alpha:
    """A dummy class for generating alpha information for a 4x4 BC1 compressed pixel chunk"""

    def __init__(self, *args: Any) -> None:
        pass

    def get(self, color: _Color) -> int:
        """returns the input alpha as DXT1 alpha is calculated from the pixel color itself"""
        return color.a


class _BC2Alpha:
    """A class for generating alpha information for a 4x4 BC2 compressed pixel chunk"""

    def __init__(self, data: bytes, ofs: int) -> None:
        """constructor - reads the chunk's alpha data"""
        self.__alphas: int = struct.unpack_from("<Q", data, ofs)[0]

    def get(self, _: Any) -> int:
        """returns the current pixel's alpha and advances the current pixel to the next pixel"""
        alpha: int = (self.__alphas & 0xf) * 0x11  # mask and expand to full 0-255 range
        self.__alphas = self.__alphas >> 4  # shift alpha by 1 pixel
        return alpha


class _BC3Alpha:
    """A class for generating alpha information for a 4x4 BC3 compressed pixel chunk"""

    def __init__(self, data: bytes, ofs: int) -> None:
        """constructor - reads the chunk's alpha data"""
        alpha0 = int.from_bytes(struct.unpack_from("c", data, ofs)[0], "little")
        alpha1 = int.from_bytes(struct.unpack_from("c", data, ofs + 1)[0], "little")
        self.__code_table: List[int] = [alpha0, alpha1]
        if alpha0 > alpha1:
            for j in range(1, 7):
                self.__code_table.append(int(((7 - j) * alpha0 + j * alpha1) / 7.0))
        else:
            for j in range(1, 5):
                self.__code_table.append(int(((5 - j) * alpha0 + j * alpha1) / 5.0))
            self.__code_table.append(0)
            self.__code_table.append(255)
        part1: int = struct.unpack_from("<I", data, ofs + 2)[0]
        part2: int = struct.unpack_from("<H", data, ofs + 6)[0]
        self.__alphas: int = (part2 << 32) | part1

    def get(self, _: Any) -> int:
        """returns the current pixel's alpha and advances the current pixel to the next pixel"""
        alpha: int = self.__code_table[int(self.__alphas & 0b111)]
        self.__alphas = self.__alphas >> 3
        return alpha


class DDSImage:
    """A class for reading compressed and uncompressed .DDS images"""

    def __init__(self, filename: str) -> None:
        """constructor"""
        self.__data: bytes = b''  # file buffer - holds entire dds file
        self.__format: Optional[DDSFormat] = None  # image format
        self.__color_mask = _Color()  # bit mask for uncompressed pixel data
        self.__mips: List[_DDSMip] = []  # information about mip maps
        self.__is_dx10: bool = False  # helper flag to make it easier to deal with DX10 format
        self.__load(filename)

    def data(self) -> bytes:
        """returns the raw image data"""
        return self.__data

    def width(self, mip: int = 0) -> int:
        """returns the width of the specified mip map number. 0 is the first (largest) mip map"""
        if 0 > mip >= len(self.__mips):
            raise IndexError
        return self.__mips[mip].width

    def height(self, mip: int = 0) -> int:
        """returns the height of the specified mip map number. 0 is the first (largest) mip map"""
        if 0 > mip >= len(self.__mips):
            raise IndexError
        return self.__mips[mip].height

    def format(self) -> Optional[DDSFormat]:
        """returns the DXT image format"""
        return self.__format

    def mip_count(self) -> int:
        """returns the number of mip maps in the file"""
        return len(self.__mips)

    def mip_map(self, mip: int = 0) -> _DDSMip:
        """returns the specified mip map as a bytes object"""
        if 0 > mip >= len(self.__mips):
            raise IndexError
        return self.__mips[mip]

    def compressed(self) -> bool:
        """returns True if the image is compressed, False if it is uncompressed"""
        return self._compressed

    def __set_format(self, pixel_format_flags: int) -> DDSFormat:
        """determines the image format, e.g. DXT1, DXT3, etc. - see DDSFormat enum for supported formats"""
        compressed = bool(pixel_format_flags & 0x4)
        self._compressed = compressed
        if (not compressed and not bool(pixel_format_flags & 0x40)) or (compressed and bool(pixel_format_flags & 0x40)):
            raise DDSException("cannot determine DXT format")
        if compressed:
            four_cc: str = self.__data[84:88].decode("utf8")
            if four_cc == "DX10":
                self.__is_dx10 = True
                try:
                    return _DxgiFormat[struct.unpack_from("<I", self.__data, DDS_HEADER_SIZE)[0]]
                except KeyError as exc:
                    raise DDSException("unsupported DXT10 format") from exc
            try:
                return DDSFormat[four_cc]
            except KeyError as exc:
                raise DDSException("unsupported DXT format") from exc
        bit_count: int = struct.unpack_from("<I", self.__data, 88)[0]
        if bit_count not in [32, 24]:
            raise DDSException("unsupported bit depth")
        return DDSFormat.UNCOMPRESSED32 if bit_count == 32 else DDSFormat.UNCOMPRESSED24

    def __load(self, filename: str) -> None:
        """loads the DDS file from the given file path"""
        with open(filename, 'rb') as fp:
            self.__data = fp.read()
        # check if proper DDS file format
        if not (self.__data[0:3] == b"DDS" and self.__data[4] == 124 and self.__data[76] == 32):
            raise DDSException("not a valid DDS image")
        # read DDS header values
        dds_flags: int = struct.unpack_from("<I", self.__data, 8)[0]
        height: int = struct.unpack_from("<I", self.__data, 12)[0]
        width: int = struct.unpack_from("<I", self.__data, 16)[0]
        mip_count: int = struct.unpack_from("<I", self.__data, 28)[0]
        pixel_format_flags: int = struct.unpack_from("<I", self.__data, 80)[0]
        if bool(dds_flags & 0x800000):
            raise DDSException("cube maps are not supported")
        self.__format = self.__set_format(pixel_format_flags)
        self.__mip_count = mip_count if mip_count > 0 else 1
        # load mip maps
        if self.__format in [DDSFormat.UNCOMPRESSED32, DDSFormat.UNCOMPRESSED24]:
            self.__get_uncompressed_mip_maps_and_color_info(width, height)
        elif self.__format in [DDSFormat.DXT1, DDSFormat.DXGI_FORMAT_BC1_UNORM_SRGB]:
            self.__get_compressed_mip_maps(width, height, 8)  # 64 bit chunks
        elif self.__format in [DDSFormat.DXT3, DDSFormat.DXGI_FORMAT_BC2_UNORM_SRGB, DDSFormat.DXT5,
                               DDSFormat.DXGI_FORMAT_BC3_UNORM_SRGB]:
            self.__get_compressed_mip_maps(width, height, 16)  # 128 bit chunks

    def __get_uncompressed_mip_maps_and_color_info(self, width: int, height: int) -> None:
        """determines color masks and gets uncompressed mip map info; width and height: dimensions of first mip map"""
        # get color and optional alpha masks
        r: int = struct.unpack_from("<I", self.__data, 92)[0]
        g: int = struct.unpack_from("<I", self.__data, 96)[0]
        b: int = struct.unpack_from("<I", self.__data, 100)[0]
        if self.__format == DDSFormat.UNCOMPRESSED32:
            a: int = struct.unpack_from("<I", self.__data, 104)[0]
            self.__color_mask = _Color(r, g, b, a)
            bytes_per_pixel = 4
        else:
            self.__color_mask = _Color(r, g, b, 255)
            self.__data += b"\x00"  # pad buffer so we can read whole 32-bit words when decoding the image
            bytes_per_pixel = 3
        # load mip maps
        offset: int = DDS_HEADER_SIZE
        for _ in range(0, self.__mip_count):
            self.__mips.append(_DDSMip(width, height, offset, width * height * bytes_per_pixel))
            offset += width * height * bytes_per_pixel
            width = int(width / 2)
            height = int(height / 2)

    def __get_compressed_mip_maps(self, width: int, height: int, chunk_size: int) -> None:
        """gets uncompressed mip map info; width and height: dimensions of first mip map"""
        offset: int = DDS_HEADER_SIZE + DXT10_HEADER_SIZE if self.__is_dx10 else DDS_HEADER_SIZE
        for _ in range(0, self.__mip_count):
            size = math.ceil(width / 4) * math.ceil(height / 4) * chunk_size
            self.__mips.append(_DDSMip(width, height, offset, size))
            offset += size
            width = int(width / 2)
            height = int(height / 2)

    def __draw_compressed(self, set_pixel_callback: SetPixelCallback, data: bytes, width: int, height: int,
                          compression: _DXTCompression) -> None:
        """outputs the given BC1/2/3 compressed mip map via the given set_pixel callback."""
        ofs = x = y = i = 0
        mapper: Dict[_DXTCompression, Tuple[Type[BCAlphaClass], int]] = {
            _DXTCompression.BC1: (_BC1Alpha, 0),  # dxt1 - if alpha is present, 1bit alpha is part of pixel data
            _DXTCompression.BC2: (_BC2Alpha, 8),  # dxt2/3 - has alpha
            _DXTCompression.BC3: (_BC3Alpha, 8),  # dxt4/5 - has alpha
        }
        while i < width * height:
            alpha = mapper[compression][0](data, ofs)
            ofs += mapper[compression][1]
            palette = _DDSPalette(data, ofs)  # generate palette
            chunk = struct.unpack_from("<I", data, ofs + 4)[0]  # read 4x4 pixel chunk
            for pos_y in range(y, y + 4):  # draw chunk
                for pos_x in range(x, x + 4):
                    color = palette.get(chunk & int('11', 2))  # 2 bits represent a palette index to look-up
                    if pos_x < width and pos_y < height:
                        set_pixel_callback(pos_x, pos_y, color.r, color.g, color.b, alpha.get(color))
                        i += 1
                    chunk = chunk >> 2  # shift color chunk by 1 pixels
            ofs += 8
            x += 4
            if x >= width:
                y += 4
                x = 0

    def __draw_uncompressed(self, set_pixel_callback: SetPixelCallback, data: bytes, width: int, height: int) -> None:
        """outputs the given uncompressed mip map via the given set_pixel callback."""
        ofs = x = y = 0
        bytes_per_pixel = 3 if self.__format == DDSFormat.UNCOMPRESSED24 else 4
        while ofs < width * height * bytes_per_pixel:
            color: int = struct.unpack_from("<I", data, ofs)[0]
            b: int = color & self.__color_mask.b
            g: int = (color & self.__color_mask.g) >> 8
            r: int = (color & self.__color_mask.r) >> 16
            a: int = (color & self.__color_mask.a) >> 24 if self.__format == DDSFormat.UNCOMPRESSED32 else 255
            set_pixel_callback(x, y, r, g, b, a)
            ofs += bytes_per_pixel
            x += 1
            if x >= width:
                y += 1
                x = 0

    def draw(self, set_pixel_callback: SetPixelCallback, mip: int) -> None:
        """
        outputs the given mip map via the given set_pixel callback.
        The callback must accept x and y coordinates and 4 8-bit RGBA values as parameters:
            set_pixel_callback(x: int, y: int, r: int, g: int, b: int, a: int) -> None
        """
        if 0 > mip >= len(self.__mips):
            raise IndexError
        data = self.__data[self.__mips[mip].offset:]
        width = self.__mips[mip].width
        height = self.__mips[mip].height
        if self.__format in [DDSFormat.UNCOMPRESSED32, DDSFormat.UNCOMPRESSED24]:
            self.__draw_uncompressed(set_pixel_callback, data, width, height)
        elif self.__format in [DDSFormat.DXT1, DDSFormat.DXGI_FORMAT_BC1_UNORM_SRGB]:
            self.__draw_compressed(set_pixel_callback, data, width, height, _DXTCompression.BC1)
        elif self.__format in [DDSFormat.DXT3, DDSFormat.DXGI_FORMAT_BC2_UNORM_SRGB]:
            self.__draw_compressed(set_pixel_callback, data, width, height, _DXTCompression.BC2)
        elif self.__format in [DDSFormat.DXT5, DDSFormat.DXGI_FORMAT_BC3_UNORM_SRGB]:
            self.__draw_compressed(set_pixel_callback, data, width, height, _DXTCompression.BC3)

__all__ = ['DDSImage', 'DDSException', 'DDSFormat']
source/static/texture/ddsImage.py
92MING/Stable-Renderer
import OpenGL.error
import glfw
import numpy as np

from static.resourcesObj import ResourcesObj
import os, time
from PIL import Image
from static.enums import *

class Texture(ResourcesObj):
    _BaseName = 'Texture'

    def __init__(self, name,
                 format:TextureFormat=TextureFormat.RGB,
                 min_filter:TextureFilter=TextureFilter.LINEAR_MIPMAP_LINEAR,
                 mag_filter:TextureFilter=TextureFilter.LINEAR,
                 s_wrap:TextureWrap=TextureWrap.REPEAT,
                 t_wrap:TextureWrap=TextureWrap.REPEAT,
                 internalFormat=None # you could specify a different internal format
                 ):
        super().__init__(name)
        self._texID = None
        self._format = format
        self._min_filter = min_filter
        self._mag_filter = mag_filter
        self._s_wrap = s_wrap
        self._t_wrap = t_wrap
        self._width = None
        self._height = None
        self._buffer = None
        self._internalFormat = internalFormat

    # region properties
    @property
    def textureID(self):
        return self._texID
    @property
    def internalFormat(self):
        return self._internalFormat or self.format.get_default_internal_format()
    @property
    def buffer(self):
        return self._buffer
    @property
    def width(self):
        return self._width
    @property
    def height(self):
        return self._height
    @property
    def format(self):
        return self._format
    @format.setter
    def format(self, value):
        if self._texID is not None:
            raise Exception('Cannot change tex format after texture is loaded.')
        self._format = value
    @property
    def min_filter(self):
        return self._min_filter
    @min_filter.setter
    def min_filter(self, value:TextureFilter):
        if self._texID is not None:
            raise Exception('Cannot change filter after texture is loaded.')
        self._min_filter = value
    @property
    def mag_filter(self):
        return self._mag_filter
    @mag_filter.setter
    def mag_filter(self, value:TextureFilter):
        if self._texID is not None:
            raise Exception('Cannot change filter after texture is loaded.')
        self._mag_filter = value
    @property
    def s_wrap(self):
        return self._s_wrap
    @s_wrap.setter
    def s_wrap(self, value):
        if self._texID is not None:
            raise Exception('Cannot change s_wrap after texture is loaded.')
        self._s_wrap = value
    @property
    def t_wrap(self):
        return self._t_wrap
    @t_wrap.setter
    def t_wrap(self, value):
        if self._texID is not None:
            raise Exception('Cannot change t_wrap after texture is loaded.')
        self._t_wrap = value
    # endregion

    def bind(self, slot:int, uniformID):
        if self._texID is None:
            raise Exception('Texture is not yet sent to GPU. Cannot bind.')
        gl.glActiveTexture(gl.GL_TEXTURE0 + slot)
        gl.glBindTexture(gl.GL_TEXTURE_2D, self._texID)
        gl.glUniform1i(uniformID, slot)

    def load(self, path):
        '''Default load by PIL'''
        image = Image.open(path).convert(self.format.get_PIL_convert_mode())
        self._buffer = image.transpose(Image.FLIP_TOP_BOTTOM).tobytes()
        self._height = image.height
        self._width = image.width
        image.close()
    def sendToGPU(self):
        '''Send data to GPU memory'''
        if self._texID is not None:
            return # already loaded
        self._texID = gl.glGenTextures(1)

        if self.format == TextureFormat.DEPTH_COMPONENT:
            type = gl.GL_UNSIGNED_INT
        elif self.format == TextureFormat.DEPTH_STENCIL:
            type = gl.GL_UNSIGNED_INT_24_8
        else:
            type = gl.GL_UNSIGNED_BYTE

        gl.glBindTexture(gl.GL_TEXTURE_2D, self._texID)
        gl.glTexImage2D(gl.GL_TEXTURE_2D, 0, self.internalFormat, self.width, self.height, 0, self.format.value, type, self.buffer)

        gl.glTexParameteri(gl.GL_TEXTURE_2D, gl.GL_TEXTURE_WRAP_S, self.s_wrap.value)
        gl.glTexParameteri(gl.GL_TEXTURE_2D, gl.GL_TEXTURE_WRAP_T, self.t_wrap.value)
        gl.glTexParameteri(gl.GL_TEXTURE_2D, gl.GL_TEXTURE_MAG_FILTER, self.mag_filter.value)
        gl.glTexParameteri(gl.GL_TEXTURE_2D, gl.GL_TEXTURE_MIN_FILTER, self.min_filter.value)
        if (self.mag_filter in (TextureFilter.LINEAR_MIPMAP_LINEAR, TextureFilter.NEAREST_MIPMAP_NEAREST) or
            self.min_filter in (TextureFilter.LINEAR_MIPMAP_LINEAR, TextureFilter.NEAREST_MIPMAP_NEAREST)):
            gl.glGenerateMipmap(gl.GL_TEXTURE_2D)
    def clear(self):
        '''
        Clear all data and release GPU memory if it has been loaded.
        This is a default implementation, you may need to override it.
        '''
        super().clear()
        if self._texID is not None:
            try:
                gl.glDeleteTextures(1, [self._texID])
            except Exception as glError:
                print('Warning: failed to delete texture. Error: {}'.format(glError))
            self._texID = None
        self._buffer = None
        self._width = None
        self._height = None
        self._internalFormat = None

    @classmethod
    def Load(cls, path, name=None,
             format:TextureFormat=TextureFormat.RGB,
             min_filter: TextureFilter = TextureFilter.LINEAR_MIPMAP_LINEAR,
             mag_filter: TextureFilter = TextureFilter.LINEAR,
             s_wrap:TextureWrap=TextureWrap.REPEAT,
             t_wrap:TextureWrap=TextureWrap.REPEAT)->'Texture':

        path, name = cls._GetPathAndName(path, name)

        if '.' in os.path.basename(path):
            ext = path.split('.')[-1]
            formatCls = cls.FindFormat(ext) # try to find a format class
            if formatCls is not None:
                texture = formatCls(name, format, min_filter, mag_filter, s_wrap, t_wrap)
                texture.load(path)
                return texture

        # default load by PIL
        texture = cls(name, format, min_filter, mag_filter, s_wrap, t_wrap)
        texture.load(path)
        return texture

source/static/texture/texture.py
92MING/Stable-Renderer
import OpenGL.GL as gl
from .ddsImage import *
from .texture import Texture

class Texture_DDS(Texture):
    _Format = 'dds'

    def load(self, path):
        dds = DDSImage(path)
        self._dds = dds
        self._width = dds.width()
        self._height = dds.height()
        self._buffer = dds.data()

    def sendToGPU(self):
        if self._texID is not None:
            return
        self._texID = gl.glGenTextures(1)
        gl.glBindTexture(gl.GL_TEXTURE_2D, self._texID)
        format = self._dds.format().getOpenGLformat()
        for level in range(0, self._dds.mip_count()):
            mipMap = self._dds.mip_map(level)
            width, height, offset, size = mipMap.width, mipMap.height, mipMap.offset, mipMap.size
            if self._dds.compressed():
                gl.glCompressedTexImage2D(gl.GL_TEXTURE_2D, level, format, width, height, 0, self._buffer[offset:offset + size])
            else:
                gl.glTexImage2D(gl.GL_TEXTURE_2D, level, format, width, height, 0, format, gl.GL_UNSIGNED_BYTE, self._buffer[offset:offset + size])

    def clear(self):
        super().clear()
        self._dds = None

source/static/texture/texture_DDS.py
92MING/Stable-Renderer
from .cross_module_class import *
from .cross_module_enum import *
from .singleton import *
from .named_obj import *
from .generic_class import *
source/utils/base_clses/__init__.py
92MING/Stable-Renderer
# -*- coding: utf-8 -*-
'''跨模組類，無論在哪裡import，都會得到相同的類'''

from ..global_utils import GetOrAddGlobalValue
_CrossModuleClassDict: dict = GetOrAddGlobalValue("_CrossModuleClassDict", dict())

class CrossModuleClassMeta(type):
    '''跨模組類的Meta class'''
    def __new__(cls, *args, **kwargs):
        clsName = args[0]
        if clsName in _CrossModuleClassDict:
            return _CrossModuleClassDict[clsName]
        else:
            thisCls = super().__new__(cls, *args, **kwargs)
            _CrossModuleClassDict[clsName] = thisCls
            return thisCls

class CrossModuleClass(metaclass=CrossModuleClassMeta):
    '''跨模組類，無論在哪裡import，都會得到相同的類'''
    pass

__all__ = ["CrossModuleClass", "CrossModuleClassMeta"]
source/utils/base_clses/cross_module_class.py
92MING/Stable-Renderer
# -*- coding: utf-8 -*-
'''跨模組Enum，無論在哪裡import，都返回同一個Enum類。名稱不能重複'''

from ..global_utils import GetOrAddGlobalValue
_CrossModuleEnumDict = GetOrAddGlobalValue("_CrossModuleEnumDict", dict())
from enum import EnumMeta, Enum

class CrossModuleEnumMeta(EnumMeta):
    def __new__(cls, *args, **kwargs):
        clsName = args[0]
        if clsName in _CrossModuleEnumDict:
            return _CrossModuleEnumDict[clsName]
        else:
            thisEnumCls = super().__new__(cls, *args, **kwargs)
            _CrossModuleEnumDict[clsName] = thisEnumCls
            return thisEnumCls

class CrossModuleEnum(Enum, metaclass=CrossModuleEnumMeta):
    '''跨模組Enum，無論在哪裡import，都返回同一個Enum類。名稱不能重複'''
    pass

__all__ = ["CrossModuleEnum", "CrossModuleEnumMeta"]
source/utils/base_clses/cross_module_enum.py
92MING/Stable-Renderer
# -*- coding: utf-8 -*-
'''this module contains generic classes.'''

class SingleGenericClass:
    _type = None
    _genericClses = {}

    DEFAULT_TYPE = None
    '''override this to set default type'''

    def __class_getitem__(cls, item):
        if not isinstance(item, type):
            raise TypeError(f'{cls.__qualname__} only accepts type as generic type, got {type(item)}')
        if item not in cls._genericClses:
            itemName = item.__qualname__.split('.')[-1] if hasattr(item, '__qualname__') else item.__name__.split('.')[-1]
            newCls = type(f'{cls.__qualname__}<{itemName}>', (cls,), {'__slots__':()})
            newCls._type = item
            cls._genericClses[item] = newCls
        return cls._genericClses[item]

    @classmethod
    def type(cls):
        if cls._type is None:
            if cls.DEFAULT_TYPE is not None:
                return cls.DEFAULT_TYPE
            raise TypeError(f'{cls.__qualname__} has no generic type. Please use {cls.__qualname__}[type] to get a generic class')
        return cls._type

class MultiGenericClass:
    _types = None
    _genericClses = {}

    DEFAULT_TYPES = None
    '''override this to set default types, e.g. (int, int)'''

    def __class_getitem__(cls, item):
        item = tuple(item)
        if item not in cls._genericClses:
            itemName = item.__qualname__.split('.')[-1] if hasattr(item, '__qualname__') else item.__name__.split('.')[-1]
            newCls = type(f'{cls.__qualname__}<{itemName}>', (cls,), {'__slots__':()})
            newCls._types = item
            cls._genericClses[item] = newCls
        return cls._genericClses[item]

    @classmethod
    def types(cls):
        if cls._types is None:
            if cls.DEFAULT_TYPES is not None:
                return cls.DEFAULT_TYPES
            raise TypeError(f'{cls.__qualname__} has no generic types. Please use {cls.__qualname__}[types] to get a generic class')
        return cls._types


__all__ = ['SingleGenericClass', 'MultiGenericClass']
source/utils/base_clses/generic_class.py
92MING/Stable-Renderer
'''NamedObj is a base cls that every instance has a name. If instance with the same name is created, raise error.'''

from .cross_module_class import CrossModuleClass

class NamedObj(CrossModuleClass):
    _instances = None

    def __new__(cls, name:str, *args, **kwargs):
        '''
        For NamedObj, the first argument is always the name of the instance. If instance with the same name is created, raise error.
        Name is not case sensitive.
        '''
        if cls._instances is None:
            cls._instances = {}
        if name in cls._instances:
            raise Exception(f'Instance with name {name} already exists')
        else:
            ins = super().__new__(cls)
            cls._instances[name] = ins
        return ins
    def __init__(self, name:str, *args, **kwargs):
        self._name = name

    @classmethod
    def GetInstance(cls, ins_name):
        '''return None if not found'''
        if cls._instances is None:
            cls._instances = {}
        return cls._instances.get(ins_name, None)
    @classmethod
    def AllInstances(cls):
        '''return a list of all instances'''
        if cls._instances is None:
            cls._instances = {}
        return list(cls._instances.values())

    def __class_getitem__(cls, item):
        '''NamedObj['name'] will return the instance with name "name". Equivalent to NamedObj.GetInstance('name')'''
        return cls.GetInstance(item)

    @property
    def name(self)->str:
        return self._name
    @name.setter
    def name(self, value:str):
        if value == self._name:
            return
        if value in self._instances:
            raise Exception(f'Instance with name {value} already exists, can not change name to it.')
        else:
            del self._instances[self._name]
            self._instances[value] = self
            self._name = value

__all__ = ['NamedObj']
source/utils/base_clses/named_obj.py
92MING/Stable-Renderer
# -*- coding: utf-8 -*-
'''單例模式基類。默認在定義時，自動生成實例（init)'''

from .cross_module_class import CrossModuleClassMeta
from functools import partial
from types import FunctionType
class SingletonMeta(CrossModuleClassMeta):
    def __init__(cls, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if cls.__qualname__!="Singleton" and cls.ifInitOnDefine():
            cls._instance = cls()
    def __getattribute__(self, item):
        obj = super().__getattribute__(item)
        if (item.startswith('__') and item.endswith('__')) or item == '_instance':
            # 特殊屬性直接返回
            return obj
        if self._instance is None:
            self._instance = self()
        if isinstance(obj, property):
            # property 也可以直接透過類訪問
            return obj.fget(self._instance)
        elif isinstance(obj, FunctionType):
            # 函數綁定到實例
            return partial(obj, self._instance)
        else:
            return obj
    def _getAttrType(self, item):
        if hasattr(self, item):
            return type(super().__getattribute__(item))
        else:
            raise AttributeError(f"{self.__name__} has no attribute {item}")
    def _getOriginObj(self, item):
        return super().__getattribute__(item)

class Singleton(metaclass=SingletonMeta):
    '''單例模式基類, 繼承CrossModuleClass'''
    _instance = None # 實例

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            # 如果沒有實例，生成實例
            if cls.__qualname__ == "Singleton":
                # 基類不可實例化
                raise Exception("Singleton can't be instantiated directly")
            cls._instance = super().__new__(cls)
            cls.__init__(cls._instance, *args, **kwargs)
            cls._instance.__init__ = lambda *args: None
            cls.__call__ = lambda *args: cls._instance
        return cls._instance

    def __getattribute__(self, item):
        # getattribute 時，返回類的屬性
        if item == '__class__':
            return super().__getattribute__(item)
        obj = getattr(self.__class__, item)
        if isinstance(obj, property):
            return obj.fget(self)
        return obj

    def __setattr__(self, key, value):
        # setattr 時，設置類的屬性
        if not hasattr(self.__class__, key):
            setattr(self.__class__, key, value)
        else:
            objType = self.__class__._getAttrType(key)
            if objType == property:
                self.__class__._getOriginObj(key).fset(self, value)
            else:
                setattr(self.__class__, key, value)

    @classmethod
    def instance(cls):
        '''如果没有实例，會自動生成'''
        return cls()
    @classmethod
    def ifInitOnDefine(cls)->bool:
        '''是否在定義時初始化，默認為True'''
        return True

__all__ = ["Singleton"]
source/utils/base_clses/singleton.py
92MING/Stable-Renderer
# -*- coding: utf-8 -*-
'''事件类，用于实现事件机制。比PYQT原生Signal輕量，不需要QObject。當QObject被刪除後，invoke event時，對應的監聽方法在運行到RuntimeError的時候，會自動刪除。'''
import heapq
import functools
try:
    # Default option
    from PySide2.QtCore import QObject, Signal
except ModuleNotFoundError:
    # Compatability with Python 3.10
    print("[INFO] PySide2 not found, attempting PySide6")
    from PySide6.QtCore import QObject, Signal
from types import FunctionType, MethodType
from typing import ForwardRef, get_origin, get_args, Union, Iterable, Literal
from inspect import getfullargspec, signature, getmro
from enum import Enum
from collections.abc import Callable

from utils.base_clses.cross_module_enum import CrossModuleEnum
from utils.type_utils import valueTypeCheck, subClassCheck

def _SignalWrapper(*args):
    class SignalWrapperCls(QObject):
        signal = Signal(*args)
    return SignalWrapperCls()
def _allArgsNotUnion(args):
    for arg in args:
        if get_origin(arg) is not None and get_origin(arg) == Union:
            return False
    return True
def _getHasUnionInput(possibleInputs:list):
    for possibleInput in possibleInputs:
        if not _allArgsNotUnion(possibleInput):
            return possibleInput
    return None
def _allPossibleInputsOk(possibleInputs):
    for possibleInput in possibleInputs:
        if not _allArgsNotUnion(possibleInput):
            return False
    return True
def _getMroClsNames(cls):
    return [cls.__qualname__ for cls in getmro(cls)]
def _mroDistance(cls, targetType):
    origin = get_origin(targetType)
    if origin is not None:
        targetType = origin
    mroClsNames = _getMroClsNames(cls)
    return mroClsNames.index(targetType.__qualname__)
def _totalMroDistance(values, targetTypes, acceptNone):
    totalDistance = 0
    for i, value in enumerate(values):
        if acceptNone and value is None:
            continue
        totalDistance += _mroDistance(type(value), targetTypes[i])
    return totalDistance
def _findMostSuitableInputType(possibleInputs, values, acceptNone):
    suitableInputs = []
    for possibleInput in possibleInputs:
        ok = True
        for i, argType in enumerate(possibleInput):
            if values[i] is None and acceptNone:
                continue
            if not valueTypeCheck(values[i], argType):
                ok = False
                break
        if ok:
            suitableInputs.append(possibleInput)
    mostSuitableDistances = [_totalMroDistance(values, suitableInput, acceptNone) for suitableInput in suitableInputs]
    mostSuitableDistance = min(mostSuitableDistances)
    return suitableInputs[mostSuitableDistances.index(mostSuitableDistance)]

class ListenerNotFoundError(Exception):
    pass
class NoneTypeNotSupportedError(Exception):
    pass

class Event:
    def __init__(self, *args, useQtSignal=False, acceptNone=False, noCheck=False):
        '''
        :param args: 事件的参数类型（1個或多個），可以是 类型 或 类型名稱 。支持所有utils.TypeUtils.simpleTypeCheck的類型。
        :param useQtSignal: 是否使用Qt的Signal作為事件機制。一般在QT需要跨線程時使用。如果使用此參數，則需要明確定義所有可能的參數類型,
                    e.g: A繼承B, 如果A和B都可能作為參數類型，則需要Event(Union[A, B], useQtSignal=True), 否則QT的Siganl會
                    導致類型坍塌為基類。
        :param acceptNone: 是否接受None作為參數。如果args中有None，則acceptNone會強制設置為True(僅限useQtSignal模式)。
        :param noCheck: addListener/invoke 時不檢查參數數量/類型是否正確
        '''
        args = list(args)
        for i, arg in enumerate(args):
            if arg is None:
                args[i] = type(None)
                if not acceptNone:
                    if not useQtSignal:
                        raise NoneTypeNotSupportedError("Event does not support NoneType as arg since 'acceptNone'=False")
                    else:
                        acceptNone = True # force acceptNone to True
            elif not isinstance(arg, (str, type, ForwardRef)):
                raise Exception("Event's arg must be type or type name, but got " + str(type(arg)))
        self._args = tuple(args)
        self._useQtSignal = useQtSignal
        self._acceptNone = acceptNone
        self._events = set()
        self._tempEvents = set()
        self._noCheck = noCheck
        self._hasEmitted = False  # to prevent sometimes Signal emit twice with unknown reason
        self._noneIndexes = set() # record the index of None in args. Since Signal is hard to emit None, we need to mark down the index of None and replace when _invoke
        if useQtSignal:
            signalArgs = []
            for arg in args:
                if subClassCheck(arg, CrossModuleEnum):
                    signalArgs.append(Enum)
                elif isinstance(arg, type):
                    signalArgs.append(arg)
                elif get_origin(arg) is not None:
                    origin = get_origin(arg)
                    if origin == Union:
                        signalArgs.append(arg)
                    elif origin != Literal:
                        signalArgs.append(origin)
                    else:
                        signalArgs.append(object)
                else:
                    signalArgs.append(object)
            possibleInputs = [signalArgs]
            while not _allPossibleInputsOk(possibleInputs):
                possibleInput = _getHasUnionInput(possibleInputs)
                possibleInputs.remove(possibleInput)
                for i, arg in enumerate(possibleInput):
                    if get_origin(arg) is not None and get_origin(arg) == Union:
                        for argType in get_args(arg): # for each type in Union
                            argTypeOrigin = get_origin(argType)
                            if argTypeOrigin is not None and argTypeOrigin != Union:
                                argType = argTypeOrigin
                            newPossibleInput = list(possibleInput)
                            newPossibleInput[i] = argType
                            possibleInputs.append(newPossibleInput.copy())
                        break
            self._possibleInputs = possibleInputs
            if len(possibleInputs) == 1:
                self._qtSignal = _SignalWrapper(*possibleInputs[0])
                self._qtSignal.signal.connect(self._invoke)
            else:
                self._qtSignal = _SignalWrapper(*possibleInputs)
                for possibleInput in possibleInputs:
                    self._qtSignal.signal.__getitem__(possibleInput).connect(self._invoke)
            if acceptNone:
                self._possibleInputs.append([type(None)] * len(signalArgs))

    def __iadd__(self, other):
        self.addListener(other)
        return self
    def __isub__(self, other):
        self.removeListener(other)
        return self
    def destroy(self):
        self._events.clear()
        self._tempEvents.clear()
        self._args = None
        self._events = None
        self._tempEvents = None
        if self._useQtSignal:
            try:
                self._qtSignal.disconnect()
            except RuntimeError:
                pass
            self._qtSignal.deleteLater()
            self._qtSignal = None
        del self

    @property
    def args(self):
        return self._args
    @property
    def argCount(self):
        return len(self._args)
    @property
    def events(self)->tuple:
        '''return a tuple of events'''
        return tuple(self._events)
    @property
    def argLength(self)->int:
        '''return the length of args'''
        return len(self.args)
    @property
    def tempEvents(self)->tuple:
        '''return a tuple of temp events'''
        return tuple(self._tempEvents)
    @property
    def acceptNone(self)->bool:
        return self._acceptNone
    @property
    def useQtSignal(self)->bool:
        return self._useQtSignal

    def _checkListener(self, listener:Callable):
        if self._noCheck:
            return
        try:
            isVarArgFunc = getfullargspec(listener).varargs is not None   # 是否是可变参数函数
            defaultsCount = len(getfullargspec(listener).defaults) if getfullargspec(listener).defaults is not None else 0
        except TypeError:
            isVarArgFunc = False
            defaultsCount = 0
        if isinstance(listener,FunctionType):
            argLength = listener.__code__.co_argcount
        elif isinstance(listener,MethodType):
            if issubclass(listener.__self__.__class__, Event):
                if listener.__qualname__.split('.')[-1] == 'invoke':
                    args = listener.__self__.args
                    for i, arg in enumerate(args):
                        if not subClassCheck(arg, self.args[i]):
                            raise Exception("Listener's arg must be type or type name")
                    argLength = len(args)
                else:
                    argLength = listener.__code__.co_argcount - 1
            else:
                argLength = listener.__code__.co_argcount - 1
        elif isinstance(listener, functools.partial):
            diff = 1 if isinstance(listener.func, MethodType) else 0
            if hasattr(listener.func, '__code__'):
                argLength = listener.func.__code__.co_argcount - len(listener.args) - len(listener.keywords) - diff
            else:
                return # really can't get arg length
        elif isinstance(listener, Callable):
            try:
                sig = signature(listener)
                argLength = len(sig.parameters)
            except ValueError: # really can't get arg length
                return
        else:
            raise Exception("Listener must be function or method. Got " + str(type(listener)))
        if not isVarArgFunc:
            possibleArgRange = range(argLength - defaultsCount, argLength + 1)
            if self.argLength not in possibleArgRange:
                raise Exception(f"Listener's arg length not match, self.argLength= {self.argLength}, but possible range is {possibleArgRange}")
        else:
            if self.argLength < (argLength - defaultsCount):
                raise Exception(f"Listener's arg length not match, self.argLength= {self.argLength}, but possible range is {argLength - defaultsCount}+")
    def _removeDestroyedListener(self, listener):
        try:
            self._events.remove(listener)
        except KeyError:
            pass
    def _removeDestroyedTempListener(self, listener):
        try:
            self._tempEvents.remove(listener)
        except KeyError:
            pass
    def addListener(self, listener:Union[callable, Iterable[callable]]):
        '''add a listener to event'''
        if isinstance(listener, Iterable):
            for l in listener:
                self.addListener(l)
        elif isinstance(listener, Callable):
            self._checkListener(listener)
            if isinstance(listener,MethodType):
                if isinstance(listener.__self__, QObject):
                    listener.__self__.destroyed.connect(functools.partial(self._removeDestroyedListener, listener))
            self._events.add(listener)
        else:
            raise TypeError("Listener must be callable, or iterable of callable")
    def addTempListener(self, listener:Union[callable, Iterable[callable]]):
        '''add a temp listener to event'''
        if isinstance(listener, Iterable):
            for l in listener:
                self.addTempListener(l)
        elif isinstance(listener, Callable):
            self._checkListener(listener)
            if isinstance(listener,MethodType):
                if isinstance(listener.__self__, QObject):
                    listener.__self__.destroyed.connect(functools.partial(self._removeDestroyedTempListener, listener))
        self._tempEvents.add(listener)

    def removeListener(self, listener:Callable, throwError=True):
        '''remove a listener from event'''
        try:
            self._events.remove(listener)
        except KeyError:
            if throwError:
                raise ListenerNotFoundError
    def removeTempListener(self, listener:Callable, throwError=True):
        '''remove a temp listener from event'''
        try:
            self._tempEvents.remove(listener)
        except KeyError:
            if throwError:
                raise ListenerNotFoundError

    def _invoke(self, *args):
        if self._hasEmitted:
            return # when use QtSignal, sometimes it will emit twice with unknown reason
        self._hasEmitted = True
        if self.acceptNone and self.useQtSignal:
            args = list(args)
            for index in self._noneIndexes:
                args[index] = None
        for event in self.events:
            try:
                event(*args)
            except RuntimeError as e:
                if 'deleted' in str(e):
                    # C/C++ object has been deleted(usually occurs in PYQT)
                    self.removeListener(event)
                else:
                    raise e
        for event in self.tempEvents:
            try:
                event(*args)
            except RuntimeError as e:
                if 'deleted' in str(e):
                    # C/C++ object has been deleted(usually occurs in PYQT)
                    continue
                else:
                    raise e
        self._tempEvents.clear()
    def _check_invoke_params(self, *args):
        if len(args) < len(self.args):
            argNeeded = self.args[len(args):]
            outputStr = ""
            for arg in argNeeded:
                outputStr += f"{arg.__qualname__}, "
            outputStr = outputStr[:-2]
            raise Exception(f"Parameter: {outputStr[:-2]} are not provided")
        elif len(args) > len(self.args):
            raise Exception(f"Too many parameters")
        for i, arg in enumerate(args):
            if isinstance(self.args[i], str):
                # check type by name
                if type(arg).__qualname__ != self.args[i]:
                    raise Exception(f"Class of parameter: {arg} is not {self.args[i]}")
            elif not valueTypeCheck(arg, self.args[i]):
                if not (arg is None and self._acceptNone):
                    raise Exception(f"Parameter: {arg} is not valid for type'{self.args[i]}'")
    def invoke(self, *args):
        '''invoke all listeners'''
        if not self._noCheck:
            self._check_invoke_params(*args)
        self._hasEmitted = False
        if not self._useQtSignal:
            self._invoke(*args)
        else:
            for i, arg in enumerate(args):
                if arg is None:
                    if self.acceptNone:
                        self._noneIndexes.add(i)
                    else:
                        raise NoneTypeNotSupportedError('''None type is not supported since you have set "acceptNone" to False.
                                                        When you want to pass None in useQtSignal mode, you must use set "acceptNone" to True.''')
            if len(self._possibleInputs) == 0:
                self._qtSignal.signal.emit(*args)
            else:
                mostSuitableInput = _findMostSuitableInputType(self._possibleInputs, args, self.acceptNone)
                self._qtSignal.signal.__getitem__(mostSuitableInput).emit(*args)
            self._noneIndexes.clear()

    def eventsCount(self):
        '''return the count of normal events'''
        return len(self._events)
    def tempEventsCount(self):
        '''return the count of temp events'''
        return len(self._tempEvents)

    def clear(self):
        '''clear all events(both temp and normal)'''
        self._events.clear()
        self._tempEvents.clear()

class DelayEvent(Event):
    '''
    When invoke delay event, it will not be executed immediately, but params will be save and execute later by calling "release".
    '''

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._last_param = None

    def invoke(self, *args):
        self._check_invoke_params(*args)
        self._last_param = args

    def release(self):
        if self._last_param is not None:
            super().invoke(*self._last_param)
        self._last_param = None

class Tasks(Event):
    '''
    Tasks is a subclass of Event, which will clear all tasks after invoke.
    Type of tasks must be callable.
    When create, the init params will be the input types of tasks.
    '''
    def _checkTask(self, task):
        if not isinstance(task, Callable):
            raise TypeError("Task must be callable")
    def addTasks(self, tasks:Union[Callable, Iterable[Callable]]):
        if isinstance(tasks, Iterable):
            for task in tasks:
                self._checkTask(task)
                self.addTempListener(task)
        else:
            self._checkTask(tasks)
            self.addTempListener(tasks)
    def removeTasks(self, tasks:Union[Callable, Iterable[Callable]]):
        if isinstance(tasks, Iterable):
            for task in tasks:
                self._checkTask(task)
                self.removeTempListener(task)
        else:
            self._checkTask(tasks)
            self.removeTempListener(tasks)
    def addForeverTasks(self, tasks:Union[Callable, Iterable[Callable]]):
        if isinstance(tasks, Iterable):
            for task in tasks:
                self._checkTask(task)
                self.addListener(task)
        else:
            self._checkTask(tasks)
            self.addListener(tasks)
    def removeForeverTasks(self, tasks:Union[Callable, Iterable[Callable]]):
        if isinstance(tasks, Iterable):
            for task in tasks:
                self._checkTask(task)
                self.removeListener(task)
        else:
            self._checkTask(tasks)
            self.removeListener(tasks)
    def execute(self, *args):
        self.invoke(*args)

class AutoSortTask(Tasks):
    '''Tasks will be cleared after execute. Tasks will be sorted by order.'''
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._tempEvents = [] # for heapq
        self._events = [] # for heapq
    def addListener(self, listener:Union[callable, Iterable[callable]], order:int=0):
        self._checkListener(listener)
        heapq.heappush(self._events, (order, listener))
    def addTempListener(self, listener:Union[callable, Iterable[callable]], order:int=0):
        self._checkListener(listener)
        heapq.heappush(self._tempEvents, (order, listener))
    def removeListener(self, listener, throwError:bool=False):
        if isinstance(listener, Iterable):
            for l in listener:
                self.removeListener(l)
        else:
            for i, l in enumerate(self._events):
                if l[1] == listener:
                    self._events.pop(i)
                    break
    def removeTempListener(self, listener, throwError:bool=False):
        if isinstance(listener, Iterable):
            for l in listener:
                self.removeTempListener(l)
        else:
            for i, l in enumerate(self._tempEvents):
                if l[1] == listener:
                    self._tempEvents.pop(i)
                    break

    def addTask(self, task:Callable, order:int=0):
        self._checkTask(task)
        self.addTempListener(task, order)
    def addTasks(self, tasks:Union[Callable, Iterable[Callable]], orders:Union[int, Iterable[int]]=0):
        if not isinstance(tasks, Iterable):
            tasks = [tasks]
        if isinstance(orders, int):
            orders = [orders] * len(tasks)
        for task, order in zip(tasks, orders):
            self.addTask(task, order)
    def addForeverTasks(self, tasks:Union[Callable, Iterable[Callable]], orders:Union[int, Iterable[int]]=0):
        if not isinstance(tasks, Iterable):
            tasks = [tasks]
        if isinstance(orders, int):
            orders = [orders] * len(tasks)
        for task, order in zip(tasks, orders):
            self.addForeverTask(task, order)
    def addForeverTask(self, task:Callable, order:int=0):
        self._checkTask(task)
        self.addListener(task, order)

    def _invoke(self, *args):
        if self._hasEmitted:
            return # when use QtSignal, sometimes it will emit twice with unknown reason
        self._hasEmitted = True
        if self.acceptNone and self.useQtSignal:
            args = list(args)
            for index in self._noneIndexes:
                args[index] = None
        for order, event in self.events:
            try:
                event(*args)
            except RuntimeError as e:
                if 'deleted' in str(e):
                    # C/C++ object has been deleted(usually occurs in PYQT)
                    self.removeListener(event)
                else:
                    raise e
        for order, event in self.tempEvents:
            try:
                event(*args)
            except RuntimeError as e:
                if 'deleted' in str(e):
                    # C/C++ object has been deleted(usually occurs in PYQT)
                    continue
                else:
                    raise e
        self._tempEvents.clear()

__all__ = ["Event", "ListenerNotFoundError", "NoneTypeNotSupportedError", "DelayEvent", "Tasks", "AutoSortTask"]

source/utils/data_struct/event.py
92MING/Stable-Renderer
# -*- coding: utf-8 -*-
'''常用的全局变量操作, 無論如何import此模塊, 都會得到同一個全局變量字典。不涉及global()或者os.environ. 如果有.env, 會自動讀取。'''

import sys, os

_needInit = True
_moduleName = __name__.split('.')[-1]
for module in sys.modules.keys():
    if module == __name__:
        break
    modulname = module.split('.')[-1] if '.' in module else module
    if modulname == _moduleName:
        _globalValues = sys.modules[module]._globalValues
        _needInit = False
        break
if _needInit:
    _globalValues = {}
    from dotenv import load_dotenv
    load_dotenv(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '.env')))

def SetGlobalValue(key: str, value: object):
    _globalValues[key] = value
def GetGlobalValue(key: str):
    try:
        return _globalValues[key]
    except KeyError:
        return None
def RemoveGlobalValue(key: str):
    try:
        _globalValues.pop(key)
    except KeyError:
        pass
def HasGlobalValue(key: str):
    return key in _globalValues
def ClearGlobalValue():
    _globalValues.clear()
def GetGlobalValueKeys():
    return tuple(_globalValues.keys())
def GetGlobalValueValues():
    return tuple(_globalValues.values())
def GetGlobalValueItems():
    return tuple(_globalValues.items())
def GetGlobalValueDict():
    '''Return a copy of global value dict.'''
    return _globalValues.copy()
def GetOrAddGlobalValue(key: str, defaultValue:object):
    if key in _globalValues:
        return _globalValues[key]
    else:
        _globalValues[key] = defaultValue
        return defaultValue

__all__ = [ 'SetGlobalValue', 'GetGlobalValue', 'RemoveGlobalValue', 'ClearGlobalValue', 'GetGlobalValueKeys', 'GetGlobalValueValues', 'GetGlobalValueItems', 'GetGlobalValueDict',
            'HasGlobalValue', 'GetOrAddGlobalValue' ]
source/utils/global_utils.py
92MING/Stable-Renderer
import os

RESOURCES_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', 'resources'))

__all__ = ['RESOURCES_DIR']
source/utils/path_utils.py
92MING/Stable-Renderer
# -*- coding: utf-8 -*-
'''常用的類型相關函數，例如類型檢查等'''

from typing import *
from inspect import getmro

# region private functions
def _checkSubClassByName(clsName, largerCls):
    return clsName in [cls.__qualname__ for cls in getmro(largerCls)]
def _directCheckSubClass(cls, largerCls):
    if get_origin(cls) == Union:
        return any([_directCheckSubClass(arg, largerCls) for arg in get_args(cls)])
    if isinstance(cls, ForwardRef):
        return _checkSubClassByName(cls.__forward_arg__, largerCls)
    elif isinstance(cls, str):
        return _checkSubClassByName(cls, largerCls)
    else:
        try:
            return issubclass(cls, largerCls)
        except TypeError:
            return False
def _checkTypeByClsName(value, clsName):
    return _directCheckSubClass(clsName, type(value))
def _directCheckType(value, targetType):
    if isinstance(targetType, ForwardRef):
        return _checkTypeByClsName(value, targetType.__forward_arg__)
    elif isinstance(targetType, str):
        return _checkTypeByClsName(value, targetType)
    else:
        return isinstance(value, targetType)
# endregion

def subClassCheck(smallerCls, largerCls):
    '''
    檢查smallerCls是否是largerCls的子類，支持Union、Optional、Sequence、Iterable、Mapping、Callable、Literal等類型。
    e.g.
        subClassCheck(int, Union[int, str]) -> True
        subClassCheck(Literal[1, 2], Union[int, str]) -> True
    '''
    def _checkTypes(smallerTypes, largerTypes):
        if Any in largerTypes:
            return True
        for smallerType in smallerTypes:
            ok = False
            for largerType in largerTypes:
                if subClassCheck(smallerType, largerType):
                    ok = True
                    break
            if not ok:
                return False
        return True
    if largerCls == Any:
        return True
    elif smallerCls == Any:
        return False
    smallerTypeOrigin = get_origin(smallerCls)
    largerTypeOrigin = get_origin(largerCls)

    if smallerTypeOrigin == Literal:
        smallerValues = get_args(smallerCls)
        if largerTypeOrigin == Literal:
            largerValues = get_args(largerCls)
            for smallerValue in smallerValues:
                if smallerValue not in largerValues:
                    return False
            return True
        else:
            for smallerValue in smallerValues:
                if not valueTypeCheck(smallerValue, largerCls):
                    return False
            return True
    elif smallerTypeOrigin is None and largerTypeOrigin is None:
        return _directCheckSubClass(smallerCls, largerCls)
    elif smallerTypeOrigin is not None and largerTypeOrigin is None:
        return _directCheckSubClass(smallerTypeOrigin, largerCls)
    elif smallerTypeOrigin is None and largerTypeOrigin is not None:
        if not _directCheckSubClass(smallerCls, largerTypeOrigin):
            return False
        if len(get_args(largerCls)) == 0:
            return True
        return False
    else:
        # both not None
        if smallerTypeOrigin in (Union, Optional):
            if largerTypeOrigin not in (Union, Optional):
                return False
            smallerTypes = get_args(smallerCls)
            largerTypes = get_args(largerCls)
            return _checkTypes(smallerTypes, largerTypes)
        else:
            if not _directCheckSubClass(smallerTypeOrigin, largerTypeOrigin):
                return False
            if _directCheckSubClass(smallerTypeOrigin,Sequence) or _directCheckSubClass(smallerTypeOrigin, Iterable):
                smallerArgType = get_args(smallerCls)[0]
                largerArgType = get_args(largerCls)[0]
                return subClassCheck(smallerArgType, largerArgType)
            elif _directCheckSubClass(smallerTypeOrigin, Mapping):
                smallerKeyType, smallerValueType = get_args(smallerCls)
                largerKeyType, largerValueType = get_args(largerCls)
                return subClassCheck(smallerKeyType, largerKeyType) and subClassCheck(smallerValueType, largerValueType)
            elif _directCheckSubClass(smallerTypeOrigin, Callable):
                smallerClsArgs = get_args(smallerCls)
                largerClsArgs = get_args(largerCls)
                smallerAgrTypes, smallerReturnType = smallerClsArgs[0], smallerClsArgs[-1]
                largerArgTypes, largerReturnType = largerClsArgs[0], largerClsArgs[-1]
                if len(smallerAgrTypes) != len(largerArgTypes):
                    return False
                return _checkTypes(smallerAgrTypes, largerArgTypes) and subClassCheck(smallerReturnType, largerReturnType)
            else:
                raise Exception('unsupported types--- smaller:', smallerTypeOrigin, ' larger:',largerTypeOrigin)

def valueTypeCheck(value, targetType):
    '''
    檢查value是否是targetType的實例，支持Union、Optional、Sequence、Iterable、Mapping、Callable、Literal等類型。
    特殊地，對於simpleTypeCheck(None, None), 返回True。
    e.g:
        valueTypeCheck(1, 'int') -> True
        valueTypeCheck(1, int) -> True
        valueTypeCheck(1, Union[int, str]) -> True
        valueTypeCheck(1, Union[str, float]) -> False
        valueTypeCheck(1, Optional[int]) -> True
        valueTypeCheck((1,), Tuple[int]) -> True
        valueTypeCheck((1,), Tuple[int, ...]) -> True
        valueTypeCheck((1, 2, 3), Tuple[int, int]) -> False
        valueTypeCheck([1, 2, 3], Sequence[int]) -> True
        valueTypeCheck("a", Literal["a","b"]) -> True
        valueTypeCheck(None, None) -> True    # special case
    '''
    if targetType == callable:
        targetType = Callable
    if value is None and targetType is None:
        return True # special case
    if get_origin(targetType) is None:
        return _directCheckType(value, targetType)
    else:
        argTypes = get_args(targetType)
        origin = get_origin(targetType)
        if origin == Union or origin == Optional:
            for argType in argTypes:
                if valueTypeCheck(value, argType):
                    return True
            return False
        elif origin == Literal:
            return value in argTypes
        else:
            if not _directCheckType(value, origin):
                return False
            if _directCheckSubClass(origin, Sequence):
                # special case for Tuple
                if _directCheckSubClass(origin, tuple):
                    if argTypes[-1] != Ellipsis and len(value) != len(argTypes):
                        return False
                    if argTypes[-1] == Ellipsis:
                        for v in value:
                            if not valueTypeCheck(v, argTypes[0]):
                                return False
                    else:
                        for i, argType in enumerate(argTypes):
                            if not valueTypeCheck(value[i], argType):
                                return False
                    return True
                else:
                    # for List, Deque, etc.
                    for v in value:
                        if not valueTypeCheck(v, argTypes[0]):
                            return False
                    return True
            elif _directCheckSubClass(origin, Mapping):
                keyType, valueType = argTypes
                for key, val in value.items():
                    if not valueTypeCheck(key, keyType) or not valueTypeCheck(val, valueType):
                        return False
                return True
            elif _directCheckSubClass(origin, Callable):
                if len(argTypes) == 0: # no any type hints of target callable type, means any callable is ok
                    return True
                else:
                    callArgTypes, returnType = *argTypes[:-1], argTypes[-1]
                    valueArgTypes = get_type_hints(value)  # here, value should be a function
                    varNames = value.__code__.co_varnames
                    for varName in varNames:
                        if varName not in valueArgTypes:
                            continue # ignore unlabeled arguments
                        if not subClassCheck(valueArgTypes[varName], callArgTypes[varNames.index(varName)]):
                            return False
                    if 'return' in valueArgTypes:
                        if not subClassCheck(valueArgTypes['return'], returnType):
                            return False
                    return True
            elif _directCheckSubClass(origin, Iterable):
                argType = argTypes[0]
                for v in value:
                    if not valueTypeCheck(v, argType):
                        return False
                return True
            else:
                raise Exception('not supported checking type: ' + str(origin))

__all__ = ['subClassCheck', 'valueTypeCheck']

source/utils/type_utils.py
92MING/Stable-Renderer