]>
code.delx.au - monosys/blob - opal-card-tool
20 CACHE_DIR
= os
.environ
.get("XDG_CACHE_HOME", os
.path
.expanduser("~/.cache/opal-card-tool"))
21 PICKLE_FILE
= os
.path
.join(CACHE_DIR
, "pickle")
23 USER_AGENT
= "Mozilla/5.0 (X11; Linux x86_64; rv:59.0) Gecko/20100101 Firefox/59.0"
24 OPAL_BASE
= "https://www.opal.com.au"
25 LOGIN_URL
= OPAL_BASE
+ "/login/registeredUserUsernameAndPasswordLogin"
26 CARD_DETAILS_URL
= OPAL_BASE
+ "/registered/getJsonCardDetailsArray"
27 TRANSACTION_LIST_URL
= OPAL_BASE
+ "/registered/opal-card-transactions/opal-card-activities-list?AMonth=-1&AYear=-1&cardIndex=%d&pageIndex=%d"
32 return " ".join(t
.strip() for t
in el
.itertext()).strip()
39 return d
.isoweekday() <= 5
42 d
= datetime
.datetime
.now() - datetime
.timedelta(days
=days
)
43 d
= d
.replace(hour
=0, minute
=0, second
=0, microsecond
=0)
47 class FatalError(Exception):
50 class Transaction(object):
55 self
.transaction_list
= []
57 def get_max_transaction(self
):
58 if self
.transaction_list
:
59 return self
.transaction_list
[0].number
63 def add_transactions(self
, l
):
64 self
.transaction_list
= l
+ self
.transaction_list
67 def __init__(self
, username
, password
):
68 self
.version
= VERSION
69 self
.username
= username
70 self
.password
= password
76 self
.session
= requests
.Session()
77 self
.session
.headers
["User-Agent"] = USER_AGENT
80 print("Attempting login ", end
="", flush
=True)
82 print(".", end
="", flush
=True)
84 if self
.check_login():
91 r
= self
.session
.post(LOGIN_URL
, {
92 "h_username": self
.username
,
93 "h_password": self
.password
,
96 raise Exception("Failed to login, error code: %d" % r
.status_code
)
99 if json
["errorMessage"]:
100 raise Exception("Failed to login: %s" % json
["errorMessage"])
102 def check_login(self
):
103 r
= self
.session
.get(CARD_DETAILS_URL
)
112 for card
in self
.cards
:
113 self
.load_transactions(card
)
115 def resolve_card_number(self
, card_number
):
116 if int(card_number
) < len(self
.cards
):
117 return self
.cards
[int(card_number
)].number
121 def get_transaction_list_for_card(self
, card_number
):
122 for card
in self
.cards
:
123 if card
.number
== card_number
:
124 return card
.transaction_list
128 def load_cards(self
):
129 r
= self
.session
.get(CARD_DETAILS_URL
)
131 raise Exception("Failed to login, error code: %d" % r
.status_code
)
133 for index
, card_json
in enumerate(r
.json()):
134 card_number
= card_json
["cardNumber"]
136 for card
in self
.cards
:
137 if card
.number
== card_number
:
141 self
.cards
.append(card
)
143 card
.number
= card_number
144 card
.name
= card_json
["cardNickName"]
147 def load_transactions(self
, card
):
148 print("Loading transactions for", card
.number
, "", end
="", flush
=True)
149 max_transaction
= card
.get_max_transaction()
150 transaction_list
= []
152 for page
in itertools
.count(1):
153 print(".", end
="", flush
=True)
154 transaction_page
= self
.fetch_transaction_page(card
.index
, page
)
155 continue_paging
= False
157 for transaction
in transaction_page
:
158 if transaction
.number
<= max_transaction
:
159 continue_paging
= False
162 transaction_list
.append(transaction
)
163 continue_paging
= True
165 if not continue_paging
:
169 card
.add_transactions(transaction_list
)
171 def parse_transaction(self
, cells
):
173 t
.number
= int(stringify(cells
["transaction number"]))
174 t
.timestamp
= datetime
.datetime
.strptime(stringify(cells
["date/time"]), "%a %d/%m/%Y %H:%M")
175 t
.mode
= get_first(cells
["mode"].xpath("img/@alt"))
176 t
.details
= stringify(cells
["details"])
177 t
.journey_number
= stringify(cells
["journey number"])
178 t
.fare_applied
= stringify(cells
["fare applied"])
179 t
.fare
= stringify(cells
["fare"])
180 t
.fare_discount
= stringify(cells
["discount"])
181 t
.amount
= stringify(cells
["amount"])
184 def fetch_transaction_page(self
, card
, page
):
185 url
= TRANSACTION_LIST_URL
% (card
, page
)
186 r
= self
.session
.get(url
)
188 raise Exception("Failed to fetch transactions, error code: %d" % r
.status_code
)
190 doc
= lxml
.html
.fromstring(r
.text
)
191 headers
= [stringify(th
).lower() for th
in doc
.xpath("//table/thead//th")]
196 for tr
in doc
.xpath("//table/tbody/tr"):
198 yield self
.parse_transaction(dict(zip(headers
, tr
.getchildren())))
200 print("Failed to parse:", headers
, lxml
.html
.tostring(tr
), file=sys
.stderr
)
204 class CommuterGraph(object):
205 class gnuplot_dialect(csv
.excel
):
209 self
.data_am_csv
, self
.data_am_file
= self
.new_csv()
210 self
.data_pm_csv
, self
.data_pm_file
= self
.new_csv()
211 self
.plot_file
= self
.new_tempfile()
212 self
.files
= [self
.data_am_file
, self
.data_pm_file
, self
.plot_file
]
214 self
.xrange_start
= None
215 self
.xrange_end
= None
217 def is_plottable(self
):
218 return self
.xrange_start
is not None and self
.xrange_end
is not None
220 def graph(self
, transaction_list
):
222 self
.write_points(transaction_list
)
223 if not self
.is_plottable():
224 print("No transactions!", file=sys
.stderr
)
226 self
.write_plot_command()
232 def new_tempfile(self
):
233 return tempfile
.NamedTemporaryFile(
236 prefix
="opal-card-tool-",
241 f
= self
.new_tempfile()
242 out
= csv
.writer(f
, dialect
=self
.gnuplot_dialect
)
245 def update_xrange(self
, ts
):
246 if self
.xrange_start
is None or ts
< self
.xrange_start
:
247 self
.xrange_start
= ts
248 if self
.xrange_end
is None or ts
> self
.xrange_end
:
251 def generate_point(self
, transaction
):
252 ts
= transaction
.timestamp
253 x_date
= ts
.strftime("%Y-%m-%dT00:00:00")
254 y_time
= ts
.strftime("2000-01-01T%H:%M:00")
255 y_label
= ts
.strftime("%H:%M")
256 return [x_date
, y_time
, y_label
]
258 def write_point(self
, ts
, point
):
259 if ts
.time() < datetime
.time(12):
260 out_csv
= self
.data_am_csv
262 out_csv
= self
.data_pm_csv
264 out_csv
.writerow(point
)
266 def write_points(self
, transaction_list
):
267 for transaction
in transaction_list
:
268 if not self
.is_commuter_transaction(transaction
):
271 self
.update_xrange(transaction
.timestamp
)
272 point
= self
.generate_point(transaction
)
273 self
.write_point(transaction
.timestamp
, point
)
275 def is_commuter_transaction(self
, transaction
):
276 if not is_weekday(transaction
.timestamp
):
278 if transaction
.details
.startswith("Auto top up"):
282 def write_plot_command(self
):
284 "data_am_filename": self
.data_am_file
.name
,
285 "data_pm_filename": self
.data_pm_file
.name
,
286 "xrange_start": self
.xrange_start
- datetime
.timedelta(hours
=24),
287 "xrange_end": self
.xrange_end
+ datetime
.timedelta(hours
=24),
289 self
.plot_file
.write(R
"""
290 set timefmt '%Y-%m-%dT%H:%M:%S'
295 set xtics 86400 scale 1.0,0.0
296 set xrange [ '{xrange_start}' : '{xrange_end}' ]
301 set yrange [ '2000-01-01T06:00:00' : '2000-01-01T23:00:00' ]
306 title 'opal-card-tool graph' \
312 '{data_pm_filename}' \
315 title 'Afternoon departure time' \
317 '{data_pm_filename}' \
323 '{data_am_filename}' \
326 title 'Morning departure time' \
328 '{data_am_filename}' \
335 def flush_files(self
):
346 def run_gnuplot(self
):
347 subprocess
.check_call([
352 def restrict_days(transaction_list
, num_days
):
353 oldest_date
= n_days_ago(num_days
)
354 for transaction
in transaction_list
:
355 if transaction
.timestamp
< oldest_date
:
359 def graph_commuter(transaction_list
):
361 g
.graph(transaction_list
)
363 def print_transaction_list(transaction_list
):
365 headers
.extend(["number", "timestamp"])
366 headers
.extend(h
for h
in sorted(transaction_list
[0].__dict
__.keys()) if h
not in headers
)
368 out
= csv
.DictWriter(sys
.stdout
, headers
)
370 for transaction
in transaction_list
:
371 out
.writerow(transaction
.__dict
__)
373 def print_cards(opal
):
374 for i
, card
in enumerate(opal
.cards
):
376 print(" number:", card
.number
)
377 print(" name:", card
.name
)
378 print(" transactions:", len(card
.transaction_list
))
382 if not os
.path
.isfile(PICKLE_FILE
):
385 with
open(PICKLE_FILE
, "rb") as f
:
386 return pickle
.load(f
)
388 def save_pickle(opal
):
389 if not os
.path
.isdir(CACHE_DIR
):
390 os
.makedirs(CACHE_DIR
)
391 with
open(PICKLE_FILE
, "wb") as f
:
396 def upgrade_opal_v2(opal
):
400 def upgrade_opal(opal
):
401 while opal
.version
< VERSION
:
402 print("Upgrading from version", opal
.version
, file=sys
.stderr
)
403 upgrade_func
= globals()["upgrade_opal_v%d" % opal
.version
]
409 opal
= try_unpickle()
415 username
= input("Username: ")
416 password
= getpass
.getpass()
417 opal
= Opal(username
, password
)
437 card_number
= args
.card_number
440 card_number
= opal
.resolve_card_number(card_number
)
443 num_days
= int(args
.num_days
)
444 elif args
.graph_commuter
:
449 transaction_list
= opal
.get_transaction_list_for_card(card_number
)
450 transaction_list
= list(restrict_days(transaction_list
, num_days
))
452 if not transaction_list
:
453 print("No transactions!", file=sys
.stderr
)
456 if args
.show_transactions
:
457 print_transaction_list(transaction_list
)
458 elif args
.graph_commuter
:
459 graph_commuter(transaction_list
)
461 print("Missing display function!", file=sys
.stderr
)
464 parser
= argparse
.ArgumentParser(description
="Opal card activity fetcher")
466 parser
.add_argument("--num-days",
467 help="restrict to NUM_DAYS of output"
469 parser
.add_argument("--card-number",
470 help="Opal card number or index (eg: 0,1,etc"
473 group
= parser
.add_mutually_exclusive_group(required
=True)
474 group
.add_argument("--load", action
="store_true",
475 help="load any new data from the Opal website"
477 group
.add_argument("--show-cards", action
="store_true",
478 help="show a list of cards"
480 group
.add_argument("--show-transactions", action
="store_true",
481 help="show transactions for card"
483 group
.add_argument("--graph-commuter", action
="store_true",
484 help="draw commuter graph for card with gnuplot"
487 args
= parser
.parse_args()
498 elif args
.show_cards
:
505 if __name__
== "__main__":
508 except (KeyboardInterrupt, BrokenPipeError
) as e
:
509 print("Exiting:", e
, file=sys
.stderr
)