from . import const from . import rpccommon from .common import MAVA_BASE_URL import jsonrpclib import hashlib import datetime import calendar import sys #--------------------------------------------------------------------------------------- class RPCObject(object): """Base class for objects read from RPC calls. It is possible to construct it from a dictionary.""" def __init__(self, value, instructions = {}): """Construct the object. value is the dictionary returned by the call. info is a mapping from names to 'instructions' on what to do with the corresponding values. If the instruction is None, it will be ignored. If the instruction is a function, the value will be passed to it and the return value will be stored in the object. For all other names, the value will be stored as the same-named attribute.""" for (key, value) in value.items(): if key in instructions: instruction = instructions[key] if instruction is None: continue try: value = instruction(value) except: print("Failed to convert value '%s' of attribute '%s':" % \ (value, key), file=sys.stderr) import traceback traceback.print_exc() setattr(self, key, value) #--------------------------------------------------------------------------------------- class Reply(RPCObject): """The generic reply structure.""" #--------------------------------------------------------------------------------------- class ScheduledFlight(RPCObject): """A scheduled flight in the time table.""" # The instructions for the construction # Type: normal flight TYPE_NORMAL = 0 # Type: VIP flight TYPE_VIP = 1 _instructions = { "id" : int, "pairID": int, "typeCode": lambda value: BookedFlight._decodeAircraftType(value), "departureTime": lambda value: ScheduledFlight._decodeTime(value), "arrivalTime": lambda value: ScheduledFlight._decodeTime(value), "duration": lambda value: ScheduledFlight._decodeDuration(value), "spec": int, "validFrom": lambda value: ScheduledFlight._decodeDate(value), "validTo": lambda value: ScheduledFlight._decodeDate(value), "date": lambda value: ScheduledFlight._decodeDate(value) } @staticmethod def _decodeTime(value): """Decode the given value as a time value.""" return datetime.datetime.strptime(value, "%H:%M:%S").time() @staticmethod def _decodeDate(value): """Decode the given value as a date value.""" if not value or value=="0000-00-00": return const.defaultDate else: return datetime.datetime.strptime(value, "%Y-%m-%d").date() @staticmethod def _decodeDuration(value): """Decode the given value as a duration. A number of seconds will be returned.""" t = datetime.datetime.strptime(value, "%H:%M:%S") return (t.hour*60 + t.minute) * 60 + t.second def __init__(self, value): """Construct the scheduled flight object from the given JSON value.""" super(ScheduledFlight, self).__init__(value, ScheduledFlight._instructions) self.aircraftType = self.typeCode del self.typeCode self.type = ScheduledFlight.TYPE_VIP if self.spec==1 \ else ScheduledFlight.TYPE_NORMAL del self.spec def compareBy(self, other, name): """Compare this flight with the other one according to the given attribute name.""" if name=="callsign": try: cs1 = int(self.callsign[2:]) cs2 = int(other.callsign[2:]) return 0 if cs1==cs2 else -1 if cs1 %d, %d>" % \ (self.id, self.pairID, BookedFlight.TYPE2TYPECODE[self.aircraftType], self.departureICAO, str(self.departureTime), self.arrivalICAO, str(self.arrivalTime), self.duration, self.type) #--------------------------------------------------------------------------------------- class ScheduledFlightPair(object): """A pair of scheduled flights. Occasionally, one of the flights may be missing.""" @staticmethod def scheduledFlights2Pairs(scheduledFlights, date): """Convert the given list of scheduled flights into a list of flight pairs.""" weekday = str(date.weekday()+1) flights = {} weekdayFlights = {} for flight in scheduledFlights: flights[flight.id] = flight if (flight.type==ScheduledFlight.TYPE_NORMAL and flight.arrivalICAO!="LHBP" and weekday in flight.days and flight.validFrom<=date and flight.validTo>=date) or \ flight.type==ScheduledFlight.TYPE_VIP: weekdayFlights[flight.id] = flight flightPairs = [] while weekdayFlights: (id, flight) = weekdayFlights.popitem() if flight.type==ScheduledFlight.TYPE_NORMAL: pairID = flight.pairID if pairID in flights: pairFlight = flights[pairID] if flight.departureICAO=="LHBP" or \ (pairFlight.departureICAO!="LHBP" and flight.callsign" % \ (self.flight0.callsign, self.flight0.departureICAO, self.flight0.arrivalICAO) #--------------------------------------------------------------------------------------- class BookedFlight(RPCObject): """A booked flight.""" # FIXME: copied from web.BookedFlight TYPECODE2TYPE = { "736" : const.AIRCRAFT_B736, "73G" : const.AIRCRAFT_B737, "738" : const.AIRCRAFT_B738, "73H" : const.AIRCRAFT_B738C, "732" : const.AIRCRAFT_B732, "733" : const.AIRCRAFT_B733, "734" : const.AIRCRAFT_B734, "735" : const.AIRCRAFT_B735, "DH4" : const.AIRCRAFT_DH8D, "762" : const.AIRCRAFT_B762, "763" : const.AIRCRAFT_B763, "CR2" : const.AIRCRAFT_CRJ2, "F70" : const.AIRCRAFT_F70, "LI2" : const.AIRCRAFT_DC3, "TU3" : const.AIRCRAFT_T134, "TU5" : const.AIRCRAFT_T154, "YK4" : const.AIRCRAFT_YK40, "146" : const.AIRCRAFT_B462 } # FIXME: copied from web.BookedFlight TYPE2TYPECODE = { const.AIRCRAFT_B736 : "736", const.AIRCRAFT_B737 : "73G", const.AIRCRAFT_B738 : "738", const.AIRCRAFT_B738C : "73H", const.AIRCRAFT_B732 : "732", const.AIRCRAFT_B733 : "733", const.AIRCRAFT_B734 : "734", const.AIRCRAFT_B735 : "735", const.AIRCRAFT_DH8D : "DH4", const.AIRCRAFT_B762 : "762", const.AIRCRAFT_B763 : "763", const.AIRCRAFT_CRJ2 : "CR2", const.AIRCRAFT_F70 : "F70", const.AIRCRAFT_DC3 : "LI2", const.AIRCRAFT_T134 : "TU3", const.AIRCRAFT_T154 : "TU5", const.AIRCRAFT_YK40 : "YK4", const.AIRCRAFT_B462 : "146" } # FIXME: copied from web.BookedFlight @staticmethod def _decodeAircraftType(typeCode): """Decode the aircraft type from the given typeCode.""" if typeCode in BookedFlight.TYPECODE2TYPE: return BookedFlight.TYPECODE2TYPE[typeCode] else: raise Exception("Invalid aircraft type code: '" + typeCode + "'") @staticmethod def _decodeStatus(status): """Decode the status from the status string.""" if status=="booked": return BookedFlight.STATUS_BOOKED elif status=="reported": return BookedFlight.STATUS_REPORTED elif status=="accepted": return BookedFlight.STATUS_ACCEPTED elif status=="rejected": return BookedFlight.STATUS_REJECTED else: raise Exception("Invalid flight status code: '" + status + "'") # FIXME: copied from web.BookedFlight @staticmethod def getDateTime(date, time): """Get a datetime object from the given textual date and time.""" return datetime.datetime.strptime(date + " " + time, "%Y-%m-%d %H:%M:%S") # FIXME: copied from web.BookedFlight STATUS_BOOKED = 1 # FIXME: copied from web.BookedFlight STATUS_REPORTED = 2 # FIXME: copied from web.BookedFlight STATUS_ACCEPTED = 3 # FIXME: copied from web.BookedFlight STATUS_REJECTED = 4 # The instructions for the construction _instructions = { "numPassengers" : int, "numCrew" : int, "bagWeight" : int, "cargoWeight" : int, "mailWeight" : int, "aircraftType" : lambda value: BookedFlight._decodeAircraftType(value), "status" : lambda value: BookedFlight._decodeStatus(value) } def __init__(self, value): """Construct the booked flight object from the given RPC result value.""" self.status = BookedFlight.STATUS_BOOKED super(BookedFlight, self).__init__(value, BookedFlight._instructions) self.departureTime = \ BookedFlight.getDateTime(self.date, self.departureTime) self.arrivalTime = \ BookedFlight.getDateTime(self.date, self.arrivalTime) if self.arrivalTime