feat(events): add support for opposition events

.pylintrc

@@ -145,7 +145,8 @@ disable=print-statement,
         too-many-locals,
         too-many-branches,
         too-few-public-methods,
-        protected-access
+        protected-access,
+        unnecessary-comprehension
 
 # Enable the message, report, category or checker with the given id(s). You can
 # either give multiple identifier separated by comma (,) or put this option

kosmorro

@@ -24,6 +24,7 @@ from kosmorrolib.version import VERSION
 from kosmorrolib import dumper
 from kosmorrolib import core
 from kosmorrolib.ephemerides import EphemeridesComputer, Position
+from kosmorrolib import events
 
 
 def main():
@@ -37,13 +38,17 @@ def main():
     month = args.month
     day = args.day
 
+    compute_date = date(year, month, day)
+
     if day is not None and month is None:
         month = date.today().month
 
     ephemeris = EphemeridesComputer(Position(args.latitude, args.longitude))
     ephemerides = ephemeris.compute_ephemerides(year, month, day)
 
-    dump = output_formats[args.format](ephemerides, date(year, month, day))
+    events_list = events.search_events(compute_date)
+
+    dump = output_formats[args.format](ephemerides, events_list, compute_date)
     print(dump.to_string())
 
     return 0

kosmorrolib/data.py

@@ -33,6 +33,10 @@ MOON_PHASES = {
     'WANING_CRESCENT': 'Waning crescent'
 }
 
+EVENTS = {
+    'OPPOSITION': {'message': '%s is in opposition'}
+}
+
 
 class MoonPhase:
     def __init__(self, identifier: str, time: Union[Time, None], next_phase_date: Union[Time, None]):
@@ -131,3 +135,20 @@ class DwarfPlanet(Planet):
 class Satellite(Object):
     def get_type(self) -> str:
         return 'satellite'
+
+
+class Event:
+    def __init__(self, event_type: str, aster: [Object], start_time: Time, end_time: Union[Time, None] = None):
+        if event_type not in EVENTS.keys():
+            raise ValueError('event_type parameter must be one of the following: %s (got %s)' % (
+                ', '.join(EVENTS.keys()),
+                event_type)
+                             )
+
+        self.event_type = event_type
+        self.object = aster
+        self.start_time = start_time
+        self.end_time = end_time
+
+    def get_description(self) -> str:
+        return EVENTS[self.event_type]['message'] % self.object.name

kosmorrolib/dumper.py

@@ -22,12 +22,13 @@ import json
 from tabulate import tabulate
 from skyfield.timelib import Time
 from numpy import int64
-from .data import Object, AsterEphemerides, MoonPhase
+from .data import Object, AsterEphemerides, MoonPhase, Event
 
 
 class Dumper(ABC):
-    def __init__(self, ephemeris: dict, date: datetime.date = datetime.date.today()):
+    def __init__(self, ephemeris: dict, events: [Event], date: datetime.date = datetime.date.today()):
         self.ephemeris = ephemeris
+        self.events = events
         self.date = date
 
     @abstractmethod
@@ -37,6 +38,7 @@ class Dumper(ABC):
 
 class JsonDumper(Dumper):
     def to_string(self):
+        self.ephemeris['events'] = self.events
         return json.dumps(self.ephemeris,
                           default=self._json_default,
                           indent=4)
@@ -60,16 +62,31 @@ class JsonDumper(Dumper):
             moon_phase['phase'] = moon_phase.pop('identifier')
             moon_phase['date'] = moon_phase.pop('time')
             return moon_phase
+        if isinstance(obj, Event):
+            event = obj.__dict__
+            event['object'] = event['object'].name
+            return event
 
         raise TypeError('Object of type "%s" could not be integrated in the JSON' % str(type(obj)))
 
 
 class TextDumper(Dumper):
     def to_string(self):
-        return '\n\n'.join(['Ephemerides of %s' % self.date.strftime('%A %B %d, %Y'),
+        text = 'Ephemerides of %s' % self.date.strftime('%A %B %d, %Y')
+        text = '\n\n'.join([text,
                             self.get_asters(self.ephemeris['details']),
-                            self.get_moon(self.ephemeris['moon_phase']),
-                            'Note: All the hours are given in UTC.'])
+                            self.get_moon(self.ephemeris['moon_phase'])
+                            ])
+
+        if len(self.events) > 0:
+            text = '\n\n'.join([text,
+                                'Expected events:',
+                                self.get_events(self.events)
+                                ])
+
+        text = '\n\n'.join([text, 'Note: All the hours are given in UTC.'])
+
+        return text
 
     @staticmethod
     def get_asters(asters: [Object]) -> str:
