From 66b4e84a88cd386929478ee16af884fbfbeaf514 Mon Sep 17 00:00:00 2001 From: Malena Trejo Date: Tue, 13 Jan 2026 14:59:18 -0300 Subject: [PATCH 1/2] . --- database.py | 140 ++++++++++++-------------- extract.py | 76 ++++++++++++-- filters.py | 195 +++++++++++++++++++++--------------- helpers.py | 48 ++++----- models.py | 280 +++++++++++++++++++++++++++++++++++----------------- write.py | 50 ++++++---- 6 files changed, 491 insertions(+), 298 deletions(-) diff --git a/database.py b/database.py index 910932c..27b9720 100644 --- a/database.py +++ b/database.py @@ -10,89 +10,77 @@ `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. +A database encapsulating collections of near-Earth objects and their close approaches. - :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 +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_name(self, name): - """Find and return an NEO by its name. +from __future__ import annotations - If no match is found, return `None` instead. +from typing import Dict, Iterable, Iterator, List, Optional, Sequence - Not every NEO in the data set has a name. No NEOs are associated with - the empty string nor with the `None` singleton. +from models import NearEarthObject, CloseApproach - 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`. +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 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. + self._neos: List[NearEarthObject] = list(neos) + self._approaches: List[CloseApproach] = list(approaches) + + # 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} + + # Link approaches <-> neos + for app in self._approaches: + neo = self._by_des.get(app._designation) + if neo: + app.neo = neo + neo.approaches.append(app) + + def get_neo_by_designation(self, designation: str) -> Optional[NearEarthObject]: + """Find and return an NEO by its primary designation, else None.""" + return self._by_des.get(designation) + + def get_neo_by_name(self, name: str) -> Optional[NearEarthObject]: + """Find and return an NEO by its name, else None.""" + return self._by_name.get(name) + + def query(self, filters: Sequence) -> Iterator[CloseApproach]: + """Yield close approaches that satisfy *all* filters. + + Parameters + ---------- + filters : Sequence[AttributeFilter] + A collection of filter predicates. + + 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..dbe7bda 100644 --- a/filters.py +++ b/filters.py @@ -15,111 +15,152 @@ 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). """ + +from __future__ import annotations + import operator +from typing import Callable, Iterable, Iterator, List, Optional, Sequence + +from models import CloseApproach class UnsupportedCriterionError(NotImplementedError): - """A filter criterion is unsupported.""" + """Raised when a filter criterion is unsupported.""" class AttributeFilter: - """A general superclass for filters on comparable attributes. + """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. + def __init__(self, op: Callable, value) -> None: + """Construct an AttributeFilter from a binary predicate and a reference value.""" + self.op = op + self.value = value - 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). + def __call__(self, approach: CloseApproach) -> bool: + """Return whether `approach` satisfies this filter.""" + return self.op(self.get(approach), self.value) - 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. + @classmethod + def get(cls, approach: CloseApproach): + """Get an attribute of interest from a close approach.""" + raise UnsupportedCriterionError - 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 __repr__(self) -> str: + return f"{self.__class__.__name__}(op=operator.{self.op.__name__}, value={self.value})" - :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) +# ---- Concrete filters ---- +class DateFilter(AttributeFilter): @classmethod - def get(cls, approach): - """Get an attribute of interest from a close approach. + def get(cls, approach: CloseApproach): + return approach.time.date() if approach.time else None - 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 +class StartDateFilter(AttributeFilter): + @classmethod + def get(cls, approach: CloseApproach): + return approach.time.date() if approach.time else None - def __repr__(self): - return f"{self.__class__.__name__}(op=operator.{self.op.__name__}, value={self.value})" + +class EndDateFilter(AttributeFilter): + @classmethod + def get(cls, approach: CloseApproach): + return approach.time.date() if approach.time else None + + +class DistanceFilter(AttributeFilter): + @classmethod + def get(cls, approach: CloseApproach): + return approach.distance + + +class VelocityFilter(AttributeFilter): + @classmethod + def get(cls, approach: CloseApproach): + return approach.velocity + + +class DiameterFilter(AttributeFilter): + @classmethod + def get(cls, approach: CloseApproach): + return approach.neo.diameter if approach.neo else float('nan') + + +class HazardousFilter(AttributeFilter): + @classmethod + def get(cls, approach: CloseApproach): + 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`. + 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))) -def limit(iterator, n=None): + 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))) + + 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. - - :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 + 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/models.py b/models.py index 302b029..d15dc67 100644 --- a/models.py +++ b/models.py @@ -16,119 +16,215 @@ 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() - @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 '' + + + # 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 - 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})" + # 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 -class CloseApproach: - """A close approach to Earth by an NEO. + # Linked approaches are filled in by NEODatabase + self.approaches: list[CloseApproach] = [] - 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. + @property + def fullname(self) -> str: + """Return a representation of the full name of this NEO.""" + 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: + """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/write.py b/write.py index 3b180ee..c5660df 100644 --- a/write.py +++ b/write.py @@ -10,36 +10,46 @@ You'll edit this file in Part 4. """ + + + +"""Write a stream of close approaches to CSV or to JSON.""" + 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. - 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. +def write_to_csv(results: Iterable[CloseApproach], filename: Path) -> None: + """Write an iterable of CloseApproach objects to a CSV file. - :param results: An iterable of `CloseApproach` objects. - :param filename: A Path-like object pointing to where the data should be saved. + 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. - 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. +def write_to_json(results: Iterable[CloseApproach], filename: Path) -> None: + """Write an iterable of CloseApproach objects to a JSON file. - :param results: An iterable of `CloseApproach` objects. - :param filename: A Path-like object pointing to where the data should be saved. + 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) From 2ce80dbad898b74e9c5bccdcc778ad58caad708c Mon Sep 17 00:00:00 2001 From: Malena Trejo Date: Tue, 13 Jan 2026 15:15:28 -0300 Subject: [PATCH 2/2] Enhance docstrings across multiple modules for improved clarity and consistency --- database.py | 26 +++++++- filters.py | 180 ++++++++++++++++++++++++++++++++++++++++++++++++++-- write.py | 22 +++++-- 3 files changed, 217 insertions(+), 11 deletions(-) diff --git a/database.py b/database.py index 27b9720..597ee0d 100644 --- a/database.py +++ b/database.py @@ -61,11 +61,33 @@ def __init__(self, neos: Sequence[NearEarthObject], approaches: Sequence[CloseAp neo.approaches.append(app) def get_neo_by_designation(self, designation: str) -> Optional[NearEarthObject]: - """Find and return an NEO by its primary designation, else None.""" + """Find and return an NEO by its primary designation. + + Parameters + ---------- + designation : str + The primary designation to search for. + + Returns + ------- + Optional[NearEarthObject] + The NEO with the given designation, or None if not found. + """ return self._by_des.get(designation) def get_neo_by_name(self, name: str) -> Optional[NearEarthObject]: - """Find and return an NEO by its name, else None.""" + """Find and return an NEO by its name. + + Parameters + ---------- + name : str + The name to search for. + + Returns + ------- + Optional[NearEarthObject] + The NEO with the given name, or None if not found. + """ return self._by_name.get(name) def query(self, filters: Sequence) -> Iterator[CloseApproach]: diff --git a/filters.py b/filters.py index dbe7bda..621009d 100644 --- a/filters.py +++ b/filters.py @@ -35,24 +35,59 @@ class UnsupportedCriterionError(NotImplementedError): - """Raised when a filter criterion is unsupported.""" + """Exception raised when a filter criterion is unsupported.""" class AttributeFilter: """A general superclass for filters on comparable attributes.""" def __init__(self, op: Callable, value) -> None: - """Construct an AttributeFilter from a binary predicate and a reference value.""" + """Construct an AttributeFilter from a binary predicate and a reference value. + + 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: CloseApproach) -> bool: - """Return whether `approach` satisfies this filter.""" + """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: CloseApproach): - """Get an attribute of interest from a close approach.""" + """Get an attribute of interest from a close approach. + + Parameters + ---------- + approach : CloseApproach + The close approach from which to extract an attribute. + + Returns + ------- + Any + The attribute value to be compared. + + Raises + ------ + UnsupportedCriterionError + This base implementation always raises this error. + """ raise UnsupportedCriterionError def __repr__(self) -> str: @@ -62,44 +97,142 @@ def __repr__(self) -> str: # ---- 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 @@ -112,6 +245,29 @@ def create_filters( ) -> Sequence[AttributeFilter]: """Create a collection of filters from user-specified criteria. + 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] @@ -150,7 +306,21 @@ def create_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. + + Yields + ------ + Any + Items from the iterator, up to n items. + + Notes + ----- + If n is 0 or None, don't limit the iterator at all. """ if not n: # No limit: pass through diff --git a/write.py b/write.py index c5660df..038208f 100644 --- a/write.py +++ b/write.py @@ -11,10 +11,6 @@ You'll edit this file in Part 4. """ - - -"""Write a stream of close approaches to CSV or to JSON.""" - import csv import json from pathlib import Path @@ -26,6 +22,15 @@ def write_to_csv(results: Iterable[CloseApproach], filename: Path) -> None: """Write an iterable of CloseApproach objects to a CSV file. + Parameters + ---------- + results : Iterable[CloseApproach] + An iterable of CloseApproach objects to write. + filename : Path + The path to the output CSV file. + + Notes + ----- Output columns: datetime_utc, distance_au, velocity_km_s, designation, name, diameter_km, potentially_hazardous """ @@ -48,6 +53,15 @@ def write_to_csv(results: Iterable[CloseApproach], filename: Path) -> None: def write_to_json(results: Iterable[CloseApproach], filename: Path) -> None: """Write an iterable of CloseApproach objects to a JSON file. + Parameters + ---------- + results : Iterable[CloseApproach] + An iterable of CloseApproach objects to write. + filename : Path + The path to the output JSON file. + + Notes + ----- Output structure: a list of dicts; each dict has approach fields and an 'neo' sub-dict. """ payload = [ca.serialize() for ca in results]