""" Code browser example. Run with: python code_browser.py PATH """ from __future__ import annotations import sys import subprocess import urllib.parse import json # Import the json module import yaml from pathlib import Path # Import Path from typing import Iterable, cast, Text# Import Iterable and cast for type hinting from rich.syntax import Syntax from rich.traceback import Traceback from rich.text import Text # Import Text for rich text rendering from rich.style import Style # Import Style for rich text styling from textual.app import App, ComposeResult from textual.containers import Container, VerticalScroll, Horizontal, Vertical, ScrollableContainer from textual.reactive import reactive, var from textual.widgets import (DirectoryTree, Footer, Header, Static, Input, Select, Switch, Button, Switch, Collapsible, TextArea ) from textual.reactive import reactive from textual import errors as textual_errors from textual.widgets._directory_tree import DirEntry # For type hinting in DirectoryTree from textual.widgets._tree import TreeNode # For type hinting in DirectoryTree from textual_autocomplete import AutoComplete, DropdownItem class FilteredDirectoryTree(DirectoryTree): """ A DirectoryTree that filters its entries based on a search query. """ # def reder_label(self, node: TreeNode[DirEntry], base_style: Style, style: Style) -> Text: # label = super().render_label(node, base_style, style) # node_label = node._label # # Check if the node has data (i.e., it's a file or directory entry) # self.notify(f"What.", title=f"Node", severity="error", timeout=15) # if node.data and isinstance(node.data, DirEntry): # path = node.data.path # self.notify(f"{path}.", title=f"Node", severity="error", timeout=15) # icon = "" # if path.is_dir(): # match path.name.lower(): # case "jobs": # icon = "πŸ“š " # case _: # icon = "πŸ“ " # if path.is_file(): # try: # with open(path, "r", encoding="utf-8") as file: # script = yaml.safe_load(file) # match script.get('status', None): # case "ready": # icon = "πŸ”΅[R] " # case "conduct": # icon = "🟣[C] " # case "setup": # icon = "🟣[S] " # case "test": # icon = "βšͺ️[T] " # case "packup": # icon = "🟀[P] " # case "done": # icon = "🟒[D] " # case _: # icon = "πŸ“„ " # except FileNotFoundError: # self.notify(f"Could not find file {self.path}.", title=f"File not found", severity="error", timeout=15) # return None # except yaml.YAMLError as e: # self.notify(f"Reading YAML format from {self.path} failed.", title=f"Failed to parse YAML", severity="error", timeout=15) # self.log.error(e) # return None # except Exception as e: # self.notify(f"Unexpected error while opening {self.path}.", title=f"Open error", severity="error", timeout=15) # self.log.error(e) # return None # # Construct the label with the custom icon # # Textual uses Rich's Text and Style. # # We insert the icon at the beginning of the label. # # You can also apply specific styles to the icon or filename. # if len(label) > 0: # Ensure the label is not empty # label.overwrite(icon, 0, 2) # # If you were adding classes for CSS, you would add them here. # # For direct icon manipulation, we just build the Text object. # label.highlight_regex(r"\.[^.]+$", Style(dim=True)) # return label # else: # # For non-file/directory nodes (e.g., the root if it's not a DirEntry) # # or if node.data is None for some reason, fall back to default behavior. # return super().render_label(node, base_style, style) def filter_paths(self, paths: Iterable[Path]) -> Iterable[Path]: """Filter paths based on the search query.""" all_files = [path for path in paths] self.log("All files:", all_files) files = [path for path in all_files if path.name.endswith(( ".yaml",".yml" ))] return files class HtmlStatic(Static): """A Static widget that renders HTML content.""" def __init__(self, content, item_path: Path, file_name: Path, **kwargs): """Initialize the widget.""" super().__init__(content, **kwargs) self.item_path = item_path # Store the actual path self.html_file_name = file_name def on_click(self) -> None: """Handle click event.""" self.app.notify(f"http://localhost:13000/{self.item_path}/{self.html_file_name}", title="Open in Browser") self.log.debug(f"Open Result url : http://localhost:13000/{self.item_path}/{self.html_file_name}") process = subprocess.Popen(f"\"/mnt/c/Program Files (x86)/Microsoft/Edge/Application/msedge.exe\" \"http://localhost:13000/{self.item_path.split("/jobs/")[1]}/{self.html_file_name}\"", shell=True) self.log.debug(f"Open with process id : ${process.pid}") class LogStatic(Static): """A Static widget that renders HTML content.""" def __init__(self, content, item_path: Path, file_name: Path, **kwargs): """Initialize the widget.""" super().__init__(content, **kwargs) self.item_path = item_path # Store the actual path self.file_name = file_name def on_click(self) -> None: """Handle click event.""" self.app.notify(f"http://localhost:14000/log?path={urllib.parse.quote(f"{self.item_path}/{self.file_name}", safe='')}#p0", title="Open in Browser") self.log.debug(f"Open Log url : http://localhost:14000/log?path={urllib.parse.quote(f"{self.item_path}/{self.file_name}", safe='')}#p0") process = subprocess.Popen(f"\"/mnt/c/Program Files (x86)/Microsoft/Edge/Application/msedge.exe\" \"http://localhost:14000/log?path={urllib.parse.quote(f"{self.item_path}/{self.file_name}", safe='')}#p0\"", shell=True) self.log.debug(f"Open with process id : ${process.pid}") class LogNavStatic(Static): """A Static widget that renders HTML content.""" def __init__(self, content, item_path: Path, **kwargs): """Initialize the widget.""" super().__init__(content, **kwargs) self.item_path = item_path # Store the actual path def on_click(self) -> None: """Handle click event.""" self.app.notify(f"http://localhost:14000/?dir={urllib.parse.quote(f"{self.item_path}/", safe='')}", title="Open in Browser") self.log.debug(f"Open Log url : http://localhost:14000/?dir={urllib.parse.quote(f"{self.item_path}/", safe='')}") process = subprocess.Popen(f"\"/mnt/c/Program Files (x86)/Microsoft/Edge/Application/msedge.exe\" \"http://localhost:14000/?dir={urllib.parse.quote(f"{self.item_path}/", safe='')}\"", shell=True) self.log.debug(f"Open with process id : ${process.pid}") class JobViewer(App): """Main application.""" CSS = """ Screen { &:inline { height: auto; } } Horizontal { width: 100%; height: auto; } Static { } Collapsible { width: 100%; } Collapsible .collapsible--body { align: left top; } HtmlStatic { border: blue; width: auto; height: auto; } HtmlStatic.click { background: darkred; } LogStatic { border: green; width: auto; height: auto; } LogNavStatic { border: yellow; width: auto; height: auto; } TextArea { height: 15%; min-height: 15; } Toast { } ToastRack { } Toast.-information { } Toast.-information .toast--title { text-style: bold; } Toast.-warning { } Toast.-error { } JobViewer.-show-tree #jobs-tree { display: block; max-width: 50%; } VerticalScroll { scrollbar-gutter: stable; overflow: auto; } /* Adjust by id */ #jobs-tree { display: none; scrollbar-gutter: stable; overflow: auto; width: auto; height: 100%; dock: left; } .job-screen { width: 100%; # border: solid blue; height: auto; } .job-meta { # border: solid green; height: auto; } .job-results { } .job-results-title { width: 20%; height: auto; } .job-meta-title{ width: 30%; height: auto; } .job-meta-data{ height: auto; } """ BINDINGS = [ ("b", "job_abort", "Abort Job"), ("r", "job_clone", "Clone a new Job"), ("c", "job_create", "Create a new"), ("s", "job_save", "Save a Job"), ("f", "toggle_files", "Toggle Files"), ("q", "quit", "Quit"), ] DEPOSIT_SAMPLE = """\ jenkins_workspace: - type: sshfs - server: 192.168.34.33 - path: /home/builduser1/jenkins-agent/workspace - user: builduser1 - password: '$xpffpclqtm12' - options: ro """ XTS_SAMPLE = """\ run cts -m CtsMediaV2TestCases run cts -m CtsMediaCodecTestCases run cts -m CtsMediaDecoderTestCases run cts -m CtsMediaExtractorTestCases run cts -m CtsNativeHardwareTestCases """ PRESCRIPT_SAMPLE = """\ source_workspace/script1 jenkins_workspace/script2 """ POSTSCRIPT_SAMPLE = """\ source_workspace/script3 jenkins_workspace/script4 """ FWDN_SAMPLE = """\ --fwdn tcc805x_fwdn.cs.json --storage emmc --low-format --storage snor --low-format -w tcc805x_boot.cs.json -w tools/mktcimg/SD_Data.fai --storage emmc --area user --no-crc -w prebuilt/tcc8050_snor.cs.rom --storage snor --area die1 --no-crc """ show_tree = var(True) path: reactive[str | None] = reactive(None) job_alias_textarea = reactive("") candidate_cts_suite = reactive([ # This cts suites update from Android CTS server and deposit cached directory. "⭐ android-cts-12_r8-linux_x86-arm", "⭐ android-cts-13_r8-linux_x86-arm", "⭐ android-cts-14_r8-linux_x86-arm", "⭐ android-cts-15_r8-linux_x86-arm", "⭐ android-cts-16_r8-linux_x86-arm", ]) candidate_board_name = reactive([ DropdownItem(main="0", prefix="🐍"), DropdownItem(main="1", prefix="πŸ“œ"), DropdownItem(main="2", prefix="πŸ”·"), ]) candidate_fwdn_path = reactive([ DropdownItem(main="0", prefix="🐍"), DropdownItem(main="1", prefix="πŸ“œ"), DropdownItem(main="2", prefix="πŸ”·"), ]) candidate_rsc_device = reactive([ DropdownItem(main="0", prefix="🐍"), DropdownItem(main="1", prefix="πŸ“œ"), DropdownItem(main="2", prefix="πŸ”·"), ]) candidate_rsc_fwdn = reactive([ DropdownItem(main="0", prefix="🐍"), DropdownItem(main="1", prefix="πŸ“œ"), DropdownItem(main="2", prefix="πŸ”·"), ]) candidate_rsc_normal = reactive([ DropdownItem(main="0", prefix="🐍"), DropdownItem(main="1", prefix="πŸ“œ"), DropdownItem(main="2", prefix="πŸ”·"), ]) def __inti__(self) -> None: super().__init__() @staticmethod def _Static(title: str, id: str, value, *args, **kwargs): return Horizontal(Static(f"{title}", classes="job-meta-title", id=f"{id}-static-title"), Static(f"{value}", classes="job-data", id=f"{id}-static-data"), name=title, id=id) @staticmethod def _LogStatic(title: str, id: str, item_path: str, file_name:str, *args, **kwargs): return Horizontal(Static(f"{title}", classes="job-meta-title", id=f"{id}-static-title"), LogStatic(f"", item_path=item_path, file_name = file_name, classes="job-log-data", id=f"{id}-logstatic-data"), name=title) @staticmethod def _AutoComplete(title: str, id: str, placeholder: str, candidates: list, *args, **kwargs): user_input=Input(placeholder=placeholder, classes="job-create-input", id=f"{id}-auto-input") return Static(f"{title}", classes="job-create-title", id=f"{id}-auto-title"), user_input, AutoComplete(user_input, name=title, candidates=candidates, id=f"{id}-auto-data") @staticmethod def _Input(title: str, id: str, placeholder: str, *args, **kwargs): return Static(f"{title}", classes="job-create-title", id=f"{id}-input-title"), Input(name=title, placeholder=placeholder, classes="job-create-input", id=f"{id}-input-data") @staticmethod def _TextArea(title: str, id: str, placeholder: str, language: str="bash", *args, **kwargs): return Static(f"{title}", classes="job-create-title", id=f"{id}-textarea-title"), TextArea(placeholder, name=title, show_line_numbers=True, language=language, id=f"{id}-textarea-data") @staticmethod def _Select(title: str, id: str, options: list|tuple, *args, **kwargs): return Static(f"{title}", classes="job-create-title", id=f"{id}-select-title"), Select(name=title, allow_blank=False, classes="job-create-select", options=((line, line) for line in options), id=f"{id}-select-data") def watch_show_tree(self, show_tree: bool) -> None: """Called when show_tree is modified.""" self.set_class(show_tree, "-show-tree") def compose(self) -> ComposeResult: """Compose our UI.""" try: self.path = "./" if len(sys.argv) < 2 else sys.argv[1] except Exception as e: self.path = "./" self.notify(f"Failed to open {self.path}: {e}.",title="Open Path Error", severity="error", timeout=15) yield Header() with ScrollableContainer(): yield FilteredDirectoryTree(Path(self.path), id="jobs-tree") # Job Generator with VerticalScroll(id="job-create"): yield from JobViewer._TextArea("μΆ”κ°€ μ €μž₯μ†Œ", placeholder=self.DEPOSIT_SAMPLE, id="deposit-user") yield from JobViewer._AutoComplete(title = "ν…ŒμŠ€νŠΈ Suite 이름", placeholder = "android-cts-14_r8-linux_x86-arm", candidates = self.candidate_cts_suite, id="cts-suite") yield from JobViewer._TextArea(title = "cts-tradefed λͺ…λ Ήμ–΄", placeholder = self.XTS_SAMPLE, id="cts-command") yield from JobViewer._AutoComplete(title = "λ³΄λ“œ μ’…λ₯˜", placeholder = "TCC8050", candidates = self.candidate_board_name, id="board-type") yield from JobViewer._AutoComplete(title = "νŽŒμ›¨μ–΄ 경둜", placeholder = "jenkins-workspace/source", candidates = self.candidate_fwdn_path, id="fwdn-path") yield from JobViewer._TextArea(title = "FWDN λͺ…λ Ήμ–΄", placeholder = self.FWDN_SAMPLE, id="fwdn-command") yield from JobViewer._AutoComplete(title = "RSC μž₯치", placeholder = "-d2", candidates = self.candidate_rsc_device, id="rsc-board-type") yield from JobViewer._AutoComplete(title = "RSC FWDN", placeholder = "0", candidates = self.candidate_rsc_fwdn, id="rsc-fwdn") yield from JobViewer._AutoComplete(title = "RSC Normal", placeholder = "2", candidates = self.candidate_rsc_normal, id="rsc-normal") yield from JobViewer._TextArea(title = "μ „ 처리 슀크립트", placeholder = self.PRESCRIPT_SAMPLE, id="pre-script") yield from JobViewer._TextArea(title = "ν›„ 처리 슀크립트", placeholder = self.POSTSCRIPT_SAMPLE, id="post-script") # Job Monitor with VerticalScroll(id = "job-view"): with Static(id = "job-meta"): yield JobViewer._Static(title = "μž‘μ—… μƒνƒœ", value = "", id="runner-status") yield JobViewer._Static(title = "λ³΄λ“œ μ’…λ₯˜", value = "", id="board-type") yield JobViewer._Static(title = "생성 일자", value = "", id="create-date") with Collapsible(id = "job-results", title = "Job Results", collapsed = False): yield Static (id = "overall-static-data") yield Static (id = "each-static-data") with Collapsible(id = "job-detail", title = "Job Details", collapsed = True): yield JobViewer._Static(title = "Script Hash", value = "", id="script-hash") yield JobViewer._Static(title = "Overseer Hash", value = "", id="overseer-hash") yield JobViewer._Static(title = "Runner Hash", value = "", id="runner-hash") with Collapsible(id = "job-source", title = "Job Script File", collapsed = True): yield Static(id = "script-static-data") yield Footer() def on_mount(self) -> None: self.query_one(DirectoryTree).focus() for _ in self.query(VerticalScroll): _.display = False def theme_change(_signal) -> None: """Force the syntax to use a different theme.""" self.watch_path(self.path) self.theme_changed_signal.subscribe(self, theme_change) def on_directory_tree_file_selected(self, event: DirectoryTree.FileSelected) -> None: """Called when the user click a file in the directory tree.""" event.stop() self.yaml_data = None self.path = str(event.path) self.log.debug(f"event stopped") self.log.debug(f"event.path: {str(event.path)}") self.log.debug(f"self.yaml_data: {self.yaml_data}") self.log.debug(f"self.path: {self.path}") self.item_path = Path(self.path) # ABS path of job.yml self.item_subdir = Path(f"{self.item_path.parent}/{Path(self.path).stem}") # ABS path of job/ self.log.debug(f"item_path: {self.item_path}") self.log.debug(f"item_subdir: {self.item_subdir}") job_view = self.query_one("#job-view") job_view.display = True self.query_one("#job-create").display = False if self.item_path.suffix not in (".yaml", ".yml"): self.log.info(f"File extension is not ends with .yaml or .yml at {self.path}") return None try: with open(self.item_path, "r", encoding="utf-8") as file: self.yaml_data = yaml.safe_load(file) if not isinstance(self.yaml_data, dict): raise yaml.YAMLError("Expected a dictionary") except FileNotFoundError: self.notify(f"Could not find file {self.path}.", title=f"File not found", severity="error", timeout=15) return None except yaml.YAMLError as e: self.notify(f"Reading YAML format from {self.path} failed.", title=f"Failed to parse YAML", severity="error", timeout=15) self.log.error(e) return None except Exception as e: self.notify(f"Unexpected error while opening {self.path}.", title=f"Open error", severity="error", timeout=15) self.log.error(e) return None if "meta" not in self.yaml_data: self.notify(f"No meta data found in {self.path}.", title=f"No meta data", severity="error", timeout=15) self.log.error(f"No meta data found in {self.path}.") job_view.query_one("#job-meta").display = False return None # if "runner" not in self.yaml_data: # self.notify(f"No runner data found in {self.path}.", title=f"No runner data", severity="warning", timeout=15) # self.log.warning(f"No runner dabtopta found in {self.path}.") # if "deposit" not in self.yaml_data: # self.notify(f"No deposit data found in {self.path}.", title=f"No deposit data", severity="warning", timeout=15) # self.log.warning(f"No deposit data found in {self.path}.") if not self.item_subdir.is_dir(): self.notify(f"No output results in name of {self.item_subdir}.", title=f"No Output found", severity="warning", timeout=15) self.log.warning(f"No output results in name of {self.item_subdir}.") self.query_one("#job-meta #runner-status-static-data").update(f"{self.yaml_data["meta"].get('status', None)}") self.query_one("#job-meta #board-type-static-data").update(f"{self.yaml_data["meta"].get('board_type', None)}") self.query_one("#job-detail #script-hash-static-data").update(f"{self.yaml_data["meta"].get('script_hash', None)}") self.query_one("#job-detail #overseer-hash-static-data").update(f"{self.yaml_data["meta"].get('overseer_hash', None)}") self.query_one("#job-detail #runner-hash-static-data").update(f"{self.yaml_data["meta"].get('runner_hash', None)}") self.query_one("#job-source #script-static-data").update( Syntax.from_path( self.path, line_numbers=True, word_wrap=False, indent_guides=True, theme="github-dark" if self.current_theme.dark else "github-light", ) ) self.query_one("#job-source #script-static-data").collapsible = False self.query_one("#job-results #overall-static-data").remove_children() if Path(f"{self.item_subdir}/runner-output.log").is_file(): self.query_one("#job-results #overall-static-data").mount( Horizontal( Static(f"Full Log", classes="job-results-title"), LogStatic(f"Instance Log", item_path=f"{self.item_subdir}", file_name=f"runner-output.log") ) ) else: self.notify(f"No runner log file found in {self.item_subdir}.", title=f"No runner log found", severity="warning", timeout=15) self.log.warning(f"No runner log file found in {self.item_subdir}/runner-output.log") self.query_one("#job-results #each-static-data").remove_children() if Path(f"{self.item_subdir}/results").is_dir(): for item in Path(f"{self.item_subdir}/results").iterdir(): self.log.verbose(f"item is {item.name}") if not Path(item).is_dir(): self.log.debug(f"item is not a directory : {item}") continue # self.log.debug(f"inv_dirs : {[inv_dir for inv_dir in Path(item).glob("inv_*")]}") self.query_one("#job-results #each-static-data").mount( Horizontal( Static(f"{item.name}", classes="job-results-title"), HtmlStatic(f"Result\nAll", file_name=f"test_result.html", item_path=f"{self.item_subdir}/results/{item.name}"), HtmlStatic(f"Result\nFails", file_name=f"test_result_failures_suite.html", item_path=f"{self.item_subdir}/results/{item.name}"), LogStatic(f"Invocation\nSumary", item_path=f"{self.item_subdir}/results/{item.name}", file_name=f"invocation_summary.txt"), LogStatic(f"Session log\nolc", item_path=f"{self.item_subdir}/logs/{item.name}/olc_server_session_logs", file_name=f"olc_server_session_log.txt"), LogNavStatic(f"Session log\ninv ", item_path=f"{self.item_subdir}/logs/{item.name}") ) ) # log_navs=[LogNavStatic(f"{inv_dir.name}", item_path=f"{item_subdir}/logs/{item.name}/{inv_dir.name}") for inv_dir in Path(f"{item_subdir}/results/{item.name}").iterdir()] # log_items.append( # Horizontal(*log_navs) # ) else: self.notify(f"No results directory found in {self.item_subdir}.", title=f"No Results found", severity="warning", timeout=15) self.log.warning(f"No results directory found in {self.item_subdir}.") self.refresh() def on_text_area_changed(self, event: TextArea.Changed) -> None: match event.control.id: case "deposit-user-textarea-data": pass case "cts-command-textarea-data": pass case "fwdn-command-textarea-data": pass case "pre-script-textarea-data": pass case "post-script-textarea-data": pass case _: pass pass def on_input_changed(self, event: Input.Changed) -> None: match event.control.id: case "fwdn-path-auto-input": pass case "cts-suite-auto-input": pass case _: pass def watch_path(self, path: str | None) -> None: """Called when path changes.""" # code_view = self.query_one("#content", Static) # if path is None: # code_view.update("") # return # try: # syntax = Syntax.from_path( # path, # line_numbers=True, # word_wrap=False, # indent_guides=True, # theme="github-dark" if self.current_theme.dark else "github-light", # ) # except Exception: # code_view.update(Traceback(theme="github-dark", width=None)) # self.sub_title = "ERROR" # else: # code_view.update(syntax) # self.query_one("#job-view").scroll_home(animate=False) self.sub_title = path def action_toggle_files(self) -> None: """Called in response to key binding.""" self.show_tree = not self.show_tree def action_job_create(self) -> None: """Called in response to key binding.""" self.query_one("#job-view").display = False job_create= self.query_one("#job-create") job_create.display = True self.title = "Create a new Job" self.sub_title = "" # self.query_one("#job-detail #runner-hash-static-data").update(f"{self.yaml_data["meta"].get('runner_hash', None)}") def on_button_pressed(self, event: Button.Pressed) -> None: self.title = str("Button pressed") self.sub_title = str(event.button) + " " + str(event.button.label) + " " + str(event.button.id) # job_alias = self.query_one("#job-alias") # job_alias.remove_children() # job_alias.mount_all([ # Static("μ €μž₯μ†Œ 별칭", classes="job-create-title"), # *JobViewer._Input("별칭 이름", placeholder="source-workspace"), # *JobViewer._Select("μ—°κ²° 방법", options=["sshfs", "rsync", "REST (Jenkins)"]), # Button("Generate", variant="primary"), # ]) # self.exit(str(event.button)) def on_select_changed(self, event: Select.Changed) -> None: self.title = str(f"Select triggered: {event.value}") # if event.sender.title == "connection-type-select": # Check if it's our specific Select # selected_type = event.value # self.log(f"Selected connection type: {selected_type}") # try: # job_alias = self.query_one("job-alias") # # Clear existing dynamic inputs # job_alias.remove_children() # self.notify("Job alias cleaned.", title=f"Alias cleaned", severity="debug") # # Add new inputs based on selection # match selected_type: # case "sshfs": # job_alias.mount( # Static("경둜", classes="job-create-title"), # Input(placeholder="jenkins-tools/android-xts-deposit/deposit"), # Static("Mount Options", classes="job-create-title"), # Input(placeholder="password_stdin,direct_io,ro,reconnect,ServerAliveInterval=15,ServerAliveCountMax=3,allow_other,StrictHostKeyChecking=no"), # Static("μ‚¬μš©μž", classes="job-create-title"), # Input(placeholder="builduser1"), # Static("λΉ„λ°€λ²ˆν˜Έ/Access Token", classes="job-create-title"), # Input(placeholder="$xpffpclqtm12"), # ) # case _: # pass # except Exception as e: # self.notify(f"{list(job_view.children)}", title=f"on_select_changed", severity="information") # self.notify(f"{len(job_view.children)}", title=f"on_select_changed", severity="information") # self.notify(f"{e}", title=f"on_select_changed", severity="information") def action_job_clone(self) -> None: if self.yaml_data is None: self.notify("A job script is not sleected.", title=f"No Job is selected", severity="information") self.log.info(f"JA job script is not sleected. Empty on YAML data. {self.yaml_data}") return None pass def action_job_abort(self) -> None: if self.yaml_data is None: self.notify("A job script is not sleected.", title=f"No Job is selected", severity="information") self.log.info(f"JA job script is not sleected. Empty on YAML data. {self.yaml_data}") return None meta_data = self.yaml_data["meta"] # TODO: Hard coded on status. Change it from future match meta_data.get("status", None): case "done": self.notify("Job is already done.", title=f"Abort Job failed.", severity="information") self.log.info(f"Job is already done. Cannot restate to \'abort\' on {self.path}.") case "abort": self.notify("Job is in abort.", title=f"Abort Job failed.", severity="information") self.log.info(f"Job is already set abort. Already \'abort\' on {self.path}.") case "packup": self.notify("Job is in packup.", title=f"Abort Job failed.", severity="information") self.log.info(f"Job is in packup. This goes to the \'done\' after it finished. {self.path}.") case _: self.yaml_data["meta"]["status"] = "abort" try: with open(self.path, 'w') as file: yaml.dump(self.yaml_data, file, default_flow_style=False, sort_keys=False) self.notify(f"Aborting \'{self.path}\'", title=f"Aborting Job.", severity="information") self.log.info(f"Aborting a job on {self.path}.") # This open is to check purpose. with open(self.path, "r", encoding="utf-8") as file: pass except FileNotFoundError: self.notify(f"Fail to save a job script of \'{self.path}\'", title=f"File not Found.", severity="error") self.log.error(f"Fail to save a job script of \'{self.path}\'") except yaml.YAMLError as e: self.notify(f"Fail to save a job script of \'{self.path}\'", title=f"YAML dump failed.", severity="error") self.log.error(f"Fail to save a job script of \'{self.path}\'. {e}") except Exception as e: self.notify(f"Unexpected failure while writing job script file \'{self.path}\'.", title=f"Unknown YAML dump failed.", severity="error") self.log.error(f"Unexpected failure while writing job script file \'{self.path}\'. {e}") if __name__ == "__main__": JobViewer().run()