Merge pull request #3 from FusionSolutions/patch

Update to v0.0.4

Author: iFA88

.github/workflows/python-package.yml

@@ -23,14 +23,17 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install mypy
+        python -m pip install mypy pyflakes
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
     - name: MyPy check
       run: |
-        mypy fsSignal
+        mypy --strict fsSignal
+    - name: PyFlakes check
+      run: |
+        pyflakes fsSignal
     - name: Install package
       run: |
         python setup.py install
-    - name: Unittest
+    - name: Test package
       run: |
-        python -m fsSignal
+        python setup.py test

.github/workflows/python-publish.yml

@@ -26,7 +26,6 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m fsSignal
         pip install build
     - name: Build package
       run: python -m build

fsSignal/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.0.3"
+__version__ = "0.0.4"
 __doc__ = """
 Signal capturer v{}
 Copyright (C) 2021 Fusion Solutions KFT <contact@fusionsolutions.io>
@@ -18,4 +18,4 @@
 """.format(__version__)
 from .fsSignal import KillSignal, SignalIterator, Signal
 
-__all__ = "KillSignal", "SignalIterator", "Signal"
+__all__ = "KillSignal", "SignalIterator", "Signal"

fsSignal/__main__.py

@@ -1,24 +1,32 @@
 # Builtin modules
 from __future__ import annotations
-import os, traceback, unittest, signal as _signal
-from threading import Timer, Event
+import traceback,  signal as _signal
+from threading import Event
 from time import monotonic, sleep
 from typing import Callable, Dict, Any, Iterator, Iterable, Optional, Union
+from types import FrameType
+# Third party modules
 # Local modules
 # Program
+Signals = _signal.Signals
+
 class KillSignal(Exception): pass
 
-class SignalIterator(Iterator):
+class SignalIterator(Iterator[Any]):
 	__slots__ = ("event", "it", "checkDelay", "lastCheck")
-	def __init__(self, event:Event, it:Iterable, checkDelay:float=1.0):
-		self.event:Event = event
-		self.it:Iterator = it.__iter__()
-		self.checkDelay:float = checkDelay
-		self.lastCheck:float = monotonic()
-	def __iter__(self) -> Iterator:
+	event:Event
+	it:Iterator[Any]
+	checkDelay:float
+	lastCheck:float
+	def __init__(self, event:Event, it:Iterable[Any], checkDelay:float=1.0):
+		self.event      = event
+		self.it         = it.__iter__()
+		self.checkDelay = checkDelay
+		self.lastCheck  = monotonic()
+	def __iter__(self) -> Iterator[Any]:
 		return self
 	def __next__(self) -> Any:
-		m:float = monotonic()
+		m = monotonic()
 		if m-self.lastCheck > self.checkDelay:
 			self.lastCheck = m
 			if self.event.is_set():
@@ -28,38 +36,50 @@ def __next__(self) -> Any:
 class BaseSignal:
 	_force:bool
 	@classmethod
-	def check(self) -> bool:
-		if not isinstance(Signal._handler, Signal):
-			return False
-		return Signal._handler._check(self._force)
+	def get(self) -> bool:
+		if isinstance(Signal._handler, Signal):
+			return Signal._handler._get(self._force)
+		return False
 	@classmethod
-	def checkSoft(self) -> bool:
-		if not isinstance(Signal._handler, Signal):
-			return False
-		return Signal._handler._check(False)
+	def getSoft(self) -> bool:
+		if isinstance(Signal._handler, Signal):
+			return Signal._handler._get(False)
+		return False
 	@classmethod
-	def checkHard(self) -> bool:
-		if not isinstance(Signal._handler, Signal):
-			return False
-		return Signal._handler._check(True)
+	def getHard(self) -> bool:
+		if isinstance(Signal._handler, Signal):
+			return Signal._handler._get(True)
+		return False
+	@classmethod
+	def check(self) -> None:
+		if isinstance(Signal._handler, Signal):
+			return Signal._handler._check(self._force)
+	@classmethod
+	def checkSoft(self) -> None:
+		if isinstance(Signal._handler, Signal):
+			return Signal._handler._check(False)
+	@classmethod
+	def checkHard(self) -> None:
+		if isinstance(Signal._handler, Signal):
+			return Signal._handler._check(True)
 	@classmethod
 	def sleep(self, seconds:Union[int, float], raiseOnKill:bool=False) -> None:
-		if not isinstance(Signal._handler, Signal):
-			return sleep(seconds)
-		return Signal._handler._sleep(seconds, raiseOnKill, self._force)
+		if isinstance(Signal._handler, Signal):
+			return Signal._handler._sleep(seconds, raiseOnKill, self._force)
+		return sleep(seconds)
 	@classmethod
-	def signalSoftKill(self, *args, **kwargs) -> None:
+	def signalSoftKill(self, *args:Any, **kwargs:Any) -> None:
 		if isinstance(Signal._handler, Signal):
 			return Signal._handler._signalSoftKill(*args, **kwargs)
 	@classmethod
