# -*- coding: utf-8 -*- import copy import datetime import ioex.datetimeex import pytz import re import yaml import yaml.representer ioex.register_yaml_str_as_unicode_constructor(yaml.Loader) ioex.register_yaml_unicode_as_str_representer(yaml.Dumper) ioex.datetimeex.register_yaml_timestamp_constructor(yaml.Loader) ioex.datetimeex.Duration.register_yaml_constructor(yaml.Loader) ioex.datetimeex.Duration.register_yaml_representer(yaml.Dumper) ioex.datetimeex.Period.register_yaml_constructor(yaml.Loader) ioex.datetimeex.Period.register_yaml_representer(yaml.Dumper) class _Object(object): def __eq__(self, other): return type(self) == type(other) and vars(self) == vars(other) def __ne__(self, other): return not (self == other) class _YamlInitConstructor(_Object, yaml.YAMLObject): @classmethod def from_yaml(cls, loader, node): return cls(**loader.construct_mapping(node, deep = True)) # return cls(**{ # k: unicode(v) if isinstance(v, str) else v # for (k, v) in loader.construct_mapping(node, deep = True).items() # }) def __eq__(self, other): return (type(self) == type(other) and vars(self) == vars(other)) def __ne__(self, other): return not (self == other) class Figure(_YamlInitConstructor): yaml_tag = u"!figure" def __init__(self, value = None, unit = None): self.value = value self.unit = unit def get_value(self): return self._value def set_value(self, value): self._value = value """ use property() instead of decorator to enable overriding """ value = property(get_value, set_value) def get_unit(self): return self._unit def set_unit(self, unit): assert type(unit) is unicode self._unit = unit """ use property() instead of decorator to enable overriding """ unit = property(get_unit, set_unit) @classmethod def to_yaml(cls, dumper, figure): return dumper.represent_mapping( cls.yaml_tag, {'unit': figure.get_unit(), 'value': figure.get_value()}, ) def __sub__(self, other): assert type(self) == type(other) assert self.unit == other.unit result = copy.deepcopy(self) result.value -= other.value return result class ScalarFigure(Figure): yaml_tag = u'!scalar' def get_value(self): return super(ScalarFigure, self).get_value() def set_value(self, value): assert type(value) is float super(ScalarFigure, self).set_value(value) """ use property() instead of decorator to enable overriding """ value = property(get_value, set_value) @classmethod def from_yaml(cls, loader, node): attr = loader.construct_scalar(node).split(' ') return cls( value = float(attr[0]), unit = attr[1], ) @classmethod def to_yaml(cls, dumper, s): return dumper.represent_scalar( cls.yaml_tag, '%s %s' % (repr(s.value), s.unit), ) class Distance(ScalarFigure): yaml_tag = u'!distance' @property def metres(self): if self.unit == 'km': return self.value * 1000 else: raise Exception() class Sum(ScalarFigure): yaml_tag = u'!sum' def __init__(self, value = None, currency = None, unit = None): if not currency is None and not unit is None: raise ValueError('either specify currency or unit') else: unit = currency if currency else unit super(Sum, self).__init__( value = value, unit = currency if currency else unit, ) @property def currency(self): return self.unit def get_unit(self): return super(Sum, self).get_unit() def set_unit(self, currency): if currency == u'€': currency = u'EUR' if currency == u'US$': currency = u'USD' assert type(currency) is unicode assert currency in [u'EUR', u'USD'] super(Sum, self).set_unit(currency) """ use property() instead of decorator to enable overriding """ unit = property(get_unit, set_unit) @classmethod def from_yaml(cls, loader, node): attr = loader.construct_scalar(node).split(' ') return cls( value = float(attr[0]), currency = attr[1], ) @staticmethod def parse_text(text): match = re.search( ur'^(?P((US)?\$|€))(?P-?\d+([\.,]\d{2})?)( (?PUSD))?$', text, re.UNICODE, ) assert not match is None, text attr = match.groupdict() currency = attr['currency'] if attr['currency'] else attr['curr_pre'] if currency == u'$': currency = u'USD' return Sum( value = float(attr['value'].replace(',', '.')), currency = currency, ) class Discount(_YamlInitConstructor): yaml_tag = u'!discount' def __init__( self, name = None, amount = None, ): assert type(name) is unicode self.name = name assert type(amount) is Sum assert amount.value >= 0 self.amount = amount class Order(_YamlInitConstructor): yaml_tag = u'!order' def __init__(self, platform, order_id, order_date, customer_id = None, items = None, discounts = None, ): assert type(platform) is unicode self.platform = platform if type(order_id) in [int]: order_id = unicode(order_id) assert type(order_id) is unicode self.order_id = order_id assert type(order_date) in [datetime.date, datetime.datetime] if type(order_date) is datetime.datetime and order_date.tzinfo: order_date = order_date.astimezone(pytz.utc) self.order_date = order_date if customer_id is not None: assert type(customer_id) is unicode self.customer_id = customer_id if items is None: self.items = [] else: assert type(items) is list assert all([isinstance(i, Item) for i in items]) self.items = items if discounts is None: self.discounts = [] else: assert type(discounts) is list assert all([isinstance(d, Discount) for d in discounts]) self.discounts = discounts class Item(_YamlInitConstructor): yaml_tag = u'!item' def __init__( self, name = None, price_brutto = None, ): if not name is None: assert type(name) is unicode self.name = name assert type(price_brutto) is Sum self.price_brutto = price_brutto class Campaign(_YamlInitConstructor): yaml_tag = u'!campaign' def __init__(self, end = None, founder = None, name = None, website_url = None, ): assert type(name) is unicode self.name = name assert type(founder) is unicode self.founder = founder if not end is None: assert type(end) is datetime.datetime assert not end.tzinfo is None, '%r' % end self.end = end if not website_url is None: assert type(website_url) is unicode self.website_url = website_url class Pledge(Item): yaml_tag = u'!pledge' def __init__(self, campaign = None, reward = None, **kwargs ): super(Pledge, self).__init__(**kwargs) assert type(campaign) is Campaign self.campaign = campaign if not reward is None: assert type(reward) is unicode self.reward = reward class Contribution(Item): yaml_tag = u'!contribution' def __init__(self, campaign = None, reward = None, **kwargs ): super(Contribution, self).__init__(**kwargs) assert type(campaign) is Campaign self.campaign = campaign if not reward is None: assert type(reward) is unicode self.reward = reward class Article(Item): yaml_tag = u'!article' def __init__( self, authors = None, color = None, delivery_date = None, depth = None, features = None, height = None, maximum_load = None, option = None, product_id = None, quantity = None, reseller = None, shipper = None, size = None, state = None, width = None, **kwargs ): super(Article, self).__init__(**kwargs) assert not self.name is None assert type(quantity) is int self.quantity = quantity if authors is not None: assert type(authors) is list self.authors = authors if state is not None: assert type(state) is unicode self.state = state if reseller is not None: assert type(reseller) is unicode self.reseller = reseller if shipper is not None: assert type(shipper) is unicode self.shipper = shipper if product_id is not None: if type(product_id) in [int]: product_id = unicode(product_id) assert type(product_id) is unicode self.product_id = product_id if option is not None: assert type(option) is unicode self.option = option if color is not None: assert type(color) is unicode self.color = color if size is not None: assert type(size) is unicode self.size = size if width is not None: assert type(width) is ScalarFigure self.width = width if depth is not None: assert type(depth) is ScalarFigure self.depth = depth if height is not None: assert type(height) is ScalarFigure self.height = height if maximum_load is not None: assert type(maximum_load) is ScalarFigure self.maximum_load = maximum_load if features is not None: assert type(features) is unicode self.features = features assert delivery_date is None or type(delivery_date) is datetime.date self.delivery_date = delivery_date class Service(Item): yaml_tag = u'!service' def __init__( self, duration = None, **kwargs ): super(Service, self).__init__(**kwargs) if duration is not None: assert type(duration) is ioex.datetimeex.Duration self.duration = duration class Transportation(Item): yaml_tag = u'!transportation' def __init__( self, departure_point = None, destination_point = None, distance = None, estimated_arrival_time = None, passenger = None, route_map = None, ticket_url = None, valid_from = None, valid_until = None, **kwargs ): super(Transportation, self).__init__(**kwargs) if departure_point is not None: assert type(departure_point) is unicode self.departure_point = departure_point if destination_point is not None: assert type(destination_point) is unicode self.destination_point = destination_point if distance is not None: assert type(distance) is Distance self.distance = distance if route_map is not None: assert type(route_map) is str self.route_map = route_map if passenger is not None: assert type(passenger) is Person self.passenger = passenger if valid_from is not None: assert type(valid_from) is datetime.datetime assert not valid_from.tzinfo is None self.valid_from = valid_from if valid_until is not None: assert type(valid_until) is datetime.datetime assert not valid_until.tzinfo is None self.valid_until = valid_until if ticket_url is not None: assert type(ticket_url) is unicode self.ticket_url = ticket_url if estimated_arrival_time is not None: assert type(estimated_arrival_time) is ioex.datetimeex.Period assert not estimated_arrival_time.start.tzinfo is None assert not estimated_arrival_time.end.tzinfo is None self.estimated_arrival_time = estimated_arrival_time class Shipping(Transportation): yaml_tag = u'!shipping' def __init__( self, **kwargs ): super(Shipping, self).__init__(**kwargs) class TaxiRide(Transportation): yaml_tag = u'!taxi-ride' def __init__( self, arrival_time = None, departure_time = None, driver = None, name = None, **kwargs ): if name is None: name = u'Taxi Ride' super(TaxiRide, self).__init__(name = name, **kwargs) assert type(driver) is unicode self.driver = driver assert arrival_time is None or type(arrival_time) is datetime.datetime self.arrival_time = arrival_time assert departure_time is None or type(departure_time) is datetime.datetime self.departure_time = departure_time class OrderRegistry(_Object, yaml.YAMLObject): yaml_tag = u'!order-registry' def __init__(self): self.registry = {} def register(self, order): assert isinstance(order, Order) if not order.platform in self.registry: self.registry[order.platform] = {} self.registry[order.platform][order.order_id] = order @classmethod def to_yaml(cls, dumper, self): return dumper.represent_mapping(cls.yaml_tag, self.registry) @classmethod def from_yaml(cls, loader, node): self = cls() self.registry = loader.construct_mapping(node) return self class Person(_YamlInitConstructor): yaml_tag = u'!person' def __init__(self, first_name = None, last_name = None): self.first_name = first_name self.last_name = last_name @property def first_name(self): return self._first_name @first_name.setter def first_name(self, first_name): assert first_name is None or type(first_name) is unicode self._first_name = first_name @property def last_name(self): return self._last_name @last_name.setter def last_name(self, last_name): assert last_name is None or type(last_name) is unicode self._last_name = last_name @classmethod def to_yaml(cls, dumper, person): return dumper.represent_mapping(cls.yaml_tag, { 'first_name': person.first_name, 'last_name': person.last_name, }) def __repr__(self): return self.__class__.__name__ + '(%s)' % ', '.join([ '%s=%r' % (k, v) for k, v in vars(self).items() ])