hangpersonal/Udacity_Nanodegree_Intermediate_Python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add starter code of Project: Exploring Near-Earth Objects

Browse Source
This commit is contained in:
Hang Cui
2025-12-26 17:19:08 -08:00
parent 241df7c9d3
commit 137a13ec61
20 changed files with 92427 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1055
Exploring_Near_Earth_Objects/README.md Normal file
View File

File diff suppressed because it is too large Load Diff

1
Exploring_Near_Earth_Objects/data/cad.json Normal file
View File

File diff suppressed because one or more lines are too long

23968
Exploring_Near_Earth_Objects/data/neos.csv Normal file
View File

File diff suppressed because it is too large Load Diff

98
Exploring_Near_Earth_Objects/database.py Normal file
View File

@@ -0,0 +1,98 @@
"""A database encapsulating collections of near-Earth objects and their close approaches.
A `NEODatabase` holds an interconnected data set of NEOs and close approaches.
It provides methods to fetch an NEO by primary designation or by name, as well
as a method to query the set of close approaches that match a collection of
user-specified criteria.
Under normal circumstances, the main module creates one NEODatabase from the
data on NEOs and close approaches extracted by `extract.load_neos` and
`extract.load_approaches`.
You'll edit this file in Tasks 2 and 3.
"""
class NEODatabase:
"""A database of near-Earth objects and their close approaches.
A `NEODatabase` contains a collection of NEOs and a collection of close
approaches. It additionally maintains a few auxiliary data structures to
help fetch NEOs by primary designation or by name and to help speed up
querying for close approaches that match criteria.
"""
def __init__(self, neos, approaches):
"""Create a new `NEODatabase`.
As a precondition, this constructor assumes that the collections of NEOs
and close approaches haven't yet been linked - that is, the
`.approaches` attribute of each `NearEarthObject` resolves to an empty
collection, and the `.neo` attribute of each `CloseApproach` is None.
However, each `CloseApproach` has an attribute (`._designation`) that
matches the `.designation` attribute of the corresponding NEO. This
constructor modifies the supplied NEOs and close approaches to link them
together - after it's done, the `.approaches` attribute of each NEO has
a collection of that NEO's close approaches, and the `.neo` attribute of
each close approach references the appropriate NEO.
:param neos: A collection of `NearEarthObject`s.
:param approaches: A collection of `CloseApproach`es.
"""
self._neos = neos
self._approaches = approaches
# TODO: What additional auxiliary data structures will be useful?
# TODO: Link together the NEOs and their close approaches.
def get_neo_by_designation(self, designation):
"""Find and return an NEO by its primary designation.
If no match is found, return `None` instead.
Each NEO in the data set has a unique primary designation, as a string.
The matching is exact - check for spelling and capitalization if no
match is found.
:param designation: The primary designation of the NEO to search for.
:return: The `NearEarthObject` with the desired primary designation, or `None`.
"""
# TODO: Fetch an NEO by its primary designation.
return None
def get_neo_by_name(self, name):
"""Find and return an NEO by its name.
If no match is found, return `None` instead.
Not every NEO in the data set has a name. No NEOs are associated with
the empty string nor with the `None` singleton.
The matching is exact - check for spelling and capitalization if no
match is found.
:param name: The name, as a string, of the NEO to search for.
:return: The `NearEarthObject` with the desired name, or `None`.
"""
# TODO: Fetch an NEO by its name.
return None
def query(self, filters=()):
"""Query close approaches to generate those that match a collection of filters.
This generates a stream of `CloseApproach` objects that match all of the
provided filters.
If no arguments are provided, generate all known close approaches.
The `CloseApproach` objects are generated in internal order, which isn't
guaranteed to be sorted meaningfully, although is often sorted by time.
:param filters: A collection of filters capturing user-specified criteria.
:return: A stream of matching `CloseApproach` objects.
"""
# TODO: Generate `CloseApproach` objects that match all of the filters.
for approach in self._approaches:
yield approach

38
Exploring_Near_Earth_Objects/extract.py Normal file
View File

@@ -0,0 +1,38 @@
"""Extract data on near-Earth objects and close approaches from CSV and JSON files.
The `load_neos` function extracts NEO data from a CSV file, formatted as
described in the project instructions, into a collection of `NearEarthObject`s.
The `load_approaches` function extracts close approach data from a JSON file,
formatted as described in the project instructions, into a collection of
`CloseApproach` objects.
The main module calls these functions with the arguments provided at the command
line, and uses the resulting collections to build an `NEODatabase`.
You'll edit this file in Task 2.
"""
import csv
import json
from models import NearEarthObject, CloseApproach
def load_neos(neo_csv_path):
"""Read near-Earth object information from a CSV file.
:param neo_csv_path: A path to a CSV file containing data about near-Earth objects.
:return: A collection of `NearEarthObject`s.
"""
# TODO: Load NEO data from the given CSV file.
return ()
def load_approaches(cad_json_path):
"""Read close approach data from a JSON file.
:param cad_json_path: A path to a JSON file containing data about close approaches.
:return: A collection of `CloseApproach`es.
"""
# TODO: Load close approach data from the given JSON file.
return ()

125
Exploring_Near_Earth_Objects/filters.py Normal file
View File

@@ -0,0 +1,125 @@
"""Provide filters for querying close approaches and limit the generated results.
The `create_filters` function produces a collection of objects that is used by
the `query` method to generate a stream of `CloseApproach` objects that match
all of the desired criteria. The arguments to `create_filters` are provided by
the main module and originate from the user's command-line options.
This function can be thought to return a collection of instances of subclasses
of `AttributeFilter` - a 1-argument callable (on a `CloseApproach`) constructed
from a comparator (from the `operator` module), a reference value, and a class
method `get` that subclasses can override to fetch an attribute of interest from
the supplied `CloseApproach`.
The `limit` function simply limits the maximum number of values produced by an
iterator.
You'll edit this file in Tasks 3a and 3c.
"""
import operator
class UnsupportedCriterionError(NotImplementedError):
"""A filter criterion is unsupported."""
class AttributeFilter:
"""A general superclass for filters on comparable attributes.
An `AttributeFilter` represents the search criteria pattern comparing some
attribute of a close approach (or its attached NEO) to a reference value. It
essentially functions as a callable predicate for whether a `CloseApproach`
object satisfies the encoded criterion.
It is constructed with a comparator operator and a reference value, and
calling the filter (with __call__) executes `get(approach) OP value` (in
infix notation).
Concrete subclasses can override the `get` classmethod to provide custom
behavior to fetch a desired attribute from the given `CloseApproach`.
"""
def __init__(self, op, value):
"""Construct a new `AttributeFilter` from an binary predicate and a reference value.
The reference value will be supplied as the second (right-hand side)
argument to the operator function. For example, an `AttributeFilter`
with `op=operator.le` and `value=10` will, when called on an approach,
evaluate `some_attribute <= 10`.
:param op: A 2-argument predicate comparator (such as `operator.le`).
:param value: The reference value to compare against.
"""
self.op = op
self.value = value
def __call__(self, approach):
"""Invoke `self(approach)`."""
return self.op(self.get(approach), self.value)
@classmethod
def get(cls, approach):
"""Get an attribute of interest from a close approach.
Concrete subclasses must override this method to get an attribute of
interest from the supplied `CloseApproach`.
:param approach: A `CloseApproach` on which to evaluate this filter.
:return: The value of an attribute of interest, comparable to `self.value` via `self.op`.
"""
raise UnsupportedCriterionError
def __repr__(self):
return f"{self.__class__.__name__}(op=operator.{self.op.__name__}, value={self.value})"
def create_filters(
date=None, start_date=None, end_date=None,
distance_min=None, distance_max=None,
velocity_min=None, velocity_max=None,
diameter_min=None, diameter_max=None,
hazardous=None
):
"""Create a collection of filters from user-specified criteria.
Each of these arguments is provided by the main module with a value from the
user's options at the command line. Each one corresponds to a different type
of filter. For example, the `--date` option corresponds to the `date`
argument, and represents a filter that selects close approaches that occurred
on exactly that given date. Similarly, the `--min-distance` option
corresponds to the `distance_min` argument, and represents a filter that
selects close approaches whose nominal approach distance is at least that
far away from Earth. Each option is `None` if not specified at the command
line (in particular, this means that the `--not-hazardous` flag results in
`hazardous=False`, not to be confused with `hazardous=None`).
The return value must be compatible with the `query` method of `NEODatabase`
because the main module directly passes this result to that method. For now,
this can be thought of as a collection of `AttributeFilter`s.
:param date: A `date` on which a matching `CloseApproach` occurs.
:param start_date: A `date` on or after which a matching `CloseApproach` occurs.
:param end_date: A `date` on or before which a matching `CloseApproach` occurs.
:param distance_min: A minimum nominal approach distance for a matching `CloseApproach`.
:param distance_max: A maximum nominal approach distance for a matching `CloseApproach`.
:param velocity_min: A minimum relative approach velocity for a matching `CloseApproach`.
:param velocity_max: A maximum relative approach velocity for a matching `CloseApproach`.
:param diameter_min: A minimum diameter of the NEO of a matching `CloseApproach`.
:param diameter_max: A maximum diameter of the NEO of a matching `CloseApproach`.
:param hazardous: Whether the NEO of a matching `CloseApproach` is potentially hazardous.
:return: A collection of filters for use with `query`.
"""
# TODO: Decide how you will represent your filters.
return ()
def limit(iterator, n=None):
"""Produce a limited stream of values from an iterator.
If `n` is 0 or None, don't limit the iterator at all.
:param iterator: An iterator of values.
:param n: The maximum number of values to produce.
:yield: The first (at most) `n` values from the iterator.
"""
# TODO: Produce at most `n` values from the given iterator.
return iterator

