path: root/git/objects/util.py

# util.py
# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors
#
# This module is part of GitPython and is released under
# the BSD License: http://www.opensource.org/licenses/bsd-license.php
"""Module for general utility functions"""
from git.util import (
                        IterableList, 
                        Actor
                    )

import re
from collections import deque as Deque

from string import digits
import time
import os

__all__ = ('get_object_type_by_name', 'parse_date', 'parse_actor_and_date', 
            'ProcessStreamAdapter', 'Traversable', 'altz_to_utctz_str', 'utctz_to_altz', 
            'verify_utctz', 'Actor')

#{ Functions

def mode_str_to_int(modestr):
    """
    :param modestr: string like 755 or 644 or 100644 - only the last 6 chars will be used
    :return:
        String identifying a mode compatible to the mode methods ids of the 
        stat module regarding the rwx permissions for user, group and other, 
        special flags and file system flags, i.e. whether it is a symlink
        for example."""
    mode = 0
    for iteration, char in enumerate(reversed(modestr[-6:])):
        mode += int(char) << iteration*3
    # END for each char
    return mode

def get_object_type_by_name(object_type_name):
    """
    :return: type suitable to handle the given object type name.
        Use the type to create new instances.
        
    :param object_type_name: Member of TYPES
        
    :raise ValueError: In case object_type_name is unknown"""
    if object_type_name == "commit":
        import commit
        return commit.Commit
    elif object_type_name == "tag":
        import tag
        return tag.TagObject
    elif object_type_name == "blob":
        import blob
        return blob.Blob
    elif object_type_name == "tree":
        import tree
        return tree.Tree
    else:
        raise ValueError("Cannot handle unknown object type: %s" % object_type_name)
        
def utctz_to_altz(utctz):
    """we convert utctz to the timezone in seconds, it is the format time.altzone
    returns. Git stores it as UTC timezone which has the opposite sign as well, 
    which explains the -1 * ( that was made explicit here )
    :param utctz: git utc timezone string, i.e. +0200"""
    return -1 * int(float(utctz)/100*3600)
    
def altz_to_utctz_str(altz):
    """As above, but inverses the operation, returning a string that can be used
    in commit objects"""
    utci = -1 * int((altz / 3600)*100)
    utcs = str(abs(utci))
    utcs = "0"*(4-len(utcs)) + utcs
    prefix = (utci < 0 and '-') or '+'
    return prefix + utcs
    

def verify_utctz(offset):
    """:raise ValueError: if offset is incorrect
    :return: offset"""
    fmt_exc = ValueError("Invalid timezone offset format: %s" % offset)
    if len(offset) != 5:
        raise fmt_exc
    if offset[0] not in "+-":
        raise fmt_exc
    if  offset[1] not in digits or \
        offset[2] not in digits or \
        offset[3] not in digits or \
        offset[4] not in digits:
        raise fmt_exc
    # END for each char
    return offset

def parse_date(string_date):
    """
    Parse the given date as one of the following
    
        * Git internal format: timestamp offset
        * RFC 2822: Thu, 07 Apr 2005 22:13:13 +0200. 
        * ISO 8601 2005-04-07T22:13:13
            The T can be a space as well
         
    :return: Tuple(int(timestamp), int(offset)), both in seconds since epoch
    :raise ValueError: If the format could not be understood
    :note: Date can also be YYYY.MM.DD, MM/DD/YYYY and DD.MM.YYYY"""
    # git time
    try:
        if string_date.count(' ') == 1 and string_date.rfind(':') == -1:
            timestamp, offset = string_date.split()
            timestamp = int(timestamp)
            return timestamp, utctz_to_altz(verify_utctz(offset))
        else:
            offset = "+0000"                    # local time by default
            if string_date[-5] in '-+':
                offset = verify_utctz(string_date[-5:])
                string_date = string_date[:-6]  # skip space as well
            # END split timezone info
            
            # now figure out the date and time portion - split time
            date_formats = list()
            splitter = -1
            if ',' in string_date:
                date_formats.append("%a, %d %b %Y")
                splitter = string_date.rfind(' ')
            else:
                # iso plus additional
                date_formats.append("%Y-%m-%d")
                date_formats.append("%Y.%m.%d")
                date_formats.append("%m/%d/%Y")
                date_formats.append("%d.%m.%Y")
                
                splitter = string_date.rfind('T')
                if splitter == -1:
                    splitter = string_date.rfind(' ')
                # END handle 'T' and ' '
            # END handle rfc or iso 
            
            assert splitter > -1
            
            # split date and time
            time_part = string_date[splitter+1:]    # skip space
            date_part = string_date[:splitter]
            
            # parse time
            tstruct = time.strptime(time_part, "%H:%M:%S")
            
            for fmt in date_formats:
                try:
                    dtstruct = time.strptime(date_part, fmt)
                    fstruct = time.struct_time((dtstruct.tm_year, dtstruct.tm_mon, dtstruct.tm_mday, 
                                                tstruct.tm_hour, tstruct.tm_min, tstruct.tm_sec,
                                                dtstruct.tm_wday, dtstruct.tm_yday, tstruct.tm_isdst))
                    return int(time.mktime(fstruct)), utctz_to_altz(offset)
                except ValueError:
                    continue
                # END exception handling
            # END for each fmt
            
            # still here ? fail
            raise ValueError("no format matched")
        # END handle format
    except Exception:
        raise ValueError("Unsupported date format: %s" % string_date)  
    # END handle exceptions

    
