Отправьте чириканье интернет-провайдеру при снижении скорости интернета
Я написал приложение на Python 3, которое контролирует скорость интернета и отправит чириканье интернет-провайдеру, когда скорость падает слишком низко. Приложение имеет конфигурационный файл, в котором могут быть настроены ISP и целевые скорости.
Программа имеет два потока одновременно. SpeedTestThread
отслеживает скорость интернета с установленными интервалами, используя модуль python speedtest-cli . TwitterThread
использует tweepy для отправки твитов, содержащих данные со скоростью тесты.
Когда SpeedTestThread
получает скорость ниже порогового значения, данные переносятся в tweet_data_queue и устанавливается глобальный tweetFlag
. TwitterThread
отслеживает tweetFlag
и, когда установлен, извлекает данные из очереди и генерирует твит, используя его.
Я удалил docstrings, чтобы сэкономить место; их можно найти в моем репозитории Github .
speed_test.py:
import speedtest
import json
import time
import csv
import os
import threading
import queue
import tweepy
import random
config = json.load(open('config.json'))
exitFlag = 0
tweetFlag = 0
# Global queue for tweet data, shared between threads
tweet_data_queue = queue.Queue()
def main():
error_logger = ErrorLogger(config['errorFilePath'])
test_thread = SpeedTestThread(1, "SpeedTestThread1", error_logger)
tweet_thread = TwitterThread(2, "TwitterThread1", error_logger)
test_thread.start()
tweet_thread.start()
class SpeedTestThread(threading.Thread):
def __init__(self, thread_id, name, error_logger):
threading.Thread.__init__(self)
self.name = name
self.thread_id = thread_id
self.s = speedtest.Speedtest()
self.targetSpeeds = config['internetSpeeds']
self.dataLogger = ErrorLogger(config['logFilePath'])
self.error_logger = error_logger
def run(self):
global exitFlag
prevError = False
while exitFlag == 0:
try:
results = self.getSpeeds()
self.checkSpeeds(results)
self.dataLogger.logCsv(results)
except Exception as e:
error = {"time": time.ctime(),
"error": "Unable to retrieve results",
"exception": e}
self.error_logger.logError(error)
prevError = True
if prevError:
self.error_logger.counter = 0
time.sleep(config['testFreq'])
def getSpeeds(self):
self.s.get_best_server()
self.s.upload()
self.s.download()
return self.s.results.dict()
def checkSpeeds(self, results):
global tweetFlag
down = results['download']
up = results['upload']
ping = results['ping']
if (down / (2**20) < self.targetSpeeds['download'] or
up / (2**20) < self.targetSpeeds['upload'] or
ping > self.targetSpeeds['ping']):
print("Unnaceptable speed results:\n"
"Download: %s\n"
"Upload: %s\n"
"Ping: %s\n" % (down, up, ping))
tweetFlag = 1
tweet_data_queue.put(results)
print("Results queued for tweet")
class TwitterThread(threading.Thread):
def __init__(self, thread_id, name, error_logger):
threading.Thread.__init__(self)
self.thread_id = thread_id
self.name = name
self.error_logger = error_logger
# Set up tweepy with twitter API authentication
self.apiData = config['twitterAPI']
auth = tweepy.OAuthHandler(
self.apiData['apiKey'], self.apiData['apiSecret'])
auth.set_access_token(self.apiData['accessToken'],
self.apiData['accessTokenSecret'])
self.twitterAPI = tweepy.API(auth)
def run(self):
global exitFlag
global tweetFlag
prevError = False
while True:
if exitFlag == 1:
break
if tweetFlag == 1:
tweet = self.getTweet()
try:
self.twitterAPI.update_status(tweet)
print("Tweet successful")
except Exception as e:
error = {"time": time.ctime(),
"error": "Unable to send tweet",
"exception": e}
self.error_logger.logError(error)
prevError = True
if prevError:
self.error_logger.counter = 0
if tweet_data_queue.qsize() == 0:
tweetFlag = 0
def getTweet(self):
data = tweet_data_queue.get()
down = round(data['download'] / (2**20), 2)
up = round(data['upload'] / (2**20), 2)
content = random.choice(config['tweetContent'])
return content.format(config['ispTwitter'], down, up)
class Logger(object):
def __init__(self, filepath):
self.filepath = filepath
def logCsv(self, data):
print("Logging ...")
with open(self.filepath, 'a') as f:
writer = csv.DictWriter(f, fieldnames=data.keys())
if os.stat(self.filepath).st_size == 0:
writer.writeheader()
writer.writerow(data)
print("Done -> '%s'" % self.filepath)
class ErrorLogger(Logger):
def __init__(self, filepath):
Logger.__init__(self, filepath)
self.counter = 0
def logError(self, errorData):
global exitFlag
if self.counter >= config['testAttempts']:
exitFlag = 1
errorData['error'] = "10 Failed test attempts, exiting."
self.counter = 0
print(errorData['error'])
self.logCsv(errorData)
4 ответа
В целом это неплохой модуль. Вот несколько вопросов юзабилити /nitpicks:
-
При запуске скрипта нет простого способа остановить его. CTRL + C не работает, мне нужно вручную убить процесс. Вероятно, это связано с тем, как обрабатывает
threading.Thread
, но я не уверен. -
Было бы неплохо, если бы было немного проще добавить другой обработчик, чем твитер. Например, я, возможно, захочу отправить мне электронное письмо, а не сразу подправить его. Чтобы исправить это, вы должны удалить все упоминания твитов из класса
SpeedTestThread
и, возможно, определить общий классEventHandler
, из которого вызываетсяTwitterThread
, который вызывает некоторый методaction(self, up, down)
, который должен быть определен в производном классе. Таким образом вы можете легко определитьMailThread(EventHandler)
с помощью другого методаaction
, который отправляет почту вместо этого. -
Вы смешиваете два стиля именования переменных,
camelCase
иlower_case_with_underscores
. Официальный стиль руководства Python, PEP8 , рекомендует придерживаться последнего для всех переменных и функции /методы. Глобальные константы должны быть вUPPER_CASE
. -
Для флагов вам нужно просто использовать
True
иFalse
, читать его немного легче. Обратите внимание, что вы можете сделатьwhile not exitFlag:
вместоwhile True: if exitFlag == 1: break
иif prevError
вместоif prevError == 1
. -
Вы никогда не увеличиваете свой
error_logger.counter
, он должен бытьif prevError: self.error_logger.counter += 1 else: self.error_logger.counter = 0
.
Это скорее обзор дизайна /алгоритма /архитектуры, чем код. Есть некоторые важные проблемы, которые другие ответы не рассматривались.
Прежде всего, трата пропускной способности сети, использующая тест скорости на повторе, кажется плохой идеей. Он будет заполнять ваше соединение каждый час (по умолчанию), поэтому вам больно лично, если вы когда-то делаете что-то в Интернете в то время. Кроме того, это добавляет к перегрузке для других пользователей вашего интернет-провайдера.
В вашем файле конфигурации в репозитории github ваши строки твитового формата по умолчанию являются слишком агрессивными и агрессивными WAY для чего-то, что может быть не ошибкой вашего интернет-провайдера:
"tweetContent": [ "{0}! I'm meant to get 52mb/s down, 10mb/s up. I got {1}mb/s down, {2}mb/s up!", "Hey {0}, think {1}mb/s down, {2}mb/s up instead of 52mb/s down, 10mb/s up is ok - it's not!", "Don't break your promise {0}. {1}mb/s down, {2}mb/s up != 52mb/s down, 10mb/s up", "{0}, how do I Netflix as expected with {1}mb/s down, {2}mb/s instead of 52mb/s down, 10mb/s up?" ]
Результаты теста скорости могут зависеть от перегруженности в других местах сети или на серверах проверки скорости. Даже скопление из ваших собственных загрузок (или просмотр Netflix), которое происходит во время выполнения теста скорости, уменьшит измеренное значение.
Разумным сообщением может быть: "Automated speed-test got {1}M down, {2}M up, much lower than {0}'s expected 52M down 10M up speed."
Это просто явное изложение факта и оставляет открытым то, что что-то пошло не так с тестом. Но люди, читающие это, все равно получат сообщение. В некотором смысле, нейтральное представление фактов more может быть воспринято всерьез, чем обвинения в обезвреживании или нытье о Netflix.
Ваш тест скорости измеряет только собственную полосу пропускания, а не общую пропускную способность вашего подключения к вашему интернет-провайдеру. . Это будет намного лучше, но потребует чего-то на вашем маршрутизаторе, чтобы отслеживать трафик, и способ запросить его из этой программы. (Вы все равно можете использовать тест скорости для генерации трафика и убедиться, что он подтолкнул общую сумму до ожидаемой пропускной способности.)
Ваш цикл while True:
никогда не спит, ожидая установки любого флага. Ожидание в ожидании без сна - огромная трата времени процессора (и электроэнергии).
Я действительно не знаю Python, поэтому у меня нет рекомендаций по использованию, но вам определенно нужен какой-либо языковой параметр /флаг синхронизации, который позволяет вам спать, пока другой поток не изменит его. (Вы могли просто спать в течение 10 секунд, а затем снова проверять переменные, но опрос сосет по сравнению с уведомлением, поддерживаемым ОС.
Вам действительно не нужно многопоточно это делать.
Любой разумный интервал между скоростными тестами достаточно длинный для возврата функции твитов (даже если она не работает). Если вы не хотите включать время, проведенное в трио, в режиме сна между проверкой скорости, вы можете проверить время до /после вызова функции твита и вычесть, что много секунд из интервала ожидания.
Вы можете реализовать такую очередь, как этот псевдокод:
while not exitFlag: # Graipher suggested this loop structure
result = speedtest()
deadline = now() + speedtest_interval
check_result(result, tweet_queue) # logs and adds to queue if slow
if (not tweet_queue.is_empty())
try_tweet(retry_queue, deadline) # loops until queue empty or deadline reached.
sleep(deadline - now())
Это намного проще, чем потоки, но все же может догнать отставание от твитов. Он продолжает повторять попытку, когда он не может чирикать, так же, как и ваша потоковая версия. Мы можем сделать это так просто, потому что не имеет большого значения, что интервал между скоростными тестами - это точно настроенное количество секунд. Быть застрявшим в тайм-аут твит-сообщений в течение дополнительной минуты - это хорошо.
Очередные твиты, которые не могут быть опубликованы, сомнительны.
Вы должны, по крайней мере, сохранить временную метку для отложенных твитов. Если я не пропустил ее, в настоящее время они появятся на твиттере только с момента их публикации, а не с того времени, когда низкая скорость была измерена. И если ваша связь была приостановлена на некоторое время, вы опубликуете поток сердитых твитов, когда вы вернетесь в онлайн (потому что результаты будут равны нулю). Или, может быть, вы справились с тем случаем, когда тест скорости приводит к ошибке вместо низкой скорости, я не выглядел так осторожно.
Я хотел бы сказать только одну (но очень важную) вещь:
Вы разрабатываете вредные привычки раньше! Точка классов для устранения (действительно) global
состояния и управления ею внутри классов и объектов. Вы действительно злоупотребляли использованием global
, который затрудняет сбор кода.
Есть случаи, когда хорошо использовать глобальные переменные, где общая обстановка вокруг объекта повсюду перевешивается неудобствами. Вы должны избегать этого, когда начинаете, пока не почувствуете, когда это уместно.
Один из способов избежать глобальных привязок - использовать явные аргументы для функций, классов, вызываемых для создания одного из их экземпляров и т. д. (это, как правило, самый ясный, поскольку он делает зависимость наиболее явной, когда это возможно и не слишком повторяется).
Что касается вашего последнего комментария, вы правы. Темы, как правило, сложны, и вы не можете ожидать интуитивного понимания порядка, в котором происходят события, когда два, три (или более) потока работают с одним и тем же значением. Язык, компилятор, ОС, процессор ... все могут сыграть свою роль и решить изменить порядок операций для скорости, практичности или любой другой причины.
Правильный способ такого использования - использовать инструменты обмена Python ( locks и друзей), или лучше, обмениваться данными, а не делиться ими.
Я думаю, вам нужно немного подумать о своих вариантах использования.
- Что это должно сделать во время отключения?
- Вы хотите вставить новый твит, если предыдущий еще не отправлен?
- Вы хотите вставить специальное сообщение для этого, например «отключение, обнаруженное в
hh:mm
, разрешено вhh:mm
" (скорректировано при отправке после отключения )? - Должен ли он вообще чирикать?
- Что именно он должен измерять?
Как говорили другие, будьте осторожны, чтобы измерить то, что вы хотите измерить. Если вы смотрите Netflix, пока ваш партнер играет Counter Strike наверху, вы можете столкнуться с своей личной скоростью , пока сеть в порядке.
Посмотрите, можете ли вы учитывать свой собственный трафик. - Что делать, если скорость упала ниже порога?
Вы хотите продолжать измерять с обычным интервалом, вы хотите проверять чаще, реже? Возможно, вы хотите чаще проверять , но tweet , или проверить ответ от своего интернет-провайдера. - Как уже упоминал еще один пользователь, я бы изменил сообщение в твиттере, чтобы уточнить, что это автоматический тест automatic . Если ваш бот пойдет на уступку, по крайней мере, всем ясно, что происходит.