44
Exploring_Near_Earth_Objects/helpers.py Normal file
View File

@@ -0,0 +1,44 @@
"""Convert datetimes to and from strings.
NASA's dataset provides timestamps as naive datetimes (corresponding to UTC).
The `cd_to_datetime` function converts a string, formatted as the `cd` field of
NASA's close approach data, into a Python `datetime`
The `datetime_to_str` function converts a Python `datetime` into a string.
Although `datetime`s already have human-readable string representations, those
representations display seconds, but NASA's data (and our datetimes!) don't
provide that level of resolution, so the output format also will not.
"""
import datetime
def cd_to_datetime(calendar_date):
"""Convert a NASA-formatted calendar date/time description into a datetime.
NASA's format, at least in the `cd` field of close approach data, uses the
English locale's month names. For example, December 31st, 2020 at noon is:
2020-Dec-31 12:00
This will become the Python object `datetime.datetime(2020, 12, 31, 12, 0)`.
:param calendar_date: A calendar date in YYYY-bb-DD hh:mm format.
:return: A naive `datetime` corresponding to the given calendar date and time.
"""
return datetime.datetime.strptime(calendar_date, "%Y-%b-%d %H:%M")
def datetime_to_str(dt):
"""Convert a naive Python datetime into a human-readable string.
The default string representation of a datetime includes seconds; however,
our data isn't that precise, so this function only formats the year, month,
date, hour, and minute values. Additionally, this function provides the date
in the usual ISO 8601 YYYY-MM-DD format to avoid ambiguities with
locale-specific month names.
:param dt: A naive Python datetime.
:return: That datetime, as a human-readable string without seconds.
"""
return datetime.datetime.strftime(dt, "%Y-%m-%d %H:%M")

397
Exploring_Near_Earth_Objects/main.py Normal file
View File

