{-# LANGUAGE DataKinds #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedLists #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeApplications #-} module Server.GTFS_RT (gtfsRealtimeServer) where import Prelude hiding (id) import API (GtfsRealtimeAPI) import Control.Monad (forM) import Control.Monad.IO.Class (MonadIO (..)) import qualified Data.ByteString.Char8 as C8 import Data.ByteString.Lazy (fromStrict) import Data.Functor ((<&>)) import qualified Data.Map as M import Data.Maybe (catMaybes) import Data.Pool (Pool) import Data.Sequence (Seq) import qualified Data.Sequence as Seq import Data.Text (Text) import qualified Data.Text as T import Data.Text.Encoding (encodeUtf8) import Data.Time.Calendar (Day, toGregorian) import Data.Time.Clock (UTCTime (utctDay), getCurrentTime, addUTCTime) import Data.Time.Clock.POSIX (getPOSIXTime) import Data.Time.Clock.System (SystemTime (systemSeconds), getSystemTime, utcToSystemTime) import qualified Data.UUID as UUID import Data.Word (Word64) import Database.Persist (Entity (..), PersistQueryRead (selectFirst), selectList, (==.)) import Database.Persist.Postgresql (SqlBackend) import GHC.Float (double2Float, int2Double) import GTFS (GTFS (..), Seconds (..), Stop (..), Trip (..), stationId, toUTC, TripID, tripsOnDay) import GTFS.Realtime.Alert as AL (Alert (..)) import GTFS.Realtime.Alert.SeverityLevel (SeverityLevel (WARNING)) import GTFS.Realtime.EntitySelector as ES (EntitySelector (..)) import GTFS.Realtime.FeedEntity as FE (FeedEntity (..)) import GTFS.Realtime.FeedHeader (FeedHeader (FeedHeader)) import GTFS.Realtime.FeedHeader.Incrementality (Incrementality (FULL_DATASET)) import GTFS.Realtime.FeedMessage as FM (FeedMessage (..)) import GTFS.Realtime.Position as POS (Position (..)) import GTFS.Realtime.TimeRange (TimeRange (TimeRange)) import GTFS.Realtime.TranslatedString (TranslatedString (TranslatedString)) import GTFS.Realtime.TranslatedString.Translation (Translation (Translation)) import GTFS.Realtime.TripDescriptor as TD (TripDescriptor (..)) import GTFS.Realtime.TripUpdate as TU (TripUpdate (..)) import GTFS.Realtime.TripUpdate.StopTimeEvent as STE (StopTimeEvent (..)) import GTFS.Realtime.TripUpdate.StopTimeUpdate as STU (StopTimeUpdate (..)) import qualified GTFS.Realtime.TripUpdate.StopTimeUpdate.ScheduleRelationship as SR import qualified GTFS.Realtime.TripDescriptor.ScheduleRelationship as TSR import GTFS.Realtime.VehicleDescriptor as VD (VehicleDescriptor (..)) import GTFS.Realtime.VehiclePosition as VP (VehiclePosition (..)) import Persist (Announcement (..), EntityField (..), Key (..), Running (..), Token (..), TrainAnchor (..), TrainPing (..), runSql) import Servant.API ((:<|>) (..)) import Text.ProtocolBuffers (Utf8 (Utf8), defaultValue) import Control.Monad.Extra (mapMaybeM) import Data.List.NonEmpty (nonEmpty) import Data.Time.Format.ISO8601 (ISO8601 (iso8601Format), iso8601Show) import Data.UUID (toASCIIBytes, toLazyASCIIBytes) import qualified Data.Vector as V import Extrapolation (Extrapolator (extrapolateAtPosition, extrapolateAtSeconds), LinearExtrapolator (..)) import GTFS (Depth (..)) import GTFS.Realtime.TripUpdate (TripUpdate (TripUpdate)) import Server.Util (Service, secondsNow) uuidUtf8 :: UUID.UUID -> Utf8 uuidUtf8 = Utf8 . fromStrict . UUID.toASCIIBytes toUtf8 :: Text -> Utf8 toUtf8 = Utf8 . fromStrict . encodeUtf8 -- | formats a day in the "stupid" format used by gtfs realtime toStupidDate :: Day -> Utf8 toStupidDate date = toUtf8 $ pad 4 year <> pad 2 month <> pad 2 day where (year, month, day) = toGregorian date pad len num = T.pack $ if ndigits < len then replicate (len - ndigits) '0' <> show num else show num where ndigits = length (show num) -- | basically unix timestamps, raw (because why not i guess) toStupidTime :: Num i => UTCTime -> i toStupidTime = fromIntegral . systemSeconds . utcToSystemTime gtfsRealtimeServer :: GTFS -> Pool SqlBackend -> Service GtfsRealtimeAPI gtfsRealtimeServer gtfs@GTFS{..} dbpool = handleServiceAlerts :<|> handleTripUpdates :<|> handleVehiclePositions where handleServiceAlerts = runSql dbpool $ do -- TODO filter: only select current & future days announcements <- selectList [] [] dFeedMessage $ Seq.fromList $ fmap mkAlert announcements where mkAlert (Entity (AnnouncementKey uuid) Announcement{..}) = (dFeedEntity (uuidUtf8 uuid)) { alert = Just $ Alert { active_period = [TimeRange Nothing Nothing defaultValue] -- TODO: is this time range reasonable, needed, etc.? , informed_entity = [dEntitySelector { ES.trip = Just (dTripDescriptor announcementTrip (Just announcementDay) Nothing) } ] , cause = Nothing , effect = Nothing , url = fmap (lang "de" . toUtf8) announcementUrl , header_text = Just $ lang "de" (toUtf8 announcementHeader) , description_text = Just $ lang "de" (toUtf8 announcementMessage) , tts_header_text = Nothing , tts_description_text = Nothing , severity_level = Nothing , image = Nothing , image_alternative_text = Nothing , AL.ext'field = defaultValue } } handleTripUpdates = runSql dbpool $ do today <- liftIO $ getCurrentTime <&> utctDay nowSeconds <- secondsNow today let running = M.toList $ tripsOnDay gtfs today anchors <- flip mapMaybeM running $ \(tripId, trip@Trip{..}) -> do entities <- selectList [TrainAnchorTrip ==. tripId, TrainAnchorDay ==. today] [] case nonEmpty (fmap entityVal entities) of Nothing -> pure Nothing Just anchors -> pure $ Just (tripId, trip, anchors) dFeedMessage $ Seq.fromList $ fmap (mkTripUpdate today nowSeconds) anchors where mkTripUpdate today nowSeconds (tripId :: Text, Trip{..} :: Trip Deep Deep, anchors) = let lastCall = extrapolateAtSeconds LinearExtrapolator anchors nowSeconds stations = tripStops <&> (\stop@Stop{..} -> fmap (, stop) $ extrapolateAtPosition LinearExtrapolator anchors (int2Double stopSequence)) -- note: these IDs should be stable across iterations, so just do tripId + runningday. TODO: breaks in case of cross-midnight? in (dFeedEntity (Utf8 $ fromStrict $ (encodeUtf8 tripId <> "-" <> (C8.pack $ iso8601Show today)))) { FE.trip_update = Just $ TripUpdate { TU.trip = dTripDescriptor tripId (Just today) (Just $ toUtf8 $ T.pack $ show $ stopDeparture $ V.head tripStops) -- TODO will break if cross-midnight train , TU.vehicle = Nothing , TU.stop_time_update = Seq.fromList $ fmap (\(TrainAnchor{..}, Stop{..}) -> StopTimeUpdate { STU.stop_sequence = Just (fromIntegral stopSequence) , STU.stop_id = Just (toUtf8 $ stationId stopStation) , STU.arrival = Just ( defaultValue { STE.delay = Just $ fromIntegral $ unSeconds $ trainAnchorDelay , STE.time = Just $ (toStupidTime (addUTCTime (fromIntegral $ unSeconds trainAnchorDelay) (toUTC stopArrival tzseries today))) , STE.uncertainty = Just 60 }) , STU.departure = Just ( defaultValue { STE.delay = Just $ fromIntegral $ unSeconds $ trainAnchorDelay , STE.time = Just $ (toStupidTime (addUTCTime (fromIntegral $ unSeconds trainAnchorDelay) (toUTC stopDeparture tzseries today))) , STE.uncertainty = Just 60 }) , STU.departure_occupancy_status = Nothing , STU.schedule_relationship = Just SR.SCHEDULED , STU.stop_time_properties = Nothing , STU.ext'field = defaultValue }) $ catMaybes $ V.toList stations , TU.delay = Nothing -- lastCall <&> (fromIntegral . unSeconds . trainAnchorDelay) , TU.timestamp = lastCall <&> (toStupidTime . trainAnchorCreated) , TU.trip_properties = Nothing , TU.ext'field = defaultValue } } handleVehiclePositions = runSql dbpool $ do (running :: [Entity Running]) <- selectList [] [] pings <- forM running $ \(Entity key entity) -> do selectFirst [TrainPingToken ==. key] [] <&> fmap (, entity) dFeedMessage $ Seq.fromList $ mkPosition <$> catMaybes pings where mkPosition (Entity (TrainPingKey key) TrainPing{..}, Running{..}) = (dFeedEntity (toUtf8 . T.pack . show $ key)) { FE.vehicle = Just $ VehiclePosition { trip = Just (dTripDescriptor runningTrip Nothing Nothing) , VP.vehicle = case runningVehicle of Nothing -> Nothing Just trainset -> Just $ VehicleDescriptor { VD.id = Nothing , VD.label = Just (toUtf8 trainset) , VD.license_plate = Nothing , VD.ext'field = defaultValue } , position = Just $ Position { latitude = double2Float trainPingLat , longitude = double2Float trainPingLong , odometer = Nothing , speed = Nothing , bearing = Nothing , POS.ext'field = defaultValue } -- TODO: at least one of these should probably be given , current_stop_sequence = Nothing , stop_id = Nothing , current_status = Nothing , timestamp = Just (toStupidTime trainPingTimestamp) , congestion_level = Nothing , occupancy_status = Nothing , occupancy_percentage = Nothing , multi_carriage_details = [] , VP.ext'field = defaultValue } } lang :: Utf8 -> Utf8 -> TranslatedString lang code msg = TranslatedString [Translation msg (Just code) defaultValue] defaultValue -- | a default FeedMessage, issued at the current system time -- TODO: do we ever need incremental updates? -- TODO: maybe instead use last update time? dFeedMessage :: MonadIO m => Seq FeedEntity -> m FeedMessage dFeedMessage entities = do now <- liftIO getSystemTime <&> systemSeconds pure $ FeedMessage { header = FeedHeader "2.0" (Just FULL_DATASET) (Just $ fromIntegral now) defaultValue , entity = entities , FM.ext'field = defaultValue } -- | a dummy FeedEntity (use record updates to add meaningful values to this) dFeedEntity :: Utf8 -> FeedEntity dFeedEntity id = FeedEntity { id , is_deleted = Nothing , trip_update = Nothing , vehicle = Nothing , alert = Nothing , shape = Nothing , FE.ext'field = defaultValue } dEntitySelector :: EntitySelector dEntitySelector = EntitySelector { agency_id = Nothing , route_id = Nothing , route_type = Nothing , trip = Nothing , stop_id = Nothing , direction_id = Nothing , ES.ext'field = defaultValue } dTripDescriptor :: TripID -> Maybe Day -> Maybe Utf8 -> TripDescriptor dTripDescriptor tripID day starttime = TripDescriptor { trip_id = Just (toUtf8 tripID) , route_id = Nothing , direction_id = Nothing , start_time = starttime , start_date = fmap toStupidDate day , schedule_relationship = Just TSR.SCHEDULED , TD.ext'field = defaultValue }