Berry/plugin_loader.py

from textual_window import Window
from textual.widgets import RichLog
from textual import work
from lupa import LuaRuntime, lua51

import os, json, asyncio


class PluginLoader(Window):
	def __init__(self):
		super().__init__(
			id="Plugin Loader",
			mode="permanent",
			icon="⚙️",
			starting_horizontal="right",
			starting_vertical="bottom",
			start_open=True,
			allow_maximize=True
		)

	@work
	async def find_plugins(self):
		log = self.query_one(RichLog)

		no_errors = True

		log.write("[b]Setting up LUA runtime..[/]")
		self.lua_runtime = lua51.LuaRuntime()
		lua_runtime_stuff = {
			"ui": {
				"notify": self.notify
			}
		}

		log.write("[b]Finding plugins...[/]")

		# Find all plugins (they're just in the plugins folder for now)
		folders = [
			os.path.join(os.path.dirname(__file__), "plugins")
		]
		# path to the folder of all correctly formatted plugins
		plugin_paths = []

		for folder in folders:
			log.write(f"Searching {folder}...")
			plugin_folders = os.listdir(folder)

			for plugin_folder in plugin_folders:
				plugin_folder = os.path.join(folder, plugin_folder)

				if not os.path.isdir(plugin_folder):
					log.write(f"[d]Ignoring {plugin_folder} because it is not a folder.[/]")
					no_errors = False
					continue

				if not os.path.isdir(os.path.join(plugin_folder, "lua")):
					log.write(f"[d]Ignoring {plugin_folder} because it has no \"lua\" folder.[/]")
					no_errors = False
					continue

				if not os.path.isfile(os.path.join(plugin_folder, "plugin.json")):
					log.write(f"[d]Ignoring {plugin_folder} because it has no plugin.json file.[/]")
					no_errors = False
					continue

				with open(os.path.join(plugin_folder, "plugin.json"), "r") as f:
					try:
						plugin_json = json.loads(f.read())
						plugin_json["name"]
						plugin_json["version"]
						plugin_json["author"]
						plugin_json["dependencies"]
					except UnicodeDecodeError:
						log.write(f"[d]Ignoring {plugin_folder} because its plugin.json file is unreadable.[/]")
						no_errors = False
						continue
					except json.JSONDecodeError:
						log.write(f"[d]Ignoring {plugin_folder} because its plugin.json file is malformed.[/]")
						no_errors = False
						continue
					except KeyError as e:
						log.write(f"[d]Ignoring {plugin_folder} because its plugin.json file is missing the field {e}.")
						no_errors = False
						continue
					except Exception as e:
						log.write(f"[d]Ignoring {plugin_folder} because of error: {e}.[/]")
						no_errors = False
						continue


				log.write(f"[b green]FOUND[/] {plugin_json['name']} ({plugin_json['version']})")
				for lua_file in os.listdir(os.path.join(plugin_folder, "lua")):
					lua_file_path = os.path.join(plugin_folder, f"lua/{lua_file}")
					
					with open(lua_file_path, "r") as f:
						code = f.read()
						sandbox = self.lua_runtime.eval("{}")
						setfenv = self.lua_runtime.eval("setfenv")

						sandbox.print = self.log #self.lua_runtime.globals().print
						sandbox.math = self.lua_runtime.globals().math
						sandbox.string = self.lua_runtime.globals().string
						sandbox.berry = lua_runtime_stuff

						setfenv(0, sandbox)
						try:
							executed_code = self.lua_runtime.execute(code)
						except lua51.LuaError as e:
							log.write(f"[b red]Error in {lua_file_path}: {e}[/]")
							self.notify("There was Lua error while loading one your installed plugins. Check the Plugin Loader window for more details.", title="Lua Error", severity="error")
							no_errors = False
							continue
						
						if executed_code.init:
							executed_code.init()
						if executed_code.main:
							self.run_worker(thread=True, work=executed_code.main)
				
				plugin_paths.append(plugin_folder)

		if no_errors:
			log.write("\n[d]Window will automatically close in 5 seconds.[/]")
			await asyncio.sleep(5.0)
			self.close_window()

	async def on_mount(self):
		self.find_plugins()
		

	def compose(self):
		yield RichLog(markup=True, id="plugins-log", wrap=True)