-	def signalHardKill(self, *args, **kwargs) -> None:
+	def signalHardKill(self, *args:Any, **kwargs:Any) -> None:
 		if isinstance(Signal._handler, Signal):
 			return Signal._handler._signalHardKill(*args, **kwargs)
 	@classmethod
-	def iter(self, it:Iterable, checkDelay:float=1.0) -> Iterable:
-		if not isinstance(Signal._handler, Signal):
-			return it
-		return Signal._handler._iter(it, checkDelay, self._force)
+	def iter(self, it:Iterable[Any], checkDelay:float=1.0) -> Iterable[Any]:
+		if isinstance(Signal._handler, Signal):
+			return Signal._handler._iter(it, checkDelay, self._force)
+		return it
 	@classmethod
 	def softKill(self) -> None:
 		if isinstance(Signal._handler, Signal):
@@ -89,16 +109,22 @@ class HardSignal(BaseSignal):
 	_force:bool = True
 
 class Signal(HardSignal):
+	_handler:Optional[Signal] = None
+	softKillFn:Optional[Callable[[Signals, FrameType], Any]]
+	hardKillFn:Optional[Callable[[Signals, FrameType], Any]]
+	forceKillCounterFn:Optional[Callable[[int, int], Any]]
+	counter:int
+	forceCounter:int
 	eSoft:Event
 	eHard:Event
-	_handler:Optional[Signal] = None
-	def __init__(self, softKillFn:Optional[Callable]=None, hardKillFn:Optional[Callable]=None,
-	forceKillCounterFn:Optional[Callable]=None, forceCounter:int=10):
-		self.softKillFn:Optional[Callable] = softKillFn
-		self.hardKillFn:Optional[Callable] = hardKillFn
-		self.forceKillCounterFn:Optional[Callable] = forceKillCounterFn
-		self.counter:int = 0
-		self.forceCounter:int = forceCounter
+	def __init__(self, softKillFn:Optional[Callable[[Signals, FrameType], Any]]=None,
+	hardKillFn:Optional[Callable[[Signals, FrameType], Any]]=None,
+	forceKillCounterFn:Optional[Callable[[int, int], Any]]=None, forceCounter:int=10):
+		self.softKillFn = softKillFn
+		self.hardKillFn = hardKillFn
+		self.forceKillCounterFn = forceKillCounterFn
+		self.counter = 0
+		self.forceCounter = forceCounter
 		self.eSoft = Event()
 		self.eHard = Event()
 		Signal._handler = self
@@ -123,16 +149,21 @@ def __setstate__(self, states:Dict[str, Any]) -> None:
 	def _activate(self) -> None:
 		_signal.signal(_signal.SIGINT, Signal.signalSoftKill)
 		_signal.signal(_signal.SIGTERM, Signal.signalHardKill)
-	def _check(self, force:bool=True) -> bool:
+	def _get(self, force:bool=True) -> bool:
 		if force:
 			return self.eHard.is_set()
 		return self.eSoft.is_set()
+	def _check(self, force:bool=True) -> None:
+		if (force and self.eHard.is_set()) or (not force and self.eSoft.is_set()):
+			raise KillSignal
+		return None
 	def _sleep(self, seconds:Union[int, float], raiseOnKill:bool=False, force:bool=True) -> None:
 		if (self.eHard if force else self.eSoft).wait(float(seconds)) and raiseOnKill:
 			raise KillSignal
-	def _iter(self, it:Iterable, checkDelay:float=1.0, force:bool=True) -> Iterator:
+		return None
+	def _iter(self, it:Iterable[Any], checkDelay:float=1.0, force:bool=True) -> Iterator[Any]:
 		return SignalIterator(self.eHard if force else self.eSoft, it, checkDelay)
-	def _signalSoftKill(self, *args, **kwargs) -> None:
+	def _signalSoftKill(self, *args:Any, **kwargs:Any) -> None:
 		self._softKill()
 		if not self.eHard.is_set():
 			self.counter += 1
@@ -143,67 +174,26 @@ def _signalSoftKill(self, *args, **kwargs) -> None:
 					traceback.print_exc()
 			if self.counter >= self.forceCounter:
 				self._hardKill()
-	def _signalHardKill(self, *args, **kwargs) -> None:
+	def _signalHardKill(self, *args:Any, **kwargs:Any) -> None:
 		self._softKill()
 		self._hardKill()
 	def _softKill(self) -> None:
 		if not self.eSoft.is_set():
 			self.eSoft.set()
 			if callable(self.softKillFn):
 				try:
-					self.softKillFn()
+					self.softKillFn() # type: ignore
 				except:
 					traceback.print_exc()
 	def _hardKill(self) -> None:
 		if not self.eHard.is_set():
 			self.eHard.set()
 			if callable(self.hardKillFn):
 				try:
