From 642b04b222698eb39cbd7b3ac128b064e068bb5c Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Thu, 27 Oct 2022 12:45:41 +0200 Subject: [PATCH 01/20] bump to 1.5.6 --- src/natlinkcore/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/natlinkcore/__init__.py b/src/natlinkcore/__init__.py index 2bd786f..83283f6 100644 --- a/src/natlinkcore/__init__.py +++ b/src/natlinkcore/__init__.py @@ -1,5 +1,5 @@ '''Python portion of Natlink, a compatibility module for Dragon Naturally Speaking''' -__version__="5.3.4" +__version__="5.3.5" #pylint:disable=C0114, W0401 from typing import Optional from pathlib import Path From 0c2d65c994ee1b49fa142e5f52eac516b39c9609 Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Thu, 27 Oct 2022 12:51:20 +0200 Subject: [PATCH 02/20] sorry previous bump number was wrong (the branch name so to say, which could cause confusion) --- src/natlinkcore/__init__.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/natlinkcore/__init__.py b/src/natlinkcore/__init__.py index 83283f6..ac216d3 100644 --- a/src/natlinkcore/__init__.py +++ b/src/natlinkcore/__init__.py @@ -1,4 +1,6 @@ -'''Python portion of Natlink, a compatibility module for Dragon Naturally Speaking''' +'''Python portion of Natlink, a compatibility module for Dragon Naturally Speaking +The python stuff including test modules''' + __version__="5.3.5" #pylint:disable=C0114, W0401 from typing import Optional From 78f87a740b373e48986ce58f7894372d8cbe159f Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Mon, 14 Nov 2022 11:55:35 +0100 Subject: [PATCH 03/20] correct import natlink lines --- src/natlinkcore/SampleMacros/_debug_natlink.py | 2 +- src/natlinkcore/config.py | 2 +- src/natlinkcore/loader.py | 2 +- src/natlinkcore/nsformat.py | 2 +- src/natlinkcore/redirect_output.py | 2 +- 5 files changed, 5 insertions(+), 5 deletions(-) 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/config.py b/src/natlinkcore/config.py index 1f43241..8f5ca48 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 diff --git a/src/natlinkcore/loader.py b/src/natlinkcore/loader.py index 11c721a..ef25a76 100644 --- a/src/natlinkcore/loader.py +++ b/src/natlinkcore/loader.py @@ -15,7 +15,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 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" ) From 28af081b4dcf3cd7870d95a41b1c2853ae031b06 Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Sun, 27 Nov 2022 16:28:08 +0100 Subject: [PATCH 04/20] improved expand_path function. Also sub directories of a packages are expanded (unimacro/unimacrogrammars). --- src/natlinkcore/config.py | 29 ++++++++++++++++++++--------- tests/test_config.py | 20 ++++++++++++++++++++ 2 files changed, 40 insertions(+), 9 deletions(-) diff --git a/src/natlinkcore/config.py b/src/natlinkcore/config.py index 8f5ca48..eb28a84 100644 --- a/src/natlinkcore/config.py +++ b/src/natlinkcore/config.py @@ -126,7 +126,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. @@ -163,15 +163,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__[0] + 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/tests/test_config.py b/tests/test_config.py index f18a37e..a4577bd 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -138,6 +138,26 @@ 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) + + def test_config_locations(): """tests the lists of possible config_locations and of valid_locations """ From 772b3c763affa024136b6d86b94fb9ade6c549ca Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Sun, 27 Nov 2022 16:28:50 +0100 Subject: [PATCH 05/20] why was this one lost?? --- build_natlinkcore.ps1 | 4 ++++ 1 file changed, 4 insertions(+) create mode 100644 build_natlinkcore.ps1 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 From 16bb5d2b0fbb4af463343273daa6d9c95e95eaa5 Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Sun, 27 Nov 2022 17:00:16 +0100 Subject: [PATCH 06/20] fixing details (directory is last item of module.__path__ (when package is editable) --- src/natlinkcore/config.py | 2 +- src/natlinkcore/natlinkstatus.py | 14 +++++--------- tests/test_config.py | 7 +++++++ 3 files changed, 13 insertions(+), 10 deletions(-) diff --git a/src/natlinkcore/config.py b/src/natlinkcore/config.py index eb28a84..5683b0e 100644 --- a/src/natlinkcore/config.py +++ b/src/natlinkcore/config.py @@ -174,7 +174,7 @@ def expand_path(input_path: str) -> str: # find path for package. not an alternative way without loading the package is to use importlib.util.findspec. try: pack = __import__(package_trunk) - package_path = pack.__path__[0] + package_path = pack.__path__[-1] if rest: dir_expanded = str(Path(package_path)/rest) return dir_expanded diff --git a/src/natlinkcore/natlinkstatus.py b/src/natlinkcore/natlinkstatus.py index a5d4569..cf96317 100644 --- a/src/natlinkcore/natlinkstatus.py +++ b/src/natlinkcore/natlinkstatus.py @@ -120,7 +120,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. @@ -165,10 +164,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,7 +397,7 @@ def getUnimacroDirectory(self): except ImportError: self.UnimacroDirectory = "" return "" - self.UnimacroDirectory = str(Path(unimacro.__file__).parent) + self.UnimacroDirectory = unimacro.__path__[-1] return self.UnimacroDirectory @@ -539,7 +537,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 +545,7 @@ def getVocolaDirectory(self): except ImportError: self.VocolaDirectory = '' return '' - self.VocolaDirectory = str(Path(vocola2.__file__).parent) + self.VocolaDirectory = vocola2.__path__[-1] return self.VocolaDirectory @@ -726,7 +723,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') diff --git a/tests/test_config.py b/tests/test_config.py index a4577bd..0f1849e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -157,6 +157,13 @@ def test_expand_path(mock_syspath,mock_userdir): 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 From 4a8662dfe7d202fa3dc374e5fc958202ec27224d Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Thu, 1 Dec 2022 16:51:20 +0100 Subject: [PATCH 07/20] added module natlinktimer (multiplexing natlink.setTimerCallback), and added natlinkstatus: getUnimacroDataDirectory --- src/natlinkcore/natlinkstatus.py | 32 +++- src/natlinkcore/natlinktimer.py | 272 +++++++++++++++++++++++++++++++ tests/unittestNatlinktimer.py | 236 +++++++++++++++++++++++++++ 3 files changed, 536 insertions(+), 4 deletions(-) create mode 100644 src/natlinkcore/natlinktimer.py create mode 100644 tests/unittestNatlinktimer.py diff --git a/src/natlinkcore/natlinkstatus.py b/src/natlinkcore/natlinkstatus.py index cf96317..af758c4 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. @@ -151,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 @@ -401,6 +405,26 @@ def getUnimacroDirectory(self): 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. @@ -690,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', @@ -762,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..33b5041 --- /dev/null +++ b/src/natlinkcore/natlinktimer.py @@ -0,0 +1,272 @@ +""" +Handling of more calls to the Natlink timer +Quintijn Hoogenboom, 25-4-2020 + +""" +#--------------------------------------------------------------------------- +import time +import traceback + +import natlink + +natlinktimer = None + + +class GrammarTimer: + """object which specifies how to call the natlinkTimer + + """ + #pylint:disable=R0913 + def __init__(self, callback, interval, startNow=False, stopAtMicOff=False, maxIterations=None): + curTime = self.starttime = round(time.time()*1000) + self.callback = callback + self.interval = interval + self.nextTime = curTime + interval + self.stopAtMicOff = stopAtMicOff + self.startNow = startNow + 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', 'stopAtMicOff', 'startNow', 'maxIterations': + value = self.__dict__.get(varname, None) + if not value is None: + L.append(f' {varname.ljust(13)}: {value}') + return "\n".join(L) + + + + +class NatlinkTimer: + """ + 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 + + """ + def __init__(self, minInterval=None): + """initialize the natlink timer instance + + Should be called only once in a session + + 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 = None + self.timerStartTime = self.getnow() + self.minInterval = minInterval or 50 + self.tolerance = min(10, int(self.minInterval/4)) + + def __del__(self): + """stop the timer, when destroyed + """ + self.stopTimer() + + def getnow(self): + """get time in milliseconds + """ + return round(time.time()*1000) + + + def addCallback(self, callback, interval, debug=None): + """add an interval + """ + self.debug = self.debug or debug + now = self.getnow() + + if interval <= 0: + self.removeCallback(callback) + return None + if interval <= self.minInterval: + if self.debug: + print(f'addCallback {callback.__name}, set interval from {interval} to minInterval: {self.minInterval}') + interval = round(interval) + gt = GrammarTimer(callback, interval) + self.callbacks[callback] = gt + if self.debug: + print(f'set new timer{callback.__name__}: {interval} ({now}') + self.hittimer() + 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__}') + + try: + del self.callbacks[callback] + except KeyError: + pass + if not self.callbacks: + if self.debug: + print("last timer removed, setTimerCallback to 0") + + self.stopTimer() + return + + self.hittimer() + + def hittimer(self): + """move to a next callback point + """ + #pylint:disable=R0914, R0912, R0915, W0702 + now = self.getnow() + nowRel = now - self.timerStartTime + if self.debug: + print("start hittimer at", nowRel) + + toBeRemoved = [] + # c = callbackFunc, g = grammarTimer + decorated = [(g.interval, c, g) for (c, g) in self.callbacks.items()] + sortedList = sorted(decorated) + + + 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: + newInterval = callbackFunc() + except: + print(f"exception in callbackFunc ({callbackFunc}), remove from list") + traceback.print_exc() + toBeRemoved.append(callbackFunc) + endCallback = None + else: + endCallback = self.getnow() + + if newInterval and newInterval >= 0: + print(f"newInterval as result of {callbackFunc.__name__}: {newInterval}") + grammarTimer.interval = interval = newInterval + elif newInterval and newInterval <= 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}") + + + + + def stopTimer(self): + """stop the natlink timer, by passing in None, 0 + """ + natlink.setTimerCallback(None, 0) + + + +def setTimerCallback(callback, interval, debug=None): + """This function sets a callback + + Interval in seconds, unless larger than 24 + callback: the function to be called + """ + #pylint:disable=W0603 + global natlinktimer + if not natlinktimer: + natlinktimer = NatlinkTimer() + 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: + gt = natlinktimer.addCallback(callback, interval, debug=debug) + return gt + natlinktimer.removeCallback(callback, debug=debug) + return None + + +def removeTimerCallback(callback, debug=None): + """This function removes a callback from the callbacks dict + + callback: the function to be called + """ + #pylint:disable=W0603 + global natlinktimer + 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 + if natlinktimer: + del natlinktimer + diff --git a/tests/unittestNatlinktimer.py b/tests/unittestNatlinktimer.py new file mode 100644 index 0000000..6072b36 --- /dev/null +++ b/tests/unittestNatlinktimer.py @@ -0,0 +1,236 @@ +"""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, R0201 + +# import sys +import unittest +import time +# import traceback # for printing exceptions + +# from pathqh import path +from pathlib import Path +import natlinkcore +from natlinkcore import natlink +from natlinkcore.natlinkutils import GrammarBase +from natlinkcore import natlinktimer +from natlinkcore import natlinkstatus + +class TestError(Exception): + """TestError""" + +ExitQuietly = 'ExitQuietly' + +# try some experiments more times, because gotBegin sometimes seems +# not to hit +nTries = 10 +natconnectOption = 1 # or 1 for threading, 0 for not. Seems to make difference + # with spurious error (if set to 1), missing gotBegin and all that... + +thisDir = natlinkcore.getThisDir(__file__) +# +# +# def getBaseFolder(globalsDict=None): +# """get the folder of the calling module. +# +# return a str... +# """ +# baseFolder = path(".").normpath() +# return baseFolder +# +# thisDir = getBaseFolder(globals()) + + +logFileName = Path(thisDir)/"Natlinktimertestresult.txt" + +# make different versions testing possible: +nlstatus = natlinkstatus.NatlinkStatus() +DNSVersion = nlstatus.getDNSVersion() + +#--------------------------------------------------------------------------- +# These tests should be run after we call natConnect +class UnittestNatlinktimer(unittest.TestCase): + def setUp(self): + if not natlink.isNatSpeakRunning(): + raise TestError('NatSpeak is not currently running') + self.connect() + # remember user and get DragonPad in front: + self.setMicState = "off" + #self.lookForDragonPad() + + def tearDown(self): + try: + # give message: + self.setMicState = "off" + # kill things + finally: + natlinktimer.stopTimerCallback() + self.disconnect() + + def connect(self): + # start with 1 for thread safety when run from pythonwin: + natlink.natConnect(natconnectOption) + + def disconnect(self): + natlink.natDisconnect() + + def log(self, t, doPlaystring=None): + # displayTest seems not to work: + natlink.displayText(t, 0) + if doPlaystring: + natlink.playString(t+'\n') + # do the global log function: + log(t) + + def wait(self, t=1): + time.sleep(t) + + #--------------------------------------------------------------------------- + # This utility subroutine executes a Python command and makes sure that + # an exception (of the expected type) is raised. Otherwise a TestError + # exception is raised + + def doTestForException(self, exceptionType,command,localVars=None): + #pylint:disable=W0122 + if localVars is None: + localVars = dict() + try: + exec(command,globals(),localVars) + except exceptionType: + return + raise TestError('Expecting an exception to be raised calling '+command) + + #--------------------------------------------------------------------------- + # Utility function which calls a routine and tests the return value + + def doTestFuncReturn(self, expected,command,localVars=None): + # account for different values in case of [None, 0] (wordFuncs) + #pylint:disable=W0123 + if localVars is None: + actual = eval(command) + else: + actual = eval(command, globals(), localVars) + + if actual != expected: + time.sleep(1) + self.assertEqual(expected, actual, 'Function call "%s" returned unexpected result\nExpected: %s, got: %s'% + (command, expected, actual)) + + def testSingleTimer(self): + #pylint:disable=W0201 + testName = "testSingleTimer" + self.log(testName) + + # Create a simple command grammar. + # This grammar then sets the timer, and after 3 times expires + + class TestGrammar(GrammarBase): + def __init__(self): + GrammarBase.__init__(self) + self.resetExperiment() + + def resetExperiment(self): + self.Hit = 0 + self.MaxHit = 5 + self.sleepTime = 0 # to be specified by calling instance, the sleeping time after each hit + self.results = [] + + + def report(self): + """print the results lines + """ + for line in self.results: + print(line) + + def doTimer(self): + self.results.append(f'doTimer {self.Hit}') + self.Hit +=1 + log(f"hit {self.Hit}") + time.sleep(self.sleepTime/1000) # sleep 10 milliseconds + if self.Hit == self.MaxHit: + expectElapsed = self.Hit * self.interval + print(f"expect duration of this timer: {expectElapsed}") + natlinktimer.removeTimerCallback(self.doTimer) + ## try to shorten interval: + currentInterval = self.grammarTimer.interval + if currentInterval > 150: + newInterval = currentInterval - 10 + return newInterval + return None + + testGram = TestGrammar() + testGram.interval = 200 + testGram.sleepTime = 30 # all milliseconds now + testGram.grammarTimer = natlinktimer.setTimerCallback(testGram.doTimer, interval=testGram.interval, debug=1) + for _i in range(5): + if testGram.Hit >= testGram.MaxHit : + printInfo = str(testGram.grammarTimer) + self.log(f"timer seems to be ready. results: {printInfo}") + break + wait(1000) + else: + self.log(f"waiting time expired, results got: {testGram.results}") + testGram.report() + self.log("End of %s"% testName) + +def wait(tmilli=100): + """wait milliseconds via waitForSpeech loop of natlink + + default 100 milliseconds, or 0.1 second + """ + tmilli = int(tmilli) + natlink.waitForSpeech(tmilli) + +def log(t, refresh=None): + """logging is a mess, just print here... + """ + #pylint:disable=W0613 + print(t) + # openOption = 'a' if not refresh else 'w' + # with open(logFileName, openOption) as lf: + # lf.write(t + '\n') + +#--------------------------------------------------------------------------- +# run +# +# # This is the main entry point. It will connect to NatSpeak and perform +# # a series of tests. In the case of an error, it will cleanly disconnect +# # from NatSpeak and print the End of testSingleTimer +# def dumpResult(testResult, logFileName): +# """dump into the logFile +# """ +# with open(logFileName, 'a') as logFile: +# if testResult.wasSuccessful(): +# mes = "all succesEnd of testSingleTimerful" +# logFile.write(mes) +# return +# logFile.write('\n--------------- errors -----------------\n') +# for case, tb in testResult.errors: +# logFile.write('\n---------- %s --------\n'% case) +# logFile.write(tb) +# +# logFile.write('\n--------------- failures -----------------\n') +# for case, tb in testResult.failures: +# logFile.write('\n---------- %s --------\n'% case) +# logFile.write(tb) + + +def run(): + # log("log messages to file: %s"% logFileName) + log('starting unittestNatlinktimer') + suite = unittest.makeSuite(UnittestNatlinktimer, 'test') + log('\nstarting tests with threading: %s\n'% natconnectOption) + unittest.TextTestRunner().run(suite) + +if __name__ == "__main__": + log("no log file, just printing on console") + run() From de953da22b248b728e52b2d128e0af2c7d3d91ca Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Thu, 1 Dec 2022 18:19:33 +0100 Subject: [PATCH 08/20] rename test script to test_natlinktimer.py --- tests/{unittestNatlinktimer.py => test_natlinktimer.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename tests/{unittestNatlinktimer.py => test_natlinktimer.py} (100%) diff --git a/tests/unittestNatlinktimer.py b/tests/test_natlinktimer.py similarity index 100% rename from tests/unittestNatlinktimer.py rename to tests/test_natlinktimer.py From 856e7e02913d0ae1b807f4f3683943eaba0f4219 Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Sat, 3 Dec 2022 16:41:54 +0100 Subject: [PATCH 09/20] typo natlinkstatus.py, improved and tested (a bit) natlinktimer --- src/natlinkcore/natlinkstatus.py | 2 +- src/natlinkcore/natlinktimer.py | 39 ++-- tests/test_natlinktimer.py | 320 ++++++++++++------------------- 3 files changed, 146 insertions(+), 215 deletions(-) diff --git a/src/natlinkcore/natlinkstatus.py b/src/natlinkcore/natlinkstatus.py index af758c4..447f127 100644 --- a/src/natlinkcore/natlinkstatus.py +++ b/src/natlinkcore/natlinkstatus.py @@ -140,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 diff --git a/src/natlinkcore/natlinktimer.py b/src/natlinkcore/natlinktimer.py index 33b5041..0fad2e6 100644 --- a/src/natlinkcore/natlinktimer.py +++ b/src/natlinkcore/natlinktimer.py @@ -1,6 +1,8 @@ """ Handling of more calls to the Natlink timer -Quintijn Hoogenboom, 25-4-2020 + +make it a Singleton class (December 2022) +Quintijn Hoogenboom """ #--------------------------------------------------------------------------- @@ -8,10 +10,11 @@ import traceback import natlink +from natlinkcore import singleton +## this variable will hold the (only) NatlinkTimer instance natlinktimer = None - class GrammarTimer: """object which specifies how to call the natlinkTimer @@ -41,21 +44,20 @@ def __repr__(self): return "\n".join(L) - - -class NatlinkTimer: +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 + 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 - Should be called only once in a session + 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 @@ -90,7 +92,7 @@ def addCallback(self, callback, interval, debug=None): return None if interval <= self.minInterval: if self.debug: - print(f'addCallback {callback.__name}, set interval from {interval} to minInterval: {self.minInterval}') + print(f'addCallback {callback.__name__}, set interval from {interval} to minInterval: {self.minInterval}') interval = round(interval) gt = GrammarTimer(callback, interval) self.callbacks[callback] = gt @@ -212,9 +214,6 @@ def hittimer(self): totaltime = nownownow - now print(f"time taken in closingphase: {timeinclosingphase}") print(f"total time spent hittimer: {totaltime}") - - - def stopTimer(self): """stop the natlink timer, by passing in None, 0 @@ -226,7 +225,9 @@ def stopTimer(self): def setTimerCallback(callback, interval, debug=None): """This function sets a callback - Interval in seconds, unless larger than 24 + Interval in milliseconds, unless smaller than 25 + + When 0 or negative: it functions as removeTimerCallback!! callback: the function to be called """ #pylint:disable=W0603 @@ -239,10 +240,11 @@ def setTimerCallback(callback, interval, debug=None): if callback is None: raise Exception("stop the timer callback with natlinktimer.removeCallback(callback)") - if interval: + if interval > 0: gt = natlinktimer.addCallback(callback, interval, debug=debug) return gt - natlinktimer.removeCallback(callback, debug=debug) + # interval is 0 (or negative), remove the callback + removeTimerCallback(callback, debug=debug) return None @@ -251,8 +253,6 @@ def removeTimerCallback(callback, debug=None): callback: the function to be called """ - #pylint:disable=W0603 - global natlinktimer if not natlinktimer: print(f'no timers active, cannot remove {callback} from natlinktimer') return @@ -269,4 +269,11 @@ def stopTimerCallback(): global natlinktimer if natlinktimer: del natlinktimer + +def getNatlinktimerStatus(): + """report how many callbacks are active, None if natlinktimer is gone + """ + if natlinktimer is None: + return None + return len(natlinktimer.callbacks) diff --git a/tests/test_natlinktimer.py b/tests/test_natlinktimer.py index 6072b36..64ff0a1 100644 --- a/tests/test_natlinktimer.py +++ b/tests/test_natlinktimer.py @@ -10,25 +10,16 @@ natlinktimer.py, for multiplexing timer instances acrross different grammars Quintijn Hoogenboom, summer 2020 """ -#pylint:disable=C0115, C0116, R0201 +#pylint:disable=C0115, C0116 +#pylint:disable=E1101 # import sys -import unittest import time -# import traceback # for printing exceptions - -# from pathqh import path from pathlib import Path -import natlinkcore -from natlinkcore import natlink +import pytest +import natlink from natlinkcore.natlinkutils import GrammarBase from natlinkcore import natlinktimer -from natlinkcore import natlinkstatus - -class TestError(Exception): - """TestError""" - -ExitQuietly = 'ExitQuietly' # try some experiments more times, because gotBegin sometimes seems # not to hit @@ -36,152 +27,126 @@ class TestError(Exception): natconnectOption = 1 # or 1 for threading, 0 for not. Seems to make difference # with spurious error (if set to 1), missing gotBegin and all that... -thisDir = natlinkcore.getThisDir(__file__) -# -# -# def getBaseFolder(globalsDict=None): -# """get the folder of the calling module. +thisDir = Path(__file__).parent + +# define TestError, and mark is to be NOT a part of pytest: +class TestError(Exception): + pass +TestError.__test__ = False + +# 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.sleepTime = 0 # to be specified by calling instance, the sleeping time after each hit + self.results = [] + + # def __del__(self): + # """try to remove the grammarTimer first""" + # del self.grammarTimer + + def doTimer(self): + self.results.append(f'doTimer {self.name}: {self.Hit}') + 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 this timer: {expectElapsed} milliseconds') + natlinktimer.removeTimerCallback(self.doTimer) + ## try to shorten interval: + currentInterval = self.grammarTimer.interval + if currentInterval > 250: + newInterval = currentInterval - 25 + return newInterval + return None +TestGrammar.__test__ = False + + +# def testSingleTimer(): +# try: +# natlink.natConnect() +# testGram = TestGrammar(name="single") +# testGram.interval = 200 # all milliseconds +# testGram.sleepTime = 30 +# assert natlinktimer.getNatlinktimerStatus() in (0, None) +# cycles = 2 +# for cycle in range(cycles): +# print(f'cycle: {cycle} of {cycles} (test is {testGram.name})') +# testGram.resetExperiment() +# testGram.grammarTimer = natlinktimer.setTimerCallback(testGram.doTimer, interval=testGram.interval) ##, debug=1) +# assert natlinktimer.getNatlinktimerStatus() == 1 +# for _ in range(5): +# if testGram.Hit >= testGram.MaxHit: +# break +# wait(1000) # 1 second +# else: +# raise TestError('not enough time to finish the testing procedure') +# 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. # -# return a str... -# """ -# baseFolder = path(".").normpath() -# return baseFolder +# 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 # -# thisDir = getBaseFolder(globals()) - - -logFileName = Path(thisDir)/"Natlinktimertestresult.txt" - -# make different versions testing possible: -nlstatus = natlinkstatus.NatlinkStatus() -DNSVersion = nlstatus.getDNSVersion() - -#--------------------------------------------------------------------------- -# These tests should be run after we call natConnect -class UnittestNatlinktimer(unittest.TestCase): - def setUp(self): - if not natlink.isNatSpeakRunning(): - raise TestError('NatSpeak is not currently running') - self.connect() - # remember user and get DragonPad in front: - self.setMicState = "off" - #self.lookForDragonPad() - - def tearDown(self): - try: - # give message: - self.setMicState = "off" - # kill things - finally: - natlinktimer.stopTimerCallback() - self.disconnect() - - def connect(self): - # start with 1 for thread safety when run from pythonwin: - natlink.natConnect(natconnectOption) - - def disconnect(self): - natlink.natDisconnect() - - def log(self, t, doPlaystring=None): - # displayTest seems not to work: - natlink.displayText(t, 0) - if doPlaystring: - natlink.playString(t+'\n') - # do the global log function: - log(t) - - def wait(self, t=1): - time.sleep(t) - - #--------------------------------------------------------------------------- - # This utility subroutine executes a Python command and makes sure that - # an exception (of the expected type) is raised. Otherwise a TestError - # exception is raised - - def doTestForException(self, exceptionType,command,localVars=None): - #pylint:disable=W0122 - if localVars is None: - localVars = dict() - try: - exec(command,globals(),localVars) - except exceptionType: - return - raise TestError('Expecting an exception to be raised calling '+command) - - #--------------------------------------------------------------------------- - # Utility function which calls a routine and tests the return value - - def doTestFuncReturn(self, expected,command,localVars=None): - # account for different values in case of [None, 0] (wordFuncs) - #pylint:disable=W0123 - if localVars is None: - actual = eval(command) - else: - actual = eval(command, globals(), localVars) - - if actual != expected: - time.sleep(1) - self.assertEqual(expected, actual, 'Function call "%s" returned unexpected result\nExpected: %s, got: %s'% - (command, expected, actual)) - - def testSingleTimer(self): - #pylint:disable=W0201 - testName = "testSingleTimer" - self.log(testName) - - # Create a simple command grammar. - # This grammar then sets the timer, and after 3 times expires - - class TestGrammar(GrammarBase): - def __init__(self): - GrammarBase.__init__(self) - self.resetExperiment() - - def resetExperiment(self): - self.Hit = 0 - self.MaxHit = 5 - self.sleepTime = 0 # to be specified by calling instance, the sleeping time after each hit - self.results = [] - - - def report(self): - """print the results lines - """ - for line in self.results: - print(line) - - def doTimer(self): - self.results.append(f'doTimer {self.Hit}') - self.Hit +=1 - log(f"hit {self.Hit}") - time.sleep(self.sleepTime/1000) # sleep 10 milliseconds - if self.Hit == self.MaxHit: - expectElapsed = self.Hit * self.interval - print(f"expect duration of this timer: {expectElapsed}") - natlinktimer.removeTimerCallback(self.doTimer) - ## try to shorten interval: - currentInterval = self.grammarTimer.interval - if currentInterval > 150: - newInterval = currentInterval - 10 - return newInterval - return None - - testGram = TestGrammar() - testGram.interval = 200 - testGram.sleepTime = 30 # all milliseconds now - testGram.grammarTimer = natlinktimer.setTimerCallback(testGram.doTimer, interval=testGram.interval, debug=1) - for _i in range(5): - if testGram.Hit >= testGram.MaxHit : - printInfo = str(testGram.grammarTimer) - self.log(f"timer seems to be ready. results: {printInfo}") +# finally: +# natlink.natDisconnect() + +def testTwoTimers(): + try: + natlink.natConnect() + testGramOne = TestGrammar(name="timer_one") + testGramOne.interval = 100 # all milliseconds + testGramOne.sleepTime = 30 + testGramTwo = TestGrammar(name="timer_two") + testGramTwo.interval = 77 # all milliseconds + testGramTwo.sleepTime = 10 + assert natlinktimer.getNatlinktimerStatus() in (0, None) + testGramOne.grammarTimer = natlinktimer.setTimerCallback(testGramOne.doTimer, interval=testGramOne.interval) ##, debug=1) + testGramTwo.grammarTimer = natlinktimer.setTimerCallback(testGramTwo.doTimer, interval=testGramTwo.interval) ##, debug=1) + assert natlinktimer.getNatlinktimerStatus() == 2 + for _ in range(5): + if testGramOne.Hit >= testGramOne.MaxHit and testGramTwo.Hit >= testGramTwo.MaxHit: break - wait(1000) + wait(1000) # 1 second else: - self.log(f"waiting time expired, results got: {testGram.results}") - testGram.report() - self.log("End of %s"% testName) - + raise TestError('not enough time to finish the testing procedure') + print(f'testGramOne.results: {testGramOne.results}') + print(f'testGramTwo.results: {testGramTwo.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 @@ -189,48 +154,7 @@ def wait(tmilli=100): """ tmilli = int(tmilli) natlink.waitForSpeech(tmilli) - -def log(t, refresh=None): - """logging is a mess, just print here... - """ - #pylint:disable=W0613 - print(t) - # openOption = 'a' if not refresh else 'w' - # with open(logFileName, openOption) as lf: - # lf.write(t + '\n') - -#--------------------------------------------------------------------------- -# run -# -# # This is the main entry point. It will connect to NatSpeak and perform -# # a series of tests. In the case of an error, it will cleanly disconnect -# # from NatSpeak and print the End of testSingleTimer -# def dumpResult(testResult, logFileName): -# """dump into the logFile -# """ -# with open(logFileName, 'a') as logFile: -# if testResult.wasSuccessful(): -# mes = "all succesEnd of testSingleTimerful" -# logFile.write(mes) -# return -# logFile.write('\n--------------- errors -----------------\n') -# for case, tb in testResult.errors: -# logFile.write('\n---------- %s --------\n'% case) -# logFile.write(tb) -# -# logFile.write('\n--------------- failures -----------------\n') -# for case, tb in testResult.failures: -# logFile.write('\n---------- %s --------\n'% case) -# logFile.write(tb) - - -def run(): - # log("log messages to file: %s"% logFileName) - log('starting unittestNatlinktimer') - suite = unittest.makeSuite(UnittestNatlinktimer, 'test') - log('\nstarting tests with threading: %s\n'% natconnectOption) - unittest.TextTestRunner().run(suite) if __name__ == "__main__": - log("no log file, just printing on console") - run() + pytest.main(['test_natlinktimer.py']) + \ No newline at end of file From 8d80a38e66a869accc81b33c4582c5590aa57662 Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Wed, 7 Dec 2022 17:56:08 +0100 Subject: [PATCH 10/20] work on natlinktimer, with (nearly) old calling method it seems to work. --- src/natlinkcore/natlinktimer.py | 284 +++++++++++++++++++------------- tests/test_natlinktimer.py | 231 ++++++++++++++++++-------- 2 files changed, 337 insertions(+), 178 deletions(-) diff --git a/src/natlinkcore/natlinktimer.py b/src/natlinkcore/natlinktimer.py index 0fad2e6..cd88759 100644 --- a/src/natlinkcore/natlinktimer.py +++ b/src/natlinkcore/natlinktimer.py @@ -5,9 +5,12 @@ Quintijn Hoogenboom """ +#pylint:disable=R0913 + #--------------------------------------------------------------------------- import time import traceback +import operator import natlink from natlinkcore import singleton @@ -20,13 +23,14 @@ class GrammarTimer: """ #pylint:disable=R0913 - def __init__(self, callback, interval, startNow=False, stopAtMicOff=False, maxIterations=None): + def __init__(self, callback, interval, stopAtMicOff=False, maxIterations=None): curTime = self.starttime = round(time.time()*1000) self.callback = callback self.interval = interval - self.nextTime = curTime + interval + + self.nextTime = curTime + interval + self.stopAtMicOff = stopAtMicOff - self.startNow = startNow self.maxIterations = maxIterations def __str__(self): @@ -37,12 +41,21 @@ def __str__(self): def __repr__(self): L = ['GrammarTimer instance:'] - for varname in 'interval', 'nextTime', 'stopAtMicOff', 'startNow', 'maxIterations': + for varname in 'interval', 'nextTime', 'stopAtMicOff', '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): """ @@ -65,40 +78,47 @@ def __init__(self, minInterval=None): The minimum interval for the timer can be specified, is 50 by default. """ self.callbacks = {} - self.debug = None + self.debug = False self.timerStartTime = self.getnow() self.minInterval = minInterval or 50 self.tolerance = min(10, int(self.minInterval/4)) + self.in_timer = False 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 addCallback(self, callback, interval, debug=None): + def addCallback(self, callback, interval, stopAtMicOff=False, maxIterations=None, debug=None): """add an interval - """ - self.debug = self.debug or debug + """ + self.debug = debug now = self.getnow() if interval <= 0: self.removeCallback(callback) return None - if interval <= self.minInterval: - if self.debug: - print(f'addCallback {callback.__name__}, set interval from {interval} to minInterval: {self.minInterval}') - interval = round(interval) - gt = GrammarTimer(callback, interval) + interval = max(round(interval), self.minInterval) + gt = GrammarTimer(callback, interval, stopAtMicOff=stopAtMicOff, maxIterations=maxIterations) self.callbacks[callback] = gt if self.debug: - print(f'set new timer{callback.__name__}: {interval} ({now}') - self.hittimer() + print(f'set new timer {callback.__name__}, {interval} ({now})') + return gt def removeCallback(self, callback, debug=None): @@ -119,121 +139,135 @@ def removeCallback(self, callback, debug=None): self.stopTimer() return - self.hittimer() def hittimer(self): """move to a next callback point """ #pylint:disable=R0914, R0912, R0915, W0702 - now = self.getnow() - nowRel = now - self.timerStartTime - if self.debug: - print("start hittimer at", nowRel) - - toBeRemoved = [] - # c = callbackFunc, g = grammarTimer - decorated = [(g.interval, c, g) for (c, g) in self.callbacks.items()] - sortedList = sorted(decorated) - - - for interval, callbackFunc, grammarTimer in sortedList: + self.in_timer = True + try: now = self.getnow() - # for printing: + nowRel = now - self.timerStartTime 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 + 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)) - 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: - newInterval = callbackFunc() - except: - print(f"exception in callbackFunc ({callbackFunc}), remove from list") - traceback.print_exc() - toBeRemoved.append(callbackFunc) - endCallback = None - else: - endCallback = self.getnow() - - if newInterval and newInterval >= 0: - print(f"newInterval as result of {callbackFunc.__name__}: {newInterval}") - grammarTimer.interval = interval = newInterval - elif newInterval and newInterval <= 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: + 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"spent too much time in {callbackFunc.__name__}, increase interval from {interval} to: {spentInCallback*2}") - grammarTimer.interval = interval = spentInCallback*2 - grammarTimer.nextTime += interval + 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: - 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: + 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("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: + 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"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}") - + 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 + + def stopTimer(self): """stop the natlink timer, by passing in None, 0 """ natlink.setTimerCallback(None, 0) - - -def setTimerCallback(callback, interval, debug=None): - """This function sets a callback +def createGrammarTimer(callback, interval=0, stopAtMicOff=False, maxIterations=None, debug=None): + """return a grammarTimer instance - Interval in milliseconds, unless smaller than 25 + parameters: + callback: a function into which natlinktime will callback + interval: a starting interval (or 0), with which the timer shall run (initially) + optional: + stopAtMicOff: 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. - When 0 or negative: it functions as removeTimerCallback!! - callback: the function to be called + 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") @@ -241,11 +275,38 @@ def setTimerCallback(callback, interval, debug=None): raise Exception("stop the timer callback with natlinktimer.removeCallback(callback)") if interval > 0: - gt = natlinktimer.addCallback(callback, interval, debug=debug) + gt = natlinktimer.addCallback(callback, interval, stopAtMicOff=stopAtMicOff, 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, stopAtMicOff=False, maxIterations=None, debug=None): + """This function sets a timercallback, nearly the same as natlink.setTimerCallback + + Interval in milliseconds, unless smaller than 25 + + When 0 or negative: it functions as removeTimerCallback!! + callback: the function to be called + + But there are extra parameters possible, which are passed on to createGrammarTimer, see there + + """ + #pylint:disable=W0603 + if interval > 0: + if natlinktimer and callback in natlinktimer.callbacks: + gt = natlinktimer.callbacks[callback] + rel_cur_time = round(time.time()*1000) - gt.starttime + + 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, stopAtMicOff=stopAtMicOff, maxIterations=maxIterations, debug=debug) + natlinktimer.hittimer() + return # interval is 0 (or negative), remove the callback - removeTimerCallback(callback, debug=debug) - return None + removeTimerCallback(callback) + return def removeTimerCallback(callback, debug=None): @@ -267,8 +328,11 @@ def stopTimerCallback(): """ #pylint:disable=W0603 global natlinktimer - if natlinktimer: + natlink.setTimerCallback(None, 0) + try: del natlinktimer + except NameError: + pass def getNatlinktimerStatus(): """report how many callbacks are active, None if natlinktimer is gone diff --git a/tests/test_natlinktimer.py b/tests/test_natlinktimer.py index 64ff0a1..7bc67c7 100644 --- a/tests/test_natlinktimer.py +++ b/tests/test_natlinktimer.py @@ -21,12 +21,6 @@ from natlinkcore.natlinkutils import GrammarBase from natlinkcore import natlinktimer -# try some experiments more times, because gotBegin sometimes seems -# not to hit -nTries = 10 -natconnectOption = 1 # or 1 for threading, 0 for not. Seems to make difference - # with spurious error (if set to 1), missing gotBegin and all that... - thisDir = Path(__file__).parent # define TestError, and mark is to be NOT a part of pytest: @@ -34,6 +28,8 @@ 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"): @@ -46,18 +42,40 @@ def resetExperiment(self): self.MaxHit = 5 self.sleepTime = 0 # to be specified by calling instance, the sleeping time after each hit self.results = [] - - # def __del__(self): - # """try to remove the grammarTimer first""" - # del self.grammarTimer + self.starttime = round(time.time()*1000) + + 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: + # 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) + + # time.sleep(self.sleepTime/1000) # sleep 10 milliseconds + if self.Hit == self.MaxHit: + natlinktimer.setTimerCallback(self.doTimerClassic, 0) def doTimer(self): - self.results.append(f'doTimer {self.name}: {self.Hit}') + """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 this timer: {expectElapsed} milliseconds') + print(f'expect duration of timer {self.name}: {expectElapsed} milliseconds') natlinktimer.removeTimerCallback(self.doTimer) ## try to shorten interval: currentInterval = self.grammarTimer.interval @@ -65,37 +83,109 @@ def doTimer(self): newInterval = currentInterval - 25 return newInterval return None + + def doTimerMicToggle(self): + """this doTimer toggles the microphone when halfway + + """ + 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 doTimerMicToggle {self.name}: {expectElapsed} milliseconds') + natlinktimer.removeTimerCallback(self.doTimer) + if self.Hit >= self.maxHit/2: + print(f'toggle mic at {relTime}') + self.toggleMicrophone() + return None + + ## 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 testSingleTimer(): +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="single") -# testGram.interval = 200 # all milliseconds -# testGram.sleepTime = 30 -# assert natlinktimer.getNatlinktimerStatus() in (0, None) -# cycles = 2 -# for cycle in range(cycles): -# print(f'cycle: {cycle} of {cycles} (test is {testGram.name})') -# testGram.resetExperiment() -# testGram.grammarTimer = natlinktimer.setTimerCallback(testGram.doTimer, interval=testGram.interval) ##, debug=1) -# assert natlinktimer.getNatlinktimerStatus() == 1 -# for _ in range(5): -# if testGram.Hit >= testGram.MaxHit: -# break -# wait(1000) # 1 second -# else: -# raise TestError('not enough time to finish the testing procedure') -# 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. +# 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(): +# def testWrongValuesTimer(): # try: # natlink.natConnect() # testGram = TestGrammar(name="wrongvalues") @@ -109,42 +199,47 @@ def doTimer(self): # # finally: # natlink.natDisconnect() - -def testTwoTimers(): - try: - natlink.natConnect() - testGramOne = TestGrammar(name="timer_one") - testGramOne.interval = 100 # all milliseconds - testGramOne.sleepTime = 30 - testGramTwo = TestGrammar(name="timer_two") - testGramTwo.interval = 77 # all milliseconds - testGramTwo.sleepTime = 10 - assert natlinktimer.getNatlinktimerStatus() in (0, None) - testGramOne.grammarTimer = natlinktimer.setTimerCallback(testGramOne.doTimer, interval=testGramOne.interval) ##, debug=1) - testGramTwo.grammarTimer = natlinktimer.setTimerCallback(testGramTwo.doTimer, interval=testGramTwo.interval) ##, debug=1) - assert natlinktimer.getNatlinktimerStatus() == 2 - 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}') - 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 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): From 48c7ecd9a7705c732ccbf29615140baf1a09a7f7 Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Fri, 9 Dec 2022 11:57:51 +0100 Subject: [PATCH 11/20] working on natlinktimer, add callback at mic off in loader --- .../SampleMacros/_sample_callback.py | 4 ++ src/natlinkcore/loader.py | 10 ++++ src/natlinkcore/natlinktimer.py | 59 ++++++++++++++----- tests/test_natlinktimer.py | 55 +++++++++++++---- 4 files changed, 103 insertions(+), 25 deletions(-) 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/loader.py b/src/natlinkcore/loader.py index ef25a76..a4c71b9 100644 --- a/src/natlinkcore/loader.py +++ b/src/natlinkcore/loader.py @@ -62,6 +62,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 @@ -72,6 +73,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) @@ -84,6 +88,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) @@ -398,6 +405,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/natlinktimer.py b/src/natlinkcore/natlinktimer.py index cd88759..b7e31ed 100644 --- a/src/natlinkcore/natlinktimer.py +++ b/src/natlinkcore/natlinktimer.py @@ -11,9 +11,13 @@ import time import traceback import operator +import logging import natlink -from natlinkcore import singleton +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 @@ -21,16 +25,17 @@ 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, stopAtMicOff=False, maxIterations=None): + 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.stopAtMicOff = stopAtMicOff + + self.callAtMicOff = callAtMicOff self.maxIterations = maxIterations def __str__(self): @@ -41,7 +46,7 @@ def __str__(self): def __repr__(self): L = ['GrammarTimer instance:'] - for varname in 'interval', 'nextTime', 'stopAtMicOff', 'maxIterations': + 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}') @@ -83,6 +88,7 @@ def __init__(self, minInterval=None): self.minInterval = minInterval or 50 self.tolerance = min(10, int(self.minInterval/4)) self.in_timer = False + natlinkmain.set_on_mic_off_callback(self.on_mic_off_callback) def __del__(self): """stop the timer, when destroyed @@ -104,7 +110,18 @@ def getnow(self): """ return round(time.time()*1000) - def addCallback(self, callback, interval, stopAtMicOff=False, maxIterations=None, debug=None): + def on_mic_off_callback(self): + """all callbacks that have callAtMicOff set, will be stopped (and deleted) + """ + to_stop = [(cb, gt) for (cb, gt) in self.callbacks.items() if gt.callAtMicOff] + if not to_stop: + print('natlinktimer: no timers to stop') + return + for cb, gt in to_stop: + print(f'natlinktimer: stopping {cb}, {gt}') + self.removeCallback(cb) + + def addCallback(self, callback, interval, callAtMicOff=False, maxIterations=None, debug=None): """add an interval """ self.debug = debug @@ -114,7 +131,7 @@ def addCallback(self, callback, interval, stopAtMicOff=False, maxIterations=None self.removeCallback(callback) return None interval = max(round(interval), self.minInterval) - gt = GrammarTimer(callback, interval, stopAtMicOff=stopAtMicOff, maxIterations=maxIterations) + gt = GrammarTimer(callback, interval, callAtMicOff=callAtMicOff, maxIterations=maxIterations) self.callbacks[callback] = gt if self.debug: print(f'set new timer {callback.__name__}, {interval} ({now})') @@ -245,14 +262,14 @@ def stopTimer(self): """ natlink.setTimerCallback(None, 0) -def createGrammarTimer(callback, interval=0, stopAtMicOff=False, maxIterations=None, debug=None): +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: - stopAtMicOff: default False. When True, the timer stops when the mic is toggled to off + 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 @@ -275,33 +292,41 @@ def createGrammarTimer(callback, interval=0, stopAtMicOff=False, maxIterations=N raise Exception("stop the timer callback with natlinktimer.removeCallback(callback)") if interval > 0: - gt = natlinktimer.addCallback(callback, interval, stopAtMicOff=stopAtMicOff, maxIterations=maxIterations, debug=debug) + 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, stopAtMicOff=False, maxIterations=None, debug=None): +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 + Interval in milliseconds, unless smaller than 25 (default) When 0 or negative: it functions as removeTimerCallback!! - callback: the function to be called + + 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, stopAtMicOff=stopAtMicOff, maxIterations=maxIterations, debug=debug) + createGrammarTimer(callback, interval, callAtMicOff=callAtMicOff, maxIterations=maxIterations, debug=debug) natlinktimer.hittimer() return # interval is 0 (or negative), remove the callback @@ -337,6 +362,10 @@ def stopTimerCallback(): 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/tests/test_natlinktimer.py b/tests/test_natlinktimer.py index 7bc67c7..159c6d4 100644 --- a/tests/test_natlinktimer.py +++ b/tests/test_natlinktimer.py @@ -40,10 +40,14 @@ def __init__(self, name="testGrammar"): 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) + def cancelMode(self): + natlinktimer.setTimerCallback(self.doTimerClassic,0) + def doTimerClassic(self): """have no introspection, but be as close as possible to the old calling method of setTimerCallback @@ -53,13 +57,15 @@ def doTimerClassic(self): self.results.append(f'{self.Hit} {self.name}: {relTime}') self.Hit +=1 if self.Hit: + if self.toggleMicAt and self.toggleMicAt == self.Hit: + self.toggleMicrophone() # 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) + natlinktimer.setTimerCallback(self.doTimerClassic, interval=self.interval, callAtMicOff=self.cancelMode) # time.sleep(self.sleepTime/1000) # sleep 10 milliseconds if self.Hit == self.MaxHit: @@ -118,18 +124,51 @@ def toggleMicrophone(self): TestGrammar.__test__ = False -def testSingleTimerClassic(): +# 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="single") + 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, debug=debug) + natlinktimer.setTimerCallback(testGram.doTimerClassic, interval=testGram.interval, callAtMicOff=testGram.cancelMode, debug=debug) ## 1 timer active: - assert natlinktimer.getNatlinktimerStatus() == 1 for _ in range(5): if testGram.Hit >= testGram.MaxHit: break @@ -139,14 +178,10 @@ def testSingleTimerClassic(): 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 len(testGram.results) == testGram.toggleMicAt assert natlinktimer.getNatlinktimerStatus() == 0 ## natlinktimer is NOT destroyed after last timer is gone. - finally: - del natlinktimer.natlinktimer natlinktimer.stopTimerCallback() natlink.natDisconnect() From c2b80792641b130cfc5b5b6aceeef929d17da98d Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Sun, 18 Dec 2022 15:28:26 +0100 Subject: [PATCH 12/20] natlinktimer.py passes test --- src/natlinkcore/natlinktimer.py | 13 +++++++++++++ tests/test_natlinktimer.py | 27 ++++----------------------- 2 files changed, 17 insertions(+), 23 deletions(-) diff --git a/src/natlinkcore/natlinktimer.py b/src/natlinkcore/natlinktimer.py index b7e31ed..6db5873 100644 --- a/src/natlinkcore/natlinktimer.py +++ b/src/natlinkcore/natlinktimer.py @@ -88,6 +88,7 @@ def __init__(self, minInterval=None): 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): @@ -145,6 +146,12 @@ def removeCallback(self, callback, debug=None): 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: del self.callbacks[callback] except KeyError: @@ -255,6 +262,12 @@ def hittimer(self): print(f"total time spent hittimer: {totaltime}") finally: self.in_timer = False + if self.timers_to_stop: + print(f'stop timers: {self.timers_to_stop}') + for callback in self.timers_to_stop: + self.removeCallback(callback) + self.timers_to_stop = set() + def stopTimer(self): diff --git a/tests/test_natlinktimer.py b/tests/test_natlinktimer.py index 159c6d4..7fbdb97 100644 --- a/tests/test_natlinktimer.py +++ b/tests/test_natlinktimer.py @@ -58,7 +58,8 @@ def doTimerClassic(self): self.Hit +=1 if self.Hit: if self.toggleMicAt and self.toggleMicAt == self.Hit: - self.toggleMicrophone() + ## 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 @@ -90,29 +91,7 @@ def doTimer(self): return newInterval return None - def doTimerMicToggle(self): - """this doTimer toggles the microphone when halfway - - """ - 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 doTimerMicToggle {self.name}: {expectElapsed} milliseconds') - natlinktimer.removeTimerCallback(self.doTimer) - if self.Hit >= self.maxHit/2: - print(f'toggle mic at {relTime}') - self.toggleMicrophone() - return None - ## try to shorten interval: - currentInterval = self.grammarTimer.interval - if currentInterval > 250: - newInterval = currentInterval - 25 - return newInterval - return None def toggleMicrophone(self): micstate = natlink.getMicState() @@ -170,6 +149,8 @@ def testStopAtMicOff(): 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 From 3a0681d71ed5d17a81beb3adaef92414e297a7d6 Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Tue, 20 Dec 2022 15:10:04 +0100 Subject: [PATCH 13/20] tidying up some print lines natlinktimer.py, seems to work well now. --- src/natlinkcore/natlinktimer.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/src/natlinkcore/natlinktimer.py b/src/natlinkcore/natlinktimer.py index 6db5873..71ad485 100644 --- a/src/natlinkcore/natlinktimer.py +++ b/src/natlinkcore/natlinktimer.py @@ -116,10 +116,12 @@ def on_mic_off_callback(self): """ to_stop = [(cb, gt) for (cb, gt) in self.callbacks.items() if gt.callAtMicOff] if not to_stop: - print('natlinktimer: no timers to stop') + if self.debug: + print('natlinktimer: no timers to stop') return for cb, gt in to_stop: - print(f'natlinktimer: stopping {cb}, {gt}') + if self.debug: + print(f'natlinktimer: stopping {cb}, {gt}') self.removeCallback(cb) def addCallback(self, callback, interval, callAtMicOff=False, maxIterations=None, debug=None): @@ -147,12 +149,13 @@ def removeCallback(self, callback, debug=None): 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') + # 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 @@ -263,7 +266,11 @@ def hittimer(self): finally: self.in_timer = False if self.timers_to_stop: - print(f'stop timers: {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() From 5b7c700001c78360d81dbb10044400e443432e2a Mon Sep 17 00:00:00 2001 From: Doug Ransom Date: Wed, 21 Dec 2022 10:10:22 -0800 Subject: [PATCH 14/20] checkpoint --- pyproject.toml | 4 +++ src/natlinkcore/SampleMacros/__init__.py | 2 ++ .../configure/natlink_extensions.py | 26 +++++++++++++++++++ 3 files changed, 32 insertions(+) create mode 100644 src/natlinkcore/SampleMacros/__init__.py create mode 100644 src/natlinkcore/configure/natlink_extensions.py 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/configure/natlink_extensions.py b/src/natlinkcore/configure/natlink_extensions.py new file mode 100644 index 0000000..95d1fbf --- /dev/null +++ b/src/natlinkcore/configure/natlink_extensions.py @@ -0,0 +1,26 @@ +"""Command Line Program to List Advertised Natlink Exensions.""" +import sys +from importlib.metadata import entry_points +import argparse + +def main(): + parser=argparse.ArgumentParser(description="Enumerate natlink extension momdules.") + args=parser.parse_args() + + 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 + + print(f"\n{n} {path}") + return 0 + +if '__main__' == __name__: + main() \ No newline at end of file From 8be8ce93f0b035c6d800f6d95026cf4bc3227304 Mon Sep 17 00:00:00 2001 From: Doug Ransom Date: Wed, 21 Dec 2022 10:13:45 -0800 Subject: [PATCH 15/20] removed incorrect return statement --- src/natlinkcore/configure/natlink_extensions.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/natlinkcore/configure/natlink_extensions.py b/src/natlinkcore/configure/natlink_extensions.py index 95d1fbf..24bb2b9 100644 --- a/src/natlinkcore/configure/natlink_extensions.py +++ b/src/natlinkcore/configure/natlink_extensions.py @@ -20,7 +20,7 @@ def main(): path = e print(f"\n{n} {path}") - return 0 + return 0 if '__main__' == __name__: main() \ No newline at end of file From 5cf9d32b206f6c2ced8a676c3bbdc994a05e80bf Mon Sep 17 00:00:00 2001 From: Doug Ransom Date: Wed, 21 Dec 2022 10:14:37 -0800 Subject: [PATCH 16/20] removed useless blank line --- src/natlinkcore/configure/natlink_extensions.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/natlinkcore/configure/natlink_extensions.py b/src/natlinkcore/configure/natlink_extensions.py index 24bb2b9..c02ac9f 100644 --- a/src/natlinkcore/configure/natlink_extensions.py +++ b/src/natlinkcore/configure/natlink_extensions.py @@ -19,7 +19,7 @@ def main(): except Exception as e: path = e - print(f"\n{n} {path}") + print(f"{n} {path}") return 0 if '__main__' == __name__: From 54b64e058182e5c64008aa94476738ccea11b2ce Mon Sep 17 00:00:00 2001 From: Doug Ransom Date: Wed, 21 Dec 2022 10:36:15 -0800 Subject: [PATCH 17/20] added extension enumeration to config cli --- src/natlinkcore/configure/__init__.py | 1 + src/natlinkcore/configure/natlink_extensions.py | 13 ++++++++----- src/natlinkcore/configure/natlinkconfig_cli.py | 12 +++++++++++- 3 files changed, 20 insertions(+), 6 deletions(-) create mode 100644 src/natlinkcore/configure/__init__.py diff --git a/src/natlinkcore/configure/__init__.py b/src/natlinkcore/configure/__init__.py new file mode 100644 index 0000000..6efb209 --- /dev/null +++ b/src/natlinkcore/configure/__init__.py @@ -0,0 +1 @@ +from 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 index c02ac9f..2c44925 100644 --- a/src/natlinkcore/configure/natlink_extensions.py +++ b/src/natlinkcore/configure/natlink_extensions.py @@ -3,10 +3,7 @@ from importlib.metadata import entry_points import argparse -def main(): - parser=argparse.ArgumentParser(description="Enumerate natlink extension momdules.") - args=parser.parse_args() - +def extensions_and_folders(): discovered_extensions=entry_points(group='natlink_extensions') for extension in discovered_extensions: @@ -17,8 +14,14 @@ def main(): path=pathfn() except Exception as e: - path = 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 diff --git a/src/natlinkcore/configure/natlinkconfig_cli.py b/src/natlinkcore/configure/natlinkconfig_cli.py index ff63282..1e09460 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 natlink_extensions 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 From 703d0915ef51d5464cb480d8204ccb2bb1853d54 Mon Sep 17 00:00:00 2001 From: Doug Ransom Date: Wed, 21 Dec 2022 10:47:55 -0800 Subject: [PATCH 18/20] tweaked imports --- src/natlinkcore/configure/__init__.py | 2 +- src/natlinkcore/configure/natlinkconfig_cli.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/natlinkcore/configure/__init__.py b/src/natlinkcore/configure/__init__.py index 6efb209..09c00fa 100644 --- a/src/natlinkcore/configure/__init__.py +++ b/src/natlinkcore/configure/__init__.py @@ -1 +1 @@ -from natlink_extensions import extensions_and_folders \ No newline at end of file +from natlinkcore.configure.natlink_extensions import extensions_and_folders \ No newline at end of file diff --git a/src/natlinkcore/configure/natlinkconfig_cli.py b/src/natlinkcore/configure/natlinkconfig_cli.py index 1e09460..7bb074e 100644 --- a/src/natlinkcore/configure/natlinkconfig_cli.py +++ b/src/natlinkcore/configure/natlinkconfig_cli.py @@ -4,7 +4,7 @@ import cmd import os import os.path -from natlink_extensions import extensions_and_folders +from natlinkcore.configure import extensions_and_folders from natlinkcore.configure import natlinkconfigfunctions From 26796a6025bac1422e75c178b3d25a4929365550 Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Wed, 21 Dec 2022 20:09:15 +0100 Subject: [PATCH 19/20] improve natlinktimer, when mic switches off! --- src/natlinkcore/natlinktimer.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/natlinkcore/natlinktimer.py b/src/natlinkcore/natlinktimer.py index 71ad485..1140753 100644 --- a/src/natlinkcore/natlinktimer.py +++ b/src/natlinkcore/natlinktimer.py @@ -114,6 +114,7 @@ def getnow(self): 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: @@ -122,7 +123,7 @@ def on_mic_off_callback(self): for cb, gt in to_stop: if self.debug: print(f'natlinktimer: stopping {cb}, {gt}') - self.removeCallback(cb) + gt.callAtMicOff() def addCallback(self, callback, interval, callAtMicOff=False, maxIterations=None, debug=None): """add an interval From fdc97aeffc65b98df57f7951ae226211862c36c3 Mon Sep 17 00:00:00 2001 From: Quintijn Hoogenboom Date: Thu, 22 Dec 2022 16:15:34 +0100 Subject: [PATCH 20/20] check lines in the cancelMode routine of test grammar of natlinktimer. --- tests/test_natlinktimer.py | 61 +++++++++++++++++++++++++++++++++++++- 1 file changed, 60 insertions(+), 1 deletion(-) diff --git a/tests/test_natlinktimer.py b/tests/test_natlinktimer.py index 7fbdb97..85cb9ec 100644 --- a/tests/test_natlinktimer.py +++ b/tests/test_natlinktimer.py @@ -44,9 +44,15 @@ def resetExperiment(self): 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 @@ -102,7 +108,58 @@ def toggleMicrophone(self): 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() @@ -161,6 +218,8 @@ def testStopAtMicOff(): 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()