@@ -98,6 +115,15 @@ class TextDumper(Dumper):
         return tabulate(data, headers=['Object', 'Rise time', 'Culmination time', 'Set time'], tablefmt='simple',
                         stralign='center', colalign=('left',))
 
+    @staticmethod
+    def get_events(events: [Event]) -> str:
+        data = []
+
+        for event in events:
+            data.append([event.start_time.utc_strftime('%H:%M'), event.get_description()])
+
+        return tabulate(data, tablefmt='plain', stralign='left')
+
     @staticmethod
     def get_moon(moon_phase: MoonPhase) -> str:
         return 'Moon phase: %s\n' \

kosmorrolib/events.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+
+#    Kosmorro - Compute The Next Ephemerides
+#    Copyright (C) 2019  Jérôme Deuchnord <jerome@deuchnord.fr>
+#
+#    This program is free software: you can redistribute it and/or modify
+#    it under the terms of the GNU Affero General Public License as
+#    published by the Free Software Foundation, either version 3 of the
+#    License, or (at your option) any later version.
+#
+#    This program is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#    GNU Affero General Public License for more details.
+#
+#    You should have received a copy of the GNU Affero General Public License
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+from datetime import date as date_type
+
+from skyfield.timelib import Time
+from skyfield.almanac import find_discrete
+
+from .data import Event, Planet
+from .core import get_timescale, get_skf_objects, ASTERS
+
+
+def _search_oppositions(start_time: Time, end_time: Time) -> [Event]:
+    earth = get_skf_objects()['earth']
+    sun = get_skf_objects()['sun']
+    aster = None
+
+    def is_oppositing(time: Time) -> [bool]:
+        earth_pos = earth.at(time)
+        sun_pos = earth_pos.observe(sun).apparent()  # Never do this without eyes protection!
+        aster_pos = earth_pos.observe(get_skf_objects()[aster.skyfield_name]).apparent()
+        _, lon1, _ = sun_pos.ecliptic_latlon()
+        _, lon2, _ = aster_pos.ecliptic_latlon()
+        return (lon1.degrees - lon2.degrees) > 180
+
+    is_oppositing.rough_period = 1.0
+    events = []
+
+    for aster in ASTERS:
+        if not isinstance(aster, Planet) or aster.name in ['Mercury', 'Venus']:
+            continue
+
+        times, _ = find_discrete(start_time, end_time, is_oppositing)
+        for time in times:
+            events.append(Event('OPPOSITION', aster, time))
+
+    return events
+
+
+def search_events(date: date_type) -> [Event]:
+    start_time = get_timescale().utc(date.year, date.month, date.day)
+    end_time = get_timescale().utc(date.year, date.month, date.day + 1)
+
+    return [
+        opposition for opposition in _search_oppositions(start_time, end_time)
+    ]

test/__init__.py

@@ -1,2 +1,3 @@
 from .dumper import *
 from .ephemerides import *
+from .events import *

test/dumper.py

@@ -1,17 +1,22 @@
 import unittest
-from kosmorrolib.data import AsterEphemerides, Planet, MoonPhase
-from kosmorrolib.dumper import JsonDumper
+from datetime import date
+
+from kosmorrolib.data import AsterEphemerides, Planet, MoonPhase, Event
+from kosmorrolib.dumper import JsonDumper, TextDumper
 from kosmorrolib.core import get_timescale
 
 
 class DumperTestCase(unittest.TestCase):
