303 lines
7.8 KiB
Perl
303 lines
7.8 KiB
Perl
package Travelynx::Command::work;
|
|
|
|
# Copyright (C) 2020 Daniel Friesel
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
use Mojo::Base 'Mojolicious::Command';
|
|
|
|
use DateTime;
|
|
use JSON;
|
|
use List::Util;
|
|
|
|
has description =>
|
|
'Perform automatic checkout when users arrive at their destination';
|
|
|
|
has usage => sub { shift->extract_usage };
|
|
|
|
sub run {
|
|
my ($self) = @_;
|
|
|
|
my $now = DateTime->now( time_zone => 'Europe/Berlin' );
|
|
my $json = JSON->new;
|
|
|
|
my $db = $self->app->pg->db;
|
|
|
|
for my $entry (
|
|
$db->select( 'in_transit_str', '*', { cancelled => 0 } )->hashes->each )
|
|
{
|
|
|
|
my $uid = $entry->{user_id};
|
|
my $dep = $entry->{dep_eva};
|
|
my $arr = $entry->{arr_eva};
|
|
my $train_id = $entry->{train_id};
|
|
|
|
# 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
|
|
# update departure data for up to 15 minutes after departure and
|
|
# delaying automatic checkout by at least 10 minutes.
|
|
|
|
eval {
|
|
if ( $now->epoch - $entry->{real_dep_ts} < 900 ) {
|
|
my $status = $self->app->iris->get_departures(
|
|
station => $dep,
|
|
lookbehind => 30,
|
|
lookahead => 30
|
|
);
|
|
if ( $status->{errstr} ) {
|
|
die("get_departures($dep): $status->{errstr}\n");
|
|
}
|
|
|
|
my ($train) = List::Util::first { $_->train_id eq $train_id }
|
|
@{ $status->{results} };
|
|
|
|
if ( not $train ) {
|
|
die("could not find train $train_id at $dep\n");
|
|
}
|
|
|
|
# selecting on user_id and train_no avoids a race condition when
|
|
# 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',
|
|
{
|
|
dep_platform => $train->platform,
|
|
real_departure => $train->departure,
|
|
route => $json->encode(
|
|
[ $self->app->iris->route_diff($train) ]
|
|
),
|
|
messages => $json->encode(
|
|
[
|
|
map { [ $_->[0]->epoch, $_->[1] ] }
|
|
$train->messages
|
|
]
|
|
),
|
|
},
|
|
{
|
|
user_id => $uid,
|
|
train_no => $train->train_no
|
|
}
|
|
);
|
|
if ( $train->departure_is_cancelled and $arr ) {
|
|
|
|
# depending on the amount of users in transit, some time may
|
|
# have passed between fetching $entry from the database and
|
|
# now. Ensure that the user is still checked into this train
|
|
# before calling checkout to mark the cancellation.
|
|
if (
|
|
$db->select(
|
|
'in_transit',
|
|
'count(*) as count',
|
|
{
|
|
user_id => $uid,
|
|
train_no => $train->train_no,
|
|
checkin_station_id => $dep,
|
|
checkout_station_id => $arr,
|
|
}
|
|
)->hash->{count}
|
|
)
|
|
{
|
|
$db->update(
|
|
'in_transit',
|
|
{
|
|
cancelled => 1,
|
|
},
|
|
{
|
|
user_id => $uid,
|
|
train_no => $train->train_no,
|
|
checkin_station_id => $dep,
|
|
checkout_station_id => $arr,
|
|
}
|
|
);
|
|
|
|
# check out (adds a cancelled journey and resets journey state
|
|
# to checkin
|
|
$self->app->checkout(
|
|
station => $arr,
|
|
force => 1,
|
|
uid => $uid
|
|
);
|
|
}
|
|
}
|
|
else {
|
|
$self->app->add_route_timestamps( $uid, $train, 1 );
|
|
}
|
|
}
|
|
};
|
|
if ($@) {
|
|
$self->app->log->error("work($uid)/departure: $@");
|
|
}
|
|
|
|
eval {
|
|
if (
|
|
$arr
|
|
and ( not $entry->{real_arr_ts}
|
|
or $now->epoch - $entry->{real_arr_ts} < 600 )
|
|
)
|
|
{
|
|
my $status = $self->app->iris->get_departures(
|
|
station => $arr,
|
|
lookbehind => 20,
|
|
lookahead => 220
|
|
);
|
|
if ( $status->{errstr} ) {
|
|
die("get_departures($arr): $status->{errstr}\n");
|
|
}
|
|
|
|
# Note that a train may pass the same station several times.
|
|
# Notable example: S41 / S42 ("Ringbahn") both starts and
|
|
# terminates at Berlin Südkreuz
|
|
my ($train) = List::Util::first {
|
|
$_->train_id eq $train_id
|
|
and $_->sched_arrival
|
|
and $_->sched_arrival->epoch > $entry->{sched_dep_ts}
|
|
}
|
|
@{ $status->{results} };
|
|
|
|
$train //= List::Util::first { $_->train_id eq $train_id }
|
|
@{ $status->{results} };
|
|
|
|
if ( not $train ) {
|
|
|
|
# If we haven't seen the train yet, its arrival is probably
|
|
# too far in the future. This is not critical.
|
|
return;
|
|
}
|
|
|
|
# selecting on user_id, train_no and checkout_station_id avoids a
|
|
# race condition when a user checks into a new train or changes
|
|
# their destination station while we are fetching times based on no
|
|
# longer valid database entries.
|
|
$db->update(
|
|
'in_transit',
|
|
{
|
|
arr_platform => $train->platform,
|
|
sched_arrival => $train->sched_arrival,
|
|
real_arrival => $train->arrival,
|
|
route => $json->encode(
|
|
[ $self->app->iris->route_diff($train) ]
|
|
),
|
|
messages => $json->encode(
|
|
[
|
|
map { [ $_->[0]->epoch, $_->[1] ] }
|
|
$train->messages
|
|
]
|
|
),
|
|
},
|
|
{
|
|
user_id => $uid,
|
|
train_no => $train->train_no,
|
|
checkout_station_id => $arr
|
|
}
|
|
);
|
|
if ( $train->arrival_is_cancelled ) {
|
|
|
|
# depending on the amount of users in transit, some time may
|
|
# have passed between fetching $entry from the database and
|
|
# now. Ensure that the user is still checked into this train
|
|
# before calling checkout to mark the cancellation.
|
|
if (
|
|
$db->select(
|
|
'in_transit',
|
|
'count(*) as count',
|
|
{
|
|
user_id => $uid,
|
|
train_no => $train->train_no,
|
|
checkout_station_id => $arr
|
|
}
|
|
)->hash->{count}
|
|
)
|
|
{
|
|
# check out (adds a cancelled journey and resets journey state
|
|
# to destination selection)
|
|
$self->app->checkout(
|
|
station => $arr,
|
|
force => 0,
|
|
uid => $uid
|
|
);
|
|
}
|
|
}
|
|
else {
|
|
$self->app->add_route_timestamps( $uid, $train, 0 );
|
|
}
|
|
}
|
|
elsif ( $entry->{real_arr_ts} ) {
|
|
my ( undef, $error ) = $self->app->checkout(
|
|
station => $arr,
|
|
force => 1,
|
|
uid => $uid
|
|
);
|
|
if ($error) {
|
|
die("${error}\n");
|
|
}
|
|
}
|
|
};
|
|
if ($@) {
|
|
$self->app->log->error("work($uid)/arrival: $@");
|
|
}
|
|
|
|
eval { }
|
|
}
|
|
|
|
for my $account_data ( $self->app->traewelling->get_pull_accounts ) {
|
|
|
|
# $account_data->{user_id} is the travelynx uid
|
|
# $account_data->{user_name} is the Träwelling username
|
|
$self->app->log->debug(
|
|
"Pulling Traewelling status for UID $account_data->{user_id}");
|
|
$self->app->traewelling_api->get_status_p(
|
|
username => $account_data->{data}{user_name},
|
|
token => $account_data->{token}
|
|
)->then(
|
|
sub {
|
|
my ($traewelling) = @_;
|
|
$self->app->traewelling_to_travelynx(
|
|
traewelling => $traewelling,
|
|
user_data => $account_data
|
|
);
|
|
}
|
|
)->catch(
|
|
sub {
|
|
my ($err) = @_;
|
|
$self->app->log->debug("Error $err");
|
|
}
|
|
)->wait;
|
|
}
|
|
|
|
for my $candidate ( $self->app->traewelling->get_pushable_accounts ) {
|
|
$self->app->log->debug(
|
|
"Pushing to Traewelling for UID $candidate->{uid}");
|
|
my $trip_id = $candidate->{journey_data}{trip_id};
|
|
if ( not $trip_id ) {
|
|
$self->app->log->debug("... trip_id is missing");
|
|
$self->app->traewelling->log(
|
|
uid => $candidate->{uid},
|
|
message =>
|
|
"Fehler bei $candidate->{train_type} $candidate->{train_no}: Keine trip_id vorhanden",
|
|
is_error => 1
|
|
);
|
|
next;
|
|
}
|
|
if ( $candidate->{data}{latest_push_ts}
|
|
and $candidate->{data}{latest_push_ts} == $candidate->{checkin_ts} )
|
|
{
|
|
$self->app->log->debug("... already handled");
|
|
next;
|
|
}
|
|
$self->app->traewelling_api->checkin( %{$candidate},
|
|
trip_id => $trip_id );
|
|
}
|
|
}
|
|
|
|
1;
|
|
|
|
__END__
|
|
|
|
=head1 SYNOPSIS
|
|
|
|
Usage: index.pl work
|
|
|
|
Work Work Work.
|
|
|
|
Should be called from a cronjob every three minutes or so.
|