12
\$\begingroup\$

I've built a small utility in Python which is going to supposedly send an SMS (using Twilio) each time an earthquake with the magnitude > 4 is to be happening.

NOTE:

This simple script was mainly done for testing purposes and learning activities. DO NOT take the information for granted as there might be big delays between the time the earthquake occurs and the time the SMS arrives.

More, this works only for Romania, Europe and it's using the data from this website which is in a testing phase (BETA).

This should eventually run as a daemon on a UNIX OS.

I'd like to receive any kind of feedback regarding this and what could possibly be improved.

"""
This projects aims to send an SMS to a specific number using Twilio
if an earthquake with magnitude > 4 is going to occur (depending on
your location, this can warn you (best case scenario within 25-30
seconds before you feel the earthquake wave).
"""

import datetime
import json
import re
import time

import requests

from exceptions import EmptyCredentialsFile

from lxml.html import fromstring
from twilio.rest import Client


BASE_URL = 'http://alerta.infp.ro'
DATA_URL = f'{BASE_URL}/server.php'
CREDENTIALS_FILE = 'credentials.json'
DELAY = 1  # seconds


def read_credentials_file(filepath: str) -> dict:
    """Return secrets from `credentials_file` file as a dict.

    The file looks like this:

        {
          "TWILIO_ACCOUNT_SID": "Your twilio account SID",
          "TWILIO_AUTH_TOKEN": "Your twilio account auth token",
          "FROM": "The number from which you'll receive the alert",
          "TO": "The number the message is sent to"
        }


    Args:
        filepath (str): Path to credentials JSON file.

    Returns:
        dict: The return value.
    """

    with open(filepath) as credentials_file:
        credentials = json.load(credentials_file)

    if not credentials:
        raise EmptyCredentialsFile('Credentials file should not be empty.')

    return credentials


def get_credential(name: str) -> str:
    """Return a specific credential value based on its name.

    Args:
        name (str): Name of the desired credential

    Returns:
        str: The return value
    """

    credentials = read_credentials_file(CREDENTIALS_FILE)
    return credentials.get(name)


def get_earthquake_data() -> dict:
    """Get earthquake data from `DATA_URL`.

    Returns:
        dict: A dict containing the following data:
              {
                'mag': '0.1',
                'heart': '2020-01-04 13:30:04 HEARTBEAT',
                'sec': '30',
                'key': 'NjY2NDYyMzAzNjMwNjM2MzM1Mz...=='
              }
    """

    session = requests.Session()
    with session as page_session:
        html_page = page_session.get(BASE_URL).content
        html_script = fromstring(html_page).xpath('//script[contains(., "source")]/text()')[0]
        key = {
            'keyto': re.search(
                r"var source = new EventSource\('server\.php\?keyto=(.*)'\);", html_script
            ).group(1)
        }
        earthquake_data = page_session.get(f'{DATA_URL}', params=key).content
        earthquake_data = earthquake_data.decode('utf8').replace("data", '"data"').strip()
        return json.loads(f'{{{earthquake_data}}}')


def send_message() -> None:
    """Send a message via Twilio if the magnitude of an earthquake
    is bigger than 4.
    """

    twilio_client = Client(
        get_credential('TWILIO_ACCOUNT_SID'),
        get_credential('TWILIO_AUTH_TOKEN')
    )

    data = get_earthquake_data().get('data')
    eq_magnitude = data.get('mag')

    if float(eq_magnitude) >= 4:
        body = f"""ATTENTION!!!

        Earthquake with magnitude: {eq_magnitude} 
        at {datetime.datetime.strftime(datetime.datetime.now(), '%Y-%m-%d %H:%M:%S')}!
        """

        try:
            twilio_client.messages.create(
                body=body,
                from_=get_credential('FROM'),
                to=get_credential('TO')
            )
        except Exception as error:
            print(f'Twilio API error: {error}')
    else:
        print('No need to worry. YET!')


def main() -> None:
    """Main entry to the program."""

    while True:
        try:
            send_message()
            time.sleep(DELAY)
        except KeyboardInterrupt:
            print('Closing the program...')


