Find the next predicted time a satellite is going to take picture of place

Question

create a function which predicts the next time a satellite image will be taken of a certain location. When there is enough data to do so, the function should print a prediction for when the next picture will be taken. and when there is no enough data then return error. if prediction date is less than current date, then return send future date.

import requests
from datetime import datetime
from datetime import timedelta

import math
import numpy as np

# NASA's API and API key
API = "https://api.nasa.gov/planetary/earth/assets"
API_KEY = "xxxxxxxxxxxxxxxxxx"

# Constants
DATE_FORMAT = '%Y-%m-%dT%H:%M:%S'

# Error Constants
INVALID_COORDINATES = 'Invalid coordinate!'
INSUFFICIENT_DATA = 'Insufficient data recorded'


def flyby(latitude, longitude, place='not specified'):
    print("place ", place)

    # if invalid coordinate print invalid
    if latitude < -90 or latitude > 90 or longitude < -180 or longitude > 180:
        print(INVALID_COORDINATES)
        return INVALID_COORDINATES

    querystring = {"api_key": API_KEY, "lat": str(latitude),
                   "lon": str(longitude)}

    response = requests.request("GET", API, params=querystring)

    if response.status_code != 200:
        return response.text

    json_response = response.json()
    total_records = json_response.get('count')
    results = json_response.get('results')

    # if count < 2, print insufficient
    if total_records < 2:
        print(INSUFFICIENT_DATA)
        return INSUFFICIENT_DATA

    avg_time_delta, last_date = latest_average_time_delta(results,
                                                          total_records)
    sd = get_standard_deviation(results, total_records)
    print("latest : {}".format(last_date))
    print("ave_time_delta : {}".format(avg_time_delta))

    dates_diff = datetime.today() - last_date
    if dates_diff.total_seconds() < 0:
        predicted = last_date
    else:
        dates_diff % avg_time_delta
        predicted = datetime.today() + (dates_diff % avg_time_delta)

    print("Next time: {}".format(predicted))
    print("Next time also can be between {} : {}".format(
        predicted - timedelta(seconds=sd),
        predicted + timedelta(seconds=sd)))
    print("-----------------------")


def latest_average_time_delta(results, count):
    dates = map(lambda x: datetime.strptime(x['date'], DATE_FORMAT), results)
    dates_list = list(dates)

    oldest = min(dates_list)
    youngest = max(dates_list)

    # Average time taken is the duration between the youngest and oldest
    # recorded date, divided by the number of periods (n - 1)
    ave_time_delta = (youngest - oldest) / (count - 1)
    return ave_time_delta, youngest


def get_standard_deviation(results, count):
    str_dates = map(lambda x: x['date'], results)
    mean_date = (np.array(list(str_dates), dtype='datetime64[s]')
                 .view('i8')
                 .mean()
                 .astype('datetime64[s]'))
    print("Average Date ", mean_date)
    dates = map(lambda x: datetime.strptime(x['date'], DATE_FORMAT), results)
    dates_list = list(dates)
    sd = 0.0
    for date in dates_list:
        date_diff = date - mean_date.astype(datetime)
        sd += date_diff.seconds ** 2
    sd = math.sqrt(sd / float(count - 1))
    return sd


lat = 36.998979
lon = -109.045183

flyby(lat, lon)

flyby(0.000000, 0.000000, "GULF OF GUINEA")
flyby(36.098592, -112.097796, "GRAND CANYON")
flyby(43.078154, -79.075891, "NIAGARA FALLS")
flyby(36.998979, -109.045183, "FOUR CORNERS")
flyby(37.7937007, -122.4039064, "DELPHIX")

# BOUNDARY/EDGE

# MINIMUM LATITUDE
flyby(-90.000001, 0.000000, "MIN LAT 1")
flyby(-90.000000, 0.000000, "MIN LAT 2")
flyby(-89.999999, 0.000000, "MIN LAT 3")

# MAXIMUM LATITUDE
flyby(89.999999, 0.000000, "MAX LAT 1")
flyby(90.000000, 0.000000, "MAX LAT 2")
flyby(90.000001, 0.000000, "MAX LAT 3")

