from textual.app import App, ComposeResult, SystemCommand from textual.widgets import Header, Footer, ContentSwitcher, DirectoryTree, Static, Button, TextArea, Tabs, Tab, RichLog, Input from textual.containers import HorizontalGroup, Vertical from textual.binding import Binding from textual_window import Window from textual import on from textual_fspicker import FileOpen, FileSave, SelectDirectory from pathlib import Path from home_page import HomePage from assets.theme_mappings import theme_mappings from plugin_loader import PluginLoader from settings import SettingsScreen from settings_store import ConfigHandler from directory_tree_custom import CustomDirectoryTree from completions_menu import CompletionsMenu from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from lsp_client import LSPClient import os, sys class Watcher(FileSystemEventHandler): def __init__(self, app: App): super().__init__() self.app = app async def on_any_event(self, event): await self.app.query_one(DirectoryTree).reload() class Berry(App): CSS_PATH = "assets/style.tcss" SUB_TITLE = "New File" def __init__(self, path: str): super().__init__() self.config_handler = ConfigHandler(self) self.bindings = eval(self.config_handler.get("editor", "bindings")) self.path = path self.BINDINGS = ( Binding(self.bindings["new"], "new", "New File"), Binding(self.bindings["open"], "open", "Open File"), Binding(self.bindings["open-folder"], "open_folder", "Open Folder"), Binding(self.bindings["save"], "save", "Save File"), Binding(self.bindings["save-as"], "save_as", "Save File As..."), Binding(self.bindings["find"], "find", "Find and Replace", priority=True), # this is priority because otherwise TextArea overrides it Binding(self.bindings["settings"], "settings", "Settings"), ) for binding in self.BINDINGS: self._bindings._add_binding(binding) self.refresh_bindings() self.current_editor = None self.file_tabs = {} self.completions_menu = None self.current_lsp = None self.running_lsps = { "py": LSPClient("jedi-language-server", [], "python") } self.last_change_pos = None def compose(self) -> ComposeResult: yield Header() with Vertical(id="sidebar"): with HorizontalGroup(id="sidebar-buttons"): yield Button("") yield Button("") with ContentSwitcher(initial="files", id="sidebar-switcher"): with Vertical(id="files"): yield Static("EXPLORER") yield CustomDirectoryTree(self.path, id="directory") with Vertical(id="editor"): first_tab = Tab("Home") first_tab.file_path = "berry://home" self.file_tabs["berry://home"] = first_tab yield Tabs( first_tab, id="file-tabs" ) with Vertical() as editor: yield HomePage() self.current_editor = editor #if os.name == "nt": #with Vertical(id="console-container"): # yield RichLog(id="console") # yield Input(placeholder="> ", id="console-input") #else: # yield Terminal(command="bash", id="terminal") yield Footer() if bool(int(self.config_handler.get("plugins", "enabled"))) == True: yield PluginLoader() def action_settings(self): self.push_screen(SettingsScreen()) def get_system_commands(self, screen): yield SystemCommand( "Quit the application", "Quit the application as soon as possible", self.action_quit, ) if screen.query("HelpPanel"): yield SystemCommand( "Hide keys and help panel", "Hide the keys and widget help panel", self.action_hide_help_panel, ) else: yield SystemCommand( "Show keys and help panel", "Show help for the focused widget and a summary of available keys", self.action_show_help_panel, ) yield SystemCommand("Settings", "Open the settings menu", self.action_settings) async def chose_file_to_open(self, result): if result == None: return result = str(result) if self.open_file == result: return def is_within_directory(file_path: str, directory: str) -> bool: file_path = Path(file_path).resolve() directory = Path(directory).resolve() return directory in file_path.parents self.switching = True tabs: Tabs = self.query_one("#file-tabs") if self.open_file not in self.unsaved_files: if self.open_file: self.file_tabs.pop(self.open_file) tabs.remove_tab(tabs.active_tab) self.open_file = result inside_dir = is_within_directory(result, self.path) self.sub_title = os.path.basename(self.open_file) if inside_dir else self.open_file if result not in self.file_tabs: new_tab = Tab(os.path.basename(result) if inside_dir else result) new_tab.tooltip = str(new_tab.label) setattr(new_tab, "file_path", result) await tabs.add_tab(new_tab) self.file_tabs[result] = new_tab tabs.active = new_tab.id else: tabs.active = self.file_tabs[result].id def action_open(self): self.app.push_screen(FileOpen(), self.chose_file_to_open) async def choose_folder_to_open(self, result): if result == None: return self.path = result tabs: Tabs = self.query_one("#file-tabs") tabs.clear() first_tab = Tab("Home") first_tab.file_path = "berry://home" self.file_tabs["berry://home"] = first_tab tabs.add_tab(first_tab) self.current_editor.remove_children() self.current_editor.mount(HomePage()) file_tree: CustomDirectoryTree = self.query_one("#directory") file_tree.path = result def action_open_folder(self): self.push_screen(SelectDirectory(), self.choose_folder_to_open) def action_find(self): try: self.query_one("#find-window") return except: find_window = Window( Vertical( HorizontalGroup( Input(placeholder="Find"), Static("0 of 0", id="num-matches"), Button("↑", flat=True), Button("↓", flat=True), ), HorizontalGroup( Input(placeholder="Replace"), ), ), icon="󰍉", start_open=True, allow_resize=False, allow_maximize=False, id="find-window", mode="temporary", name="Find & Replace" ) self.mount(find_window) async def on_directory_tree_file_selected(self, event: DirectoryTree.FileSelected): if self.open_file == str(event.path): return self.file_clicked = True self.switching = True tabs: Tabs = self.query_one("#file-tabs") if self.open_file not in self.unsaved_files: if self.open_file: self.file_tabs.pop(self.open_file) tabs.remove_tab(tabs.active_tab) self.open_file = str(event.path) self.sub_title = os.path.basename(self.open_file) if str(event.path) not in self.file_tabs: new_tab = Tab(os.path.basename(str(event.path))) new_tab.tooltip = str(new_tab.label) setattr(new_tab, "file_path", str(event.path)) await tabs.add_tab(new_tab) self.file_tabs[str(event.path)] = new_tab tabs.active = new_tab.id else: tabs.active = self.file_tabs[str(event.path)].id async def open_text_editor(self, file_content: bytes, file_extension: str): code_editor = TextArea.code_editor(placeholder="This file is empty.", classes="code-editor", theme="css", disabled=True, soft_wrap=bool(int(self.config_handler.get("editor", "word_wrap"))), show_line_numbers=bool(int(self.config_handler.get("editor", "line_numbers")))) await self.current_editor.mount(code_editor) # start lsp if file_extension in self.running_lsps: lsp = self.running_lsps[file_extension] self.current_lsp = file_extension if not lsp.running: await lsp.start(self.open_file) else: await lsp.open_file(self.open_file) else: self.current_lsp = None # setup text editor code_editor.language = theme_mappings.get(file_extension, None) try: code_editor.text = file_content.decode("utf-8") code_editor.disabled = False except UnicodeDecodeError: code_editor.text = "This file is in binary, it can't be opened. Sorry." code_editor.language = None code_editor.disabled = True code_editor.focus() async def open_editor(self, file_extension: str, file_content: bytes): default_editors: dict = eval(self.config_handler.get("editor", "default_editors")) editor_for_file = default_editors.get(file_extension, None) self.current_editor.remove_children() if not editor_for_file: await self.open_text_editor(file_content, file_extension) @on(Tabs.TabActivated) async def on_tab_shown(self, event: Tabs.TabActivated): if self.file_clicked: self.file_clicked = False else: self.open_file = getattr(event.tab, "file_path") if self.open_file and self.open_file.startswith(r"berry://"): return self.switching = True content = b"" file_extension = "" if self.open_file: try: f = open(self.open_file, "rb") except Exception as e: self.notify(f"Failed to open the file: {e}") return # if the file is unsaved, use the content we wrote into it # otherwise, read the CURRENT state of the file if self.open_file in self.unsaved_files: content = self.unsaved_files[self.open_file]["current"] else: content = f.read() # get the file extension file_extension = self.open_file.split(".")[-1] if "." in self.open_file else "" # close the file f.close() await self.open_editor(file_extension, content) @on(Window.Minimized) def window_minimized(self, event: Window.Minimized): event.window.remove_window() async def on_text_area_changed(self, event: TextArea.Changed): if not event.text_area.has_class("code-editor"): return if self.switching: self.switching = False return tabs: Tabs = self.query_one("#file-tabs") if self.open_file: if not self.open_file in self.unsaved_files: with open(self.open_file, "r", encoding="utf-8") as f: if f.read() == event.text_area.text: # TODO: figure out why im guetting what seems like a race conidition which is making this if statement needed return self.unsaved_files[self.open_file] = {"current": event.text_area.text.encode(), "original": f.read()} tabs.active_tab.tooltip = f"Unsaved changes in {tabs.active_tab.label}" tabs.active_tab.label = "[d orange]●[/] " + str(tabs.active_tab.label) else: self.unsaved_files[self.open_file]["current"] = event.text_area.text if self.unsaved_files[self.open_file]["original"] == self.unsaved_files[self.open_file]["current"]: tabs.active_tab.label = os.path.basename(self.open_file) tabs.active_tab.tooltip = str(tabs.active_tab.label) self.unsaved_files.pop(self.open_file) await self.run_completions() async def run_completions(self): code_editor: TextArea = self.current_editor.query_one(TextArea) if len(code_editor.text.strip()) == 0: return lsp: LSPClient = self.running_lsps[self.current_lsp] line, col = code_editor.cursor_location # ensure we're on the latest version of the file when we apply completions or else we will error if self.last_change_pos: await lsp.apply_change(code_editor.text) self.last_change_pos = (line, col) self.log("waiting for completions") completions = await lsp.get_completions((line,col-1)) self.log("god completions") self.notify(str(completions.get("error", None))) if not completions["result"]: if self.completions_menu: self.completions_menu.hide() return else: if self.completions_menu: self.completions_menu.show() completion_choices = [{"text": result["label"], "type": result["kind"]} for result in completions["result"]["items"]] if not self.completions_menu: self.completions_menu = CompletionsMenu(completion_choices, code_editor) await self.current_editor.mount(self.completions_menu) else: self.completions_menu.align_to_cursor() self.completions_menu.update(completion_choices) def action_new(self): tabs: Tabs = self.query_one("#file-tabs") new_tab = Tab("New File") setattr(new_tab, "file_path", None) tabs.add_tab(new_tab) tabs.active = new_tab.id self.open_file = None self.switching = True # open empty editor self.open_editor("", b"") def done_saving(self, result): if result is None: return #with open(result, "w", encoding="utf-8") as f: # f.write(self.query_one("#code-editor").text) tabs: Tabs = self.query_one("#file-tabs") tabs.active_tab.label = os.path.basename(result) self.notify(f"Saved to {result} successfully.", title="Done!", markup=False) self.query_one(DirectoryTree).reload() def action_save_as(self): self.push_screen(FileSave(), callback=self.done_saving) def action_save(self): if self.open_file == None: self.action_save_as() # dont bother saving if there are no new changes if not self.open_file in self.unsaved_files: return with open(self.open_file, "w") as f: f.write(self.unsaved_files[self.open_file]["current"]) tabs: Tabs = self.query_one("#file-tabs") tabs.active_tab.label = os.path.basename(self.open_file) tabs.active_tab.tooltip = str(tabs.active_tab.label) self.unsaved_files.pop(self.open_file) self.notify("Saved.") def action_quit(self): self.observer.stop() self.observer.join() return super().action_quit() async def on_ready(self): self.open_file = None self.unsaved_files = {} # list of paths self.switching = False self.file_clicked = False self.observer = Observer() self.observer.schedule(Watcher(self), path=self.path) self.observer.start() self.config_handler.apply_settings() if __name__ == "__main__": working_path = os.getcwd() if len(sys.argv) == 1 else sys.argv[1] app = Berry(working_path) app.run()