if __name__ == '__main__':
    main()

And the custom exception from exceptions.py looks like this:

"""
Main module for all the exceptions across project.
"""


class EmptyCredentialsFile(Exception):
    """
    Exception to be raised when credentials file is empty and valid
    """
    pass

NOTE 2:

If you'd like to run the whole script and don't have a Twilio account, just call get_earthquake_data() instead of send_message() function.

\$\endgroup\$

3 Answers 3

11
\$\begingroup\$

I don't really see much that's notable. This is some nice looking code.

Minor things though:

If you're going to use type-hints, I think I'd take it a step further and use the generic Dict to specify what types the dictionaries hold:

from typing import Dict

. . .

def get_earthquake_data() -> Dict[str, str]:
   . . .

And then similarly for the other cases. dict (currently?) doesn't support generic type hints, so Dict, a generic type alias, can be used. Now, for example:

data = get_earthquake_data()
val = data[some_key]  # It can typecheck the key now
val  # And it knows that this is a String

That also better communicates to the reader what types are involved.


I'm on my phone so it's a pain to check, but here:

except Exception as error: 
    print(f'Twilio API error: {error}')

Is there really not a more specific exception that Twilio throws? I'd see if it has its own base class that you can catch. You don't want to accidentally catch a non-API error there and mask a bug.


data = get_earthquake_data().get('data')
eq_magnitude = data.get('mag')

I'd probably use subscripting there instead of get. Sure, that will allow for a KeyError, but that will cause a more sensical error than it will if None is returned by get and you end up with float(None). I only use get if I'm expecting a potentially invalid key and I want to handle it by checking the return for None (or a sentinal that I specify). I think avoiding the KeyError like this causes more issues than it solves by allowing bad data to be passed on instead of just dying where the error happened.

\$\endgroup\$
2
  • \$\begingroup\$ Nice answer! +1. Those are all things that I've given a thought but didn't really do anything about it. What can you say about the naming? For me (and probably most developers) that's the hardest thing ^_^ \$\endgroup\$ Commented Jan 6, 2020 at 6:56
  • \$\begingroup\$ @GrajdeanuAlex. No complaints they're all PEP8 compliant and descriptive. \$\endgroup\$ Commented Jan 6, 2020 at 14:42
4
\$\begingroup\$

In your main loop, you catch KeyBoardInterrupt without handling it or exiting, so if you press Ctrl-C during execution it prints the message and continues looping forever. You should add a break statement after the print statement to get out of this.

\$\endgroup\$
1
  • 1
    \$\begingroup\$ Oupsiee, fair enough ^_^ That's why I love CR. Haven't even spotted something that obvious. \$\endgroup\$ Commented Jan 9, 2020 at 16:19
3
\$\begingroup\$

NOTE 2:

If you'd like to run the whole script and don't have a Twilio account, just call get_earthquake_data() instead of send_message() function.

You could improve this part by providing a command line option, so that

python eatrhquake.py

Would run get_earthquake_data();

python earthquake.py --twilio

Would run send_message() with the default (credentials.json) file; and

python earthquake.py --twilio /path/to/my/credentials.json

Would run send_message() with a user provided file path.


Python provides the argparse module for such case that you can define along the lines of

def command_line_parser():
    parser = argparse.ArgumentParser(description='Earthquake Listener')
    parser.add_argument(
            '-t', '--twilio',
            nargs='?',
            type=argparse.FileType('r'),
            const='credentials.json',
            help='<describe the file and its content here>')
    return parser

You can then use parser.parse_args().twilio to know which function to call depending on wether it is None or an opened file object.


You also call get_credentials() a lot during the script lifetime, when you could call it only once and store the resulting dict in a variable instead. This is especially wasteful as it opens the file each time.

In the same vein, you create a Twilio client at each loop even if you don't send anything. Art the very least you could create it only when you need to send something; at best you could create it once at the beginning of the script and reuse the same for each send, but I'm not familiar with the API and there may be some timeouts limiting this option.


