diff --git a/database.py b/database.py index 910932c..597ee0d 100644 --- a/database.py +++ b/database.py @@ -10,89 +10,99 @@ `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? +A database encapsulating collections of near-Earth objects and their close approaches. - # TODO: Link together the NEOs and their close approaches. +NEODatabase links NEOs with their close approaches and provides fast lookups +by designation or name, plus a streaming query that yields matching approaches. +""" - def get_neo_by_designation(self, designation): - """Find and return an NEO by its primary designation. +from __future__ import annotations - If no match is found, return `None` instead. +from typing import Dict, Iterable, Iterator, List, Optional, Sequence - Each NEO in the data set has a unique primary designation, as a string. +from models import NearEarthObject, CloseApproach - 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`. +class NEODatabase: + """A database of near-Earth objects and their close approaches.""" + + def __init__(self, neos: Sequence[NearEarthObject], approaches: Sequence[CloseApproach]) -> None: + """Create a new NEODatabase and link NEOs with their close approaches. + + Parameters + ---------- + neos : Sequence[NearEarthObject] + Collection of NEOs (unlinked). + approaches : Sequence[CloseApproach] + Collection of close approaches (unlinked, with '_designation' present). + + Notes + ----- + After construction: + - Each CloseApproach.neo references its matching NearEarthObject. + - Each NearEarthObject.approaches contains its approaches. + - Auxiliary indices are built for designation and name. """ - # TODO: Fetch an NEO by its primary designation. - return None + self._neos: List[NearEarthObject] = list(neos) + self._approaches: List[CloseApproach] = list(approaches) - def get_neo_by_name(self, name): - """Find and return an NEO by its name. + # Auxiliary indices + self._by_des: Dict[str, NearEarthObject] = {neo.designation: neo for neo in self._neos} + self._by_name: Dict[str, NearEarthObject] = {neo.name: neo for neo in self._neos if neo.name} - If no match is found, return `None` instead. + # Link approaches <-> neos + for app in self._approaches: + neo = self._by_des.get(app._designation) + if neo: + app.neo = neo + neo.approaches.append(app) - Not every NEO in the data set has a name. No NEOs are associated with - the empty string nor with the `None` singleton. + def get_neo_by_designation(self, designation: str) -> Optional[NearEarthObject]: + """Find and return an NEO by its primary designation. - The matching is exact - check for spelling and capitalization if no - match is found. + Parameters + ---------- + designation : str + The primary designation to search for. - :param name: The name, as a string, of the NEO to search for. - :return: The `NearEarthObject` with the desired name, or `None`. + Returns + ------- + Optional[NearEarthObject] + The NEO with the given designation, or None if not found. """ - # TODO: Fetch an NEO by its name. - return None + return self._by_des.get(designation) - def query(self, filters=()): - """Query close approaches to generate those that match a collection of filters. + def get_neo_by_name(self, name: str) -> Optional[NearEarthObject]: + """Find and return an NEO by its name. + + Parameters + ---------- + name : str + The name to search for. - This generates a stream of `CloseApproach` objects that match all of the - provided filters. + Returns + ------- + Optional[NearEarthObject] + The NEO with the given name, or None if not found. + """ + return self._by_name.get(name) - If no arguments are provided, generate all known close approaches. + def query(self, filters: Sequence) -> Iterator[CloseApproach]: + """Yield close approaches that satisfy *all* filters. - The `CloseApproach` objects are generated in internal order, which isn't - guaranteed to be sorted meaningfully, although is often sorted by time. + Parameters + ---------- + filters : Sequence[AttributeFilter] + A collection of filter predicates. - :param filters: A collection of filters capturing user-specified criteria. - :return: A stream of matching `CloseApproach` objects. + Yields + ------ + CloseApproach + Matching approaches, lazily (no precomputation of full result set). """ - # TODO: Generate `CloseApproach` objects that match all of the filters. - for approach in self._approaches: - yield approach + for app in self._approaches: + if all(f(app) for f in filters): + yield app diff --git a/extract.py b/extract.py index 59f7192..e654e59 100644 --- a/extract.py +++ b/extract.py @@ -11,28 +11,84 @@ line, and uses the resulting collections to build an `NEODatabase`. You'll edit this file in Task 2. + +---------------------------------------------------------------------------- + +Extract data on near-Earth objects and close approaches from CSV and JSON files. + +- load_neos: read NEOs from CSV and return a collection of NearEarthObject. +- load_approaches: read close approaches from JSON and return a collection of CloseApproach. """ + +from __future__ import annotations + import csv import json +from pathlib import Path +from typing import Iterable, List from models import NearEarthObject, CloseApproach -def load_neos(neo_csv_path): +def load_neos(neo_csv_path: Path) -> List[NearEarthObject]: """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. + Parameters + ---------- + neo_csv_path : Path + Path to the CSV file containing near-Earth objects. + + Returns + ------- + List[NearEarthObject] + A list of NEO instances. """ - # TODO: Load NEO data from the given CSV file. - return () + neos: List[NearEarthObject] = [] + with open(neo_csv_path, mode="r", encoding="utf-8") as f: + reader = csv.DictReader(f) + for row in reader: + # Keep only the fields we care about; ignore extraneous columns + info = { + "pdes": row.get("pdes"), + "name": row.get("name"), + "diameter": row.get("diameter"), + "pha": row.get("pha"), + } + neos.append(NearEarthObject(**info)) + return neos -def load_approaches(cad_json_path): +def load_approaches(cad_json_path: Path) -> List[CloseApproach]: """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. + The NASA CAD file uses a 'fields' header list and a 'data' body list of rows. + We map indices for 'des', 'cd', 'dist', 'v_rel' and build CloseApproach objects. + + Parameters + ---------- + cad_json_path : Path + Path to the JSON file containing close approach data. + + Returns + ------- + List[CloseApproach] + A list of CloseApproach instances. """ - # TODO: Load close approach data from the given JSON file. - return () + approaches: List[CloseApproach] = [] + with open(cad_json_path, mode="r", encoding="utf-8") as f: + raw = json.load(f) + + fields = raw.get("fields", []) + data_rows = raw.get("data", []) + + idx = {name: i for i, name in enumerate(fields)} + # Expected field names in NASA CAD: 'des', 'cd', 'dist', 'v_rel' + for row in data_rows: + info = { + "des": row[idx.get("des", -1)] if idx.get("des") is not None else None, + "cd": row[idx.get("cd", -1)] if idx.get("cd") is not None else None, + "dist": row[idx.get("dist", -1)] if idx.get("dist") is not None else None, + "v_rel": row[idx.get("v_rel", -1)] if idx.get("v_rel") is not None else None, + } + approaches.append(CloseApproach(**info)) + return approaches diff --git a/filters.py b/filters.py index 61e09b3..5944720 100644 --- a/filters.py +++ b/filters.py @@ -15,111 +15,323 @@ iterator. You'll edit this file in Tasks 3a and 3c. + +---------------------------------------------------------------------------- + +Provide filters for querying close approaches and limit the generated results. + +- AttributeFilter: base class wrapping an operator and a reference value. +- Subclasses override `get(approach)` to return attribute to compare. +- create_filters: build a collection of filters from CLI args. +- limit: yield at most n values from an iterator (stream-safe). """ -import operator +from __future__ import annotations -class UnsupportedCriterionError(NotImplementedError): - """A filter criterion is unsupported.""" +import operator +from typing import Callable, Iterable, Iterator, List, Optional, Sequence +from models import CloseApproach -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. +class UnsupportedCriterionError(NotImplementedError): + """Exception raised when a filter criterion is unsupported.""" - 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. +class AttributeFilter: + """A general superclass for filters on comparable attributes.""" - 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`. + def __init__(self, op: Callable, value) -> None: + """Construct an AttributeFilter from a binary predicate and a reference value. - :param op: A 2-argument predicate comparator (such as `operator.le`). - :param value: The reference value to compare against. + Parameters + ---------- + op : Callable + A binary comparison operator (e.g., operator.eq, operator.ge). + value : Any + The reference value to compare against. """ self.op = op self.value = value - def __call__(self, approach): - """Invoke `self(approach)`.""" + def __call__(self, approach: CloseApproach) -> bool: + """Return whether the approach satisfies this filter. + + Parameters + ---------- + approach : CloseApproach + The close approach to evaluate. + + Returns + ------- + bool + True if the approach satisfies the filter criterion. + """ return self.op(self.get(approach), self.value) @classmethod - def get(cls, approach): + def get(cls, approach: CloseApproach): """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`. + Parameters + ---------- + approach : CloseApproach + The close approach from which to extract an attribute. - :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`. + Returns + ------- + Any + The attribute value to be compared. + + Raises + ------ + UnsupportedCriterionError + This base implementation always raises this error. """ raise UnsupportedCriterionError - def __repr__(self): + def __repr__(self) -> str: + """Return a string representation of this AttributeFilter.""" return f"{self.__class__.__name__}(op=operator.{self.op.__name__}, value={self.value})" +# ---- Concrete filters ---- + +class DateFilter(AttributeFilter): + """Filter close approaches by exact date.""" + + @classmethod + def get(cls, approach: CloseApproach): + """Extract the date from the approach's time. + + Parameters + ---------- + approach : CloseApproach + The close approach. + + Returns + ------- + date or None + The date of the approach, or None if time is missing. + """ + return approach.time.date() if approach.time else None + + +class StartDateFilter(AttributeFilter): + """Filter close approaches by start date (inclusive).""" + + @classmethod + def get(cls, approach: CloseApproach): + """Extract the date from the approach's time. + + Parameters + ---------- + approach : CloseApproach + The close approach. + + Returns + ------- + date or None + The date of the approach, or None if time is missing. + """ + return approach.time.date() if approach.time else None + + +class EndDateFilter(AttributeFilter): + """Filter close approaches by end date (inclusive).""" + + @classmethod + def get(cls, approach: CloseApproach): + """Extract the date from the approach's time. + + Parameters + ---------- + approach : CloseApproach + The close approach. + + Returns + ------- + date or None + The date of the approach, or None if time is missing. + """ + return approach.time.date() if approach.time else None + + +class DistanceFilter(AttributeFilter): + """Filter close approaches by distance.""" + + @classmethod + def get(cls, approach: CloseApproach): + """Extract the distance from the approach. + + Parameters + ---------- + approach : CloseApproach + The close approach. + + Returns + ------- + float + The distance in astronomical units. + """ + return approach.distance + + +class VelocityFilter(AttributeFilter): + """Filter close approaches by velocity.""" + + @classmethod + def get(cls, approach: CloseApproach): + """Extract the velocity from the approach. + + Parameters + ---------- + approach : CloseApproach + The close approach. + + Returns + ------- + float + The velocity in km/s. + """ + return approach.velocity + + +class DiameterFilter(AttributeFilter): + """Filter close approaches by NEO diameter.""" + + @classmethod + def get(cls, approach: CloseApproach): + """Extract the diameter from the approach's NEO. + + Parameters + ---------- + approach : CloseApproach + The close approach. + + Returns + ------- + float + The NEO's diameter in km, or NaN if unknown or NEO is missing. + """ + return approach.neo.diameter if approach.neo else float('nan') + + +class HazardousFilter(AttributeFilter): + """Filter close approaches by NEO hazardous status.""" + + @classmethod + def get(cls, approach: CloseApproach): + """Extract the hazardous flag from the approach's NEO. + + Parameters + ---------- + approach : CloseApproach + The close approach. + + Returns + ------- + bool + True if the NEO is potentially hazardous, False otherwise. + """ + return approach.neo.hazardous if approach.neo else False + + 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 -): + 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 +) -> Sequence[AttributeFilter]: """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`. + Parameters + ---------- + date : date, optional + Filter by exact date. + start_date : date, optional + Filter by start date (inclusive). + end_date : date, optional + Filter by end date (inclusive). + distance_min : float, optional + Minimum distance in AU. + distance_max : float, optional + Maximum distance in AU. + velocity_min : float, optional + Minimum velocity in km/s. + velocity_max : float, optional + Maximum velocity in km/s. + diameter_min : float, optional + Minimum diameter in km. + diameter_max : float, optional + Maximum diameter in km. + hazardous : bool, optional + Filter by potentially hazardous status. + + Returns + ------- + Sequence[AttributeFilter] + Filters compatible with NEODatabase.query. """ - # TODO: Decide how you will represent your filters. - return () + filters: List[AttributeFilter] = [] + + if date is not None: + filters.append(DateFilter(operator.eq, date)) + if start_date is not None: + filters.append(StartDateFilter(operator.ge, start_date)) + if end_date is not None: + filters.append(EndDateFilter(operator.le, end_date)) + + if distance_min is not None: + filters.append(DistanceFilter(operator.ge, float(distance_min))) + if distance_max is not None: + filters.append(DistanceFilter(operator.le, float(distance_max))) + + if velocity_min is not None: + filters.append(VelocityFilter(operator.ge, float(velocity_min))) + if velocity_max is not None: + filters.append(VelocityFilter(operator.le, float(velocity_max))) + + if diameter_min is not None: + filters.append(DiameterFilter(operator.ge, float(diameter_min))) + if diameter_max is not None: + filters.append(DiameterFilter(operator.le, float(diameter_max))) + if hazardous is not None: + filters.append(HazardousFilter(operator.eq, bool(hazardous))) -def limit(iterator, n=None): + return tuple(filters) + + +def limit(iterator: Iterable, n: Optional[int] = None) -> Iterator: """Produce a limited stream of values from an iterator. - If `n` is 0 or None, don't limit the iterator at all. + Parameters + ---------- + iterator : Iterable + The source iterator to limit. + n : Optional[int], optional + Maximum number of items to yield. If 0 or None, no limit is applied. - :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. + Yields + ------ + Any + Items from the iterator, up to n items. + + Notes + ----- + If n is 0 or None, don't limit the iterator at all. """ - # TODO: Produce at most `n` values from the given iterator. - return iterator + if not n: + # No limit: pass through + for item in iterator: + yield item + return + + count = 0 + for item in iterator: + if count >= n: + break + yield item + count += 1 diff --git a/helpers.py b/helpers.py index 945e990..bc98132 100644 --- a/helpers.py +++ b/helpers.py @@ -9,36 +9,38 @@ 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 +Convert datetimes to and from strings. - This will become the Python object `datetime.datetime(2020, 12, 31, 12, 0)`. +NASA's dataset provides timestamps as naive datetimes (UTC). +This module converts NASA calendar date strings to datetime, and vice versa. +""" - :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") +import datetime -def datetime_to_str(dt): - """Convert a naive Python datetime into a human-readable string. +def cd_to_datetime(calendar_date: str) -> datetime.datetime: + """Convert a NASA-formatted calendar date/time description into a datetime. - 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. + Expected NASA format example: '2020-Dec-31 12:00'. - :param dt: A naive Python datetime. - :return: That datetime, as a human-readable string without seconds. + Notes + ----- + Some datasets may use ISO-like 'YYYY-MM-DD HH:MM'. We try both. """ + if not calendar_date: + return None # type: ignore + for fmt in ("%Y-%b-%d %H:%M", "%Y-%m-%d %H:%M"): + try: + return datetime.datetime.strptime(calendar_date, fmt) + except ValueError: + continue + # If none matched, raise for visibility + raise ValueError(f"Unsupported date format: {calendar_date!r}") + + +def datetime_to_str(dt: datetime.datetime) -> str: + """Convert a naive Python datetime into a human-readable string (YYYY-MM-DD HH:MM).""" return datetime.datetime.strftime(dt, "%Y-%m-%d %H:%M") diff --git a/main.py b/main.py index 59746e0..79029fb 100755 --- a/main.py +++ b/main.py @@ -246,6 +246,7 @@ class NEOShell(cmd.Cmd): 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) ' diff --git a/models.py b/models.py index 302b029..6182ec8 100644 --- a/models.py +++ b/models.py @@ -16,119 +16,209 @@ quirks of the data set, such as missing names and unknown diameters. You'll edit this file in Task 1. + +------------------------ + + +This module defines two core data models: + +- NearEarthObject (NEO): a near-Earth object with primary designation, optional + name, optional diameter (km), and hazardous flag. +- CloseApproach: a close approach event with instant (UTC), distance (au), + velocity (km/s), and a link to its NEO. + +Docstring style: PEP 257 (module/class/function). See internal coding standards +for PEP 8 and PEP 257 references. + """ + +from __future__ import annotations + +import math +from typing import Any, Dict, Optional + from helpers import cd_to_datetime, datetime_to_str class NearEarthObject: - """A near-Earth object (NEO). + """Represent 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 encapsulates semantic and physical parameters about the + object, such as: + - primary designation (string, required, unique), + - IAU name (string or None), + - diameter in kilometers (float, may be NaN if unknown), + - hazardous flag (bool). - A `NearEarthObject` also maintains a collection of its close approaches - - initialized to an empty collection, but eventually populated in the - `NEODatabase` constructor. + It also maintains a collection of its close approaches that will be + populated later by 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. + def __init__(self, **info: Any) -> None: + """Create a new NearEarthObject. + + Parameters + ---------- + info : dict + Arbitrary mapping with keys from the CSV row. Expected keys include + 'pdes' (designation), 'name', 'diameter', 'pha'. + + Notes + ----- + - Empty or missing `name` becomes None (we default to '' to safely call `.strip()`). + - Unknown or invalid `diameter` becomes `float('nan')`. + - `pha` is 'Y' or 'N'; coerced to bool (True only for 'Y'). + """ - # 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 = [] + # Primary designation: always present in dataset + self.designation: str = (info.get('pdes') or info.get('designation') or '').strip() + + # Name may be empty or None in the dataset + name_str = (info.get('name') or '').strip() + # The 'name' field may be missing, None, or whitespace in source data. + # Normalize to None when empty, and strip surrounding whitespace otherwise. + # Avoid calling .strip() on None by defaulting to ''. + self.name: Optional[str] = name_str or None + + # Diameter (km): may be missing or non-numeric -> NaN + raw_diameter = info.get('diameter', '') + try: + self.diameter: float = float(raw_diameter) if str(raw_diameter).strip() != '' else math.nan + except (TypeError, ValueError): + self.diameter = math.nan + + # Potentially hazardous: NASA uses 'Y'/'N' + pha = (info.get('pha') or info.get('hazardous') or '').strip() + self.hazardous: bool = True if str(pha).upper() == 'Y' else False + + # Linked approaches are filled in by NEODatabase + self.approaches: list[CloseApproach] = [] @property - def fullname(self): + def fullname(self) -> str: """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})" + return f"{self.designation} ({self.name})" if self.name else self.designation + + def __str__(self) -> str: + """Return a human-readable string representation.""" + diameter_txt = f"{self.diameter:.3f} km" if not math.isnan(self.diameter) else "unknown diameter" + haz_txt = "is potentially hazardous" if self.hazardous else "is not potentially hazardous" + return f"NEO {self.fullname} has {diameter_txt} and {haz_txt}." + + def __repr__(self) -> str: + """Return a computer-readable string representation.""" + return ( + f"NearEarthObject(designation={self.designation!r}, name={self.name!r}, " + f"diameter={self.diameter:.3f}, hazardous={self.hazardous!r})" + ) + + def serialize(self) -> Dict[str, Any]: + """Serialize the NEO into a dict suitable for JSON output.""" + return { + "designation": self.designation, + "name": self.name, + "diameter_km": self.diameter, + "potentially_hazardous": self.hazardous, + } 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. + """Represent a close approach to Earth by an NEO. - 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. + A CloseApproach encapsulates the moment of closest approach (UTC), nominal + distance in astronomical units, relative velocity in km/s, and a link to + its NEO (populated later in NEODatabase). """ - # 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. + def __init__(self, **info: Any) -> None: + """Create a new CloseApproach. + + Parameters + ---------- + info : dict + Arbitrary mapping with keys from the JSON row. Expected keys include + 'des' (designation), 'cd' (calendar date string), 'dist' (au), + 'v_rel' (km/s). """ - # 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 + self._designation: str = (info.get('des') or '').strip() - @property - def time_str(self): - """Return a formatted representation of this `CloseApproach`'s approach time. + # Time: parse via helpers; may be absent in malformed rows + cd = info.get('cd') + self.time = cd_to_datetime(cd) if cd else None + + # Distance & velocity: coerce to float with sensible defaults + try: + self.distance: float = float(info.get('dist', 0.0)) + except (TypeError, ValueError): + self.distance = 0.0 - 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. + try: + self.velocity: float = float(info.get('v_rel', 0.0)) + except (TypeError, ValueError): + self.velocity = 0.0 - 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. + # Linked NEO reference; filled in by NEODatabase + self.neo: Optional[NearEarthObject] = None + + @property + def time_str(self) -> str: + """Return a formatted representation of the approach time (YYYY-MM-DD HH:MM).""" + return datetime_to_str(self.time) if self.time else "" + + def __str__(self) -> str: + """Return a human-readable string representation.""" + desig = self.neo.fullname if self.neo else self._designation or "" + return ( + f"On {self.time_str}, '{desig}' approaches Earth at a distance of " + f"{self.distance:.2f} au and a velocity of {self.velocity:.2f} km/s." + ) + + def __repr__(self) -> str: + """Return a computer-readable string representation.""" + return ( + f"CloseApproach(time={self.time_str!r}, distance={self.distance:.2f}, " + f"velocity={self.velocity:.2f}, neo={self.neo!r})" + ) + + def to_csv_row(self) -> Dict[str, Any]: + """Return a flat dict row for CSV export.""" + neo = self.neo + return { + "datetime_utc": self.time_str, + "distance_au": self.distance, + "velocity_km_s": self.velocity, + "designation": (neo.designation if neo else self._designation), + "name": (neo.name if neo else None), + "diameter_km": (float("nan") if (not neo or math.isnan(neo.diameter)) else neo.diameter), + "potentially_hazardous": (neo.hazardous if neo else False), + } + + def serialize(self) -> Dict[str, Any]: + """Return a nested dict ready for JSON export. + + Structure + --------- + { + "datetime_utc": str, # 'YYYY-MM-DD HH:MM' ('' si falta) + "distance_au": float, + "velocity_km_s": float, + "neo": { + "designation": str, + "name": Optional[str], + "diameter_km": float, # float('nan') si desconocido + "potentially_hazardous": bool + } + } + + Returns + ------- + Dict[str, Any] + JSON-serializable mapping of approach and NEO attributes. """ - # 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})" + neo = self.neo + return { + "datetime_utc": self.time_str, + "distance_au": self.distance, + "velocity_km_s": self.velocity, + "neo": (neo.serialize() if neo else {"designation": self._designation}), + } diff --git a/tests/test_write.py b/tests/test_write.py index 1e2d50f..d07d191 100644 --- a/tests/test_write.py +++ b/tests/test_write.py @@ -102,7 +102,6 @@ def test_csv_data_has_header(self): 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) @@ -127,7 +126,8 @@ def test_csv_data_header_matches_requirements(self): 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') + 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())) diff --git a/write.py b/write.py index 3b180ee..038208f 100644 --- a/write.py +++ b/write.py @@ -10,36 +10,60 @@ You'll edit this file in Part 4. """ + import csv import json +from pathlib import Path +from typing import Iterable + +from models import CloseApproach -def write_to_csv(results, filename): - """Write an iterable of `CloseApproach` objects to a CSV file. +def write_to_csv(results: Iterable[CloseApproach], filename: Path) -> None: + """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. + Parameters + ---------- + results : Iterable[CloseApproach] + An iterable of CloseApproach objects to write. + filename : Path + The path to the output CSV file. - :param results: An iterable of `CloseApproach` objects. - :param filename: A Path-like object pointing to where the data should be saved. + Notes + ----- + Output columns: + datetime_utc, distance_au, velocity_km_s, designation, name, diameter_km, potentially_hazardous """ fieldnames = ( - 'datetime_utc', 'distance_au', 'velocity_km_s', - 'designation', 'name', 'diameter_km', 'potentially_hazardous' + "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. + with open(filename, mode="w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + for ca in results: + writer.writerow(ca.to_csv_row()) -def write_to_json(results, filename): - """Write an iterable of `CloseApproach` objects to a JSON file. +def write_to_json(results: Iterable[CloseApproach], filename: Path) -> None: + """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. + Parameters + ---------- + results : Iterable[CloseApproach] + An iterable of CloseApproach objects to write. + filename : Path + The path to the output JSON file. - :param results: An iterable of `CloseApproach` objects. - :param filename: A Path-like object pointing to where the data should be saved. + Notes + ----- + Output structure: a list of dicts; each dict has approach fields and an 'neo' sub-dict. """ - # TODO: Write the results to a JSON file, following the specification in the instructions. + payload = [ca.serialize() for ca in results] + with open(filename, mode="w", encoding="utf-8") as f: + json.dump(payload, f, indent=2)