@@ -0,0 +1,397 @@
#!/usr/bin/env python3
"""Explore a dataset of near-Earth objects and their close approaches to Earth.
See `README.md` for a detailed discussion of this project.
This script can be invoked from the command line::
$ python3 main.py {inspect,query,interactive} [args]
The `inspect` subcommand looks up an NEO by name or by primary designation, and
optionally lists all of that NEO's known close approaches:
$ python3 main.py inspect --pdes 1P
$ python3 main.py inspect --name Halley
$ python3 main.py inspect --verbose --name Halley
The `query` subcommand searches for close approaches that match given criteria:
$ python3 main.py query --date 1969-07-29
$ python3 main.py query --start-date 2020-01-01 --end-date 2020-01-31 --max-distance 0.025
$ python3 main.py query --start-date 2050-01-01 --min-distance 0.2 --min-velocity 50
$ python3 main.py query --date 2020-03-14 --max-velocity 25 --min-diameter 0.5 --hazardous
$ python3 main.py query --start-date 2000-01-01 --max-diameter 0.1 --not-hazardous
$ python3 main.py query --hazardous --max-distance 0.05 --min-velocity 30
The set of results can be limited in size and/or saved to an output file in CSV
or JSON format:
$ python3 main.py query --limit 5 --outfile results.csv
$ python3 main.py query --limit 15 --outfile results.json
The `interactive` subcommand loads the NEO database and spawns an interactive
command shell that can repeatedly execute `inspect` and `query` commands without
having to wait to reload the database each time. However, it doesn't hot-reload.
If needed, the script can load data from data files other than the default with
`--neofile` or `--cadfile`.
"""
import argparse
import cmd
import datetime
import pathlib
import shlex
import sys
import time
from extract import load_neos, load_approaches
from database import NEODatabase
from filters import create_filters, limit
from write import write_to_csv, write_to_json
# Paths to the root of the project and the `data` subfolder.
PROJECT_ROOT = pathlib.Path(__file__).parent.resolve()
DATA_ROOT = PROJECT_ROOT / 'data'
# The current time, for use with the kill-on-change feature of the interactive shell.
_START = time.time()
def date_fromisoformat(date_string):
"""Return a `datetime.date` corresponding to a string in YYYY-MM-DD format.
In Python 3.7+, there is `datetime.date.fromisoformat`, but alas - we're
supporting Python 3.6+.
:param date_string: A date in the format YYYY-MM-DD.
:return: A `datetime.date` correspondingo the given date string.
"""
try:
return datetime.datetime.strptime(date_string, '%Y-%m-%d').date()
except ValueError:
raise argparse.ArgumentTypeError(f"'{date_string}' is not a valid date. Use YYYY-MM-DD.")
def make_parser():
"""Create an ArgumentParser for this script.
:return: A tuple of the top-level, inspect, and query parsers.
"""
parser = argparse.ArgumentParser(
description="Explore past and future close approaches of near-Earth objects."
)
# Add arguments for custom data files.
parser.add_argument('--neofile', default=(DATA_ROOT / 'neos.csv'),
type=pathlib.Path,
help="Path to CSV file of near-Earth objects.")
parser.add_argument('--cadfile', default=(DATA_ROOT / 'cad.json'),
type=pathlib.Path,
help="Path to JSON file of close approach data.")
subparsers = parser.add_subparsers(dest='cmd')
# Add the `inspect` subcommand parser.
inspect = subparsers.add_parser('inspect',
description="Inspect an NEO by primary designation or by name.")
inspect.add_argument('-v', '--verbose', action='store_true',
help="Additionally, print all known close approaches of this NEO.")
inspect_id = inspect.add_mutually_exclusive_group(required=True)
inspect_id.add_argument('-p', '--pdes',
help="The primary designation of the NEO to inspect (e.g. '433').")
inspect_id.add_argument('-n', '--name',
help="The IAU name of the NEO to inspect (e.g. 'Halley').")
# Add the `query` subcommand parser.
query = subparsers.add_parser('query',
description="Query for close approaches that "
"match a collection of filters.")
filters = query.add_argument_group('Filters',
description="Filter close approaches by their attributes "
"or the attributes of their NEOs.")
filters.add_argument('-d', '--date', type=date_fromisoformat,
help="Only return close approaches on the given date, "
"in YYYY-MM-DD format (e.g. 2020-12-31).")
filters.add_argument('-s', '--start-date', type=date_fromisoformat,
help="Only return close approaches on or after the given date, "
"in YYYY-MM-DD format (e.g. 2020-12-31).")
filters.add_argument('-e', '--end-date', type=date_fromisoformat,
help="Only return close approaches on or before the given date, "
"in YYYY-MM-DD format (e.g. 2020-12-31).")
filters.add_argument('--min-distance', dest='distance_min', type=float,
help="In astronomical units. Only return close approaches that "
"pass as far or farther away from Earth as the given distance.")
filters.add_argument('--max-distance', dest='distance_max', type=float,
help="In astronomical units. Only return close approaches that "
"pass as near or nearer to Earth as the given distance.")
filters.add_argument('--min-velocity', dest='velocity_min', type=float,
help="In kilometers per second. Only return close approaches "
"whose relative velocity to Earth at approach is as fast or faster "
"than the given velocity.")
filters.add_argument('--max-velocity', dest='velocity_max', type=float,
help="In kilometers per second. Only return close approaches "
"whose relative velocity to Earth at approach is as slow or slower "
"than the given velocity.")
filters.add_argument('--min-diameter', dest='diameter_min', type=float,
help="In kilometers. Only return close approaches of NEOs with "
"diameters as large or larger than the given size.")
filters.add_argument('--max-diameter', dest='diameter_max', type=float,
help="In kilometers. Only return close approaches of NEOs with "
"diameters as small or smaller than the given size.")
filters.add_argument('--hazardous', dest='hazardous', default=None, action='store_true',
help="If specified, only return close approaches of NEOs that "
"are potentially hazardous.")
filters.add_argument('--not-hazardous', dest='hazardous', default=None, action='store_false',
help="If specified, only return close approaches of NEOs that "
"are not potentially hazardous.")
query.add_argument('-l', '--limit', type=int,
help="The maximum number of matches to return. "
"Defaults to 10 if no --outfile is given.")
query.add_argument('-o', '--outfile', type=pathlib.Path,
help="File in which to save structured results. "
"If omitted, results are printed to standard output.")
repl = subparsers.add_parser('interactive',
description="Start an interactive command session "
"to repeatedly run `interact` and `query` commands.")
repl.add_argument('-a', '--aggressive', action='store_true',
help="If specified, kill the session whenever a project file is modified.")
return parser, inspect, query
def inspect(database, pdes=None, name=None, verbose=False):
"""Perform the `inspect` subcommand.
This function fetches an NEO by designation or by name. If a matching NEO is
found, information about the NEO is printed (additionally, information for
all of the NEO's known close approaches is printed if `verbose=True`).
Otherwise, a message is printed noting that there are no matching NEOs.
At least one of `pdes` and `name` must be given. If both are given, prefer
to look up the NEO by the primary designation.
:param database: The `NEODatabase` containing data on NEOs and their close approaches.
:param pdes: The primary designation of an NEO for which to search.
:param name: The name of an NEO for which to search.
:param verbose: Whether to additionally print all of a matching NEO's close approaches.
:return: The matching `NearEarthObject`, or None if not found.
"""
# Fetch the NEO of interest.
if pdes:
neo = database.get_neo_by_designation(pdes)
else:
neo = database.get_neo_by_name(name)
# Ensure that we have received an NEO.
if not neo:
print("No matching NEOs exist in the database.", file=sys.stderr)
return None
# Display information about this NEO, and optionally its close approaches if verbose.
print(neo)
if verbose:
for approach in neo.approaches:
print(f"- {approach}")
return neo
def query(database, args):
"""Perform the `query` subcommand.
Create a collection of filters with `create_filters` and supply them to the
database's `query` method to produce a stream of matching results.
If an output file wasn't given, print these results to stdout, limiting to
10 entries if no limit was specified. If an output file was given, use the
file's extension to infer whether the file should hold CSV or JSON data, and
then write the results to the output file in that format.
:param database: The `NEODatabase` containing data on NEOs and their close approaches.
:param args: All arguments from the command line, as parsed by the top-level parser.
"""
# Construct a collection of filters from arguments supplied at the command line.
filters = create_filters(
date=args.date, start_date=args.start_date, end_date=args.end_date,
distance_min=args.distance_min, distance_max=args.distance_max,
velocity_min=args.velocity_min, velocity_max=args.velocity_max,
diameter_min=args.diameter_min, diameter_max=args.diameter_max,
hazardous=args.hazardous
)
# Query the database with the collection of filters.
results = database.query(filters)
if not args.outfile:
# Write the results to stdout, limiting to 10 entries if not specified.
for result in limit(results, args.limit or 10):
print(result)
else:
# Write the results to a file.
if args.outfile.suffix == '.csv':
write_to_csv(limit(results, args.limit), args.outfile)
elif args.outfile.suffix == '.json':
write_to_json(limit(results, args.limit), args.outfile)
else:
print("Please use an output file that ends with `.csv` or `.json`.", file=sys.stderr)
class NEOShell(cmd.Cmd):
"""Perform the `interactive` subcommand.
This is a `cmd.Cmd` shell - a specialized tool for command-based REPL sessions.
It wraps the `inspect` and `query` parsers to parse flags for those commands
as if they were supplied at the command line.
The primary purpose of this shell is to allow users to repeatedly perform
inspect and query commands, while only loading the data (which can be quite
slow) once.
"""
intro = ("Explore close approaches of near-Earth objects. "
"Type `help` or `?` to list commands and `exit` to exit.\n")
prompt = '(neo) '
def __init__(self, database, inspect_parser, query_parser, aggressive=False, **kwargs):
"""Create a new `NEOShell`.
Creating this object doesn't start the session - for that, use `.cmdloop()`.
:param database: The `NEODatabase` containing data on NEOs and their close approaches.
:param inspect_parser: The subparser for the `inspect` subcommand.
:param query_parser: The subparser for the `query` subcommand.
:param aggressive: Whether to kill the session whenever a project file is changed.
:param kwargs: A dictionary of excess keyword arguments passed to the superclass.
"""
super().__init__(**kwargs)
self.db = database
self.inspect = inspect_parser
self.query = query_parser
self.aggressive = aggressive
@classmethod
def parse_arg_with(cls, arg, parser):
"""Parse the additional text passed to a command, using a given parser.
If any error is encountered (in lexical parsing or argument parsing),
print the error to stderr and return None.
:param arg: The additional text supplied after the command.
:param parser: An `argparse.ArgumentParser` to parse the arguments.
:return: A `Namespace` of the arguments (produced by `parse_args`) or None.
"""
# Lexically parse the additional text with POSIX shell-like syntax.
try:
args = shlex.split(arg)
except ValueError as err:
print(err, file=sys.stderr)
return None
# Use the ArgumentParser to parse the shell arguments.
try:
return parser.parse_args(args)
except SystemExit as err:
# The `parse_args` method doesn't actually surface `ArgumentError`s
# nor `ArgumentTypeError`s - instead, it calls its own `error`
# method which prints the error message and then calls `sys.exit`.
return None
def do_i(self, arg):
"""Shorthand for `inspect`."""
self.do_inspect(arg)
def do_inspect(self, arg):
"""Perform the `inspect` subcommand within the REPL session.
Inspect an NEO by designation or by name:
(neo) inspect --pdes 1P
(neo) inspect --name Halley
Additionally, list all known close approaches:
(neo) inspect --verbose --name Eros
"""
args = self.parse_arg_with(arg, self.inspect)
if not args:
return
# Run the `inspect` subcommand.
inspect(self.db,
pdes=args.pdes, name=args.name,
verbose=args.verbose)
def do_q(self, arg):
"""Shorthand for `query`."""
self.do_query(arg)
def do_query(self, arg):
"""Perform the `query` subcommand within the REPL session.
This command behaves the same as the `query` subcommand from the command
line. For example, to query close approaches on January 1st, 2020:
(neo) query --date 2020-01-01
You can use any of the other filters: `--start-date`, `--end-date`,
`--min-distance`, `--max-distance`, `--min-velocity`, `--max-velocity`,
`--min-diameter`, `--max-diameter`, `--hazardous`, `--not-hazardous`.
The number of results shown can be limited to a maximum number with `--limit`:
(neo) query --limit 2
The results can be saved to a file (instead of displayed to stdout) with
`--outfile`:
(neo) query --limit 5 --outfile results.csv
(neo) query --limit 5 --outfile results.json
"""
args = self.parse_arg_with(arg, self.query)
if not args:
return
# Run the `inspect` subcommand.
query(self.db, args)
def do_EOF(self, _arg):
"""Exit the interactive session."""
return True
# Alternative ways to quit.
do_exit = do_EOF
do_quit = do_EOF
def precmd(self, line):
"""Watch for changes to the files in this project."""
changed = [f for f in PROJECT_ROOT.glob('*.py') if f.stat().st_mtime > _START]
if changed:
print("The following file(s) have been modified since this interactive session began: "
f"{', '.join(str(f.relative_to(PROJECT_ROOT)) for f in changed)}.",
file=sys.stderr)
if not self.aggressive:
print("To include these changes, please exit and restart this interactive session.",
file=sys.stderr)
else:
print("Preemptively terminating the session aggressively.", file=sys.stderr)
return 'exit'
return line
def main():
"""Run the main script."""
parser, inspect_parser, query_parser = make_parser()
args = parser.parse_args()
# Extract data from the data files into structured Python objects.
database = NEODatabase(load_neos(args.neofile), load_approaches(args.cadfile))
# Run the chosen subcommand.
if args.cmd == 'inspect':
inspect(database, pdes=args.pdes, name=args.name, verbose=args.verbose)
elif args.cmd == 'query':
query(database, args)
elif args.cmd == 'interactive':
NEOShell(database, inspect_parser, query_parser, aggressive=args.aggressive).cmdloop()
if __name__ == '__main__':
main()

