WIP: HAFAS support

5 files changed, 308 insertions, 26 deletions

diff --git a/lib/Travelynx/Command/work.pm b/lib/Travelynx/Command/work.pm
index 23d2925..ac4416d 100644
--- a/lib/Travelynx/Command/work.pm
+++ b/lib/Travelynx/Command/work.pm
@@ -37,6 +37,70 @@ sub run {
 		my $arr      = $entry->{arr_eva};
 		my $train_id = $entry->{train_id};
 
+		if ( $train_id =~ m{[|]} ) {
+
+			$self->app->hafas->get_journey_p( trip_id => $train_id )->then(
+				sub {
+					my ($journey) = @_;
+
+					my $found_dep;
+					my $found_arr;
+					for my $stop ( $journey->route ) {
+						if ( $stop->eva == $dep ) {
+							$found_dep = $stop;
+						}
+						if ( $arr and $stop->eva == $arr ) {
+							$found_arr = $stop;
+							last;
+						}
+					}
+					if ( not $found_dep ) {
+						return Mojo::Promise->reject(
+							"Did not find $dep within journey $train_id");
+					}
+
+					if ( $found_dep->{rt_dep} ) {
+						$self->app->in_transit->update_departure_hafas(
+							uid     => $uid,
+							journey => $journey,
+							stop    => $found_dep,
+							dep_eva => $dep,
+							arr_eva => $arr
+						);
+					}
+
+					if ( $found_arr and $found_arr->{rt_arr} ) {
+						$self->app->in_transit->update_arrival_hafas(
+							uid     => $uid,
+							journey => $journey,
+							stop    => $found_arr,
+							dep_eva => $dep,
+							arr_eva => $arr
+						);
+					}
+				}
+			)->catch(
+				sub {
+					my ($err) = @_;
+					$self->app->log->error("work($uid)/journey: $err");
+				}
+			)->wait;
+
+			if (    $arr
+				and $entry->{real_arr_ts}
+				and $now->epoch - $entry->{real_arr_ts} > 600 )
+			{
+				$self->app->checkout_p(
+					station => $arr,
+					force   => 2,
+					dep_eva => $dep,
+					arr_eva => $arr,
+					uid     => $uid
+				)->wait;
+			}
+			next;
+		}
+
 		# Note: IRIS data is not always updated in real-time. Both departure and
 		# arrival delays may take several minutes to appear, especially in case
 		# of large-scale disturbances. We work around this by continuing to
@@ -183,7 +247,7 @@ sub run {
 				)->catch(
 					sub {
 						my ($error) = @_;
-						$self->app->log->error("work($uid)/arrival: $@");
+						$self->app->log->error("work($uid)/arrival: $error");
 						$errors += 1;
 					}
 				)->wait;
@@ -194,7 +258,7 @@ sub run {
 			$errors += 1;
 		}
 
-		eval { }
+		eval { };
 	}
 
 	my $started_at       = $now;