# MINIMUM LONGITUDE
flyby(0.000000, -180.000001, "MIN LON 1")
flyby(0.000000, -180.000000, "MIN LON 2")
flyby(0.000000, -179.999999, "MIN LON 3")

# MAXIMUM LONGITUDE
flyby(0.000000, 179.999999, "MAX LON 1")
flyby(0.000000, 180.000000, "MAX LON 2")
flyby(0.000000, 180.000001, "MAX LON 3")

# EDGES COMBINATION
flyby(-90.000000, -180.000000, "MIN LAT, MIN LON")
flyby(-90.000000, 180.000000, "MIN LAT, MAX LON")
flyby(90.000000, -180.000000, "MAX LAT, MIN LON")
flyby(90.000000, 180.000000, "MAX LAT, MAX LON")

Any feedback on style, flaws in the code, or how to improve the algorithm would be greatly appreciated.

Answer 1

Specific suggestions:

1. You never use the return value of flyby, so it might as well return None or raise InvalidCoordinateError(). Raising exceptions is generally the most useful way to provide information for users to act on, because it tells them not just what went wrong ("Invalid coordinate!") but where because of the stack trace.
2. Ditto for return response.text - it would be more useful as something like raise UnhandledResponseError(response.text).
3. The API key should be configuration, not code. In general such keys should not be anywhere in your repository, for a host of reasons which should be spelled out in the license. This could include rate limiting, whether you are even allowed to share the key, and which groups of people it can be used by (for example, excluding commercial enterprises).
4. dates_diff % avg_time_delta doesn't actually do anything with the result of the calculation, so that line can be deleted.
5. It is nice that you have test cases, but it would be better to put those in a real TestCase class and being explicit about what they should do. This will show that the code is hard to test because the main side effect is simply printing, and testing print statements is much hairier than testing return values. It seems a useful return value would be the tuple (predicted, sd) or an object.

General suggestions:

1. black can automatically format your code to be more idiomatic.
2. isort can group and sort your imports automatically.

3. flake8 with a strict complexity limit will give you more hints to write idiomatic Python:

   [flake8]
   max-complexity = 4
   ignore = W503,E203
   

   (That limit is not absolute by any means, but it's worth thinking hard whether you can keep it low whenever validation fails. For example, I'm working with a team on an application since a year now, and our complexity limit is up to 7 in only one place. Conversely, on an ugly old piece of code I wrote without static analysis support I recently found the complexity reaches 87!)

4. I would then recommend adding type hints everywhere and validating them using a strict mypy configuration:

   [mypy]
   check_untyped_defs = true
   disallow_untyped_defs = true
   ignore_missing_imports = true
   no_implicit_optional = true
   warn_redundant_casts = true
   warn_return_any = true
   warn_unused_ignores = true
   

Answer 2

code after some adding some exception handling.

import math
from datetime import datetime
from datetime import timedelta

import numpy as np
import requests

# NASA's API and API key
API = "https://api.nasa.gov/planetary/earth/assets"

# This can go to some configuration file
API_KEY = "9Jz6tLIeJ0yY9vjbEUWaH9fsXA930J9hspPchute"

# Constants
DATE_FORMAT = '%Y-%m-%dT%H:%M:%S'

# Error Constants
INVALID_COORDINATES = 'ERROR_INVALID_COORDINATE'
INSUFFICIENT_DATA = 'ERROR_INSUFFICIENT_DATA_RECORDED'
UNHANDLED_RESPONSE = 'ERROR_UNHANDLED_RESPONSE'


class BaseException(Exception):
    """
    Base exception class
    """
    err_code = None
    err_str = None

    def to_dict(self):
        """
        return json object
        """
        return {
            'err_code': self.err_code,
            'err_str': self.err_str
        }


class UnhandledResponseError(BaseException):
    """
    Unhandled response
    """
    err_code = 1000
    err_str = UNHANDLED_RESPONSE


class InvalidCoordinateError(BaseException):
    """
    Invalid Coordinate exception
    """
    err_code = 1001
    err_str = INVALID_COORDINATES


class InsufficientDataError(BaseException):
    """
    Insufficient data recorded
    """
    err_code = 1002
    err_str = INSUFFICIENT_DATA


