Python Cron Task (Measure Internet Speed)
Let's measure internet speed with Python and learn some cron basics
One evening I had poor internet and I decided to remind how to run periodic tasks on Python. First let's make a small script that will measure download and upload speed and save this data to csv file. With a little help of generative AI I had this script.
import logging
import os
import pandas as pd
from config import CSV_FILENAME, LOG_FILENAME, STORE_PATH
from speedtest import Speedtest
def _add_log() -> None:
log = logging.getLogger(__name__)
log.setLevel(logging.INFO)
if not os.path.exists(STORE_PATH):
os.makedirs(STORE_PATH)
file_handler = logging.FileHandler(LOG_FILENAME)
stream_handler = logging.StreamHandler()
file_handler.setLevel(logging.WARNING)
stream_handler.setLevel(logging.INFO)
logging_format = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(logging_format)
stream_handler.setFormatter(logging_format)
log.addHandler(file_handler)
log.addHandler(stream_handler)
def humansize(nbytes: int) -> str:
suffixes = ['b', 'Kb', 'Mb', 'Gb', 'Tb', 'Pb']
i = 0
while nbytes >= 1024 and i <= len(suffixes)-1:
nbytes /= 1024
i += 1
return f"{nbytes:.2f}{suffixes[i]}ps"
def main() -> None:
log = logging.getLogger(__name__)
log.info('Start measure')
st = Speedtest()
ds = humansize(st.download())
us = humansize(st.upload())
recorded_time = pd.Timestamp.now()
if not os.path.exists(CSV_FILENAME):
log.warn(
f"File '{CSV_FILENAME}' not found. Creating a new CSV file.")
columns = ["Datetime", "Download", "Upload"]
df = pd.DataFrame(columns=columns)
else:
df = pd.read_csv(CSV_FILENAME, parse_dates=True)
new_data = {"Datetime": recorded_time, "Download": ds, "Upload": us}
df.loc[recorded_time] = new_data
df.to_csv(CSV_FILENAME, index=False)
log.info('Done')
log.info(f"{new_data}")
if __name__ == "__main__":
log = logging.getLogger(__name__)
_add_log()
try:
main()
except Exception as e:
log.error(e)
Lets look it closely. This is a helper function to store errors in file due to our process will be running in crontab we will not see output of our script.
def _add_log() -> None:
log = logging.getLogger(__name__)
log.setLevel(logging.INFO)
if not os.path.exists(STORE_PATH):
os.makedirs(STORE_PATH)
file_handler = logging.FileHandler(LOG_FILENAME)
stream_handler = logging.StreamHandler()
file_handler.setLevel(logging.WARNING)
stream_handler.setLevel(logging.INFO)
logging_format = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(logging_format)
stream_handler.setFormatter(logging_format)
log.addHandler(file_handler)
log.addHandler(stream_handler)
Next one is humanize of number. Like 1024 will be 1.00kbps
def humansize(nbytes: int) -> str:
suffixes = ['b', 'Kb', 'Mb', 'Gb', 'Tb', 'Pb']
i = 0
while nbytes >= 1024 and i <= len(suffixes)-1:
nbytes /= 1024
i += 1
return f"{nbytes:.2f}{suffixes[i]}ps"
And here we measure download and upload speed. Using pandas read file and add new line of our result with datetime. I install lib calles speedtest-cli to measure internet speed.
def main() -> None:
log = logging.getLogger(__name__)
log.info('Start measure')
st = Speedtest()
ds = humansize(st.download())
us = humansize(st.upload())
recorded_time = pd.Timestamp.now()
if not os.path.exists(CSV_FILENAME):
log.warn(
f"File '{CSV_FILENAME}' not found. Creating a new CSV file.")
columns = ["Datetime", "Download", "Upload"]
df = pd.DataFrame(columns=columns)
else:
df = pd.read_csv(CSV_FILENAME, parse_dates=True)
new_data = {"Datetime": recorded_time, "Download": ds, "Upload": us}
df.loc[recorded_time] = new_data
df.to_csv(CSV_FILENAME, index=False)
log.info('Done')
log.info(f"{new_data}")
Almost all. You should create config.py where you will describe path for logs and .csv file. And now lets create periodic job using cron.
crontab -e
This will open editon for your cron jobs. Add a new line using cron syntacs
* * * * * <venv/bin/python> <path>/main.py
And that is it, easy
For more please visit Cron Doc