diff --git a/lib/Travelynx/Controller/Traveling.pm b/lib/Travelynx/Controller/Traveling.pm
index 3e051a1..39c8e8f 100755
--- a/lib/Travelynx/Controller/Traveling.pm
+++ b/lib/Travelynx/Controller/Traveling.pm
@@ -747,7 +747,12 @@ sub travel_action {
 		else {
 			my $redir = '/';
 			if ( $status->{checked_in} or $status->{cancelled} ) {
-				$redir = '/s/' . $status->{dep_ds100};
+				if ( $status->{dep_ds100} ) {
+					$redir = '/s/' . $status->{dep_ds100};
+				}
+				else {
+					$redir = '/s/' . $status->{dep_eva} . '?hafas=1';
+				}
 			}
 			$self->render(
 				json => {
@@ -880,7 +885,7 @@ sub station {
 	if ($use_hafas) {
 		$promise = $self->hafas->get_departures_p(
 			eva        => $station,
-			lookbehind => 120,
+			lookbehind => 30,
 			lookahead  => 30,
 		);
 	}
diff --git a/lib/Travelynx/Helper/HAFAS.pm b/lib/Travelynx/Helper/HAFAS.pm
index de5bd1e..1c36925 100644
--- a/lib/Travelynx/Helper/HAFAS.pm
+++ b/lib/Travelynx/Helper/HAFAS.pm
@@ -98,6 +98,43 @@ sub get_departures_p {
 	);
 }
 
+sub get_journey_p {
+	my ( $self, %opt ) = @_;
+
+	my $promise = Mojo::Promise->new;
+	my $now     = DateTime->now( time_zone => 'Europe/Berlin' );
+
+	Travel::Status::DE::HAFAS->new_p(
+		journey => {
+			id => $opt{trip_id},
+		},
+		with_polyline => 0,
+		cache         => $self->{realtime_cache},
+		promise       => 'Mojo::Promise',
+		user_agent    => $self->{user_agent}->request_timeout(10),
+	)->then(
+		sub {
+			my ($hafas) = @_;
+			my $journey = $hafas->result;
+
+			if ($journey) {
+				$promise->resolve($journey);
+				return;
+			}
+			$promise->reject('no journey');
+			return;
+		}
+	)->catch(
+		sub {
+			my ($err) = @_;
+			$promise->reject($err);
+			return;
+		}
+	)->wait;
+
+	return $promise;
+}
+
 sub get_route_timestamps_p {
 	my ( $self, %opt ) = @_;
 
@@ -133,7 +170,6 @@ sub get_route_timestamps_p {
 					rt_dep    => _epoch( $stop->{rt_dep} ),
 					arr_delay => $stop->{arr_delay},
 					dep_delay => $stop->{dep_delay},
-					eva       => $stop->{eva},
 					load      => $stop->{load}
 				};
 				if (    ( $stop->{arr_cancelled} or not $stop->{sched_arr} )
diff --git a/lib/Travelynx/Model/InTransit.pm b/lib/Travelynx/Model/InTransit.pm
index 26f689f..aec193f 100644
--- a/lib/Travelynx/Model/InTransit.pm
+++ b/lib/Travelynx/Model/InTransit.pm
@@ -27,6 +27,12 @@ my %visibility_atoi = (
 	private   => 10,
 );
 
+sub _epoch {
+	my ($dt) = @_;
+
+	return $dt ? $dt->epoch : 0;
+}
+
 sub epoch_to_dt {
 	my ($epoch) = @_;
 
@@ -78,33 +84,80 @@ sub add {
 	my $uid                = $opt{uid};
 	my $db                 = $opt{db} // $self->{pg}->db;
 	my $train              = $opt{train};
+	my $journey            = $opt{journey};
+	my $stop               = $opt{stop};
 	my $checkin_station_id = $opt{departure_eva};
 	my $route              = $opt{route};
 
 	my $json = JSON->new;
 
-	$db->insert(
-		'in_transit',
-		{
-			user_id   => $uid,
-			cancelled => $train->departure_is_cancelled
-			? 1
-			: 0,
-			checkin_station_id => $checkin_station_id,
-			checkin_time       => DateTime->now( time_zone => 'Europe/Berlin' ),
-			dep_platform       => $train->platform,
-			train_type         => $train->type,
-			train_line         => $train->line_no,
-			train_no           => $train->train_no,
-			train_id           => $train->train_id,
-			sched_departure    => $train->sched_departure,
-			real_departure     => $train->departure,
-			route              => $json->encode($route),
-			messages           => $json->encode(
-				[ map { [ $_->[0]->epoch, $_->[1] ] } $train->messages ]
-			)
+	if ($train) {
+		$db->insert(
+			'in_transit',
+			{
+				user_id   => $uid,
+				cancelled => $train->departure_is_cancelled
+				? 1
+				: 0,
+				checkin_station_id => $checkin_station_id,
+				checkin_time => DateTime->now( time_zone => 'Europe/Berlin' ),
+				dep_platform => $train->platform,
+				train_type   => $train->type,
+				train_line   => $train->line_no,
+				train_no     => $train->train_no,
+				train_id     => $train->train_id,
+				sched_departure => $train->sched_departure,
+				real_departure  => $train->departure,
+				route           => $json->encode($route),
+				messages        => $json->encode(
+					[ map { [ $_->[0]->epoch, $_->[1] ] } $train->messages ]
+				)
+			}
+		);
+	}
+	elsif ( $journey and $stop ) {
+		my @route;
+		for my $j_stop ( $journey->route ) {
+			push(
+				@route,
+				[
+					$j_stop->name,
+					$j_stop->eva,
+					{
+						sched_arr => _epoch( $j_stop->{sched_arr} ),
+						sched_dep => _epoch( $j_stop->{sched_dep} ),
+						rt_arr    => _epoch( $j_stop->{rt_arr} ),
+						rt_dep    => _epoch( $j_stop->{rt_dep} ),
+						arr_delay => $j_stop->{arr_delay},
+						dep_delay => $j_stop->{dep_delay},
+						load      => $j_stop->{load}
+					}
+				]
+			);
 		}
-	);
+		$db->insert(
+			'in_transit',
+			{
+				user_id   => $uid,
+				cancelled => $stop->{dep_cancelled}
+				? 1
+				: 0,
+				checkin_station_id => $stop->eva,
+				checkin_time => DateTime->now( time_zone => 'Europe/Berlin' ),
+				dep_platform => $stop->{platform},
+				train_type   => $journey->type,
+				train_line   => $journey->line_no,
+				train_no     => $journey->number // q{},
+				train_id     => $journey->id,
+				sched_departure => $stop->{sched_dep},
+				real_departure  => $stop->{rt_dep} // $stop->{sched_dep},
+				route           => $json->encode( [@route] ),
+			}
+		);
+	}
+	else {
+		die('neither train nor journey specified');
+	}
 }
 
 sub add_from_journey {
@@ -576,6 +629,33 @@ sub update_departure_cancelled {
 	return $rows;
 }
 
+sub update_departure_hafas {
+	my ( $self, %opt ) = @_;
+	my $uid     = $opt{uid};
+	my $db      = $opt{db} // $self->{pg}->db;
+	my $dep_eva = $opt{dep_eva};
+	my $arr_eva = $opt{arr_eva};
+	my $journey = $opt{journey};
+	my $stop    = $opt{stop};
+	my $json    = JSON->new;
+
+	# selecting on user_id and train_no avoids a race condition if a user checks
+	# into a new train while we are fetching data for their previous journey. In
+	# this case, the new train would receive data from the previous journey.
+	$db->update(
+		'in_transit',
+		{
+			real_departure => $stop->{rt_dep},
+		},
+		{
+			user_id             => $uid,
+			train_id            => $journey->id,
+			checkin_station_id  => $dep_eva,
+			checkout_station_id => $arr_eva,
+		}
+	);
+}
+
 sub update_arrival {
 	my ( $self, %opt ) = @_;
 	my $uid     = $opt{uid};
@@ -618,6 +698,67 @@ sub update_arrival {
 	return $rows;
 }
 
+sub update_arrival_hafas {
+	my ( $self, %opt ) = @_;
+	my $uid     = $opt{uid};
+	my $db      = $opt{db} // $self->{pg}->db;
+	my $dep_eva = $opt{dep_eva};
+	my $arr_eva = $opt{arr_eva};
+	my $journey = $opt{journey};
+	my $stop    = $opt{stop};
+	my $json    = JSON->new;
+
+	# TODO use old rt data if available
+	my @route;
+	for my $j_stop ( $journey->route ) {
+		push(
+			@route,
+			[
+				$j_stop->name,
+				$j_stop->eva,
+				{
+					sched_arr => _epoch( $j_stop->{sched_arr} ),
+					sched_dep => _epoch( $j_stop->{sched_dep} ),
+					rt_arr    => _epoch( $j_stop->{rt_arr} ),
+					rt_dep    => _epoch( $j_stop->{rt_dep} ),
+					arr_delay => $j_stop->{arr_delay},
+					dep_delay => $j_stop->{dep_delay},
+					load      => $j_stop->{load}
+				}
+			]
+		);
+	}
+
+	my $res_h = $db->select( 'in_transit', ['route'], { user_id => $uid } )
+	  ->expand->hash;
+	my $old_route = $res_h ? $res_h->{route} : [];
+
+	for my $i ( 0 .. $#route ) {
+		if ( $old_route->[$i] and $old_route->[$i][1] == $route[$i][1] ) {
+			for my $k (qw(rt_arr rt_dep arr_delay dep_delay)) {
+				$route[$i][2]{$k} //= $old_route->[$i][2]{$k};
+			}
+		}
+	}
+
+	# selecting on user_id and train_no avoids a race condition if a user checks
+	# into a new train while we are fetching data for their previous journey. In
+	# this case, the new train would receive data from the previous journey.
+	$db->update(
+		'in_transit',
+		{
+			real_arrival => $stop->{rt_arr},
+			route        => $json->encode( [@route] ),
+		},
+		{
+			user_id             => $uid,
+			train_id            => $journey->id,
+			checkin_station_id  => $dep_eva,
+			checkout_station_id => $arr_eva,
+		}
+	);
+}
+
 sub update_data {
 	my ( $self, %opt ) = @_;
 
diff --git a/lib/Travelynx/Model/Stations.pm b/lib/Travelynx/Model/Stations.pm
index af318ee..75b4174 100644
--- a/lib/Travelynx/Model/Stations.pm
+++ b/lib/Travelynx/Model/Stations.pm
@@ -14,6 +14,42 @@ sub new {
 	return bless( \%opt, $class );
 }
 
+sub add_or_update {
+	my ( $self, %opt ) = @_;
+	my $stop   = $opt{stop};
+	my $source = 1;
+	my $db     = $opt{db} // $self->{pg}->db;
+
+	if ( my $s = $self->get_by_eva( $stop->eva, db => $db ) ) {
+		if ( $source == 1 and $s->{source} == 0 and not $s->{archived} ) {
+			return;
+		}
+		$db->update(
+			'stations',
+			{
+				name     => $stop->name,
+				lat      => $stop->lat,
+				lon      => $stop->lon,
+				source   => $source,
+				archived => 0
+			},
+			{ eva => $stop->eva }
+		);
+		return;
+	}
+	$db->insert(
+		'stations',
+		{
+			eva      => $stop->eva,
+			name     => $stop->name,
+			lat      => $stop->lat,
+			lon      => $stop->lon,
+			source   => $source,
+			archived => 0
+		}
+	);
+}
+
 # Fast
 sub get_by_eva {
 	my ( $self, $eva, %opt ) = @_;