134
Exploring_Near_Earth_Objects/models.py Normal file
View File

@@ -0,0 +1,134 @@
"""Represent models for near-Earth objects and their close approaches.
The `NearEarthObject` class represents a near-Earth object. Each has a unique
primary designation, an optional unique name, an optional diameter, and a flag
for whether the object is potentially hazardous.
The `CloseApproach` class represents a close approach to Earth by an NEO. Each
has an approach datetime, a nominal approach distance, and a relative approach
velocity.
A `NearEarthObject` maintains a collection of its close approaches, and a
`CloseApproach` maintains a reference to its NEO.
The functions that construct these objects use information extracted from the
data files from NASA, so these objects should be able to handle all of the
quirks of the data set, such as missing names and unknown diameters.
You'll edit this file in Task 1.
"""
from helpers import cd_to_datetime, datetime_to_str
class NearEarthObject:
"""A near-Earth object (NEO).
An NEO encapsulates semantic and physical parameters about the object, such
as its primary designation (required, unique), IAU name (optional), diameter
in kilometers (optional - sometimes unknown), and whether it's marked as
potentially hazardous to Earth.
A `NearEarthObject` also maintains a collection of its close approaches -
initialized to an empty collection, but eventually populated in the
`NEODatabase` constructor.
"""
# TODO: How can you, and should you, change the arguments to this constructor?
# If you make changes, be sure to update the comments in this file.
def __init__(self, **info):
"""Create a new `NearEarthObject`.
:param info: A dictionary of excess keyword arguments supplied to the constructor.
"""
# TODO: Assign information from the arguments passed to the constructor
# onto attributes named `designation`, `name`, `diameter`, and `hazardous`.
# You should coerce these values to their appropriate data type and
# handle any edge cases, such as a empty name being represented by `None`
# and a missing diameter being represented by `float('nan')`.
self.designation = ''
self.name = None
self.diameter = float('nan')
self.hazardous = False
# Create an empty initial collection of linked approaches.
self.approaches = []
@property
def fullname(self):
"""Return a representation of the full name of this NEO."""
# TODO: Use self.designation and self.name to build a fullname for this object.
return ''
def __str__(self):
"""Return `str(self)`."""
# TODO: Use this object's attributes to return a human-readable string representation.
# The project instructions include one possibility. Peek at the __repr__
# method for examples of advanced string formatting.
return f"A NearEarthObject ..."
def __repr__(self):
"""Return `repr(self)`, a computer-readable string representation of this object."""
return f"NearEarthObject(designation={self.designation!r}, name={self.name!r}, " \
f"diameter={self.diameter:.3f}, hazardous={self.hazardous!r})"
class CloseApproach:
"""A close approach to Earth by an NEO.
A `CloseApproach` encapsulates information about the NEO's close approach to
Earth, such as the date and time (in UTC) of closest approach, the nominal
approach distance in astronomical units, and the relative approach velocity
in kilometers per second.
A `CloseApproach` also maintains a reference to its `NearEarthObject` -
initially, this information (the NEO's primary designation) is saved in a
private attribute, but the referenced NEO is eventually replaced in the
`NEODatabase` constructor.
"""
# TODO: How can you, and should you, change the arguments to this constructor?
# If you make changes, be sure to update the comments in this file.
def __init__(self, **info):
"""Create a new `CloseApproach`.
:param info: A dictionary of excess keyword arguments supplied to the constructor.
"""
# TODO: Assign information from the arguments passed to the constructor
# onto attributes named `_designation`, `time`, `distance`, and `velocity`.
# You should coerce these values to their appropriate data type and handle any edge cases.
# The `cd_to_datetime` function will be useful.
self._designation = ''
self.time = None # TODO: Use the cd_to_datetime function for this attribute.
self.distance = 0.0
self.velocity = 0.0
# Create an attribute for the referenced NEO, originally None.
self.neo = None
@property
def time_str(self):
"""Return a formatted representation of this `CloseApproach`'s approach time.
The value in `self.time` should be a Python `datetime` object. While a
`datetime` object has a string representation, the default representation
includes seconds - significant figures that don't exist in our input
data set.
The `datetime_to_str` method converts a `datetime` object to a
formatted string that can be used in human-readable representations and
in serialization to CSV and JSON files.
"""
# TODO: Use this object's `.time` attribute and the `datetime_to_str` function to
# build a formatted representation of the approach time.
# TODO: Use self.designation and self.name to build a fullname for this object.
return ''
def __str__(self):
"""Return `str(self)`."""
# TODO: Use this object's attributes to return a human-readable string representation.
# The project instructions include one possibility. Peek at the __repr__
# method for examples of advanced string formatting.
return f"A CloseApproach ..."
def __repr__(self):
"""Return `repr(self)`, a computer-readable string representation of this object."""
return f"CloseApproach(time={self.time_str!r}, distance={self.distance:.2f}, " \
f"velocity={self.velocity:.2f}, neo={self.neo!r})"

4
Exploring_Near_Earth_Objects/tests/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
"""Let Python know that the `tests/` folder is a package for Test Discovery [1].
[1]: https://docs.python.org/3/library/unittest.html#unittest-test-discovery
"""

61122
Exploring_Near_Earth_Objects/tests/test-cad-2020.json Normal file
View File

File diff suppressed because it is too large Load Diff

4227
Exploring_Near_Earth_Objects/tests/test-neos-2020.csv Normal file
View File

File diff suppressed because it is too large Load Diff

62
Exploring_Near_Earth_Objects/tests/test_data_files.py Normal file
View File

@@ -0,0 +1,62 @@
"""Check that the data files exist and are readable, nonempty, and well-formatted.
To run these tests from the project root, run:
$ python3 -m unittest --verbose tests.test_data_files
These tests should pass on the starter code.
"""
import collections
import csv
import json
import os
import pathlib
import unittest
# The root of the project, containing `main.py`.
PROJECT_ROOT = pathlib.Path(__file__).parent.parent.resolve()
class TestDataFiles(unittest.TestCase):
def setUp(self):
self.data_root = PROJECT_ROOT / 'data'
self.neo_file = self.data_root / 'neos.csv'
self.cad_file = self.data_root / 'cad.json'
def test_data_files_exist(self):
self.assertTrue(self.neo_file.exists())
self.assertTrue(self.cad_file.exists())
def test_data_files_are_readable(self):
self.assertTrue(os.access(self.neo_file, os.R_OK))
self.assertTrue(os.access(self.cad_file, os.R_OK))
def test_data_files_are_not_empty(self):
try:
self.assertTrue(self.neo_file.stat().st_size > 0, "Empty NEO file.")
self.assertTrue(self.cad_file.stat().st_size > 0, "Empty CAD file.")
except OSError:
self.fail("Unexpected OSError.")
def test_data_files_are_well_formatted(self):
# Check that the NEO data is CSV-formatted.
try:
with self.neo_file.open() as f:
# Consume the entire sequence into length-0 deque.
collections.deque(csv.reader(f), maxlen=0)
except csv.Error as err:
raise self.failureException(f"{self.neo_file!r} is not a well-formated CSV.") from err
# Check that the CAD data is JSON-formatted.
try:
with self.cad_file.open() as f:
json.load(f)
json.loads(self.cad_file.read_text())
except json.JSONDecodeError as err:
raise self.failureException(f"{self.cad_file!r} is not a valid JSON document.") from err
if __name__ == '__main__':
unittest.main()

119
Exploring_Near_Earth_Objects/tests/test_database.py Normal file
View File