Proposed improvements:

"""
This projects aims to send an SMS to a specific number using Twilio
if an earthquake with magnitude > 4 is going to occur (depending on
your location, this can warn you (best case scenario within 25-30
seconds before you feel the earthquake wave).
"""

import argparse
import datetime
import json
import re
import time
from functools import partial

import requests
from lxml.html import fromstring
from twilio.rest import Client


BASE_URL = 'http://alerta.infp.ro'
DATA_URL = f'{BASE_URL}/server.php'


def command_line_parser():
    parser = argparse.ArgumentParser(description='Earthquake Listener')
    parser.add_argument(
            '-t', '--twilio',
            nargs='?',
            type=credentials,
            const='credentials.json',
            help='<???>')
    parser.add_argument(
            '-d', '--delay',
            type=float,
            default=1.0,
            help='delay in seconds between two calls to the earthquake API')
    return parser


def credentials(filepath: str) -> dict:
    """Return secrets from `filepath` file as a dict.

    The file looks like this:

        {
          "TWILIO_ACCOUNT_SID": "Your twilio account SID",
          "TWILIO_AUTH_TOKEN": "Your twilio account auth token",
          "FROM": "The number from which you'll receive the alert",
          "TO": "The number the message is sent to"
        }


    Args:
        filepath (str): Path to credentials JSON file.

    Returns:
        dict: The return value.
    """

    with open(filepath) as credentials_file:
        credentials = json.load(credentials_file)

    if not credentials:
        raise ValueError('Credentials file should not be empty.')

    return credentials


def get_earthquake_data() -> dict:
    """Get earthquake data from `DATA_URL`.

    Returns:
        dict: A dict containing the following data:
              {
                'mag': '0.1',
                'heart': '2020-01-04 13:30:04 HEARTBEAT',
                'sec': '30',
                'key': 'NjY2NDYyMzAzNjMwNjM2MzM1Mz...=='
              }
    """

    session = requests.Session()
    with session as page_session:
        html_page = page_session.get(BASE_URL).content
        html_script = fromstring(html_page).xpath('//script[contains(., "source")]/text()')[0]
        key = {
            'keyto': re.search(
                r"var source = new EventSource\('server\.php\?keyto=(.*)'\);", html_script
            ).group(1)
        }
        earthquake_data = page_session.get(f'{DATA_URL}', params=key).content
        earthquake_data = earthquake_data.decode('utf8').replace("data", '"data"').strip()
        return json.loads(f'{{{earthquake_data}}}')


def send_message(twilio_client, send_to, sent_from) -> None:
    """Send a message via Twilio if the magnitude of an earthquake
    is bigger than 4.
    """

    data = get_earthquake_data().get('data')
    eq_magnitude = data.get('mag')

    if float(eq_magnitude) >= 4:
        body = f"""ATTENTION!!!

        Earthquake with magnitude: {eq_magnitude} 
        at {datetime.datetime.strftime(datetime.datetime.now(), '%Y-%m-%d %H:%M:%S')}!
        """

        try:
            twilio_client.messages.create(body=body, from_=sent_from, to=send_to)
        except Exception as error:
            print(f'Twilio API error: {error}')
    else:
        print('No need to worry. YET!')


def main(credentials=None, delay=1.0) -> None:
    """Main entry to the program."""

    if credentials is None:
        action = get_earthquake_data
    else:
        twilio_client = Client(
                credentials['TWILIO_ACCOUNT_SID'],
                credentials['TWILIO_AUTH_TOKEN'])
        sender = credentials['FROM']
        receiver = credentials['TO']
        action = partial(send_message, twilio_client, receiver, sender)

    while True:
        try:
            action()
            time.sleep(delay)
        except KeyboardInterrupt:
            print('Closing the program...')


if __name__ == '__main__':
    args = command_line_parser.parse_args()
    main(args.twilio, args.delay)

Note that I changed your custom exception to a ValueError to integrate better with argparse's exception handling.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.