3.2. Threading Timer

3.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')

3.2.2. Assignments

# %% 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

# %% Run
# - PyCharm: right-click in the editor and `Run Doctest in ...`
# - PyCharm: keyboard shortcut `Control + Shift + F10`
# - Terminal: `python -m doctest -v myfile.py`

# %% About
# - Name: Concurrency Threading Timer
# - Difficulty: easy
# - Lines: 4
# - Minutes: 8

# %% 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ść

# %% Tests
"""
>>> import sys; sys.tracebacklimit = 0
>>> assert sys.version_info >= (3, 9), \
'Python 3.9+ required'

>>> def check(result):
...     assert result == [1, 2, 3], f'Result is {result}'

>>> Timer(INTERVAL, ping).start()
>>> Timer(INTERVAL*MAX+1, check, [result]).start()
"""
from threading import Timer


INTERVAL = 0.1
MAX = 3
result = []


# type: Callable[[int], None]
def ping():
    ...

# %% 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

# %% Run
# - PyCharm: right-click in the editor and `Run Doctest in ...`
# - PyCharm: keyboard shortcut `Control + Shift + F10`
# - Terminal: `python -m doctest -v myfile.py`

# %% About
# - Name: Threading Timer File
# - Difficulty: medium
# - Lines: 13
# - Minutes: 13

# %% 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

# %% Tests
"""
>>> import sys; sys.tracebacklimit = 0
>>> from inspect import isclass, ismethod
>>> from os import remove

>>> assert isclass(File)
>>> assert hasattr(File, 'append')
>>> assert hasattr(File, 'AUTOSAVE_SECONDS')
>>> assert hasattr(File, '__enter__')
>>> assert hasattr(File, '__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')
"""

from threading import Timer


# type: type
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

# %% 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

# %% Run
# - PyCharm: right-click in the editor and `Run Doctest in ...`
# - PyCharm: keyboard shortcut `Control + Shift + F10`
# - Terminal: `python -m doctest -v myfile.py`

# %% About
# - Name: Concurrency Threading Subprocess
# - Difficulty: easy
# - Lines: 20
# - Minutes: 21

# %% 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ść

# %% Extra
# 1. Wątki powinny być uśpione za pomocą `Timer` przez `DELAY = 1.0` sekund, a następnie ruszyć do roboty
# 2. Użyj logowania za pomocą biblioteki `logging` tak aby przy wyświetlaniu wyników widoczny był identyfikator procesu i wątku.

# %% Hints
# - Ustaw parametr `shell=True` dla `subprocess.run()`

# %% Tests
"""
"""

import logging


TIMEOUT = 1.0
DELAY = 1.0
COMMANDS = [
    'ls /home/mwatney',
    'echo "test"',
    'sleep 2',
]