+    def setUp(self) -> None:
+        self.maxDiff = None
+
     def test_json_dumper_returns_correct_json(self):
         data = self._get_data()
         self.assertEqual('{\n'
                          '    "moon_phase": {\n'
-                         '        "next_phase_date": "2019-11-20T00:00:00Z",\n'
+                         '        "next_phase_date": "2019-10-21T00:00:00Z",\n'
                          '        "phase": "FULL_MOON",\n'
-                         '        "date": "2019-11-11T00:00:00Z"\n'
+                         '        "date": "2019-10-14T00:00:00Z"\n'
                          '    },\n'
                          '    "details": [\n'
                          '        {\n'
@@ -22,13 +27,51 @@ class DumperTestCase(unittest.TestCase):
                          '                "set_time": null\n'
                          '            }\n'
                          '        }\n'
+                         '    ],\n'
+                         '    "events": [\n'
+                         '        {\n'
+                         '            "event_type": "OPPOSITION",\n'
+                         '            "object": "Mars",\n'
+                         '            "start_time": "2018-07-27T05:12:00Z",\n'
+                         '            "end_time": null\n'
+                         '        }\n'
                          '    ]\n'
-                         '}', JsonDumper(data).to_string())
+                         '}', JsonDumper(data,
+                                         [Event('OPPOSITION', Planet('Mars', 'MARS'),
+                                                get_timescale().utc(2018, 7, 27, 5, 12))]
+                                         ).to_string())
+
+    def test_text_dumper_without_events(self):
+        ephemerides = self._get_data()
+        self.assertEqual('Ephemerides of Monday October 14, 2019\n\n'
+                         'Object     Rise time    Culmination time    Set time\n'
+                         '--------  -----------  ------------------  ----------\n'
+                         'Mars           -               -               -\n\n'
+                         'Moon phase: Full Moon\n'
+                         'Last Quarter on Mon Oct 21, 2019 00:00\n\n'
+                         'Note: All the hours are given in UTC.',
+                         TextDumper(ephemerides, [], date=date(2019, 10, 14)).to_string())
+
+    def test_text_dumper_with_events(self):
+        ephemerides = self._get_data()
+        self.assertEqual('Ephemerides of Monday October 14, 2019\n\n'
+                         'Object     Rise time    Culmination time    Set time\n'
+                         '--------  -----------  ------------------  ----------\n'
+                         'Mars           -               -               -\n\n'
+                         'Moon phase: Full Moon\n'
+                         'Last Quarter on Mon Oct 21, 2019 00:00\n\n'
+                         'Expected events:\n\n'
+                         '05:12  Mars is in opposition\n\n'
+                         'Note: All the hours are given in UTC.',
+                         TextDumper(ephemerides, [Event('OPPOSITION',
+                                                        Planet('Mars', 'MARS'),
+                                                        get_timescale().utc(2018, 7, 27, 5, 12))
+                                                  ], date=date(2019, 10, 14)).to_string())
 
     @staticmethod
     def _get_data():
         return {
-            'moon_phase': MoonPhase('FULL_MOON', get_timescale().utc(2019, 11, 11), get_timescale().utc(2019, 11, 20)),
+            'moon_phase': MoonPhase('FULL_MOON', get_timescale().utc(2019, 10, 14), get_timescale().utc(2019, 10, 21)),
             'details': [Planet('Mars', 'MARS', AsterEphemerides(None, None, None))]
         }
 

test/events.py

@@ -0,0 +1,33 @@
+import unittest
+
+from datetime import date
+
+from kosmorrolib import events
+from kosmorrolib.data import Event
+from kosmorrolib.core import get_timescale
+
+
+class MyTestCase(unittest.TestCase):
+    def test_event_only_accepts_valid_values(self):
+        with self.assertRaises(ValueError):
+            Event('SUPERNOVA', None, get_timescale().now())
+
+    def test_find_oppositions(self):
+        # Test case: Mars opposition
+        # Source of the information: https://promenade.imcce.fr/en/pages6/887.html#mar
+        o1 = (events.search_events(date(2020, 10, 13)), '^2020-10-13T23:25')
+        o2 = (events.search_events(date(2022, 12, 8)), '^2022-12-08T05:41')
+        o3 = (events.search_events(date(2025, 1, 16)), '^2025-01-16T02:38')
+        o4 = (events.search_events(date(2027, 2, 19)), '^2027-02-19T15:50')
+
+        for (o, expected_date) in [o1, o2, o3, o4]:
+            self.assertEqual(1, len(o), 'Expected 1 event for %s, got %d' % (expected_date, len(o)))
+            self.assertEqual('OPPOSITION', o[0].event_type)
+            self.assertEqual('MARS', o[0].object.skyfield_name)
+            self.assertRegex(o[0].start_time.utc_iso(), expected_date)
+            self.assertIsNone(o[0].end_time)
+            self.assertEqual('Mars is in opposition', o[0].get_description())
+
+
+if __name__ == '__main__':
+    unittest.main()