testedevelocidade.py

import time
import speedtest
from datetime import datetime
import sqlite3
import logging
from threading import Timer


logging.basicConfig(format='%(levelname)s - %(asctime)s - %(message)s', level=logging.INFO)

def testevelocidade():
    """
    Função para teste de velocidade.
    """
    s = speedtest.Speedtest()
    data_atual = datetime.now().strftime('%d/%m/%Y')
    hora_atual = datetime.now().strftime('%H:%M')
    logging.info("Coletando dados...")
    velocidade = s.download(threads=None)*(10**-6) # Mbps
    S = Sql()
    S.salva_sql(data_atual,hora_atual,velocidade)
    logging.info("Dados coletados com sucesso!")
    Timer(15,testevelocidade).start() 
   
class Sql():
    '''Classe com métodos para salvar e ler em um banco local (sqlite).
    '''
    def salva_sql(self,*dados):
        '''Método para salvar na base local do SQL
            
        Args:
        *dados: Sequência de inserção na tabela do banco de dados.

        Returns:
         None
        '''
        
        logging.debug("Conectando ao DB local...")
        conn = sqlite3.connect(f'{BANCO}.db')
        logging.debug("Conectado com sucesso!")
        c = conn.cursor()
        
        c.execute(f"""SELECT count(name) FROM sqlite_master WHERE type='table' AND name='{TABELA}' """)
        if c.fetchone()[0]==1 : 
            logging.debug(f'Tabela {TABELA} já existente, nada a fazer.')
        else :
            logging.debug(f'Tabela {TABELA} não existente, criando...')
            try:
                conn.execute(f"""CREATE TABLE {TABELA} 
                (ID INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, 
                DATAATUAL  TEXT  NOT NULL, 
                HORAATUAL  TEXT  NOT NULL, 
                VELOCIDADE REAL);""")
                logging.debug(f'Tabela {TABELA} criada!')
            except Exception as e:
                logging.error(f'Erro ao criar a tabela {TABELA}:" {e}"')
                
        logging.debug("Salvando dados...")
        conn.execute(f"""INSERT INTO {TABELA} 
        (DATAATUAL,HORAATUAL,VELOCIDADE)
        VALUES ('{dados[0]}','{dados[1]}',{dados[2]})""")
        conn.commit()
        logging.info("Dados salvos com sucesso!")
        conn.close()


    def le_sql(self):
        '''Método para ler na base local do SQL e mostrar os últimos 5 valores

        Args:
         None

        Returns:
         None
        '''
        logging.debug("Conectando ao DB local...")
        conn = sqlite3.connect(f'{BANCO}.db')
        logging.debug("Conectado com sucesso!")

        print ("Mostrando os 5 últimos registros:\n")
        cursor = conn.execute(f"SELECT * from {TABELA} order by ID desc LIMIT 5;")
        for row in cursor:
            print ("ID = ", row[0])
            print ("DATA = ", row[1])
            print ("HORA = ", row[2])
            print ("VELOCIDADE(MBps) = ", row[3],'\n----------')

        logging.info("\nOperação realizada com sucesso")
        conn.close()

if __name__ == "__main__":
    BANCO = "dados"
    TABELA = "dadosinternet"
    testevelocidade()
    #S = Sql()
    #S.le_sql()