def flyby(latitude, longitude):
    try:
        if latitude < -90 or latitude > 90 or longitude < -180 or longitude > 180:
            raise InvalidCoordinateError

        querystring = {"api_key": API_KEY, "lat": str(latitude),
                       "lon": str(longitude)}

        response = requests.request("GET", API, params=querystring)

        if response.status_code != 200:
            raise UnhandledResponseError

        json_response = response.json()
        total_records = json_response.get('count')
        results = json_response.get('results')

        if total_records < 2:
            raise InsufficientDataError

        avg_time_delta, last_date = \
            get_last_date_and_average_time_delta(results, total_records)

        sd = get_standard_deviation(results, total_records)
        print("latest : {}".format(last_date))
        print("average time delta : {}".format(avg_time_delta))

        dates_diff = datetime.today() - last_date
        if dates_diff.total_seconds() < 0:
            predicted = last_date
        else:
            dates_diff % avg_time_delta
            predicted = datetime.today() + (dates_diff % avg_time_delta)

        print("Next time: {}".format(predicted))
        print("Next time also can be between {} : {}".format(
            predicted - timedelta(seconds=sd),
            predicted + timedelta(seconds=sd)))
        print("-----------------------")
        return predicted

    except (InvalidCoordinateError, InsufficientDataError,
            UnhandledResponseError) as err:
        print(err.to_dict())
        print("-----------------------")
        return err.to_dict()


def get_last_date_and_average_time_delta(results, count):
    dates = map(lambda x: datetime.strptime(x['date'], DATE_FORMAT), results)
    dates_list = list(dates)

    oldest = min(dates_list)
    youngest = max(dates_list)

    # Average time taken is the duration between the youngest and oldest
    # recorded date, divided by the number of periods (n - 1)
    ave_time_delta = (youngest - oldest) / (count - 1)
    return ave_time_delta, youngest


def get_standard_deviation(results, count):
    str_dates = map(lambda x: x['date'], results)
    mean_date = (np.array(list(str_dates), dtype='datetime64[s]')
                 .view('i8')
                 .mean()
                 .astype('datetime64[s]'))
    dates = map(lambda x: datetime.strptime(x['date'], DATE_FORMAT), results)
    dates_list = list(dates)
    sd = 0.0
    for date in dates_list:
        date_diff = date - mean_date.astype(datetime)
        sd += date_diff.seconds ** 2
    sd = math.sqrt(sd / float(count - 1))
    return sd


if __name__ == "__main__":
    print("GULF OF GUINEA")
    result = flyby(0.000000, 0.000000)
    assert result['err_str'] == INSUFFICIENT_DATA

    print("GRAND CANYON")
    result = flyby(36.098592, -112.097796)
    assert result is not None

    print("NIAGARA FALLS")
    result = flyby(43.078154, -79.075891)
    assert result is not None

    print("FOUR CORNERS")
    result = flyby(36.998979, -109.045183)
    assert result is not None

    print("DELPHIX")
    result = flyby(37.7937007, -122.4039064)
    assert result is not None

    print("Invalid Latitude Negative")
    result = flyby(-90.000001, 0.000000, )
    assert result['err_str'] == INVALID_COORDINATES
    print("Minimum Latitude")
    result = flyby(-90.000000, 0.000000)
    assert result['err_str'] == INSUFFICIENT_DATA

    print("Invalid Latitude Positive")
    result = flyby(90.000001, 0.000000, )
    assert result['err_str'] == INVALID_COORDINATES
    print("Maximum Latitude")
    result = flyby(90.000000, 0.000000)
    assert result['err_str'] == INSUFFICIENT_DATA

    print("Invalid Longitude Negative")
    result = flyby(0.000000, -180.000001)
    assert result['err_str'] == INVALID_COORDINATES
    print("Minimum Longitude")
    result = flyby(0.000000, -180.000000)
    assert result['err_str'] == INSUFFICIENT_DATA

    print("Invalid Longitude Positive")
    result = flyby(0.000000, 180.000001, )
    assert result['err_str'] == INVALID_COORDINATES
    print("Maximum Longitude")
    result = flyby(0.000000, 180.000000, )
    assert result['err_str'] == INSUFFICIENT_DATA