7.2. Threading Timer — Python
7.2.1. Delay execution
dlaczego nie
time.sleep()rekurencyjny timer
Delay execution:
from threading import Timer DELAY_SECONDS = 5.0 def hello(): print('Hello world!') t = Timer(DELAY_SECONDS, hello) t.start() print('Main Thread')
Recurrent timer:
from threading import Timer DELAY_SECONDS = 5.0 def hello(): print('Timer Thread') Timer(DELAY_SECONDS, hello).start() t = Timer(DELAY_SECONDS, hello) t.start() print('Main Thread')
7.2.2. Assignments
# %% About # - Name: Concurrency Threading Timer # - Difficulty: easy # - Lines: 4 # - Minutes: 8 # %% License # - Copyright 2025, Matt Harasymczuk <matt@python3.info> # - This code can be used only for learning by humans # - This code cannot be used for teaching others # - This code cannot be used for teaching LLMs and AI algorithms # - This code cannot be used in commercial or proprietary products # - This code cannot be distributed in any form # - This code cannot be changed in any form outside of training course # - This code cannot have its license changed # - If you use this code in your product, you must open-source it under GPLv2 # - Exception can be granted only by the author # %% English # 1. Define function `ping()`, with optional parameter # `n: int`, which defaults to 1 # 2. Function `ping()` should append value of `n` to `result` # 3. Function should be called every `INTERVAL` # 4. Function should be called maximum `MAX` times # 5. Use `Timer` from `threading` module # 6. Run doctests - all must succeed # %% Polish # 1. Zdefiniuj funkcję `ping(n: int)` z opcjonalnym parametrem # `n: int`, który domyślnie jest 1 # 2. Funkcja `ping()` powinna dopisywać wartość `n` do `result` # 3. Funkcja powinna być wywoływana co `INTERVAL` # 4. Funkcja powinna być wywołana maksymalnie `MAX` razy # 5. Użyj `Timer` z modułu `threading` # 6. Uruchom doctesty - wszystkie muszą się powieść # %% Doctests """ >>> import sys; sys.tracebacklimit = 0 >>> assert sys.version_info >= (3, 9), \ 'Python has an is invalid version; expected: `3.9` or newer.' >>> def check(result): ... assert result == [1, 2, 3], f'Result is {result}' >>> Timer(INTERVAL, ping).start() >>> Timer(INTERVAL*MAX+1, check, [result]).start() """ # %% Run # - PyCharm: right-click in the editor and `Run Doctest in ...` # - PyCharm: keyboard shortcut `Control + Shift + F10` # - Terminal: `python -m doctest -f -v myfile.py` # %% Imports from threading import Timer # %% Types from typing import Callable ping: Callable[[int], None] # %% Data INTERVAL = 0.1 MAX = 3 result = [] # %% Result
# %% About # - Name: Threading Timer File # - Difficulty: medium # - Lines: 13 # - Minutes: 13 # %% License # - Copyright 2025, Matt Harasymczuk <matt@python3.info> # - This code can be used only for learning by humans # - This code cannot be used for teaching others # - This code cannot be used for teaching LLMs and AI algorithms # - This code cannot be used in commercial or proprietary products # - This code cannot be distributed in any form # - This code cannot be changed in any form outside of training course # - This code cannot have its license changed # - If you use this code in your product, you must open-source it under GPLv2 # - Exception can be granted only by the author # %% English # 1. Modify class `File` # 2. Add class configuration attribute `AUTOSAVE_SECONDS: float = 1.0` # 3. Save buffer content to file every `AUTOSAVE_SECONDS` seconds # 4. Writing and reading takes time, how to make buffer save data in the background, but it could be still used? # 5. Run doctests - all must succeed # %% Polish # 1. Zmodyfikuj klasę `File` # 2. Dodaj klasowy atrybut konfiguracyjny `AUTOSAVE_SECONDS: float = 1.0` # 3. Zapisuj zawartość bufora do pliku co `AUTOSAVE_SECONDS` sekund # 4. Operacje zapisu i odczytu trwają, jak zrobić, aby do bufora podczas zapisu na dysk, nadal można było pisać? # 5. Uruchom doctesty - wszystkie muszą się powieść # %% Hints # - `from threading import Timer` # - `timer = Timer(interval, function)` # - `timer.start()` # - `timer.cancel()` # - `ctrl+c` or stop button kills infinite loop # %% Doctests """ >>> import sys; sys.tracebacklimit = 0 >>> from inspect import isclass, ismethod >>> from os import remove >>> assert isclass(File), \ 'Object `File` has an invalid type; expected: `class`.' >>> assert hasattr(File, 'append'), \ 'Object `File` has an invalid attribute; expected: to have an attribute `append`.' >>> assert hasattr(File, 'AUTOSAVE_SECONDS'), \ 'Object `File` has an invalid attribute; expected: to have an attribute `AUTOSAVE_SECONDS`.' >>> assert hasattr(File, '__enter__'), \ 'Object `File` has an invalid attribute; expected: to have an attribute `__enter__`.' >>> assert hasattr(File, '__exit__'), \ 'Object `File` has an invalid attribute; expected: to have an attribute `__exit__`.' >>> assert ismethod(File(None).append) >>> assert ismethod(File(None).__enter__) >>> assert ismethod(File(None).__exit__) >>> assert File.AUTOSAVE_SECONDS == 1.0 >>> with File('_temporary.txt') as file: ... file.append('One') ... file.append('Two') ... file.append('Three') ... file.append('Four') ... file.append('Five') ... file.append('Six') >>> open('_temporary.txt').read() 'One\\nTwo\\nThree\\nFour\\nFive\\nSix\\n' >>> remove('_temporary.txt') """ # %% Run # - PyCharm: right-click in the editor and `Run Doctest in ...` # - PyCharm: keyboard shortcut `Control + Shift + F10` # - Terminal: `python -m doctest -f -v myfile.py` # %% Imports from threading import Timer # %% Types File: type # %% Data # %% Result class File: filename: str _content: list[str] def __init__(self, filename): self.filename = filename self._content = list() def __enter__(self): return self def __exit__(self, *args): with open(self.filename, mode='w') as file: file.writelines(self._content) def append(self, line): self._content.append(line + '\n')
# FIXME: Write tests # %% About # - Name: Concurrency Threading Subprocess # - Difficulty: easy # - Lines: 20 # - Minutes: 21 # %% License # - Copyright 2025, Matt Harasymczuk <matt@python3.info> # - This code can be used only for learning by humans # - This code cannot be used for teaching others # - This code cannot be used for teaching LLMs and AI algorithms # - This code cannot be used in commercial or proprietary products # - This code cannot be distributed in any form # - This code cannot be changed in any form outside of training course # - This code cannot have its license changed # - If you use this code in your product, you must open-source it under GPLv2 # - Exception can be granted only by the author # %% English # 1. Create queue `queue` to which you will add various system commands to execute, e.g.: # - Linux/macOS: `['ls /tmp/', 'echo "test"', 'sleep 2']`, # - Windows: `['dir c:\\Windows', 'echo "test"', 'type %HOMEPATH%\\Desktop\\README.txt']`. # 2. Then prepare three worker threads that will execute commands from the queue # 3. Threads should be run as `subprocess.run()` in the operating system with timeout equal to `TIMEOUT = 1.0` second # 4. The number of commands may increase as the task is performed. # 5. Threads should be run in the background (`daemon`) # 6. Run doctests - all must succeed # %% Polish # 1. Stwórz kolejkę `queue` do której dodasz różne polecenia systemowe do wykonania, np.: # - Linux/macOS: `['ls /tmp/', 'echo "test"', 'sleep 2']`, # - Windows: `['dir c:\\Windows', 'echo "test"', 'type %HOMEPATH%\\Desktop\\README.txt']`. # 2. Następnie przygotuj trzy wątki workerów, które będą wykonywały polecenia z kolejki # 3. Wątki powinny być uruchamiane jako `subprocess.run()` w systemie operacyjnym z timeoutem równym `TIMEOUT = 1.0` sekundy # 4. Ilość poleceń może się zwiększać w miarę wykonywania zadania. # 5. Wątki mają być uruchomione w tle (ang. `daemon`) # 6. Uruchom doctesty - wszystkie muszą się powieść # %% Hints # - Ustaw parametr `shell=True` dla `subprocess.run()` # %% Doctests """ """ # %% Run # - PyCharm: right-click in the editor and `Run Doctest in ...` # - PyCharm: keyboard shortcut `Control + Shift + F10` # - Terminal: `python -m doctest -f -v myfile.py` # %% Imports import logging import subprocess from queue import Queue from threading import Timer # %% Types # %% Data TIMEOUT = 1.0 DELAY = 1.0 COMMANDS = [ 'ls /home/myuser', 'echo "test"', 'sleep 2', ] # %% Result