# precompiled regex
_re_actor_epoch = re.compile(r'^.+? (.*) (\d+) ([+-]\d+).*$')

def parse_actor_and_date(line):
    """Parse out the actor (author or committer) info from a line like::
    
        author Tom Preston-Werner <tom@mojombo.com> 1191999972 -0700
    
    :return: [Actor, int_seconds_since_epoch, int_timezone_offset]"""
    m = _re_actor_epoch.search(line)
    actor, epoch, offset = m.groups()
    return (Actor._from_string(actor), int(epoch), utctz_to_altz(offset))
    

#} END functions


#{ Classes 
    
class ProcessStreamAdapter(object):
    """Class wireing all calls to the contained Process instance.
    
    Use this type to hide the underlying process to provide access only to a specified 
    stream. The process is usually wrapped into an AutoInterrupt class to kill 
    it if the instance goes out of scope."""
    __slots__ = ("_proc", "_stream")
    def __init__(self, process, stream_name):
        self._proc = process
        self._stream = getattr(process, stream_name)
    
    def __getattr__(self, attr):
        return getattr(self._stream, attr)
        
        
class Traversable(object):
    """Simple interface to perforam depth-first or breadth-first traversals 
    into one direction.
    Subclasses only need to implement one function.
    Instances of the Subclass must be hashable"""
    __slots__ = tuple()
    
    @classmethod
    def _get_intermediate_items(cls, item):
        """
        Returns:
            List of items connected to the given item.
            Must be implemented in subclass
        """
        raise NotImplementedError("To be implemented in subclass")
            
    def list_traverse(self, *args, **kwargs):
        """
        :return: IterableList with the results of the traversal as produced by
            traverse()"""
        out = IterableList(self._id_attribute_)
        out.extend(self.traverse(*args, **kwargs))
        return out
    
    def traverse( self, predicate = lambda i,d: True,
                           prune = lambda i,d: False, depth = -1, branch_first=True,
                           visit_once = True, ignore_self=1, as_edge = False ):
        """:return: iterator yieling of items found when traversing self
            
        :param predicate: f(i,d) returns False if item i at depth d should not be included in the result
            
        :param prune: 
            f(i,d) return True if the search should stop at item i at depth d.
            Item i will not be returned.
            
        :param depth:
            define at which level the iteration should not go deeper
            if -1, there is no limit
            if 0, you would effectively only get self, the root of the iteration
            i.e. if 1, you would only get the first level of predessessors/successors
            
        :param branch_first:
            if True, items will be returned branch first, otherwise depth first
            
        :param visit_once:
            if True, items will only be returned once, although they might be encountered
            several times. Loops are prevented that way.
        
        :param ignore_self:
            if True, self will be ignored and automatically pruned from
            the result. Otherwise it will be the first item to be returned.
            If as_edge is True, the source of the first edge is None
            
        :param as_edge:
            if True, return a pair of items, first being the source, second the 
            destinatination, i.e. tuple(src, dest) with the edge spanning from 
            source to destination"""
        visited = set()
        stack = Deque()
        stack.append( ( 0 ,self, None ) )       # self is always depth level 0
    
        def addToStack( stack, item, branch_first, depth ):
            lst = self._get_intermediate_items( item )
            if not lst:
                return
            if branch_first:
                stack.extendleft( ( depth , i, item ) for i in lst )
            else:
                reviter = ( ( depth , lst[i], item ) for i in range( len( lst )-1,-1,-1) )
                stack.extend( reviter )
        # END addToStack local method
    
        while stack:
            d, item, src = stack.pop()          # depth of item, item, item_source
            
            if visit_once and item in visited:
                continue
                
            if visit_once:
                visited.add(item)
            
            rval = ( as_edge and (src, item) ) or item
            if prune( rval, d ):
                continue
    
            skipStartItem = ignore_self and ( item is self )
            if not skipStartItem and predicate( rval, d ):
                yield rval
    
            # only continue to next level if this is appropriate !
            nd = d + 1
            if depth > -1 and nd > depth:
                continue
    
            addToStack( stack, item, branch_first, nd )
        # END for each item on work stack
        

class Serializable(object):
    """Defines methods to serialize and deserialize objects from and into a data stream"""
    __slots__ = tuple()
    
    def _serialize(self, stream):
        """Serialize the data of this object into the given data stream
        :note: a serialized object would ``_deserialize`` into the same objet
        :param stream: a file-like object
        :return: self"""
        raise NotImplementedError("To be implemented in subclass")
        
    def _deserialize(self, stream):
        """Deserialize all information regarding this object from the stream
        :param stream: a file-like object
        :return: self"""
        raise NotImplementedError("To be implemented in subclass")