finoex/finoex/__init__.py

import datetime
import ioex
import ioex.calcex
import ioex.reex
import locale
import pytz
import re
import yaml


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)

    def __repr__(self):
        return self.__class__.__name__ + '(%s)' % ', '.join([
            '%s=%r' % (k, v) for k, v in vars(self).items()
        ])


class _YamlInitConstructor(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()
        #     })

    @classmethod
    def register_yaml_constructor(cls, loader, tag=None):
        loader.add_constructor(
            cls.yaml_tag if tag is None else tag,
            cls.from_yaml,
        )


class _YamlVarsRepresenter(yaml.YAMLObject):

    @classmethod
    def to_yaml(cls, dumper, obj):
        return dumper.represent_mapping(
            cls.yaml_tag,
            {k: v for k, v in vars(obj).items()
                if v and (not isinstance(v, list) or len(v) > 0)},
        )


class Sum(ioex.calcex.Figure):

    currency_symbol_map = {
        '€': 'EUR',
        'US$': 'USD',
        '￥': 'CNY',
    }

    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('missing currency')
        else:
            unit = currency if currency else unit
            super(Sum, self).__init__(
                value=value,
                unit=currency if currency else unit,
            )

    def get_value(self):
        return super(Sum, self).get_value()

    def set_value(self, value):
        assert type(value) is float
        super(Sum, self).set_value(value)

    """ use property() instead of decorator to enable overriding """
    value = property(get_value, set_value)

    def get_unit(self):
        return super(Sum, self).get_unit()

    def set_unit(self, currency):
        currency = Sum.currency_symbol_map.get(currency, currency)
        assert type(currency) is str
        super(Sum, self).set_unit(currency)

    """ use property() instead of decorator to enable overriding """
    unit = property(get_unit, set_unit)
    currency = property(get_unit, set_unit)

    value_regex = r"-?\d+([\.,]\d+)?"
    currency_regex = r"[^\d\s-]+"
    sum_regex_value_first = r'\$?(?P<value>{}) (?P<currency>{})'.format(
        value_regex, currency_regex,
    )
    sum_regex_currency_first = r'(?P<currency>{}) ?(?P<value>{})'.format(
        currency_regex, value_regex,
    )
    sum_regex = r'({})'.format('|'.join([
        ioex.reex.rename_groups(
            sum_regex_value_first,
            lambda n: {'value': 'pre_value', 'currency': 'post_currency'}[n]
        ),
        ioex.reex.rename_groups(
            sum_regex_currency_first,
            lambda n: {'currency': 'pre_currency', 'value': 'post_value'}[n]
        ),
    ]))

    @staticmethod
    def parse_text(text):
        match = re.search('^{}$'.format(Sum.sum_regex), text, re.UNICODE)
        assert not match is None, '\n' + '\n'.join([
            'regex: {}'.format(Sum.sum_regex),
            'text: {}'.format(text),
            'text repr: {!r}'.format(text),
        ])
        attr = ioex.dict_collapse(
            match.groupdict(),
            lambda k: k.split('_')[-1],
        )
        return Sum(
            value=locale.atof(attr['value']),
            currency=attr['currency'],
        )


class _ItemCollection():

    def __init__(self, items=None):
        if items is not None:
            assert isinstance(items, list)
            assert all([isinstance(i, Item) for i in items])
            self.items = items
        else:
            self.items = []


class Invoice(_Object, _ItemCollection, _YamlInitConstructor, _YamlVarsRepresenter):

    yaml_tag = u'!invoice'

    def __init__(self,
                 creditor,
                 debitor_id,
                 invoice_date,
                 invoice_id,
                 discounts=None,
                 invoice_url=None,
                 items=None
                 ):
        assert isinstance(creditor, str)
        self.creditor = creditor
        assert isinstance(invoice_id, str)
        self.invoice_id = invoice_id
        assert (isinstance(invoice_date, datetime.date)
                or isinstance(invoice_date, datetime.datetime))
        self.invoice_date = invoice_date
        assert isinstance(debitor_id, str)
        self.debitor_id = debitor_id
        if discounts:
            assert isinstance(discounts, list)
            assert all([isinstance(d, Discount) for d in discounts])
            self.discounts = discounts
        else:
            self.discounts = []
        if invoice_url:
            assert isinstance(invoice_url, str)
            self.invoice_url = invoice_url
        _ItemCollection.__init__(self, items)


class Order(_Object, _ItemCollection, _YamlInitConstructor, _YamlVarsRepresenter):

    yaml_tag = u'!order'

    def __init__(self, platform, order_id, order_date,
                 customer_id=None,
                 items=None,
                 discounts=None,
                 ):
        assert type(platform) is str
        self.platform = platform
        if type(order_id) in [int]:
            order_id = str(order_id)
        assert type(order_id) is str
        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 str
            self.customer_id = customer_id
        if discounts is None:
            self.discounts = []
        else:
            assert type(discounts) is list
            assert all([isinstance(d, Discount) for d in discounts])
            self.discounts = discounts
        _ItemCollection.__init__(self, items)


class Distance(ioex.calcex.Figure):

    yaml_tag = u'!distance'

    def get_value(self):
        return super(Distance, self).get_value()

    def set_value(self, value):
        assert type(value) is float
        super(Distance, self).set_value(value)

    """ use property() instead of decorator to enable overriding """
    value = property(get_value, set_value)

    @property
    def metres(self):
        if self.unit == 'm':
            return Distance(self.value, 'm')
        elif self.unit == 'km':
            return Distance(self.value * 1000, 'm')
        else:
            raise Exception()


class Discount(_Object, _YamlInitConstructor):

    yaml_tag = u'!discount'

    def __init__(self, name=None, amount=None, code=None):
        assert type(name) is str
        self.name = name
        assert type(amount) is Sum
        assert amount.value >= 0
        self.amount = amount
        if code:
            assert isinstance(code, str)
            self.code = code


class Item(_Object, _YamlInitConstructor, _YamlVarsRepresenter):

    yaml_tag = u'!item'

    def __init__(self,
                 name=None,
                 price_brutto=None,
                 sub_items=None,
                 ):
        if not name is None:
            assert type(name) is str
            self.name = name
        assert type(price_brutto) is Sum
        if sub_items is None:
            self.sub_items = []
        else:
            assert isinstance(sub_items, list)
            assert all([isinstance(i, Item) for i in sub_items])
            self.sub_items = sub_items
        self.price_brutto = price_brutto


class Campaign(_Object, _YamlInitConstructor):

    yaml_tag = u'!campaign'

    def __init__(self,
                 end=None,
                 founder=None,
                 name=None,
                 website_url=None,
                 ):
        assert type(name) is str
        self.name = name
        assert type(founder) is str
        self.founder = founder
        if not end is None:
            assert type(end) is datetime.datetime
            self.end = end
        if not website_url is None:
            assert type(website_url) is str
            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 str
            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 str
            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 str
            self.state = state
        if reseller is not None:
            assert type(reseller) is str
            self.reseller = reseller
        if shipper is not None:
            assert type(shipper) is str
            self.shipper = shipper
        if product_id is not None:
            if type(product_id) in [int]:
                product_id = str(product_id)
            assert type(product_id) is str
            self.product_id = product_id
        if option is not None:
            assert type(option) is str
            self.option = option
        if color is not None:
            assert type(color) is str
            self.color = color
        if size is not None:
            assert type(size) is str
            self.size = size
        if width is not None:
            assert type(width) is Distance
            self.width = width
        if depth is not None:
            assert type(depth) is Distance
            self.depth = depth
        if height is not None:
            assert type(height) is Distance
            self.height = height
        if maximum_load is not None:
            assert type(maximum_load) is ioex.calcex.Figure, type(maximum_load)
            self.maximum_load = maximum_load
        if features is not None:
            assert type(features) is str
            self.features = features
        if delivery_date is not None:
            assert type(delivery_date) is datetime.date
            self.delivery_date = delivery_date


class Service(Item):

    yaml_tag = u'!service'

    def __init__(self,
                 duration=None,
                 ip_addresses=None,
                 location=None,
                 period=None,
                 state=None,
                 **kwargs
                 ):
        super(Service, self).__init__(**kwargs)
        assert not (duration and period)
        if duration:
            assert isinstance(duration, ioex.datetimeex.Duration)
            self.duration = duration
        if ip_addresses:
            assert isinstance(ip_addresses, list)
            assert all([isinstance(a, str) for a in ip_addresses])
            self.ip_addresses = ip_addresses
        if location:
            assert isinstance(location, str)
            self.location = location
        if period:
            assert isinstance(period, ioex.datetimeex.Period)
            self.period = period
        if state:
            assert isinstance(state, str)
            self.state = state


class HostingService(Service):

    yaml_tag = u'!hosting-service'

    def __init__(self,
                 operating_system=None,
                 **kwargs
                 ):
        super(HostingService, self).__init__(**kwargs)
        if operating_system:
            assert isinstance(operating_system, str)
            self.operating_system = operating_system


class CloudMining(Service):

    yaml_tag = u'!cloud-mining'

    def __init__(self,
                 hashrate=None,
                 **kwargs
                 ):
        super(CloudMining, self).__init__(**kwargs)
        if hashrate:
            assert isinstance(hashrate, ioex.calcex.Figure)
            self.hashrate = hashrate


class Transportation(Item):

    yaml_tag = u'!transportation'

    def __init__(self,
                 arrival_time=None,
                 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 arrival_time is not None:
            assert isinstance(arrival_time, datetime.datetime) \
                    or isinstance(arrival_time, ioex.datetimeex.Period)
            self.arrival_time = arrival_time
        if departure_point is not None:
            assert type(departure_point) is str
            self.departure_point = departure_point
        if destination_point is not None:
            assert type(destination_point) is str
            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 bytes
            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 str
            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,
                 tracking_number=None,
                 **kwargs
                 ):
        super(Shipping, self).__init__(**kwargs)
        if tracking_number:
            assert isinstance(tracking_number, str)
            self.tracking_number = tracking_number


class TaxiRide(Transportation):

    yaml_tag = u'!taxi-ride'

    def __init__(self,
                 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 str
        self.driver = driver
        assert departure_time is None or type(
            departure_time) is datetime.datetime
        self.departure_time = departure_time


class Person(_Object, _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 str
        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 str
        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()
        ])