-					self.hardKillFn()
+					self.hardKillFn() # type: ignore
 				except:
 					traceback.print_exc()
 	def _reset(self) -> None:
 		self.eSoft.clear()
 		self.eHard.clear()
 		self.counter = 0
-
-class SignalTest(unittest.TestCase):
-	rootSignal:Signal
-	@classmethod
-	def setUpClass(self) -> None:
-		self.rootSignal = Signal()
-	def tearDown(self) -> None:
-		self.rootSignal.reset()
-	def killmeTimer(self) -> None:
-		def suicide():
-			os.kill(os.getpid(), _signal.SIGINT)
-		Timer(1, suicide).start()
-	def test_sleep(self) -> None:
-		t:float = monotonic()
-		self.rootSignal.sleep(2)
-		self.assertGreater(monotonic()-t, 2.0)
-	def test_sleepRaise(self) -> None:
-		self.killmeTimer()
-		with self.assertRaises(KillSignal):
-			self.rootSignal.getSoftSignal().sleep(2, raiseOnKill=True)
-	def test_iter(self) -> None:
-		s:list = list(range(5))
-		d:list = []
-		i:int
-		signal:SoftSignal = self.rootSignal.getSoftSignal()
-		self.killmeTimer()
-		with self.assertRaises(KillSignal):
-			for i in s:
-				signal.sleep(0.5, raiseOnKill=True)
-				d.append(i)
-	def test_hardkill(self) -> None:
-		self.killmeTimer()
-		sleep(0.1)
-		self.killmeTimer()
-		sleep(0.1)
-		self.killmeTimer()
-		sleep(0.1)
-		self.rootSignal.forceCounter = 3
-		with self.assertRaises(KillSignal):
-			self.rootSignal.sleep(10, raiseOnKill=True)
-		self.rootSignal.forceCounter = 10

fsSignal/fsSignal.py

@@ -0,0 +1,65 @@
+# Builtin modules
+import os, unittest, signal as _signal
+from threading import Timer
+from time import monotonic, sleep
+from typing import List
+# Third party modules
+# Local modules
+from .. import Signal, KillSignal
+# Program
+class SignalTest(unittest.TestCase):
+	rootSignal:Signal
+	@classmethod
+	def setUpClass(self) -> None:
+		self.rootSignal = Signal()
+		return None
+	def tearDown(self) -> None:
+		self.rootSignal.reset()
+		return None
+	def killmeTimer(self) -> None:
+		def suicide() -> None:
+			os.kill(os.getpid(), _signal.SIGINT)
+			return None
+		Timer(1, suicide).start()
+		return None
+	def test_sleep(self) -> None:
+		t = monotonic()
+		self.rootSignal.sleep(2)
+		self.assertGreater(monotonic()-t, 2.0)
+		return None
+	def test_sleepRaise(self) -> None:
+		self.killmeTimer()
+		with self.assertRaises(KillSignal):
+			self.rootSignal.getSoftSignal().sleep(2, raiseOnKill=True)
+		return None
+	def test_iter(self) -> None:
+		s = list(range(5))
+		d:List[int] = []
+		signal = self.rootSignal.getSoftSignal()
+		self.killmeTimer()
+		with self.assertRaises(KillSignal):
+			for i in s:
+				signal.sleep(0.5, raiseOnKill=True)
+				d.append(i)
+		return None
+	def test_hardkill(self) -> None:
+		self.killmeTimer()
+		sleep(0.1)
+		self.killmeTimer()
+		sleep(0.1)
+		self.killmeTimer()
+		sleep(0.1)
+		self.rootSignal.forceCounter = 3
+		with self.assertRaises(KillSignal):
+			self.rootSignal.sleep(10, raiseOnKill=True)
+		self.rootSignal.forceCounter = 10
+		return None
+	def test_check(self) -> None:
+		signal = self.rootSignal.getSoftSignal()
+		signal.check()
+		self.assertEqual(signal.get(), False)
+		self.killmeTimer()
+		signal.sleep(2, raiseOnKill=False)
+		with self.assertRaises(KillSignal):
+			signal.check()
+		self.assertEqual(signal.get(), True)

fsSignal/py.typed

@@ -17,12 +17,13 @@
 	author_email = "ifa@fusionsolutions.io",
 	url = "https://github.com/FusionSolutions/python-fssignal",
 	license = "GPL-3",
-	package_dir={"fsSignal": "fsSignal"},
 	packages=["fsSignal"],
 	long_description=open(os.path.join(pwd, "README.md")).read(),
 	long_description_content_type="text/markdown",
 	zip_safe=False,
 	python_requires=">=3.7.0",
+	test_suite="fsSignal.test",
+	package_data={ "":["py.typed"] },
 	classifiers=[ # https://pypi.org/pypi?%3Aaction=list_classifiers
 		"Development Status :: 4 - Beta",
 		"Topic :: Utilities",