Last active
March 6, 2024 22:18
-
-
Save JarbasAl/9c19ee61796e42827af70989b4fef0a2 to your computer and use it in GitHub Desktop.
modernize mycroft skills
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import os | |
| import shutil | |
| from os.path import basename, dirname | |
| import subprocess | |
| from yapf.yapflib.yapf_api import FormatCode | |
| IMPORTS_MAP = { | |
| "IntentQueryApi": "from ovos_utils.intents.intent_service_interface import IntentQueryApi # TODO - deprecated utils 0.1.0", | |
| "save_settings": "save_settings = lambda a, b: None # TODO - DEPRECATED use self.settings.store()", | |
| "make_priority_skill": "make_priority_skill = lambda k: False # TODO - DEPRECATED, add RuntimeRequirements classproperty instead", | |
| "update_mycroft_config": "from ovos_config.config import update_mycroft_config", | |
| "create_daemon": "from ovos_utils import create_daemon", | |
| "wait_for_exit_signal": "from ovos_utils import wait_for_exit_signal", | |
| "IntentLayers": "from ovos_workshop.decorators.layers import IntentLayers", | |
| "IntentBuilder": "from ovos_workshop.intents import IntentBuilder", | |
| "AdaptIntent": "from ovos_workshop.intents import Intent as AdaptIntent", | |
| "MycroftSkill": "from ovos_workshop.skills import OVOSSkill", | |
| "FallbackSkill": "from ovos_workshop.skills.fallback import FallbackSkill", | |
| "CommonQuerySkill": "from ovos_workshop.skills.common_query_skill import CommonQuerySkill", | |
| "AutotranslatableSkill": "from ovos_workshop.skills.auto_translatable import UniversalSkill", | |
| "AutotranslatableFallback": "from ovos_workshop.skills.auto_translatable import UniversalFallbackSkill", | |
| "CQSMatchLevel": "from ovos_workshop.skills.common_query_skill import CQSMatchLevel", | |
| "MessageBusClient": "from ovos_bus_client import MessageBusClient", | |
| "Message": "from ovos_bus_client.message import Message", | |
| "dig_for_message": "from ovos_bus_client.message import dig_for_message", | |
| "resolve_resource_file": "from ovos_workshop.resource_files import resolve_resource_file", | |
| "DeviceApi": "from ovos_backend_client import DeviceApi", | |
| "LOG": "from ovos_utils.log import LOG", | |
| "Configuration": "from ovos_config import Configuration", | |
| "ConfigurationManager": "from ovos_config import Configuration", | |
| "FileSystemAccess": "from ovos_workshop.filesystem import FileSystemAccess", | |
| "getLogger": "from ovos_utils.log import LOG\ngetLogger = lambda k: LOG", | |
| "resting_screen_handler": "from ovos_workshop.decorators import resting_screen_handler", | |
| "intent_handler": "from ovos_workshop.decorators import intent_handler", | |
| "intent_file_handler": "from ovos_workshop.decorators import intent_handler", | |
| "adds_context": "from ovos_workshop.decorators import adds_context", | |
| "removes_context": "from ovos_workshop.decorators import removes_context", | |
| "wait_while_speaking": "from time import sleep\nwait_while_speaking = lambda: sleep(1) # DEPRECATED - TODO replace me", | |
| "is_speaking": "is_speaking = lambda : False # DEPRECATED - TODO replace me", | |
| "stop_speaking": "stop_speaking = lambda : False # DEPRECATED - TODO replace me", | |
| "create_signal": "create_signal = lambda : False # DEPRECATED - TODO replace me", | |
| "match_one": "from ovos_utils.parse import match_one", | |
| "fuzzy_match": "from ovos_utils.parse import fuzzy_match", | |
| "get_cache_directory": "from ovos_utils.file_utils import get_cache_directory", | |
| "extract_datetime": "from lingua_franca.parse import extract_datetime", | |
| "nice_duration": "from lingua_franca.format import nice_duration", | |
| "extract_duration": "from lingua_franca.parse import extract_duration", | |
| "normalize": "from lingua_franca.parse import normalize", | |
| "nice_number": "from lingua_franca.format import nice_number", | |
| "nice_date": "from lingua_franca.format import nice_date", | |
| "nice_date_time": "from lingua_franca.format import nice_date_time", | |
| "nice_time": "from lingua_franca.format import nice_time", | |
| "pronounce_number": "from lingua_franca.format import pronounce_number", | |
| "date_time_format": "from lingua_franca.format import date_time_format", | |
| "join_list": "from lingua_franca.format import join_list", | |
| "extract_number": "from lingua_franca.parse import extract_number", | |
| "extractnumber": "from lingua_franca.parse import extract_number as extractnumber # TODO - update usage", | |
| "extract_numbers": "from lingua_franca.parse import extract_numbers", | |
| "AudioService": "from ovos_bus_client.apis.ocp import OCPAudioServiceInterface as AudioService", | |
| "get_ipc_directory": "from ovos_utils.signal import get_ipc_directory", | |
| "read_vocab_file": "from ovos_utils.file_utils import read_vocab_file", | |
| "play_wav": "from ovos_utils.sound import play_audio as play_wav", | |
| "play_mp3": "from ovos_utils.sound import play_audio as play_mp3", | |
| "play_audio_file": "from ovos_utils.sound import play_audio as play_audio_file", | |
| "now_local": "from ovos_utils.time import now_local", | |
| "now_utc": "from ovos_utils.time import now_utc", | |
| "to_utc": "from ovos_utils.time import to_utc", | |
| "to_local": "from ovos_utils.time import to_local", | |
| "to_system": "from ovos_utils.time import to_system", | |
| "TTSFactory": "from ovos_plugin_manager.tts import OVOSTTSFactory as TTSFactory", | |
| "skill_api_method": "from ovos_workshop.decorators import skill_api_method", | |
| "LocalConf": "from ovos_config.models import LocalConf", | |
| "USER_CONFIG": "from ovos_config.locations import USER_CONFIG", | |
| "OLD_USER_CONFIG": "from ovos_config.locations import USER_CONFIG as OLD_USER_CONFIG # TODO remove me", | |
| "DEFAULT_CONFIG": "from ovos_config.locations import DEFAULT_CONFIG", | |
| "SYSTEM_CONFIG": "from ovos_config.locations import SYSTEM_CONFIG", | |
| "is_paired": "from ovos_backend_client.pairing import is_paired", | |
| "IdentityManager": "from ovos_backend_client.identity import IdentityManager", | |
| "connected": "from ovos_utils.network_utils import is_connected as connected", | |
| "CORE_VERSION_MAJOR": "from ovos_core.version import CORE_VERSION_MAJOR", | |
| "CORE_VERSION_MINOR": "from ovos_core.version import CORE_VERSION_MINOR", | |
| "CORE_VERSION_BUILD": "from ovos_core.version import CORE_VERSION_BUILD", | |
| "CORE_VERSION_STR": "from ovos_core.version import CORE_VERSION_STR", | |
| "CORE_VERSION_TUPLE": "from ovos_core.version import CORE_VERSION_TUPLE" | |
| } | |
| class SkillConverter: | |
| def __init__(self, folder): | |
| self.clazz_name = "" | |
| self.base_skill = "" | |
| self.warnings = [] | |
| self.path = folder | |
| self.skill_id = basename(folder) | |
| def fix_skill(self, init_file): | |
| self.clazz_name = "" | |
| self.base_skill = "" | |
| self.warnings = [] | |
| with open(init_file) as f: | |
| self.code = f.read() | |
| self.lines = self.code.split("\n") | |
| if "print " in self.code: # TODO better python2 vs 3 detection | |
| subprocess.call(f"2to3 -w {init_file}", shell=True) | |
| with open(init_file) as f: | |
| self.code = f.read() | |
| self.lines = self.code.split("\n") | |
| self.warnings.append("[MODERNIZE] from python2 to python3") | |
| self.has_stop = "def stop(self):" in self.code | |
| self.has_shutdown = "def shutdown(self):" in self.code | |
| self.has_converse = "def converse(self," in self.code | |
| # ancient versions of mycroft | |
| self.very_old = "self.emitter" in self.code and "self.bus" not in self.code | |
| self.fix_import_hacks() | |
| self.fix_bus() | |
| self.fix_skill_id() | |
| self.fix_star_imports() | |
| self.fix_imports() | |
| self.fix_class() | |
| self.fix_init() | |
| self.fix_initialize() | |
| self.fix_decorators() | |
| self.fix_settings() | |
| self.fix_wait_while_waiting() | |
| self.fix_stop_speaking() | |
| self.fix_config() | |
| self.fix_stop() | |
| self.fix_create_func() | |
| self.fix_translate() | |
| code = "\n".join([l for l in self.lines if l is not None]) | |
| try: | |
| code, changed = FormatCode(code) | |
| except: | |
| pass | |
| return code, list(self.warnings) | |
| def fix_import_hacks(self): | |
| """undo try: except: import hacks common in early mycroft days, | |
| copy pasted around in several skills in the wild""" | |
| old_skill_shim = """try: | |
| from mycroft.skills.core import MycroftSkill | |
| except: | |
| class MycroftSkill: | |
| pass""" | |
| if old_skill_shim in self.code: | |
| self.code = self.code.replace(old_skill_shim, "from mycroft.skills.core import MycroftSkill") | |
| self.lines = self.code.split("\n") | |
| self.warnings.append("[MODERNIZE] undo skill import shim (very old mycroft)") | |
| old_play_shim = """try: | |
| from mycroft.skills.audioservice import AudioService | |
| except: | |
| from mycroft.util import play_mp3 | |
| AudioService = None""" | |
| if old_play_shim in self.code: | |
| self.code = self.code.replace(old_play_shim, "from mycroft.skills.audioservice import AudioService\nfrom mycroft.util import play_mp3") | |
| self.lines = self.code.split("\n") | |
| self.warnings.append("[MODERNIZE] undo audioservice import shim (very old mycroft)") | |
| def fix_star_imports(self): | |
| for idx, l in enumerate(self.lines): | |
| if l is None or " import *" not in l: | |
| continue | |
| if l.startswith("from mycroft.skills.context "): | |
| self.lines[idx] = "from ovos_workshop.decorators import adds_context, removes_context" | |
| self.warnings.append("[MODERNIZE] expand star import: mycroft.skills.context -> adds_context, removes_context") | |
| def fix_skill_id(self): | |
| for idx, l in enumerate(self.lines): | |
| if "_author__ =" in l: | |
| author = l.split("=")[-1].strip()[1:-1] | |
| self.warnings.append(f"[INFO] line {idx} - detected author: {author}") | |
| if "." not in self.skill_id: | |
| author = author.lower().\ | |
| replace("_", "-"). \ | |
| replace(" ", "-"). \ | |
| replace(".", "-"). \ | |
| replace(",", "-"). \ | |
| replace("/", "-"). \ | |
| replace("|", "-"). \ | |
| replace("'", ""). \ | |
| replace('"', "") | |
| name = self.skill_id.lower().\ | |
| replace("_", "-"). \ | |
| replace(" ", "-"). \ | |
| replace(".", "-"). \ | |
| replace(",", "-"). \ | |
| replace("/", "-"). \ | |
| replace("|", "-"). \ | |
| replace("'", ""). \ | |
| replace('"', "") | |
| self.skill_id = f"{name}.{author}" | |
| self.warnings.append(f"[INFO] skill_id: {self.skill_id}") | |
| def fix_stop_speaking(self): | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| if l.strip() in ["stop_speaking()", "mycroft.audio.stop_speaking()"]: | |
| l = l.replace("mycroft.audio.stop_speaking()", "stop_speaking()") | |
| prefix = l.split("stop_speaking()")[0] | |
| self.lines[idx] = f"{prefix}self.bus.emit(Message('mycroft.audio.speech.stop'))" | |
| self.warnings.append(f"[MODERNIZE] line {idx} - stop_speaking() -> 'mycroft.audio.speech.stop'") | |
| def fix_initialize(self): | |
| self.detected_events = [] | |
| try: | |
| start, end = self.find_skill_method("initialize") | |
| except: | |
| return # not defined | |
| self.warnings.append(f"[INFO] line {start} / {end} - detected initialize method") | |
| for idx, l in enumerate(self.lines): | |
| if idx < start or idx > end: | |
| continue | |
| # in very OLD mycroft this was needed, it was made part of base class later | |
| # some skills still have this copy pasted from the original mycroft template | |
| if l.strip() == "self.init_dialog(dirname(__file__))": | |
| self.lines[idx] = None | |
| self.warnings.append( | |
| f"[REMOVED] line {idx} - deprecated (very old mycroft) initialize boilerplate self.init_dialog(dirname(__file__))") | |
| elif l.strip() == "self.load_data_files(dirname(__file__))": | |
| self.lines[idx] = None | |
| self.warnings.append( | |
| f"[REMOVED] line {idx} - deprecated (very old mycroft) initialize boilerplate self.load_data_files(dirname(__file__))") | |
| elif l.strip().startswith("self.load_vocab_files(join(dirname(__file__), 'vocab',"): | |
| self.lines[idx] = None | |
| self.warnings.append( | |
| f"[REMOVED] line {idx} - deprecated (very old mycroft) initialize boilerplate self.load_vocab_files(join(dirname(__file__), 'vocab', self.lang))") | |
| elif l.strip().startswith("self.load_regex_files(join(dirname(__file__), 'regex',"): | |
| self.lines[idx] = None | |
| self.warnings.append( | |
| f"[REMOVED] line {idx} - deprecated (very old mycroft) initialize boilerplate self.load_regex_files(join(dirname(__file__), 'regex', self.lang))") | |
| elif "super(" in l and ".initialize()" in l: | |
| self.lines[idx] = None | |
| self.warnings.append(f"[REMOVED] line {idx} - deprecated call to super() in initialize") | |
| elif "self.bus.on(" in l: | |
| event = l.split("self.bus.on(")[-1].split(",")[0].strip()[1:-1] | |
| self.detected_events.append(event) | |
| self.warnings.append(f"[INFO] line {idx} - detected bus event listener: {event}") | |
| self.lines[idx] = l.replace("self.bus.on(", "self.add_event(") | |
| self.warnings.append(f"[MODERNIZE] line {idx} - self.bus.on -> self.add_event") | |
| # check for empty method | |
| lines = [l for l in self.lines[start+1:end] if l] | |
| if len(lines) == 0: | |
| self.lines[start] = None | |
| def fix_translate(self): | |
| if "from mtranslate import translate" not in self.code: | |
| return | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| if "from mtranslate import translate" in l: | |
| self.lines[idx] = "# " + l | |
| self.warnings.append( | |
| f"[WARNING] line {idx} - mtranslate detected, broken and unmaintained") | |
| elif " translate(" in l: | |
| self.lines[idx] = l.replace(" translate(", | |
| " self.translator.translate(") | |
| self.warnings.append( | |
| f"[MODERNIZE] line {idx} - mtranslate.translate -> self.translator.translate") | |
| def fix_stop(self): | |
| try: | |
| start, end = self.find_skill_method("stop") | |
| except: | |
| return # not defined | |
| self.warnings.append(f"[INFO] line {start} / {end} - detected stop method") | |
| if end - start == 1: # single line stop definition | |
| if self.lines[end].strip() == "return True": | |
| self.warnings.append(f"[FIX ERROR] line {end} - stop method always returned True") | |
| self.lines[start] = None | |
| self.lines[end] = None | |
| if self.lines[end].strip() in ["return False", "pass"]: | |
| self.warnings.append(f"[REMOVED] line {end} - stop method boilerplate") | |
| self.lines[start] = None | |
| self.lines[end] = None | |
| elif not any("return " in s for s in self.lines[start:end]): | |
| self.warnings.append( | |
| f"[WARNING] line {end} - stop method does not return True when consuming mycroft.stop") | |
| def fix_config(self): | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| if "ConfigurationManager.get()" in l: | |
| self.warnings.append(f"[MODERNIZE] line {idx} - ConfigurationManager.get() -> Configuration()") | |
| self.lines[idx] = l.replace("ConfigurationManager.get()", | |
| "Configuration()") | |
| if "Configuration.get()" in l: | |
| self.warnings.append(f"[MODERNIZE] line {idx} - Configuration.get() -> Configuration()") | |
| self.lines[idx] = l.replace("Configuration.get()", | |
| "Configuration()") | |
| def fix_bus(self): | |
| # VERY old mycroft-core skills | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| if "self.emitter" in l: | |
| self.lines[idx] = l.replace("self.emitter", | |
| "self.bus") | |
| self.warnings.append(f"[MODERNIZE] line {idx} - deprecated (very old mycroft) self.emitter -> self.bus") | |
| def fix_imports(self): | |
| mline = False | |
| prefix = "" | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| if l.startswith("#"): | |
| continue | |
| if l.strip().startswith("import mycroft."): | |
| self.lines[idx] += " # TODO - mycroft is DEPRECATED" | |
| continue | |
| # handle imports broken into lines by \ | |
| # TODO - improve this to handle multi line better | |
| # tested skills never used more than one but its valid | |
| if "import " in l and l.strip().endswith("\\"): | |
| self.lines[idx] = l = l.strip()[:-1] + " " + self.lines[idx + 1].strip() | |
| self.lines[idx + 1] = None | |
| if l.startswith("from ") and " import " in l: | |
| if not any(s in l for s in ["mycroft", "adapt", "msm", | |
| "jarbas_utils", | |
| "ovos_utils.messagebus", | |
| "ovos_utils.skills", | |
| "ovos_utils.signal", | |
| "ovos_utils.intents"]): | |
| continue # not a mycroft import, dont touch it | |
| if "mycroft.skills.msm_wrapper" in l: | |
| self.warnings.append( | |
| f"[ERROR] line {idx} - msm has been deprecated, this import can't be fixed automatically") | |
| if "(" in l: | |
| mline = True | |
| l = l.replace("(", "") | |
| if l.strip().endswith(" import"): | |
| self.lines[idx] = None | |
| continue | |
| else: | |
| mline = False | |
| prefix = l.split(" import ")[0] | |
| imported = [i.strip() for i in l.split(" import ")[-1].split(",") if i] | |
| ovos_imps = [i for i in imported if IMPORTS_MAP and i] | |
| new_imports = [IMPORTS_MAP[i] for i in ovos_imps if IMPORTS_MAP.get(i)] | |
| for imp in [i for i in imported if i not in ovos_imps]: | |
| new_imports.append(f"{prefix} import {imp}") | |
| if new_imports: | |
| self.lines[idx] = "\n".join(new_imports) | |
| self.warnings.append(f"[MODERNIZE] line {idx} - update imports {ovos_imps}") | |
| elif imported: | |
| self.lines[idx] += " # TODO update this import" | |
| self.warnings.append(f"[ERROR] line {idx} - unhandled mycroft imports {imported}") | |
| elif mline and prefix: | |
| if ")" in l: | |
| mline = False | |
| l = l.replace(")", "") | |
| imported = [i.strip() for i in l.split(",") if i] | |
| ovos_imps = [i for i in imported if i in IMPORTS_MAP and i] | |
| new_imports = [IMPORTS_MAP[i] for i in ovos_imps] | |
| for imp in [i for i in imported if i not in ovos_imps]: | |
| new_imports.append(f"{prefix} import {imp}") | |
| self.lines[idx] = "\n".join(new_imports) | |
| self.warnings.append(f"[MODERNIZE] line {idx} - update imports {new_imports}") | |
| def fix_class(self): | |
| for idx, l in enumerate(self.lines): | |
| if l is None or l.strip().startswith("#"): | |
| continue | |
| if "class " in l and ("Skill" in l or "Fallback" in l): | |
| clazz_idx = idx | |
| clazz_ident = l.split("class ")[0] | |
| self.warnings.append(f"[INFO] line {idx} - detected skill class: {l.strip()}") | |
| if "CommonQuerySkill" in l: | |
| self.base_skill = "CommonQuerySkill" | |
| elif "FallbackSkill" in l: | |
| self.base_skill = "FallbackSkill" | |
| elif "MycroftSkill" in l: | |
| self.base_skill = "MycroftSkill" | |
| self.lines[idx] = l.replace("MycroftSkill", "OVOSSkill") | |
| self.warnings.append(f"[MODERNIZE] line {idx} - MycroftSkill -> OVOSSkill") | |
| elif "AutotranslatableSkill" in l: | |
| self.base_skill = "AutotranslatableSkill" | |
| self.lines[idx] = l.replace("AutotranslatableSkill", "UniversalSkill") | |
| self.warnings.append(f"[MODERNIZE] line {idx} - AutotranslatableSkill -> UniversalSkill") | |
| elif "AutotranslatableFallback" in l: | |
| self.base_skill = "AutotranslatableFallback" | |
| self.lines[idx] = l.replace("AutotranslatableFallback", "UniversalFallbackSkill") | |
| self.warnings.append(f"[MODERNIZE] line {idx} - AutotranslatableFallback -> UniversalFallbackSkill") | |
| elif "OVOSSkill" in l: | |
| self.base_skill = "OVOSSkill" | |
| def fix_init(self): | |
| try: | |
| start, end = self.find_skill_method("__init__") | |
| self.warnings.append(f"[INFO] line {start} / {end} - detected __init__ method") | |
| except: | |
| return # init not defined | |
| # index of init def | |
| ins = -1 | |
| ine = -1 | |
| # index of super call | |
| ss = -1 | |
| se = -1 | |
| for idx, l in enumerate(self.lines): | |
| if idx < start or idx > end: | |
| continue | |
| if l is None: | |
| continue | |
| if l.strip().startswith("#"): | |
| continue | |
| if "def __init__(" in l and l.strip() != "def __init__(self, *args, **kwargs):": | |
| ins = idx | |
| ident = l.split("def ")[0] | |
| self.lines[idx] = f"{ident}def __init__(self, *args, **kwargs):" | |
| self.warnings.append(f"[MODERNIZE] line {idx} - __init__(self, *args, **kwargs)") | |
| if l.strip().endswith("):"): | |
| ine = idx | |
| continue | |
| # remove multi-line __init__ definition leftovers | |
| elif ine < ins: | |
| if l.strip().endswith("):"): | |
| ine = idx | |
| self.lines[idx] = None | |
| if "super(" in l: | |
| ss = idx | |
| ident = l.split("super(")[0] | |
| # handle repeated super calls, some skills do this... | |
| if f"{ident}super().__init__(*args, **kwargs)" in self.lines: | |
| self.lines[idx] = None | |
| else: | |
| self.lines[idx] = f"{ident}super().__init__(*args, **kwargs)" | |
| self.warnings.append(f"[MODERNIZE] line {idx} - super().__init__(*args, **kwargs)") | |
| if l.strip().endswith(")"): | |
| se = idx | |
| # common in OLD mycroft skills | |
| elif f"{self.base_skill}.__init__(self" in l: | |
| ss = se = idx | |
| ident = l.split(f"{self.base_skill}.__init__(self")[0] | |
| # handle repeated super calls, some skills do this... | |
| if f"{ident}super().__init__(*args, **kwargs)" in self.lines: | |
| self.lines[idx] = None | |
| else: | |
| self.lines[idx] = f"{ident}super().__init__(*args, **kwargs)" | |
| self.warnings.append(f"[MODERNIZE] line {idx} - super().__init__(*args, **kwargs)") | |
| # remove multi-line super call leftovers | |
| elif se < ss: | |
| if l.strip().endswith(")"): | |
| se = idx | |
| self.lines[idx] = None | |
| start, end = self.find_skill_method("__init__") | |
| if end - start == 1: # single line boilerplate definition | |
| if self.lines[end].strip() == "super().__init__(*args, **kwargs)": | |
| self.warnings.append(f"[REMOVED] line {idx} - __init__ boilerplate") | |
| self.lines[start] = None | |
| self.lines[end] = None | |
| def fix_settings(self): | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| if l.strip() == "save_settings(self.root_dir, self.settings)": | |
| ident = l.split("save_settings(")[0] | |
| self.lines[idx] = f"{ident}self.settings.store()" | |
| self.warnings.append( | |
| f"[MODERNIZE] line {idx} - save_settings(self.root_dir, self.settings) -> self.settings.store()") | |
| def fix_wait_while_waiting(self): | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| prev = self.lines[idx - 1] if idx > 0 else "" | |
| if l.strip() in ["wait_while_speaking()", | |
| "mycroft.audio.wait_while_speaking()"]: | |
| if ((prev.strip().startswith("self.speak(") or | |
| prev.strip().startswith("self.speak_dialog(")) | |
| and "wait=" not in prev): | |
| prev = prev[:-1] + ", wait=True)" | |
| self.lines[idx - 1] = prev | |
| self.lines[idx] = None | |
| self.warnings.append( | |
| f"[REMOVED] line {idx} - deprecated wait_while_speaking() -> set wait=True speak flag") | |
| self.warnings.append(f"[MODERNIZE] line {idx - 1} - added wait=True speak flag") | |
| else: | |
| self.lines[idx] = f"#{l} # TODO - DEPRECATED" | |
| self.warnings.append(f"[REMOVED] line {idx} - deprecated wait_while_speaking()") | |
| def fix_decorators(self): | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| if l.strip().startswith("@intent_file_handler("): | |
| self.lines[idx] = l.replace("@intent_file_handler(", | |
| "@intent_handler(") | |
| self.warnings.append(f"[MODERNIZE] line {idx - 1} - @intent_file_handler -> @intent_handler") | |
| def find_skill_method(self, name): | |
| clazz_idx = -1 | |
| clazz_ident = "" | |
| method_idx = -1 | |
| method_ident = "" | |
| end_idx = -1 | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| if l.strip().startswith("#"): | |
| continue | |
| # skill starts here | |
| if "class " in l and "Skill" in l: | |
| clazz_idx = idx | |
| clazz_ident = l.split("class ")[0] | |
| if clazz_idx < 0: # not inside skill yet | |
| continue | |
| # outside of skill! | |
| if clazz_idx and "class " in l and "Skill" not in l: | |
| if method_idx >= 0: | |
| end_idx = idx - 1 | |
| return method_idx, end_idx | |
| break | |
| if f"def {name}(self" in l: # skill method | |
| method_ident = l.split("def ")[0] | |
| method_idx = idx | |
| elif method_idx >= 0: | |
| if any(s in l for s in ["def ", "@"]) and l.startswith(method_ident): | |
| # this is the next skill method | |
| end_idx = idx - 1 | |
| return method_idx, end_idx | |
| if method_idx >= 0: | |
| return method_idx, len(self.lines) - 1 | |
| def fix_create_func(self): | |
| start_idx = -1 | |
| end_idx = -1 | |
| for idx, l in enumerate(self.lines): | |
| if l is None: | |
| continue | |
| if l.strip() == "def create_skill():": | |
| start_idx = idx | |
| self.warnings.append(f"[REMOVED] line {idx} - deprecated create_skill() boilerplate") | |
| if start_idx > 0: | |
| self.lines[idx] = None | |
| if l.strip().startswith("return "): | |
| end_idx = idx | |
| break | |
| def inspect(self): | |
| init_file = self.path + "/__init__.py" | |
| fixed_code, warnings = self.fix_skill(init_file) | |
| return warnings, fixed_code | |
| def export(self, folder): | |
| init_file = self.path + "/__init__.py" | |
| fixed_code, warnings = self.fix_skill(init_file) | |
| folder += f"/{self.skill_id}" | |
| shutil.copytree(self.path, folder, dirs_exist_ok=True, ignore_dangling_symlinks=True) | |
| with open(folder + f"/__init__.py", "w") as f: | |
| f.write(fixed_code) | |
| with open(folder + f"/conversion.log", "w") as f: | |
| f.write("\n".join(warnings)) | |
| from pprint import pprint | |
| verified = os.listdir("./ocp") | |
| for name in os.listdir("./marketplace"): | |
| if name in verified or name.endswith(".py") or name.endswith(".sh"): | |
| continue | |
| p = f"./marketplace/{name}" | |
| s = SkillConverter(p) | |
| print("\n### CONVERTED CODE:\n") | |
| s.export(f"./converted") | |
| pprint(s.warnings) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment