<s> import logging <EOL> import time <EOL> import os <EOL> import glob <EOL> import shutil <EOL> import lxml . etree as etree <EOL> import lxml . builder as builder <EOL> import antd . plugin as plugin <EOL> import antd . garmin as garmin <EOL> _log = logging . getLogger ( "<STR_LIT>" ) <EOL> E = builder . ElementMaker ( nsmap = { <EOL> None : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } ) <EOL> X = builder . ElementMaker ( namespace = "<STR_LIT>" ) <EOL> class TcxPlugin ( plugin . Plugin ) : <EOL> tcx_output_dir = "<STR_LIT:.>" <EOL> def data_available ( self , device_sn , format , files ) : <EOL> if "<STR_LIT>" != format : return files <EOL> processed = [ ] <EOL> result = [ ] <EOL> try : <EOL> for file in files : <EOL> _log . info ( "<STR_LIT>" , file ) <EOL> try : <EOL> dir = self . tcx_output_dir % { "<STR_LIT>" : hex ( device_sn ) } <EOL> if not os . path . exists ( dir ) : os . makedirs ( dir ) <EOL> files = export_tcx ( device_sn , file , dir ) <EOL> result . extend ( files ) <EOL> processed . append ( file ) <EOL> except Exception : <EOL> _log . warning ( "<STR_LIT>" , file , exc_info = True ) <EOL> plugin . publish_data ( device_sn , "<STR_LIT>" , result ) <EOL> finally : <EOL> return processed <EOL> def format_time ( gmtime ) : <EOL> return time . strftime ( "<STR_LIT>" , gmtime ) <EOL> def format_intensity ( intensity ) : <EOL> if intensity == <NUM_LIT:1> : <EOL> return "<STR_LIT>" <EOL> else : <EOL> return "<STR_LIT>" <EOL> def format_trigger_method ( trigger_method ) : <EOL> if trigger_method == <NUM_LIT:0> : return "<STR_LIT>" <EOL> elif trigger_method == <NUM_LIT:1> : return "<STR_LIT>" <EOL> elif trigger_method == <NUM_LIT:2> : return "<STR_LIT>" <EOL> elif trigger_method == <NUM_LIT:3> : return "<STR_LIT>" <EOL> elif trigger_method == <NUM_LIT:4> : return "<STR_LIT>" <EOL> def format_sport ( sport ) : <EOL> if sport == <NUM_LIT:0> : return "<STR_LIT>" <EOL> elif sport == <NUM_LIT:1> : return "<STR_LIT>" <EOL> elif sport == <NUM_LIT:2> : return "<STR_LIT>" <EOL> def format_sensor_state ( sensor ) : <EOL> if sensor : return "<STR_LIT>" <EOL> else : return "<STR_LIT>" <EOL> def create_wpt ( wpt , sport_type ) : <EOL> elements = [ E . Time ( format_time ( wpt . time . gmtime ) ) ] <EOL> if wpt . posn . valid : <EOL> elements . extend ( [ <EOL> E . Position ( <EOL> E . LatitudeDegrees ( str ( wpt . posn . deglat ) ) , <EOL> E . LongitudeDegrees ( str ( wpt . posn . deglon ) ) ) ] ) <EOL> if wpt . alt is not None : <EOL> elements . append ( E . AltitudeMeters ( str ( wpt . alt ) ) ) <EOL> if wpt . distance is not None : <EOL> elements . append ( E . DistanceMeters ( str ( wpt . distance ) ) ) <EOL> if wpt . heart_rate : <EOL> elements . append ( E . HeartRateBpm ( E . Value ( str ( wpt . heart_rate ) ) ) ) <EOL> if wpt . cadence is not None and sport_type != <NUM_LIT:0> : <EOL> elements . append ( E . Cadence ( str ( wpt . cadence ) ) ) <EOL> if wpt . cadence is not None and sport_type == <NUM_LIT:0> : <EOL> elements . append ( E . Extensions ( X . TPX ( X . RunCadence ( str ( wpt . cadence ) ) ) ) ) <EOL> return E . Trackpoint ( * elements ) <EOL> def create_lap ( lap , sport_type ) : <EOL> elements = [ <EOL> E . TotalTimeSeconds ( "<STR_LIT>" % ( lap . total_time / <NUM_LIT> ) ) , <EOL> E . DistanceMeters ( str ( lap . total_dist ) ) , <EOL> E . MaximumSpeed ( str ( lap . max_speed ) ) , <EOL> E . Calories ( str ( lap . calories ) ) ] <EOL> if lap . avg_heart_rate or lap . max_heart_rate : <EOL> elements . extend ( [ <EOL> E . AverageHeartRateBpm ( E . Value ( str ( lap . avg_heart_rate ) ) ) , <EOL> E . MaximumHeartRateBpm ( E . Value ( str ( lap . max_heart_rate ) ) ) ] ) <EOL> elements . append ( <EOL> E . Intensity ( format_intensity ( lap . intensity ) ) ) <EOL> if lap . avg_cadence is not None and sport_type != <NUM_LIT:0> : <EOL> elements . append ( <EOL> E . Cadence ( str ( lap . avg_cadence ) ) ) <EOL> elements . append ( E . TriggerMethod ( format_trigger_method ( lap . trigger_method ) ) ) <EOL> wpts = [ el for el in ( create_wpt ( w , sport_type ) for w in lap . wpts ) if el is not None ] <EOL> if wpts : <EOL> elements . append ( E . Track ( * wpts ) ) <EOL> if lap . avg_cadence is not None and sport_type == <NUM_LIT:0> : <EOL> elements . append ( E . Extensions ( X . LX ( X . AvgRunCadence ( str ( lap . avg_cadence ) ) ) ) ) <EOL> return E . Lap ( <EOL> { "<STR_LIT>" : format_time ( lap . start_time . gmtime ) } , <EOL> * elements ) <EOL> def create_creator ( device ) : <EOL> major = device . device_id . software_version / <NUM_LIT:100> <EOL> minor = device . device_id . software_version % <NUM_LIT:100> <EOL> return E . Creator ( { "<STR_LIT>" : "<STR_LIT>" } , <EOL> E . Name ( "<STR_LIT>" . join ( device . device_id . description ) ) , <EOL> E . UnitId ( str ( device . stream . device_id ) ) , <EOL> E . ProductID ( str ( device . device_id . product_id ) ) , <EOL> E . Version ( E . VersionMajor ( str ( major ) ) , <EOL> E . VersionMinor ( str ( minor ) ) , <EOL> E . BuildMajor ( "<STR_LIT:0>" ) , <EOL> E . BuildMinor ( "<STR_LIT:0>" ) ) ) <EOL> def create_activity ( device , run ) : <EOL> laps = list ( create_lap ( l , run . sport_type ) for l in run . laps ) <EOL> return E . Activity ( <EOL> { "<STR_LIT>" : format_sport ( run . sport_type ) } , <EOL> E . Id ( format_time ( run . time . gmtime ) ) , <EOL> * ( laps + [ create_creator ( device ) ] ) ) <EOL> def create_document ( device , runs ) : <EOL> doc = E . TrainingCenterDatabase ( <EOL> E . Activities ( <EOL> * list ( create_activity ( device , r ) for r in runs ) ) ) <EOL> return doc <EOL> def export_tcx ( device_sn , raw_file_name , output_dir ) : <EOL> """<STR_LIT>""" <EOL> with open ( raw_file_name ) as file : <EOL> result = [ ] <EOL> host = garmin . MockHost ( file . read ( ) ) <EOL> host . device_id = device_sn <EOL> device = garmin . Device ( host ) <EOL> run_pkts = device . get_runs ( ) <EOL> runs = garmin . extract_runs ( device , run_pkts ) <EOL> for run in runs : <EOL> tcx_name = time . strftime ( "<STR_LIT>" , run . time . gmtime ) <EOL> tcx_full_path = os . path . sep . join ( [ output_dir , tcx_name ] ) <EOL> _log . info ( "<STR_LIT>" , os . path . basename ( raw_file_name ) , tcx_full_path ) <EOL> with open ( tcx_full_path , "<STR_LIT:w>" ) as file : <EOL> doc = create_document ( device , [ run ] ) <EOL> file . write ( etree . tostring ( doc , pretty_print = True , xml_declaration = True , encoding = "<STR_LIT>" ) ) <EOL> result . append ( tcx_full_path ) <EOL> return result </s>
