diff --git a/build_natlinkcore.ps1 b/build_natlinkcore.ps1 new file mode 100644 index 0000000..80e5d33 --- /dev/null +++ b/build_natlinkcore.ps1 @@ -0,0 +1,4 @@ +#powershell to run the tests, then build the python package. +$ErrorActionPreference = "Stop" +pytest --capture=tee-sys +flit build --format sdist \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 2cb826c..37228ac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,8 +50,12 @@ testpaths= [ [project.scripts] natlinkconfig_cli = "natlinkcore.configure.natlinkconfig_cli:main_cli" +natlink_extensions = "natlinkcore.configure.natlink_extensions:main" [project.gui-scripts] natlinkconfig_gui = "natlinkcore.configure.natlinkconfig_gui:main_gui" +[project.entry-points.natlink_extensions] +natlink_sample_macros = "natlinkcore.SampleMacros:locateme" + [project.urls] Home = "https://github.com/dictation-toolbox/natlinkcore/" diff --git a/src/natlinkcore/SampleMacros/__init__.py b/src/natlinkcore/SampleMacros/__init__.py new file mode 100644 index 0000000..c369f13 --- /dev/null +++ b/src/natlinkcore/SampleMacros/__init__.py @@ -0,0 +1,2 @@ +def locateme(): + return __path__[0] \ No newline at end of file diff --git a/src/natlinkcore/SampleMacros/_debug_natlink.py b/src/natlinkcore/SampleMacros/_debug_natlink.py index e64e224..a8267c8 100644 --- a/src/natlinkcore/SampleMacros/_debug_natlink.py +++ b/src/natlinkcore/SampleMacros/_debug_natlink.py @@ -3,7 +3,7 @@ """ #pylint:disable=W0611, W0613, C0115, C0116 -from natlink import _natlink_core as natlink +import natlink from natlinkcore import natlinkpydebug as pd from natlinkcore import natlinkutils from natlinkcore import gramparser as gp diff --git a/src/natlinkcore/SampleMacros/_sample_callback.py b/src/natlinkcore/SampleMacros/_sample_callback.py index 19920fc..1b10e19 100644 --- a/src/natlinkcore/SampleMacros/_sample_callback.py +++ b/src/natlinkcore/SampleMacros/_sample_callback.py @@ -21,6 +21,8 @@ def post_load_callback(self): print('------- post_load_callback') def on_mic_on_callback(self): print('------- on_mic_on_callback') + def on_mic_off_callback(self): + print('------- on_mic_off_callback') def on_begin_utterance_callback(self): print('------- on_begin_utterance_callback') @@ -28,6 +30,7 @@ def on_begin_utterance_callback(self): thisGrammar.initialize() natlinkmain.set_on_begin_utterance_callback(thisGrammar.on_begin_utterance_callback) natlinkmain.set_on_mic_on_callback(thisGrammar.on_mic_on_callback) +natlinkmain.set_on_mic_off_callback(thisGrammar.on_mic_off_callback) natlinkmain.set_pre_load_callback(thisGrammar.pre_load_callback) natlinkmain.set_post_load_callback(thisGrammar.post_load_callback) @@ -36,6 +39,7 @@ def unload(): if thisGrammar: natlinkmain.delete_on_begin_utterance_callback(thisGrammar.on_begin_utterance_callback) natlinkmain.delete_on_mic_on_callback(thisGrammar.on_mic_on_callback) + natlinkmain.delete_on_mic_off_callback(thisGrammar.on_mic_off_callback) natlinkmain.delete_pre_load_callback(thisGrammar.pre_load_callback) natlinkmain.delete_post_load_callback(thisGrammar.post_load_callback) # extraneous deletes do not harm: diff --git a/src/natlinkcore/config.py b/src/natlinkcore/config.py index b91e73d..e10313b 100644 --- a/src/natlinkcore/config.py +++ b/src/natlinkcore/config.py @@ -6,7 +6,7 @@ from enum import IntEnum from typing import List, Iterable, Dict from pathlib import Path -from natlink import _natlink_core as natlink +import natlink class NoGoodConfigFoundException(natlink.NatError): pass @@ -143,7 +143,7 @@ def expand_path(input_path: str) -> str: Paths can be: - - the name of a python package, to be found along sys.path (typically in site-packages) + - the name of a python package, or a sub directory of a python package - natlink_userdir/...: the directory where natlink.ini is is searched for, either %(NATLINK_USERDIR) or ~/.natlink - ~/...: the home directory - some environment variable: this environment variable is expanded. @@ -180,15 +180,26 @@ def expand_path(input_path: str) -> str: print(f'natlink_userdir does not expand to a valid directory: "{nud}"') return normpath(nud) - if not (input_path.find('/') >= 0 or input_path.find('\\') >= 0): + + # try if package: + if input_path.find('/') > 0: + package_trunk, rest = input_path.split('/', 1) + elif input_path.find('\\') > 0: + package_trunk, rest = input_path.split('\\', 1) + else: + package_trunk, rest = input_path, '' # find path for package. not an alternative way without loading the package is to use importlib.util.findspec. - try: - pack = __import__(input_path) - except ModuleNotFoundError: - print(f'expand_path, package name "{input_path}" is not found') - return input_path - return pack.__path__[0] - + try: + pack = __import__(package_trunk) + package_path = pack.__path__[-1] + if rest: + dir_expanded = str(Path(package_path)/rest) + return dir_expanded + return package_path + + except ModuleNotFoundError: + pass + env_expanded = expandvars(input_path) # print(f'env_expanded: "{env_expanded}", from envvar: "{input_path}"') return normpath(env_expanded) diff --git a/src/natlinkcore/configure/__init__.py b/src/natlinkcore/configure/__init__.py new file mode 100644 index 0000000..09c00fa --- /dev/null +++ b/src/natlinkcore/configure/__init__.py @@ -0,0 +1 @@ +from natlinkcore.configure.natlink_extensions import extensions_and_folders \ No newline at end of file diff --git a/src/natlinkcore/configure/natlink_extensions.py b/src/natlinkcore/configure/natlink_extensions.py new file mode 100644 index 0000000..2c44925 --- /dev/null +++ b/src/natlinkcore/configure/natlink_extensions.py @@ -0,0 +1,29 @@ +"""Command Line Program to List Advertised Natlink Exensions.""" +import sys +from importlib.metadata import entry_points +import argparse + +def extensions_and_folders(): + discovered_extensions=entry_points(group='natlink_extensions') + + for extension in discovered_extensions: + n=extension.name + ep=extension.value + try: + pathfn=extension.load() + path=pathfn() + + except Exception as e: + path = e + yield n,path + + +def main(): + parser=argparse.ArgumentParser(description="Enumerate natlink extension momdules.") + args=parser.parse_args() + for n,path in extensions_and_folders(): + print(f"{n} {path}") + return 0 + +if '__main__' == __name__: + main() \ No newline at end of file diff --git a/src/natlinkcore/configure/natlinkconfig_cli.py b/src/natlinkcore/configure/natlinkconfig_cli.py index ff63282..7bb074e 100644 --- a/src/natlinkcore/configure/natlinkconfig_cli.py +++ b/src/natlinkcore/configure/natlinkconfig_cli.py @@ -4,6 +4,7 @@ import cmd import os import os.path +from natlinkcore.configure import extensions_and_folders from natlinkcore.configure import natlinkconfigfunctions @@ -18,7 +19,7 @@ def _main(Options=None): """ cli = CLI() cli.Config = natlinkconfigfunctions.NatlinkConfig() - shortOptions = "DVNOHKaAiIxXbBuq" + shortOptions = "DVNOHKaAiIxXbBuqe" shortArgOptions = "d:v:n:o:h:k:" if Options: if isinstance(Options, str): @@ -129,6 +130,8 @@ def usage(self): [AutoHotkey] h/H - set/clear the AutoHotkey exe directory. k/K - set/clear the User Directory for AutoHotkey scripts. +[Extensions] +e - give a list of python modules registered as extensions. [Other] u/usage - give this list @@ -153,6 +156,13 @@ def do_j(self, arg): self.Config.printPythonPath() + def do_e(self,arg): + print("extensions and folders for registered natlink extensions:") + ef="" + for n,f in extensions_and_folders(): + ef+= f"\n{n} {f}" + print(ef) + def help_i(self): print('-'*60) print("""The command info (i) gives an overview of the settings that are diff --git a/src/natlinkcore/loader.py b/src/natlinkcore/loader.py index 56e7d96..2a473cf 100644 --- a/src/natlinkcore/loader.py +++ b/src/natlinkcore/loader.py @@ -16,7 +16,7 @@ from types import ModuleType from typing import List, Dict, Set, Iterable, Any, Tuple, Callable -from natlink import _natlink_core as natlink +import natlink from natlinkcore.config import LogLevel, NatlinkConfig, expand_path from natlinkcore.readwritefile import ReadWriteFile from natlinkcore.callbackhandler import CallbackHandler @@ -65,6 +65,7 @@ def __init__(self, logger: Any=None, config: Any = None): self._pre_load_callback = CallbackHandler('pre_load') self._post_load_callback = CallbackHandler('post_load') self._on_mic_on_callback = CallbackHandler('on_mic_on') + self._on_mic_off_callback = CallbackHandler('on_mic_off') self._on_begin_utterance_callback = CallbackHandler('on_begin_utterance') self.seen: Set[Path] = set() # start empty in trigger_load self.bom = self.encoding = self.config_text = '' # getconfigsetting and writeconfigsetting @@ -76,6 +77,9 @@ def set_on_begin_utterance_callback(self, func: Callable[[], None]) -> None: def set_on_mic_on_callback(self, func: Callable[[], None]) -> None: self._on_mic_on_callback.set(func) + def set_on_mic_off_callback(self, func: Callable[[], None]) -> None: + self._on_mic_off_callback.set(func) + def set_pre_load_callback(self, func: Callable[[], None]) -> None: self._pre_load_callback.set(func) @@ -88,6 +92,9 @@ def delete_on_begin_utterance_callback(self, func: Callable[[], None]) -> None: def delete_on_mic_on_callback(self, func: Callable[[], None]) -> None: self._on_mic_on_callback.delete(func) + def delete_on_mic_off_callback(self, func: Callable[[], None]) -> None: + self._on_mic_off_callback.delete(func) + def delete_pre_load_callback(self, func: Callable[[], None]) -> None: self._pre_load_callback.delete(func) @@ -401,6 +408,9 @@ def on_change_callback(self, change_type: str, args: Any) -> None: if self.config.load_on_mic_on: self.trigger_load() + elif change_type == 'mic' and args == 'off': + self.logger.debug('on_change_callback called with: "mic", "off"') + self._on_mic_off_callback.run() else: self.logger.debug(f'on_change_callback unhandled: change_type: "{change_type}", args: "{args}"') diff --git a/src/natlinkcore/natlinkstatus.py b/src/natlinkcore/natlinkstatus.py index 4de8004..68a3691 100644 --- a/src/natlinkcore/natlinkstatus.py +++ b/src/natlinkcore/natlinkstatus.py @@ -57,8 +57,8 @@ Also users that have their own custom grammar files can use this user directory getUnimacroDirectory: get the directory where the Unimacro system is. - When git cloned, relative to the Core directory, otherwise somewhere or in the site-packages (if pipped). This grammar will (and should) hold the _control.py grammar - and needs to be included in the load directories list of James' natlinkmain + This directory is normally in the site-packages area of Python (name "unimacro"), but can be + "linked" to your cloned source code when you installed the packages with "pip install -e ." getUnimacroGrammarsDirectory: get the directory, where the user can put his Unimacro grammars. This directory will be located in the `ActiveGrammars` subdirectory of the `~/.unimacro' or `%NATLINK_USERDIR%/.unimacro`). @@ -66,6 +66,9 @@ getUnimacroUserDirectory: get the directory of Unimacro INI files, if not return '' or the Unimacro user directory +getUnimacroDataDirectory: get the directory where Unimacro grammars can store data, this should be per computer, and is set + into the natlink_user area + getVocolaDirectory: get the directory where the Vocola system is. When cloned from git, in Vocola, relative to the Core directory. Otherwise (when pipped) in some site-packages directory. It holds (and should hold) the grammar _vocola_main.py. @@ -120,7 +123,6 @@ "esp": "may\xfas"} thisDir, thisFile = os.path.split(__file__) -thisDirSymlink = natlinkcore.getThisDir(__file__) class NatlinkStatus(metaclass=singleton.Singleton): """this class holds the Natlink status functions. @@ -138,7 +140,7 @@ class NatlinkStatus(metaclass=singleton.Singleton): 'vocoladirectory', 'vocolagrammarsdirectory'] def __init__(self): - """initialise all instance variables, in this singleton class, hoeinstance + """initialise all instance variables, in this singleton class, (only one instance) """ self.natlinkmain = natlinkmain # global self.DNSVersion = None @@ -152,6 +154,7 @@ def __init__(self): self.UnimacroDirectory = None self.UnimacroUserDirectory = None self.UnimacroGrammarsDirectory = None + self.UnimacroDataDirectory = None ## Vocola: self.VocolaUserDirectory = None self.VocolaDirectory = None @@ -165,10 +168,9 @@ def __init__(self): self.symlink_line = '' if self.NatlinkDirectory is None: - self.NatlinkDirectory = natlink.__path__[0] - self.NatlinkcoreDirectory = thisDirSymlink # equal to thisDir if no symlinking is there. - if thisDirSymlink != thisDir: - self.symlink_line = f'NatlinkcoreDirectory is symlinked, for developing purposes.\n\tFiles seem to be in "{thisDirSymlink}",\n\tbut they can be edited in "{thisDir}".\n\tWhen debugging files from this directory, open and set breakpoints in files in the first (site-packages) directory!' + self.NatlinkDirectory = natlink.__path__[-1] + if len(natlinkcore.__path__) > 0: + self.symlink_line = 'NatlinkcoreDirectory is editable' def refresh(self): """rerun the __init__, refreshing all variables @@ -399,10 +401,30 @@ def getUnimacroDirectory(self): except ImportError: self.UnimacroDirectory = "" return "" - self.UnimacroDirectory = str(Path(unimacro.__file__).parent) + self.UnimacroDirectory = unimacro.__path__[-1] return self.UnimacroDirectory + def getUnimacroDataDirectory(self): + """return the path to the directory where grammars can store data. + + Expected in "UnimacroData" of the natlink user directory + (November 2022) + + """ + if self.UnimacroDataDirectory is not None: + return self.UnimacroDataDirectory + + natlink_user_dir = self.getNatlink_Userdir() + + um_data_dir = Path(natlink_user_dir)/'UnimacroData' + if not um_data_dir.is_dir(): + um_data_dir.mkdir() + um_data_dir = str(um_data_dir) + self.UnimacroDataDirectory = um_data_dir + + return um_data_dir + def getUnimacroGrammarsDirectory(self): """return the path to the directory where the ActiveGrammars of Unimacro are located. @@ -539,7 +561,6 @@ def getVocolaUserDirectory(self): return '' def getVocolaDirectory(self): - isdir, isfile, join, abspath = os.path.isdir, os.path.isfile, os.path.join, os.path.abspath if self.VocolaDirectory is not None: return self.VocolaDirectory @@ -548,7 +569,7 @@ def getVocolaDirectory(self): except ImportError: self.VocolaDirectory = '' return '' - self.VocolaDirectory = str(Path(vocola2.__file__).parent) + self.VocolaDirectory = vocola2.__path__[-1] return self.VocolaDirectory @@ -693,7 +714,7 @@ def getNatlinkStatusDict(self): for key in ['DNSIniDir', 'WindowsVersion', 'DNSVersion', 'PythonVersion', 'DNSName', 'NatlinkIni', 'Natlink_Userdir', - 'UnimacroDirectory', 'UnimacroUserDirectory', 'UnimacroGrammarsDirectory', + 'UnimacroDirectory', 'UnimacroUserDirectory', 'UnimacroGrammarsDirectory', 'UnimacroDataDirectory', 'VocolaDirectory', 'VocolaUserDirectory', 'VocolaGrammarsDirectory', 'VocolaTakesLanguages', 'VocolaTakesUnimacroActions', 'UserDirectory', @@ -726,7 +747,6 @@ def getNatlinkStatusString(self): L = [] D = self.getNatlinkStatusDict() if self.symlink_line: - L.append('--- warning:') L.append(self.symlink_line) L.append('--- properties:') self.appendAndRemove(L, D, 'user') @@ -766,7 +786,7 @@ def getNatlinkStatusString(self): ## Unimacro: if D['unimacroIsEnabled']: self.appendAndRemove(L, D, 'unimacroIsEnabled', "---Unimacro is enabled") - for key in ('UnimacroUserDirectory', 'UnimacroDirectory', 'UnimacroGrammarsDirectory'): + for key in ('UnimacroUserDirectory', 'UnimacroDirectory', 'UnimacroGrammarsDirectory', 'UnimacroDataDirectory'): self.appendAndRemove(L, D, key) else: self.appendAndRemove(L, D, 'unimacroIsEnabled', "---Unimacro is disabled") diff --git a/src/natlinkcore/natlinktimer.py b/src/natlinkcore/natlinktimer.py new file mode 100644 index 0000000..1140753 --- /dev/null +++ b/src/natlinkcore/natlinktimer.py @@ -0,0 +1,393 @@ +""" +Handling of more calls to the Natlink timer + +make it a Singleton class (December 2022) +Quintijn Hoogenboom + +""" +#pylint:disable=R0913 + +#--------------------------------------------------------------------------- +import time +import traceback +import operator +import logging + +import natlink +from natlinkcore import singleton, loader, config +Logger = logging.getLogger('natlink') +Config = config.NatlinkConfig.from_first_found_file(loader.config_locations()) +natlinkmain = loader.NatlinkMain(Logger, Config) + +## this variable will hold the (only) NatlinkTimer instance +natlinktimer = None + +class GrammarTimer: + """object which specifies how to call the natlinkTimer + + The function to be called when the mic switches off needs to be set only once, and is then preserved + """ + #pylint:disable=R0913 + def __init__(self, callback, interval, callAtMicOff=False, maxIterations=None): + curTime = self.starttime = round(time.time()*1000) + self.callback = callback + self.interval = interval + + self.nextTime = curTime + interval + + self.callAtMicOff = callAtMicOff + self.maxIterations = maxIterations + + def __str__(self): + """make string with nextTime value relative to the starttime of the grammarTimer instance + """ + result = f'grammartimer, interval: {self.interval}, nextTime (relative): {self.nextTime - self.starttime}' + return result + + def __repr__(self): + L = ['GrammarTimer instance:'] + for varname in 'interval', 'nextTime', 'callAtMicOff', 'maxIterations': + value = self.__dict__.get(varname, None) + if not value is None: + L.append(f' {varname.ljust(13)}: {value}') + return "\n".join(L) + + def start(self, newInterval=0): + """start (or continue), optionally with new interval + """ + if newInterval: + oldInterval, self.interval = self.interval, newInterval + self.nextTime = self.nextTime - oldInterval + newInterval + if not natlinktimer.in_timer: + natlinktimer.hittimer() + + +class NatlinkTimer(metaclass=singleton.Singleton): + """ + This class utilises :meth:`natlink.setTimerCallback`, but multiplexes + + In this way, more grammars can use the single Natlink timer together. + + First written by Christo Butcher for Dragonfly, now enhanced by Quintijn Hoogenboom, May 2020/December 2022 + + """ + def __init__(self, minInterval=None): + """initialize the natlink timer instance + + This is singleton class, with only one instance, so more "calls" automatically connect + to the same instance. + + The grammar callback functions are the keys of the self.callbacks dict, + The corresponding values are GrammarTimer instances, which specify interval and possibly other parameters + + The minimum interval for the timer can be specified, is 50 by default. + """ + self.callbacks = {} + self.debug = False + self.timerStartTime = self.getnow() + self.minInterval = minInterval or 50 + self.tolerance = min(10, int(self.minInterval/4)) + self.in_timer = False + self.timers_to_stop = set() # will be used when a timer wants to be removed when in_timer is True + natlinkmain.set_on_mic_off_callback(self.on_mic_off_callback) + + def __del__(self): + """stop the timer, when destroyed + """ + self.stopTimer() + + def setDebug(self, debug): + """set debug option + """ + if debug: + self.debug = True + def clearDebug(self): + """clear debug option + """ + self.debug = False + + def getnow(self): + """get time in milliseconds + """ + return round(time.time()*1000) + + def on_mic_off_callback(self): + """all callbacks that have callAtMicOff set, will be stopped (and deleted) + """ + # print('on_mic_off_callback') + to_stop = [(cb, gt) for (cb, gt) in self.callbacks.items() if gt.callAtMicOff] + if not to_stop: + if self.debug: + print('natlinktimer: no timers to stop') + return + for cb, gt in to_stop: + if self.debug: + print(f'natlinktimer: stopping {cb}, {gt}') + gt.callAtMicOff() + + def addCallback(self, callback, interval, callAtMicOff=False, maxIterations=None, debug=None): + """add an interval + """ + self.debug = debug + now = self.getnow() + + if interval <= 0: + self.removeCallback(callback) + return None + interval = max(round(interval), self.minInterval) + gt = GrammarTimer(callback, interval, callAtMicOff=callAtMicOff, maxIterations=maxIterations) + self.callbacks[callback] = gt + if self.debug: + print(f'set new timer {callback.__name__}, {interval} ({now})') + + return gt + + def removeCallback(self, callback, debug=None): + """remove a callback function + """ + self.debug = self.debug or debug + if self.debug: + print(f'remove timer for {callback.__name__}') + + if self.in_timer: + # print(f'removeCallback, in_timer: {self.in_timer}, add {callback} to timers_to_stop') + self.timers_to_stop.add(callback) + return + + # outside in_timer: + try: + print('remove 1 timer') + del self.callbacks[callback] + except KeyError: + pass + if not self.callbacks: + if self.debug: + print("last timer removed, setTimerCallback to 0") + + self.stopTimer() + return + + + def hittimer(self): + """move to a next callback point + """ + #pylint:disable=R0914, R0912, R0915, W0702 + self.in_timer = True + try: + now = self.getnow() + nowRel = now - self.timerStartTime + if self.debug: + print(f'start hittimer at {nowRel}') + + toBeRemoved = [] + # c = callbackFunc, g = grammarTimer + # sort for shortest interval times first, only sort on interval: + decorated = [(g.interval, c, g) for (c, g) in self.callbacks.items()] + sortedList = sorted(decorated, key=operator.itemgetter(0)) + + + for interval, callbackFunc, grammarTimer in sortedList: + now = self.getnow() + # for printing: + if self.debug: + nowRel, nextTimeRel = now - self.timerStartTime, grammarTimer.nextTime - self.timerStartTime + if grammarTimer.nextTime > (now + self.tolerance): + if self.debug: + print(f'no need for {callbackFunc.__name__}, now: {nowRel}, nextTime: {nextTimeRel}') + continue + + # now treat the callback, grammarTimer.nextTime > now - tolerance: + hitTooLate = now - grammarTimer.nextTime + + if self.debug: + print(f"do callback {callbackFunc.__name__} at {nowRel}, was expected at: {nextTimeRel}, interval: {interval}") + + ## now do the callback function: + newInterval = None + startCallback = now + try: + newIntervalOrNone = callbackFunc() + except: + print(f"exception in callbackFunc ({callbackFunc}), remove from list") + traceback.print_exc() + toBeRemoved.append(callbackFunc) + endCallback = None + newIntervalOrNone = None + else: + endCallback = self.getnow() + + if newIntervalOrNone is None: + pass + elif newIntervalOrNone <= 0: + print(f"newInterval <= 0, as result of {callbackFunc.__name__}: {newInterval}, remove the callback function") + toBeRemoved.append(callbackFunc) + continue + + # if cbFunc ended correct, but took too much time, its interval should be doubled: + if endCallback is None: + pass + else: + spentInCallback = endCallback - startCallback + if spentInCallback > interval: + if self.debug: + print(f"spent too much time in {callbackFunc.__name__}, increase interval from {interval} to: {spentInCallback*2}") + grammarTimer.interval = interval = spentInCallback*2 + grammarTimer.nextTime += interval + if self.debug: + nextTimeRelative = grammarTimer.nextTime - endCallback + print(f"new nextTime: {nextTimeRelative}, interval: {interval}, from gt instance: {grammarTimer.interval}") + + for gt in toBeRemoved: + del self.callbacks[gt] + + if not self.callbacks: + if self.debug: + print("no callbackFunction any more, switch off the natlink timerCallback") + self.stopTimer() + return + + nownow = self.getnow() + timeincallbacks = nownow - now + if self.debug: + print(f'time in callbacks: {timeincallbacks}') + nextTime = min(gt.nextTime-nownow for gt in self.callbacks.values()) + if nextTime < self.minInterval: + if self.debug: + print(f"warning, nextTime too small: {nextTime}, set at minimum {self.minInterval}") + nextTime = self.minInterval + if self.debug: + print(f'set nextTime to: {nextTime}') + natlink.setTimerCallback(self.hittimer, nextTime) + if self.debug: + nownownow = self.getnow() + timeinclosingphase = nownownow - nownow + totaltime = nownownow - now + print(f"time taken in closingphase: {timeinclosingphase}") + print(f"total time spent hittimer: {totaltime}") + finally: + self.in_timer = False + if self.timers_to_stop: + n_timers = len(self.timers_to_stop) + if n_timers == 1: + print('stop 1 timer (at end of hittimer)') + else: + print(f'stop {n_timers} timers (at end of hittimer)') + for callback in self.timers_to_stop: + self.removeCallback(callback) + self.timers_to_stop = set() + + + + def stopTimer(self): + """stop the natlink timer, by passing in None, 0 + """ + natlink.setTimerCallback(None, 0) + +def createGrammarTimer(callback, interval=0, callAtMicOff=False, maxIterations=None, debug=None): + """return a grammarTimer instance + + parameters: + callback: a function into which natlinktime will callback + interval: a starting interval (or 0), with which the timer shall run (initially) + optional: + callAtMicOff: default False. When True, the timer stops when the mic is toggled to off + maxIterations: default None. When a positive int, stop when this count is exceeded + debug: sets the debug status for the whole natlinktimer instance, including the grammarTimer instance + + Note: the grammarTimer is NOT started, but when other timers are running it is taken in the flow. + Call grammarTimer.start(newInterval=None) to start the timer (starting with the waiting interval) or + grammarTimer.startNow(newInterval=None) to start immediate and then hits at each interval. + + Note: all intervals are in milliseconds. + """ + #pylint:disable=W0603 + global natlinktimer + if not natlinktimer: + natlinktimer = NatlinkTimer() + if debug: + natlinktimer.setDebug(debug) + if not natlinktimer: + raise Exception("NatlinkTimer cannot be started") + + if callback is None: + raise Exception("stop the timer callback with natlinktimer.removeCallback(callback)") + + if interval > 0: + gt = natlinktimer.addCallback(callback, interval, callAtMicOff=callAtMicOff, maxIterations=maxIterations, debug=debug) + return gt + raise ValueError(f'Did not start grammarTimer instance {callback}, because the interval is not a positive value') + + +def setTimerCallback(callback, interval, callAtMicOff=None, maxIterations=None, debug=None): + """This function sets a timercallback, nearly the same as natlink.setTimerCallback + + Interval in milliseconds, unless smaller than 25 (default) + + When 0 or negative: it functions as removeTimerCallback!! + + callAtMicOff: the function that will be called when the mic switches off + + But there are extra parameters possible, which are passed on to createGrammarTimer, see there + + """ + #pylint:disable=W0603 + try: + natlinktimer + except NameError: + print('natlinktimer is gone') + return + + if interval > 0: + if natlinktimer and callback in natlinktimer.callbacks: + gt = natlinktimer.callbacks[callback] + rel_cur_time = round(time.time()*1000) - gt.starttime + if callAtMicOff: + gt.callAtMicOff = callAtMicOff + if natlinktimer.debug: + print(f'{gt}\n\ttime: {rel_cur_time}, new interval: {interval}, nextTime: {gt.nextTime-gt.starttime}') + gt.start(newInterval=interval) + else: + createGrammarTimer(callback, interval, callAtMicOff=callAtMicOff, maxIterations=maxIterations, debug=debug) + natlinktimer.hittimer() + return + # interval is 0 (or negative), remove the callback + removeTimerCallback(callback) + return + + +def removeTimerCallback(callback, debug=None): + """This function removes a callback from the callbacks dict + + callback: the function to be called + """ + if not natlinktimer: + print(f'no timers active, cannot remove {callback} from natlinktimer') + return + + if callback is None: + raise Exception("please stop the timer callback with removeTimerCallback(callback)\n or with setTimerCallback(callback, 0)") + + natlinktimer.removeCallback(callback, debug=debug) + +def stopTimerCallback(): + """should be called at destroy of Natlink + """ + #pylint:disable=W0603 + global natlinktimer + natlink.setTimerCallback(None, 0) + try: + del natlinktimer + except NameError: + pass + +def getNatlinktimerStatus(): + """report how many callbacks are active, None if natlinktimer is gone + """ + try: + natlinktimer + except NameError: + return None + if natlinktimer is None: + return None + return len(natlinktimer.callbacks) + diff --git a/src/natlinkcore/nsformat.py b/src/natlinkcore/nsformat.py index d439b31..5b4da1b 100644 --- a/src/natlinkcore/nsformat.py +++ b/src/natlinkcore/nsformat.py @@ -11,7 +11,7 @@ """ #pylint:disable=C0116, C0123, R0911, R0912, R0915, R0916 import copy -from natlink import _natlink_core as natlink +import natlink flag_useradded = 0 flag_varadded = 1 diff --git a/src/natlinkcore/redirect_output.py b/src/natlinkcore/redirect_output.py index aca68eb..2aef0af 100644 --- a/src/natlinkcore/redirect_output.py +++ b/src/natlinkcore/redirect_output.py @@ -12,7 +12,7 @@ outputDebugString(f"{__file__} importing _natlink_core" ) -from natlink import _natlink_core as natlink +import natlink outputDebugString(f"{__file__} imported _natlink_core" ) diff --git a/tests/test_config.py b/tests/test_config.py index 4602560..098bacd 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -155,6 +155,33 @@ def test_expand_path(mock_syspath,mock_userdir): result = expand_path('natlink_userdir/invalid_dir') assert not os.path.isdir(result) + # try package + result = expand_path('natlinkcore') + assert os.path.isdir(result) + + result = expand_path('natlinkcore/DefaultConfig') + assert os.path.isdir(result) + + result = expand_path('natlinkcore\\DefaultConfig') + assert os.path.isdir(result) + + result = expand_path('natlinkcore/NonExisting') + assert not os.path.isdir(result) + + result = expand_path('natlinkcore\\NonExisting') + assert not os.path.isdir(result) + + result = expand_path('/natlinkcore') + assert not os.path.isdir(result) + + result = expand_path('unimacro') + assert os.path.isdir(result) + + result = expand_path('unimacro/unimacrogrammars') + assert os.path.isdir(result) + + + def test_config_locations(): """tests the lists of possible config_locations and of valid_locations """ diff --git a/tests/test_natlinktimer.py b/tests/test_natlinktimer.py new file mode 100644 index 0000000..85cb9ec --- /dev/null +++ b/tests/test_natlinktimer.py @@ -0,0 +1,330 @@ +"""unittestNatlinktimer + +Python Macro Language for Dragon NaturallySpeaking + (c) Copyright 1999 by Joel Gould + Portions (c) Copyright 1999 by Dragon Systems, Inc. + +unittestNatlinktimer.py + + This script performs tests of the Natlinktimer module + natlinktimer.py, for multiplexing timer instances acrross different grammars + Quintijn Hoogenboom, summer 2020 +""" +#pylint:disable=C0115, C0116 +#pylint:disable=E1101 + +# import sys +import time +from pathlib import Path +import pytest +import natlink +from natlinkcore.natlinkutils import GrammarBase +from natlinkcore import natlinktimer + +thisDir = Path(__file__).parent + +# define TestError, and mark is to be NOT a part of pytest: +class TestError(Exception): + pass +TestError.__test__ = False + +debug = 1 + +# make a TestGrammar, which can be called for different instances +class TestGrammar(GrammarBase): + def __init__(self, name="testGrammar"): + GrammarBase.__init__(self) + self.name = name + self.resetExperiment() + + def resetExperiment(self): + self.Hit = 0 + self.MaxHit = 5 + self.toggleMicAt = None + self.sleepTime = 0 # to be specified by calling instance, the sleeping time after each hit + self.results = [] + self.starttime = round(time.time()*1000) + self.startCancelMode = False + self.endCancelMode = False + + def cancelMode(self): + """timer should stop, and the 2 instance variables set to True + """ + self.startCancelMode = True + natlinktimer.setTimerCallback(self.doTimerClassic,0) + self.endCancelMode = True + + def doTimerClassic(self): + """have no introspection, but be as close as possible to the old calling method of setTimerCallback + + """ + now = round(time.time()*1000) + relTime = now - self.starttime + self.results.append(f'{self.Hit} {self.name}: {relTime}') + self.Hit +=1 + if self.Hit: + if self.toggleMicAt and self.toggleMicAt == self.Hit: + ## run the on_mic_off_callback function: + natlinktimer.natlinktimer.on_mic_off_callback() + # decrease the interval at each step. There should be + # a bottom, depending on the time the routine is taking (eg minimal 3 times the time the callback takes). + # this is tested by setting the sleeptime + self.interval -= 10 + if self.sleepTime: + time.sleep(self.sleepTime/1000) + natlinktimer.setTimerCallback(self.doTimerClassic, interval=self.interval, callAtMicOff=self.cancelMode) + + # time.sleep(self.sleepTime/1000) # sleep 10 milliseconds + if self.Hit == self.MaxHit: + natlinktimer.setTimerCallback(self.doTimerClassic, 0) + + def doTimer(self): + """the doTimer function can remove itself, return None or another interval + + """ + relTime = round(time.time()*1000) - self.starttime + self.results.append(f'{self.Hit} {self.name} ({self.grammarTimer.interval}): {relTime}') + self.Hit +=1 + time.sleep(self.sleepTime/1000) # sleep 10 milliseconds + if self.Hit == self.MaxHit: + expectElapsed = self.Hit * self.interval + print(f'expect duration of timer {self.name}: {expectElapsed} milliseconds') + natlinktimer.removeTimerCallback(self.doTimer) + ## try to shorten interval: + currentInterval = self.grammarTimer.interval + if currentInterval > 250: + newInterval = currentInterval - 25 + return newInterval + return None + + + + def toggleMicrophone(self): + micstate = natlink.getMicState() + if micstate == 'off': + natlink.setMicState('on') + time.sleep(0.1) + natlink.setMicState('off') + +TestGrammar.__test__ = False + + def testIcons(self): + """in the subdirectory icons of Unimacro there are 4 icons + + These should show up when natlink.setTrayIcon('_repeat.ico', 'tooltip', func) is called, + but this is not working any more. + There are also predefined icons, which are now stored in defaulticons directory of unimacro (fork) + + I don't know how to call these again... + +It was tested in C:\DT\NatlinkDoug\pythonsrc\tests\unittestNatlink.py, this test could be taken out and put into +a pytest module! + +**From natlink.txt**: + +setTrayIcon( iconName, toolTip, callback ) + This function, provided by Jonathan Epstein, will draw an icon in the + tray section of the tackbar. + + Pass in the absolute path to a Windows icon file (.ico) or pass in one + of the following predefined names: + 'right', 'right2', 'down', 'down2', + 'left', 'left2', 'up', 'up2', 'nodir' + You can also pass in an empty string (or nothing) to remove the tray + icon. + + The toolTip parameter is optional. It is the text which is displayed + as a tooltip when the mouse is over the tray icon. If missing, a generic + tooltip is used. + + The callback parameter is optional. When used, it should be a Python + function which will be called when a mouse event occurs for the tray + icon. The function should take one parameters which is the type of + mouse event: + wm_lbuttondown, wm_lbuttonup, wm_lbuttondblclk, wm_rbuttondown, + wm_rbuttonup, wm_rbuttondblclk, wm_mbuttondown, wm_mbuttonup, + or wm_mbuttondblclk (all defined in natlinkutils) + + Raises ValueError if the iconName is invalid. + +The following functions are used in the natlinkmain base module. You +should only used these if you are control NatSpeak using the NatLink module +instead of using Python as a command and control subsystem for NatSpeak. In +the later case, users programs should probably not use either of these two +functions because they replace the callback used by the natlinkmain module +which could prevent proper module (re)loading and user changes. + + """ + natlink.setTrayIcon('_repeat.ico') + natlink.setTrayIcon('_down') # ???? + + + +# def testSingleTimerClassic(): +# try: +# natlink.natConnect() +# testGram = TestGrammar(name="single") +# testGram.resetExperiment() +# testGram.interval = 100 # all milliseconds +# testGram.sleepTime = 20 +# testGram.MaxHit = 6 +# +# assert natlinktimer.getNatlinktimerStatus() in (0, None) +# natlinktimer.setTimerCallback(testGram.doTimerClassic, interval=testGram.interval, debug=debug) +# ## 1 timer active: +# assert natlinktimer.getNatlinktimerStatus() == 1 +# for _ in range(5): +# if testGram.Hit >= testGram.MaxHit: +# break +# wait(500) # 0.5 second +# if debug: +# print(f'waited 0.1 second for timer to finish testGram, Hit: {testGram.Hit} ({testGram.MaxHit})') +# else: +# raise TestError(f'not enough time to finish the testing procedure (came to {testGram.Hit} of {testGram.MaxHit})') +# print(f'testGram.results: {testGram.results}') +# assert len(testGram.results) == testGram.MaxHit +# assert testGram.interval == 40 +# assert testGram.sleepTime == 20 +# assert natlinktimer.getNatlinktimerStatus() == 0 ## natlinktimer is NOT destroyed after last timer is gone. +# +# finally: +# del natlinktimer.natlinktimer +# natlinktimer.stopTimerCallback() +# natlink.natDisconnect() + + +def testStopAtMicOff(): + try: + natlink.natConnect() + testGram = TestGrammar(name="stop_at_mic_off") + testGram.resetExperiment() + testGram.interval = 100 # all milliseconds + testGram.sleepTime = 20 + testGram.MaxHit = 6 + testGram.toggleMicAt = 3 + assert natlinktimer.getNatlinktimerStatus() in (0, None) + natlinktimer.setTimerCallback(testGram.doTimerClassic, interval=testGram.interval, callAtMicOff=testGram.cancelMode, debug=debug) + ## 1 timer active: + for _ in range(5): + if testGram.toggleMicAt and testGram.Hit >= testGram.toggleMicAt: + break + if testGram.Hit >= testGram.MaxHit: + break + wait(500) # 0.5 second + if debug: + print(f'waited 0.1 second for timer to finish testGram, Hit: {testGram.Hit} ({testGram.MaxHit})') + else: + raise TestError(f'not enough time to finish the testing procedure (came to {testGram.Hit} of {testGram.MaxHit})') + print(f'testGram.results: {testGram.results}') + assert len(testGram.results) == testGram.toggleMicAt + assert natlinktimer.getNatlinktimerStatus() == 0 ## natlinktimer is NOT destroyed after last timer is gone. + assert testGram.startCancelMode is True + assert testGram.endCancelMode is True + + finally: + natlinktimer.stopTimerCallback() + natlink.natDisconnect() + + +# def testStopAtMicOff(): +# try: +# natlink.natConnect() +# testGram = TestGrammar(name="single") +# testGram.interval = 100 # all milliseconds +# testGram.sleepTime = 20 +# testGram.MaxHit = 10 +# assert natlinktimer.getNatlinktimerStatus() in (0, None) +# testGram.resetExperiment() +# +# gt = testGram.grammarTimerMicOff = natlinktimer.setTimerCallback(testGram.doTimer, interval=testGram.interval, debug=debug) +# gtmicoff = testGram.grammarTimerMicOff = natlinktimer.setTimerCallback(testGram.doTimerMicToggle, interval=testGram.interval*3, debug=debug) +# gtstr = str(gt) +# gtmicoffstr = str(gtmicoff) +# assert gtstr == 'grammartimer, interval: 100, nextTime (relative): 100' +# assert gtmicoffstr == 'grammartimer, interval: 300, nextTime (relative): 300' +# ## 2 timers active: +# assert natlinktimer.getNatlinktimerStatus() == 2 +# for _ in range(5): +# if testGram.Hit >= testGram.MaxHit: +# break +# wait(1000) # 0.1 second +# if debug: +# print(f'waited 0.1 second for timer to finish testGram, Hit: {testGram.Hit} ({testGram.MaxHit})') +# else: +# raise TestError(f'not enough time to finish the testing procedure (came to {testGram.Hit} of {testGram.MaxHit})') +# print(f'testGram.results: {testGram.results}') +# assert len(testGram.results) == testGram.MaxHit +# +# assert natlinktimer.getNatlinktimerStatus() == 0 ## natlinktimer is NOT destroyed after last timer is gone. +# +# finally: +# natlink.natDisconnect() +# +# def testWrongValuesTimer(): +# try: +# natlink.natConnect() +# testGram = TestGrammar(name="wrongvalues") +# testGram.interval = -200 # all milliseconds +# testGram.sleepTime = 30 +# assert natlinktimer.getNatlinktimerStatus() in (0, None) +# # testGram.resetExperiment() +# assert testGram.Hit == 0 +# testGram.grammarTimer = natlinktimer.setTimerCallback(testGram.doTimer, interval=testGram.interval) ##, debug=1) +# assert natlinktimer.getNatlinktimerStatus() == 0 +# +# finally: +# natlink.natDisconnect() +# +# +# def testThreeTimersMinimalSleepTime(): +# try: +# if debug: +# print('testThreeTimersMinimalSleepTime') +# natlink.natConnect() +# testGramOne = TestGrammar(name="min_sleeptime_one") +# testGramOne.interval = 100 # all milliseconds +# testGramOne.sleepTime = 2 +# testGramTwo = TestGrammar(name="min_sleeptime_two") +# testGramTwo.interval = 50 # all milliseconds +# testGramTwo.sleepTime = 2 +# testGramThree = TestGrammar(name="min_sleeptime_three") +# testGramThree.interval = 100 # all milliseconds +# testGramThree.sleepTime = 2 +# testGramThree.MaxHit = 2 +# assert natlinktimer.getNatlinktimerStatus() in (0, None) +# testGramOne.grammarTimer = natlinktimer.setTimerCallback(testGramOne.doTimer, interval=testGramOne.interval, debug=debug) +# testGramTwo.grammarTimer = natlinktimer.setTimerCallback(testGramTwo.doTimer, interval=testGramTwo.interval, debug=debug) +# testGramThree.grammarTimer = natlinktimer.setTimerCallback(testGramThree.doTimer, interval=testGramThree.interval, debug=debug) +# assert natlinktimer.getNatlinktimerStatus() == 3 +# for _ in range(5): +# if testGramOne.Hit >= testGramOne.MaxHit and testGramTwo.Hit >= testGramTwo.MaxHit: +# break +# wait(1000) # 1 second +# else: +# raise TestError('not enough time to finish the testing procedure') +# print(f'testGramOne.results: {testGramOne.results}') +# print(f'testGramTwo.results: {testGramTwo.results}') +# print(f'testGramThree.results: {testGramThree.results}') +# assert len(testGramOne.results) == testGramOne.MaxHit +# assert len(testGramTwo.results) == testGramTwo.MaxHit +# +# assert natlinktimer.getNatlinktimerStatus() == 0 ## natlinktimer is NOT destroyed after last timer is gone. +# +# finally: +# natlink.natDisconnect() +# + + + + +def wait(tmilli=100): + """wait milliseconds via waitForSpeech loop of natlink + + default 100 milliseconds, or 0.1 second + """ + tmilli = int(tmilli) + natlink.waitForSpeech(tmilli) + +if __name__ == "__main__": + pytest.main(['test_natlinktimer.py']) + \ No newline at end of file