]>
code.delx.au - monosys/blob - opal-card-tool
19 CACHE_DIR
= os
.environ
.get("XDG_CACHE_HOME", os
.path
.expanduser("~/.cache/opal-card-tool"))
20 PICKLE_FILE
= os
.path
.join(CACHE_DIR
, "pickle")
22 OPAL_BASE
= "https://www.opal.com.au"
23 LOGIN_URL
= OPAL_BASE
+ "/login/registeredUserUsernameAndPasswordLogin"
24 CARD_DETAILS_URL
= OPAL_BASE
+ "/registered/getJsonCardDetailsArray"
25 TRANSACTION_LIST_URL
= OPAL_BASE
+ "/registered/opal-card-transactions/opal-card-activities-list?AMonth=-1&AYear=-1&cardIndex=%d&pageIndex=%d"
30 return " ".join(t
.strip() for t
in el
.itertext()).strip()
37 return d
.isoweekday() <= 5
40 d
= datetime
.datetime
.now() - datetime
.timedelta(days
=days
)
41 d
= d
.replace(hour
=0, minute
=0, second
=0, microsecond
=0)
45 class FatalError(Exception):
48 class Transaction(object):
53 self
.transaction_list
= []
55 def get_max_transaction(self
):
56 if self
.transaction_list
:
57 return self
.transaction_list
[0].number
61 def add_transactions(self
, l
):
62 self
.transaction_list
= l
+ self
.transaction_list
65 def __init__(self
, username
, password
):
66 self
.version
= VERSION
67 self
.username
= username
68 self
.password
= password
70 self
.session
= requests
.Session()
74 r
= self
.session
.post(LOGIN_URL
, {
75 "h_username": self
.username
,
76 "h_password": self
.password
,
79 raise Exception("Failed to login, error code: %d" % r
.status_code
)
82 if json
["errorMessage"]:
83 raise Exception("Failed to login: %s" % json
["errorMessage"])
87 for card
in self
.cards
:
88 self
.load_transactions(card
)
90 def resolve_card_number(self
, card_number
):
91 if int(card_number
) < len(self
.cards
):
92 return self
.cards
[int(card_number
)].number
96 def get_transaction_list_for_card(self
, card_number
):
97 for card
in self
.cards
:
98 if card
.number
== card_number
:
99 return card
.transaction_list
103 def load_cards(self
):
104 r
= self
.session
.get(CARD_DETAILS_URL
)
106 raise Exception("Failed to login, error code: %d" % r
.status_code
)
108 for index
, card_json
in enumerate(r
.json()):
109 card_number
= card_json
["cardNumber"]
111 for card
in self
.cards
:
112 if card
.number
== card_number
:
116 self
.cards
.append(card
)
118 card
.number
= card_number
119 card
.name
= card_json
["cardNickName"]
122 def load_transactions(self
, card
):
123 print("Loading transactions for", card
.number
, "", end
="", flush
=True)
124 max_transaction
= card
.get_max_transaction()
125 transaction_list
= []
127 for page
in itertools
.count(1):
128 print(".", end
="", flush
=True)
129 transaction_page
= self
.fetch_transaction_page(card
.index
, page
)
130 continue_paging
= False
132 for transaction
in transaction_page
:
133 if transaction
.number
<= max_transaction
:
134 continue_paging
= False
137 transaction_list
.append(transaction
)
138 continue_paging
= True
140 if not continue_paging
:
144 card
.add_transactions(transaction_list
)
146 def parse_transaction(self
, cells
):
148 t
.number
= int(stringify(cells
["transaction number"]))
149 t
.timestamp
= datetime
.datetime
.strptime(stringify(cells
["date/time"]), "%a %d/%m/%Y %H:%M")
150 t
.mode
= get_first(cells
["mode"].xpath("img/@alt"))
151 t
.details
= stringify(cells
["details"])
152 t
.journey_number
= stringify(cells
["journey number"])
153 t
.fare_applied
= stringify(cells
["fare applied"])
154 t
.fare
= stringify(cells
["fare"])
155 t
.fare_discount
= stringify(cells
["discount"])
156 t
.amount
= stringify(cells
["amount"])
159 def fetch_transaction_page(self
, card
, page
):
160 url
= TRANSACTION_LIST_URL
% (card
, page
)
161 r
= self
.session
.get(url
)
163 raise Exception("Failed to fetch transactions, error code: %d" % r
.status_code
)
165 doc
= lxml
.html
.fromstring(r
.text
)
166 headers
= [stringify(th
).lower() for th
in doc
.xpath("//table/thead//th")]
171 for tr
in doc
.xpath("//table/tbody/tr"):
173 yield self
.parse_transaction(dict(zip(headers
, tr
.getchildren())))
175 print("Failed to parse:", headers
, lxml
.html
.tostring(tr
), file=sys
.stderr
)
179 class CommuterGraph(object):
180 class gnuplot_dialect(csv
.excel
):
184 self
.data_am_csv
, self
.data_am_file
= self
.new_csv()
185 self
.data_pm_csv
, self
.data_pm_file
= self
.new_csv()
186 self
.plot_file
= self
.new_tempfile()
187 self
.files
= [self
.data_am_file
, self
.data_pm_file
, self
.plot_file
]
189 self
.xrange_start
= None
190 self
.xrange_end
= None
192 def is_plottable(self
):
193 return self
.xrange_start
is not None and self
.xrange_end
is not None
195 def graph(self
, transaction_list
):
197 self
.write_points(transaction_list
)
198 if not self
.is_plottable():
199 print("No transactions!", file=sys
.stderr
)
201 self
.write_plot_command()
207 def new_tempfile(self
):
208 return tempfile
.NamedTemporaryFile(
211 prefix
="opal-card-tool-",
216 f
= self
.new_tempfile()
217 out
= csv
.writer(f
, dialect
=self
.gnuplot_dialect
)
220 def update_xrange(self
, ts
):
221 if self
.xrange_start
is None or ts
< self
.xrange_start
:
222 self
.xrange_start
= ts
223 if self
.xrange_end
is None or ts
> self
.xrange_end
:
226 def generate_point(self
, transaction
):
227 ts
= transaction
.timestamp
228 x_date
= ts
.strftime("%Y-%m-%dT00:00:00")
229 y_time
= ts
.strftime("2000-01-01T%H:%M:00")
230 y_label
= ts
.strftime("%H:%M")
231 return [x_date
, y_time
, y_label
]
233 def write_point(self
, ts
, point
):
234 if ts
.time() < datetime
.time(12):
235 out_csv
= self
.data_am_csv
237 out_csv
= self
.data_pm_csv
239 out_csv
.writerow(point
)
241 def write_points(self
, transaction_list
):
242 for transaction
in transaction_list
:
243 if not self
.is_commuter_transaction(transaction
):
246 self
.update_xrange(transaction
.timestamp
)
247 point
= self
.generate_point(transaction
)
248 self
.write_point(transaction
.timestamp
, point
)
250 def is_commuter_transaction(self
, transaction
):
251 if not is_weekday(transaction
.timestamp
):
253 if transaction
.details
.startswith("Auto top up"):
257 def write_plot_command(self
):
259 "data_am_filename": self
.data_am_file
.name
,
260 "data_pm_filename": self
.data_pm_file
.name
,
261 "xrange_start": self
.xrange_start
- datetime
.timedelta(hours
=24),
262 "xrange_end": self
.xrange_end
+ datetime
.timedelta(hours
=24),
264 self
.plot_file
.write(R
"""
265 set timefmt '%Y-%m-%dT%H:%M:%S'
270 set xtics 86400 scale 1.0,0.0
271 set xrange [ '{xrange_start}' : '{xrange_end}' ]
276 set yrange [ '2000-01-01T06:00:00' : '2000-01-01T23:00:00' ]
281 title 'opal-card-tool graph' \
287 '{data_pm_filename}' \
290 title 'Afternoon departure time' \
292 '{data_pm_filename}' \
298 '{data_am_filename}' \
301 title 'Morning departure time' \
303 '{data_am_filename}' \
310 def flush_files(self
):
321 def run_gnuplot(self
):
322 subprocess
.check_call([
327 def restrict_days(transaction_list
, num_days
):
328 oldest_date
= n_days_ago(num_days
)
329 for transaction
in transaction_list
:
330 if transaction
.timestamp
< oldest_date
:
334 def graph_commuter(transaction_list
):
336 g
.graph(transaction_list
)
338 def print_transaction_list(transaction_list
):
340 headers
.extend(["number", "timestamp"])
341 headers
.extend(h
for h
in sorted(transaction_list
[0].__dict
__.keys()) if h
not in headers
)
343 out
= csv
.DictWriter(sys
.stdout
, headers
)
345 for transaction
in transaction_list
:
346 out
.writerow(transaction
.__dict
__)
348 def print_cards(opal
):
349 for i
, card
in enumerate(opal
.cards
):
351 print(" number:", card
.number
)
352 print(" name:", card
.name
)
353 print(" transactions:", len(card
.transaction_list
))
357 if not os
.path
.isfile(PICKLE_FILE
):
360 with
open(PICKLE_FILE
, "rb") as f
:
361 return pickle
.load(f
)
363 def save_pickle(opal
):
364 if not os
.path
.isdir(CACHE_DIR
):
365 os
.makedirs(CACHE_DIR
)
366 with
open(PICKLE_FILE
, "wb") as f
:
371 def upgrade_opal_v2(opal
):
375 def upgrade_opal(opal
):
376 while opal
.version
< VERSION
:
377 print("Upgrading from version", opal
.version
, file=sys
.stderr
)
378 upgrade_func
= globals()["upgrade_opal_v%d" % opal
.version
]
384 opal
= try_unpickle()
389 username
= input("Username: ")
390 password
= getpass
.getpass()
391 opal
= Opal(username
, password
)
411 card_number
= args
.card_number
414 card_number
= opal
.resolve_card_number(card_number
)
417 num_days
= int(args
.num_days
)
418 elif args
.graph_commuter
:
423 transaction_list
= opal
.get_transaction_list_for_card(card_number
)
424 transaction_list
= list(restrict_days(transaction_list
, num_days
))
426 if not transaction_list
:
427 print("No transactions!", file=sys
.stderr
)
430 if args
.show_transactions
:
431 print_transaction_list(transaction_list
)
432 elif args
.graph_commuter
:
433 graph_commuter(transaction_list
)
435 print("Missing display function!", file=sys
.stderr
)
438 parser
= argparse
.ArgumentParser(description
="Opal card activity fetcher")
440 parser
.add_argument("--num-days",
441 help="restrict to NUM_DAYS of output"
443 parser
.add_argument("--card-number",
444 help="Opal card number or index (eg: 0,1,etc"
447 group
= parser
.add_mutually_exclusive_group(required
=True)
448 group
.add_argument("--load", action
="store_true",
449 help="load any new data from the Opal website"
451 group
.add_argument("--show-cards", action
="store_true",
452 help="show a list of cards"
454 group
.add_argument("--show-transactions", action
="store_true",
455 help="show transactions for card"
457 group
.add_argument("--graph-commuter", action
="store_true",
458 help="draw commuter graph for card with gnuplot"
461 args
= parser
.parse_args()
472 elif args
.show_cards
:
479 if __name__
== "__main__":
482 except (KeyboardInterrupt, BrokenPipeError
) as e
:
483 print("Exiting:", e
, file=sys
.stderr
)