@@ -0,0 +1,119 @@
"""Check that an `NEODatabase` can be constructed and responds to inspect queries.
The `NEODatabase` constructor should cross-link NEOs and their close approaches,
as well as prepare any additional metadata needed to support the `get_neo_by_*`
methods.
To run these tests from the project root, run:
$ python3 -m unittest --verbose tests.test_database
These tests should pass when Task 2 is complete.
"""
import pathlib
import math
import unittest
from extract import load_neos, load_approaches
from database import NEODatabase
# Paths to the test data files.
TESTS_ROOT = (pathlib.Path(__file__).parent).resolve()
TEST_NEO_FILE = TESTS_ROOT / 'test-neos-2020.csv'
TEST_CAD_FILE = TESTS_ROOT / 'test-cad-2020.json'
class TestDatabase(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.neos = load_neos(TEST_NEO_FILE)
cls.approaches = load_approaches(TEST_CAD_FILE)
cls.db = NEODatabase(cls.neos, cls.approaches)
def test_database_construction_links_approaches_to_neos(self):
for approach in self.approaches:
self.assertIsNotNone(approach.neo)
def test_database_construction_ensures_each_neo_has_an_approaches_attribute(self):
for neo in self.neos:
self.assertTrue(hasattr(neo, 'approaches'))
def test_database_construction_ensures_neos_collectively_exhaust_approaches(self):
approaches = set()
for neo in self.neos:
approaches.update(neo.approaches)
self.assertEqual(approaches, set(self.approaches))
def test_database_construction_ensures_neos_mutually_exclude_approaches(self):
seen = set()
for neo in self.neos:
for approach in neo.approaches:
if approach in seen:
self.fail(f"{approach} appears in the approaches of multiple NEOs.")
seen.add(approach)
def test_get_neo_by_designation(self):
cerberus = self.db.get_neo_by_designation('1865')
self.assertIsNotNone(cerberus)
self.assertEqual(cerberus.designation, '1865')
self.assertEqual(cerberus.name, 'Cerberus')
self.assertEqual(cerberus.diameter, 1.2)
self.assertEqual(cerberus.hazardous, False)
adonis = self.db.get_neo_by_designation('2101')
self.assertIsNotNone(adonis)
self.assertEqual(adonis.designation, '2101')
self.assertEqual(adonis.name, 'Adonis')
self.assertEqual(adonis.diameter, 0.60)
self.assertEqual(adonis.hazardous, True)
tantalus = self.db.get_neo_by_designation('2102')
self.assertIsNotNone(tantalus)
self.assertEqual(tantalus.designation, '2102')
self.assertEqual(tantalus.name, 'Tantalus')
self.assertEqual(tantalus.diameter, 1.649)
self.assertEqual(tantalus.hazardous, True)
def test_get_neo_by_designation_neos_with_year(self):
bs_2020 = self.db.get_neo_by_designation('2020 BS')
self.assertIsNotNone(bs_2020)
self.assertEqual(bs_2020.designation, '2020 BS')
self.assertEqual(bs_2020.name, None)
self.assertTrue(math.isnan(bs_2020.diameter))
self.assertEqual(bs_2020.hazardous, False)
py1_2020 = self.db.get_neo_by_designation('2020 PY1')
self.assertIsNotNone(py1_2020)
self.assertEqual(py1_2020.designation, '2020 PY1')
self.assertEqual(py1_2020.name, None)
self.assertTrue(math.isnan(py1_2020.diameter))
self.assertEqual(py1_2020.hazardous, False)
def test_get_neo_by_designation_missing(self):
nonexistent = self.db.get_neo_by_designation('not-real-designation')
self.assertIsNone(nonexistent)
def test_get_neo_by_name(self):
lemmon = self.db.get_neo_by_name('Lemmon')
self.assertIsNotNone(lemmon)
self.assertEqual(lemmon.designation, '2013 TL117')
self.assertEqual(lemmon.name, 'Lemmon')
self.assertTrue(math.isnan(lemmon.diameter))
self.assertEqual(lemmon.hazardous, False)
jormungandr = self.db.get_neo_by_name('Jormungandr')
self.assertIsNotNone(jormungandr)
self.assertEqual(jormungandr.designation, '471926')
self.assertEqual(jormungandr.name, 'Jormungandr')
self.assertTrue(math.isnan(jormungandr.diameter))
self.assertEqual(jormungandr.hazardous, True)
def test_get_neo_by_name_missing(self):
nonexistent = self.db.get_neo_by_name('not-real-name')
self.assertIsNone(nonexistent)
if __name__ == '__main__':
unittest.main()

122
Exploring_Near_Earth_Objects/tests/test_extract.py Normal file
View File

@@ -0,0 +1,122 @@
"""Check that data can be extracted from structured data files.
The `load_neos` function should load a collection of `NearEarthObject`s from a
CSV file, and the `load_approaches` function should load a collection of
`CloseApproach` objects from a JSON file.
To run these tests from the project root, run:
$ python3 -m unittest --verbose tests.test_extract
These tests should pass when Task 2 is complete.
"""
import collections.abc
import datetime
import pathlib
import math
import unittest
from extract import load_neos, load_approaches
from models import NearEarthObject, CloseApproach
TESTS_ROOT = (pathlib.Path(__file__).parent).resolve()
TEST_NEO_FILE = TESTS_ROOT / 'test-neos-2020.csv'
TEST_CAD_FILE = TESTS_ROOT / 'test-cad-2020.json'
class TestLoadNEOs(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.neos = load_neos(TEST_NEO_FILE)
cls.neos_by_designation = {neo.designation: neo for neo in cls.neos}
@classmethod
def get_first_neo_or_none(cls):
try:
# Don't use __getitem__ in case the object is a set or a stream.
return next(iter(cls.neos))
except StopIteration:
return None
def test_neos_are_collection(self):
self.assertIsInstance(self.neos, collections.abc.Collection)
def test_neos_contain_near_earth_objects(self):
neo = self.get_first_neo_or_none()
self.assertIsNotNone(neo)
self.assertIsInstance(neo, NearEarthObject)
def test_neos_contain_all_elements(self):
self.assertEqual(len(self.neos), 4226)
def test_neos_contain_2019_SC8_no_name_no_diameter(self):
self.assertIn('2019 SC8', self.neos_by_designation)
neo = self.neos_by_designation['2019 SC8']
self.assertEqual(neo.designation, '2019 SC8')
self.assertEqual(neo.name, None)
self.assertTrue(math.isnan(neo.diameter))
self.assertEqual(neo.hazardous, False)
def test_asclepius_has_name_no_diameter(self):
self.assertIn('4581', self.neos_by_designation)
neo = self.neos_by_designation['4581']
self.assertEqual(neo.designation, '4581')
self.assertEqual(neo.name, 'Asclepius')
self.assertTrue(math.isnan(neo.diameter))
self.assertEqual(neo.hazardous, True)
def test_adonis_is_potentially_hazardous(self):
self.assertIn('2101', self.neos_by_designation)
neo = self.neos_by_designation['2101']
self.assertEqual(neo.designation, '2101')
self.assertEqual(neo.name, 'Adonis')
self.assertEqual(neo.diameter, 0.6)
self.assertEqual(neo.hazardous, True)
class TestLoadApproaches(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.approaches = load_approaches(TEST_CAD_FILE)
@classmethod
def get_first_approach_or_none(cls):
try:
# Don't __getitem__, in case it's a set or a stream.
return next(iter(cls.approaches))
except StopIteration:
return None
def test_approaches_are_collection(self):
self.assertIsInstance(self.approaches, collections.abc.Collection)
def test_approaches_contain_close_approaches(self):
approach = self.get_first_approach_or_none()
self.assertIsNotNone(approach)
self.assertIsInstance(approach, CloseApproach)
def test_approaches_contain_all_elements(self):
self.assertEqual(len(self.approaches), 4700)
def test_approach_time_is_datetime(self):
approach = self.get_first_approach_or_none()
self.assertIsNotNone(approach)
self.assertIsInstance(approach.time, datetime.datetime)
def test_approach_distance_is_float(self):
approach = self.get_first_approach_or_none()
self.assertIsNotNone(approach)
self.assertIsInstance(approach.distance, float)
def test_approach_velocity_is_float(self):
approach = self.get_first_approach_or_none()
self.assertIsNotNone(approach)
self.assertIsInstance(approach.velocity, float)
if __name__ == '__main__':
unittest.main()

54
Exploring_Near_Earth_Objects/tests/test_limit.py Normal file
View File

@@ -0,0 +1,54 @@
"""Check that the `limit` function limits iterables.
To run these tests from the project root, run:
$ python3 -m unittest --verbose tests.test_limit
It isn't guaranteed that `limit` is a generator function - it's possible to
implement it imperatively with the tools from the `itertools` module.
These tests should pass when Task 3c is complete.
"""
import collections.abc
import unittest
from filters import limit
class TestLimit(unittest.TestCase):
def setUp(self):
self.iterable = tuple(range(5))
def test_limit_iterable_with_limit(self):
self.assertEqual(tuple(limit(self.iterable, 3)), (0, 1, 2))
def test_limit_iterable_without_limit(self):
self.assertEqual(tuple(limit(self.iterable)), (0, 1, 2, 3, 4))
self.assertEqual(tuple(limit(self.iterable, 0)), (0, 1, 2, 3, 4))
self.assertEqual(tuple(limit(self.iterable, None)), (0, 1, 2, 3, 4))
def test_limit_iterator_with_smaller_limit(self):
self.assertEqual(tuple(limit(iter(self.iterable), 3)), (0, 1, 2))
def test_limit_iterator_with_matching_limit(self):
self.assertEqual(tuple(limit(iter(self.iterable), 5)), (0, 1, 2, 3, 4))
def test_limit_iterator_with_larger_limit(self):
self.assertEqual(tuple(limit(iter(self.iterable), 10)), (0, 1, 2, 3, 4))
def test_limit_iterator_without_limit(self):
self.assertEqual(tuple(limit(iter(self.iterable))), (0, 1, 2, 3, 4))
self.assertEqual(tuple(limit(iter(self.iterable), 0)), (0, 1, 2, 3, 4))
self.assertEqual(tuple(limit(iter(self.iterable), None)), (0, 1, 2, 3, 4))
def test_limit_produces_an_iterable(self):
self.assertIsInstance(limit(self.iterable, 3), collections.abc.Iterable)
self.assertIsInstance(limit(self.iterable, 5), collections.abc.Iterable)
self.assertIsInstance(limit(self.iterable, 10), collections.abc.Iterable)
self.assertIsInstance(limit(self.iterable), collections.abc.Iterable)
self.assertIsInstance(limit(self.iterable, 0), collections.abc.Iterable)
self.assertIsInstance(limit(self.iterable, None), collections.abc.Iterable)
if __name__ == '__main__':
unittest.main()

49
Exploring_Near_Earth_Objects/tests/test_python_version.py Normal file
View File

@@ -0,0 +1,49 @@
"""Check that the Python version is at least up to a minimum threshold of 3.6.
The instructions explicitly invoke each command using `python3` on the command
line, but a student's local setup might not default to using Python 3.6+, which
is required for this project. Additionally, some students may accidentally be in
the habit of using bare `python`, which could invoke Python 2.x if their
environment isn't set up correctly.
Other modules in this project aggressively assume Python 3.6+, so this unit test
is our only cession to the possibility that students might be running a lower
version of Python.
To run these tests from the project root, run:
$ python3 -m unittest --verbose tests.test_python_version
These tests should (successfully) fail, but not crash, when invoked with Python 2:
$ /usr/bin/python2.7 -m unittest --verbose tests.test_python_version
"""
import sys
import unittest
class TestPythonVersion(unittest.TestCase):
"""Check that the Python version is >= 3.6."""
def test_python_version_is_at_least_3_6(self):
self.assertTrue(sys.version_info >= (3, 6),
msg="""Unsupported Python version.
It looks like you're using a version of Python that's too old.
This project requires Python 3.6+. You're currently using Python {}.{}.{}.
Make sure that you have a compatible version of Python and that you're using
`python3` at the command-line (or that your environment resolves `python` to
some Python3.6+ version if you have a custom setup).
Remember, you can always ask Python to display its version with:
$ python3 -V
Python 3.X.Y
""".format(*sys.version_info[:3]))
if __name__ == '__main__':
unittest.main()

533
Exploring_Near_Earth_Objects/tests/test_query.py Normal file
View File

@@ -0,0 +1,533 @@
"""Check that `query`ing an `NEODatabase` accurately produces close approaches.
There are a plethora of ways to combine the arguments to `create_filters`, which
correspond to different command-line options. This modules tests the options in
isolation, in pairs, and in more complicated combinations. Althought the tests
are not entirely exhaustive, any implementation that passes all of these tests
is most likely up to snuff.
To run these tests from the project root, run::
$ python3 -m unittest --verbose tests.test_query
These tests should pass when Tasks 3a and 3b are complete.
"""
import datetime
import pathlib
import unittest
from database import NEODatabase
from extract import load_neos, load_approaches
from filters import create_filters
TESTS_ROOT = (pathlib.Path(__file__).parent).resolve()
TEST_NEO_FILE = TESTS_ROOT / 'test-neos-2020.csv'
TEST_CAD_FILE = TESTS_ROOT / 'test-cad-2020.json'
class TestQuery(unittest.TestCase):
# Set longMessage to True to enable lengthy diffs between set comparisons.
longMessage = False
@classmethod
def setUpClass(cls):
cls.neos = load_neos(TEST_NEO_FILE)
cls.approaches = load_approaches(TEST_CAD_FILE)
cls.db = NEODatabase(cls.neos, cls.approaches)
def test_query_all(self):
expected = set(self.approaches)
self.assertGreater(len(expected), 0)
filters = create_filters()
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
###############################################
# Single filters and pairs of related filters #
###############################################
def test_query_approaches_on_march_2(self):
date = datetime.date(2020, 3, 2)
expected = set(
approach for approach in self.approaches
if approach.time.date() == date
)
self.assertGreater(len(expected), 0)
filters = create_filters(date=date)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_after_april(self):
start_date = datetime.date(2020, 4, 1)
expected = set(
approach for approach in self.approaches
if start_date <= approach.time.date()
)
self.assertGreater(len(expected), 0)
filters = create_filters(start_date=start_date)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_before_july(self):
end_date = datetime.date(2020, 6, 30)
expected = set(
approach for approach in self.approaches
if approach.time.date() <= end_date
)
self.assertGreater(len(expected), 0)
filters = create_filters(end_date=end_date)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_in_march(self):
start_date = datetime.date(2020, 3, 1)
end_date = datetime.date(2020, 3, 31)
expected = set(
approach for approach in self.approaches
if start_date <= approach.time.date() <= end_date
)
self.assertGreater(len(expected), 0)
filters = create_filters(start_date=start_date, end_date=end_date)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_conflicting_date_bounds(self):
start_date = datetime.date(2020, 10, 1)
end_date = datetime.date(2020, 4, 1)
expected = set()
filters = create_filters(start_date=start_date, end_date=end_date)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_bounds_and_a_specific_date(self):
start_date = datetime.date(2020, 2, 1)
date = datetime.date(2020, 3, 2)
end_date = datetime.date(2020, 4, 1)
expected = set(
approach for approach in self.approaches
if approach.time.date() == date
)
self.assertGreater(len(expected), 0)
filters = create_filters(date=date, start_date=start_date, end_date=end_date)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_max_distance(self):
distance_max = 0.4
expected = set(
approach for approach in self.approaches
if approach.distance <= distance_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(distance_max=distance_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_min_distance(self):
distance_min = 0.1
expected = set(
approach for approach in self.approaches
if distance_min <= approach.distance
)
self.assertGreater(len(expected), 0)
filters = create_filters(distance_min=distance_min)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_max_distance_and_min_distance(self):
distance_max = 0.4
distance_min = 0.1
expected = set(
approach for approach in self.approaches
if distance_min <= approach.distance <= distance_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(distance_min=distance_min, distance_max=distance_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_max_distance_and_min_distance_conflicting(self):
distance_max = 0.1
distance_min = 0.4
expected = set()
filters = create_filters(distance_min=distance_min, distance_max=distance_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_max_velocity(self):
velocity_max = 20
expected = set(
approach for approach in self.approaches
if approach.velocity <= velocity_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(velocity_max=velocity_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_min_velocity(self):
velocity_min = 10
expected = set(
approach for approach in self.approaches
if velocity_min <= approach.velocity
)
self.assertGreater(len(expected), 0)
filters = create_filters(velocity_min=velocity_min)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_max_velocity_and_min_velocity(self):
velocity_max = 20
velocity_min = 10
expected = set(
approach for approach in self.approaches
if velocity_min <= approach.velocity <= velocity_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(velocity_min=velocity_min, velocity_max=velocity_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_max_velocity_and_min_velocity_conflicting(self):
velocity_max = 10
velocity_min = 20
expected = set()
filters = create_filters(velocity_min=velocity_min, velocity_max=velocity_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_max_diameter(self):
diameter_max = 1.5
expected = set(
approach for approach in self.approaches
if approach.neo.diameter <= diameter_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(diameter_max=diameter_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_min_diameter(self):
diameter_min = 0.5
expected = set(
approach for approach in self.approaches
if diameter_min <= approach.neo.diameter
)
self.assertGreater(len(expected), 0)
filters = create_filters(diameter_min=diameter_min)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_max_diameter_and_min_diameter(self):
diameter_max = 1.5
diameter_min = 0.5
expected = set(
approach for approach in self.approaches
if diameter_min <= approach.neo.diameter <= diameter_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(diameter_min=diameter_min, diameter_max=diameter_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_max_diameter_and_min_diameter_conflicting(self):
diameter_max = 0.5
diameter_min = 1.5
expected = set()
filters = create_filters(diameter_min=diameter_min, diameter_max=diameter_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_hazardous(self):
expected = set(
approach for approach in self.approaches
if approach.neo.hazardous
)
self.assertGreater(len(expected), 0)
filters = create_filters(hazardous=True)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_with_not_hazardous(self):
expected = set(
approach for approach in self.approaches
if not approach.neo.hazardous
)
self.assertGreater(len(expected), 0)
filters = create_filters(hazardous=False)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
###########################
# Combinations of filters #
###########################
def test_query_approaches_on_march_2_with_max_distance(self):
date = datetime.date(2020, 3, 2)
distance_max = 0.4
expected = set(
approach for approach in self.approaches
if approach.time.date() == date
and approach.distance <= distance_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(date=date, distance_max=distance_max)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_on_march_2_with_min_distance(self):
date = datetime.date(2020, 3, 2)
distance_min = 0.1
expected = set(
approach for approach in self.approaches
if approach.time.date() == date
and distance_min <= approach.distance
)
self.assertGreater(len(expected), 0)
filters = create_filters(date=date, distance_min=distance_min)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_in_march_with_min_distance_and_max_distance(self):
start_date = datetime.date(2020, 3, 1)
end_date = datetime.date(2020, 3, 31)
distance_max = 0.4
distance_min = 0.1
expected = set(
approach for approach in self.approaches
if start_date <= approach.time.date() <= end_date
and distance_min <= approach.distance <= distance_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(
start_date=start_date, end_date=end_date,
distance_min=distance_min, distance_max=distance_max,
)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_in_march_with_distance_bounds_and_max_velocity(self):
start_date = datetime.date(2020, 3, 1)
end_date = datetime.date(2020, 3, 31)
distance_max = 0.4
distance_min = 0.1
velocity_max = 20
expected = set(
approach for approach in self.approaches
if start_date <= approach.time.date() <= end_date
and distance_min <= approach.distance <= distance_max
and approach.velocity <= velocity_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(
start_date=start_date, end_date=end_date,
distance_min=distance_min, distance_max=distance_max,
velocity_max=velocity_max
)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_in_march_with_distance_and_velocity_bounds(self):
start_date = datetime.date(2020, 3, 1)
end_date = datetime.date(2020, 3, 31)
distance_max = 0.4
distance_min = 0.1
velocity_max = 20
velocity_min = 10
expected = set(
approach for approach in self.approaches
if start_date <= approach.time.date() <= end_date
and distance_min <= approach.distance <= distance_max
and velocity_min <= approach.velocity <= velocity_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(
start_date=start_date, end_date=end_date,
distance_min=distance_min, distance_max=distance_max,
velocity_min=velocity_min, velocity_max=velocity_max
)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_in_spring_with_distance_and_velocity_bounds_and_max_diameter(self):
start_date = datetime.date(2020, 3, 1)
end_date = datetime.date(2020, 5, 31)
distance_max = 0.5
distance_min = 0.05
velocity_max = 25
velocity_min = 5
diameter_max = 1.5
expected = set(
approach for approach in self.approaches
if start_date <= approach.time.date() <= end_date
and distance_min <= approach.distance <= distance_max
and velocity_min <= approach.velocity <= velocity_max
and approach.neo.diameter <= diameter_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(
start_date=start_date, end_date=end_date,
distance_min=distance_min, distance_max=distance_max,
velocity_min=velocity_min, velocity_max=velocity_max,
diameter_max=diameter_max
)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_in_spring_with_distance_velocity_and_diameter_bounds(self):
start_date = datetime.date(2020, 3, 1)
end_date = datetime.date(2020, 5, 31)
distance_max = 0.5
distance_min = 0.05
velocity_max = 25
velocity_min = 5
diameter_max = 1.5
diameter_min = 0.5
expected = set(
approach for approach in self.approaches
if start_date <= approach.time.date() <= end_date
and distance_min <= approach.distance <= distance_max
and velocity_min <= approach.velocity <= velocity_max
and diameter_min <= approach.neo.diameter <= diameter_max
)
self.assertGreater(len(expected), 0)
filters = create_filters(
start_date=start_date, end_date=end_date,
distance_min=distance_min, distance_max=distance_max,
velocity_min=velocity_min, velocity_max=velocity_max,
diameter_min=diameter_min, diameter_max=diameter_max
)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_in_spring_with_all_bounds_and_potentially_hazardous_neos(self):
start_date = datetime.date(2020, 3, 1)
end_date = datetime.date(2020, 5, 31)
distance_max = 0.5
distance_min = 0.05
velocity_max = 25
velocity_min = 5
diameter_max = 1.5
diameter_min = 0.5
expected = set(
approach for approach in self.approaches
if start_date <= approach.time.date() <= end_date
and distance_min <= approach.distance <= distance_max
and velocity_min <= approach.velocity <= velocity_max
and diameter_min <= approach.neo.diameter <= diameter_max
and approach.neo.hazardous
)
self.assertGreater(len(expected), 0)
filters = create_filters(
start_date=start_date, end_date=end_date,
distance_min=distance_min, distance_max=distance_max,
velocity_min=velocity_min, velocity_max=velocity_max,
diameter_min=diameter_min, diameter_max=diameter_max,
hazardous=True
)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
def test_query_approaches_in_spring_with_all_bounds_and_not_potentially_hazardous_neos(self):
start_date = datetime.date(2020, 3, 1)
end_date = datetime.date(2020, 5, 31)
distance_max = 0.5
distance_min = 0.05
velocity_max = 25
velocity_min = 5
diameter_max = 1.5
diameter_min = 0.5
expected = set(
approach for approach in self.approaches
if start_date <= approach.time.date() <= end_date
and distance_min <= approach.distance <= distance_max
and velocity_min <= approach.velocity <= velocity_max
and diameter_min <= approach.neo.diameter <= diameter_max
and not approach.neo.hazardous
)
self.assertGreater(len(expected), 0)
filters = create_filters(
start_date=start_date, end_date=end_date,
distance_min=distance_min, distance_max=distance_max,
velocity_min=velocity_min, velocity_max=velocity_max,
diameter_min=diameter_min, diameter_max=diameter_max,
hazardous=False
)
received = set(self.db.query(filters))
self.assertEqual(expected, received, msg="Computed results do not match expected results.")
if __name__ == '__main__':
unittest.main()

230
Exploring_Near_Earth_Objects/tests/test_write.py Normal file
View File

@@ -0,0 +1,230 @@
"""Check that streams of results can be written to files.
The `write_to_csv` and `write_to_json` methods should follow a specific output
format, described in the project instructions.
There's some sketchy file-like manipulation in order to avoid writing anything
to disk and avoid letting a context manager in the implementation eagerly close
the in-memory file - so be warned that the workaround is gnarly.
To run these tests from the project root, run:
$ python3 -m unittest --verbose tests.test_write
These tests should pass when Task 4 is complete.
"""
import collections
import collections.abc
import contextlib
import csv
import datetime
import io
import json
import pathlib
import unittest
import unittest.mock
from extract import load_neos, load_approaches
from database import NEODatabase
from write import write_to_csv, write_to_json
TESTS_ROOT = (pathlib.Path(__file__).parent).resolve()
TEST_NEO_FILE = TESTS_ROOT / 'test-neos-2020.csv'
TEST_CAD_FILE = TESTS_ROOT / 'test-cad-2020.json'
def build_results(n):
neos = tuple(load_neos(TEST_NEO_FILE))
approaches = tuple(load_approaches(TEST_CAD_FILE))
# Only needed to link together these objects.
NEODatabase(neos, approaches)
return approaches[:n]
@contextlib.contextmanager
def UncloseableStringIO(value=''):
"""A context manager for an uncloseable `io.StringIO`.
This produces an almost-normal `io.StringIO`, except the `close` method has
been patched out with a no-op. The context manager takes care of restoring
the monkeypatch and closing the buffer, but this prevents other nested
context managers (such as `open` from the implementation of `write_to_*`)
from preemptively closing the `StringIO` before we can rewind it and read
its value.
"""
buf = io.StringIO(value)
buf._close = buf.close
buf.close = lambda: False
yield buf
buf.close = buf._close
delattr(buf, '_close')
buf.close()
class TestWriteToCSV(unittest.TestCase):
@classmethod
@unittest.mock.patch('write.open')
def setUpClass(cls, mock_file):
results = build_results(5)
with UncloseableStringIO() as buf:
mock_file.return_value = buf
try:
write_to_csv(results, None)
except csv.Error as err:
raise cls.failureException("Unable to write results to CSV.") from err
except ValueError as err:
raise cls.failureException("Unexpected failure while writing to CSV.") from err
else:
# Rewind the unclosed buffer to save its contents.
buf.seek(0)
cls.value = buf.getvalue()
def test_csv_data_is_well_formed(self):
# Now, we have the value in memory, and can _actually_ start testing.
buf = io.StringIO(self.value)
# Check that the output is well-formed.
try:
# Consume the output and immediately discard it.
collections.deque(csv.DictReader(buf), maxlen=0)
except csv.Error as err:
raise self.failureException("write_to_csv produced an invalid CSV format.") from err
def test_csv_data_has_header(self):
try:
self.assertTrue(csv.Sniffer().has_header(self.value))
return
except csv.Error as err:
raise self.failureException("Unable to sniff for headers.") from err
def test_csv_data_has_five_rows(self):
# Now, we have the value in memory, and can _actually_ start testing.
buf = io.StringIO(self.value)
# Check that the output is well-formed.
try:
reader = csv.DictReader(buf)
rows = tuple(reader)
except csv.Error as err:
raise self.failureException("write_to_csv produced an invalid CSV format.") from err
self.assertEqual(len(rows), 5)
def test_csv_data_header_matches_requirements(self):
# Now, we have the value in memory, and can _actually_ start testing.
buf = io.StringIO(self.value)
# Check that the output is well-formed.
try:
reader = csv.DictReader(buf)
rows = tuple(reader)
except csv.Error as err:
raise self.failureException("write_to_csv produced an invalid CSV format.") from err
fieldnames = ('datetime_utc', 'distance_au', 'velocity_km_s', 'designation', 'name', 'diameter_km', 'potentially_hazardous')
self.assertGreater(len(rows), 0)
self.assertSetEqual(set(fieldnames), set(rows[0].keys()))
class TestWriteToJSON(unittest.TestCase):
@classmethod
@unittest.mock.patch('write.open')
def setUpClass(cls, mock_file):
results = build_results(5)
with UncloseableStringIO() as buf:
mock_file.return_value = buf
try:
write_to_json(results, None)
except csv.Error as err:
raise cls.failureException("Unable to write results to CSV.") from err
except ValueError as err:
raise cls.failureException("Unexpected failure while writing to CSV.") from err
else:
# Rewind the unclosed buffer to fetch the contents saved to "disk".
buf.seek(0)
cls.value = buf.getvalue()
def test_json_data_is_well_formed(self):
# Now, we have the value in memory, and can _actually_ start testing.
buf = io.StringIO(self.value)
try:
json.load(buf)
except json.JSONDecodeError as err:
raise self.failureException("write_to_json produced an invalid JSON document") from err
def test_json_data_is_a_sequence(self):
buf = io.StringIO(self.value)
try:
data = json.load(buf)
except json.JSONDecodeError as err:
raise self.failureException("write_to_json produced an invalid JSON document") from err
self.assertIsInstance(data, collections.abc.Sequence)
def test_json_data_has_five_elements(self):
buf = io.StringIO(self.value)
try:
data = json.load(buf)
except json.JSONDecodeError as err:
raise self.failureException("write_to_json produced an invalid JSON document") from err
self.assertEqual(len(data), 5)
def test_json_element_is_associative(self):
buf = io.StringIO(self.value)
try:
data = json.load(buf)
except json.JSONDecodeError as err:
raise self.failureException("write_to_json produced an invalid JSON document") from err
approach = data[0]
self.assertIsInstance(approach, collections.abc.Mapping)
def test_json_element_has_nested_attributes(self):
buf = io.StringIO(self.value)
try:
data = json.load(buf)
except json.JSONDecodeError as err:
raise self.failureException("write_to_json produced an invalid JSON document") from err
approach = data[0]
self.assertIn('datetime_utc', approach)
self.assertIn('distance_au', approach)
self.assertIn('velocity_km_s', approach)
self.assertIn('neo', approach)
neo = approach['neo']
self.assertIn('designation', neo)
self.assertIn('name', neo)
self.assertIn('diameter_km', neo)
self.assertIn('potentially_hazardous', neo)
def test_json_element_decodes_to_correct_types(self):
buf = io.StringIO(self.value)
try:
data = json.load(buf)
except json.JSONDecodeError as err:
raise self.failureException("write_to_json produced an invalid JSON document") from err
approach = data[0]
try:
datetime.datetime.strptime(approach['datetime_utc'], '%Y-%m-%d %H:%M')
except ValueError:
self.fail("The `datetime_utc` key isn't in YYYY-MM-DD HH:MM` format.")
self.assertIsInstance(approach['distance_au'], float)
self.assertIsInstance(approach['velocity_km_s'], float)
self.assertIsInstance(approach['neo']['designation'], str)
self.assertNotEqual(approach['neo']['name'], 'None')
if approach['neo']['name']:
self.assertIsInstance(approach['neo']['name'], str)
self.assertIsInstance(approach['neo']['diameter_km'], float)
self.assertIsInstance(approach['neo']['potentially_hazardous'], bool)
if __name__ == '__main__':
unittest.main()

45
Exploring_Near_Earth_Objects/write.py Normal file
View File

@@ -0,0 +1,45 @@
"""Write a stream of close approaches to CSV or to JSON.
This module exports two functions: `write_to_csv` and `write_to_json`, each of
which accept an `results` stream of close approaches and a path to which to
write the data.
These functions are invoked by the main module with the output of the `limit`
function and the filename supplied by the user at the command line. The file's
extension determines which of these functions is used.
You'll edit this file in Part 4.
"""
import csv
import json
def write_to_csv(results, filename):
"""Write an iterable of `CloseApproach` objects to a CSV file.
The precise output specification is in `README.md`. Roughly, each output row
corresponds to the information in a single close approach from the `results`
stream and its associated near-Earth object.
:param results: An iterable of `CloseApproach` objects.
:param filename: A Path-like object pointing to where the data should be saved.
"""
fieldnames = (
'datetime_utc', 'distance_au', 'velocity_km_s',
'designation', 'name', 'diameter_km', 'potentially_hazardous'
)
# TODO: Write the results to a CSV file, following the specification in the instructions.
def write_to_json(results, filename):
"""Write an iterable of `CloseApproach` objects to a JSON file.
The precise output specification is in `README.md`. Roughly, the output is a
list containing dictionaries, each mapping `CloseApproach` attributes to
their values and the 'neo' key mapping to a dictionary of the associated
NEO's attributes.
:param results: An iterable of `CloseApproach` objects.
:param filename: A Path-like object pointing to where the data should be saved.
"""
# TODO: Write the results to a JSON file, following the specification in the instructions.