adb_python/pyadb/internal/directory.py

import posixpath


class File:
    """File abstraction. Only real method is to get fullpath"""

    def __init__(self, name, parent_dir):
        self.name = name
        self.parent_dir = parent_dir
        self.fullpath = None

    def get_fullpath(self, real_time=False):
        if self.fullpath is None or real_time:
            if self.parent is None:
                self.fullpath = self.name
            else:
                self.fullpath = posixpath.join(
                    self.parent_dir.get_fullpath(real_time), self.name
                )
        return self.fullpath


class Directory(File):
    """Directory abstraction with optional listener"""

    def __init__(self, name, is_listening=False):
        super().__init__(name, None)
        self.children = {}
        self.is_listening = is_listening
        if is_listening:
            self.modified = False

    def __index__(self, key):
        try:
            return self.children[key]
        except KeyError:
            raise FileNotFoundError

    def mark_modified(self):
        if self.is_listening:
            self.modified = True
        if self.parent_dir:
            self.parent_dir.mark_modified()

    def traverse(self, path, create_intermediate=False, notify=True):
        """Traverse filesystem returning either file or directory"""
        if type(path) == str:
            path = path.split("/")

        if len(path) == 1:
            return self[path[0]]

        _next, *rest = path
        if _next == "..":
            if self.parent_dir:
                selected = self.parent_dir
            else:
                selected = self
        elif _next == ".":
            selected = self
        else:
            try:
                selected = self[_next]
            except FileNotFoundError as e:
                if create_intermediate:
                    new = Directory(_next, self.is_listening)
                    self.add_child(new, notify)
                    selected = new
                else:
                    raise e
        return selected.traverse(
            rest, create_intermediate=create_intermediate, notify=notify
        )

    def add_child(self, child, notify=True):
        if notify and self.is_listening:
            self.mark_modified()
        self.children[child.name] = child
        child.parent_dir = self

    def remove_child(self, child_name, notify=True):
        if notify and self.is_listening:
            self.mark_modified()
        del self[child_name]

    def iter_files(self):
        for child in self.children.values():
            if isinstance(child, Directory):
                yield from child.iter_files()
            else:
                yield child

    @classmethod
    def from_dir_listing(cls, listing, rootname="", sep="/", is_listening=False):
        """Create a Directory object with files from file listing"""
        root = Directory(rootname, is_listening=is_listening)
        for file in listing:
            parts, file = file.split(sep)
            parent = root.traverse(parts, create_intermediate=True, notify=False)
            file = File(file)
            parent.add_child(file, notify=False)
        return root


def get_unmodified(dirtree: Directory):
    if dirtree.modified:
        for child in dirtree.children.values():
            if isinstance(child, Directory):
                yield from get_unmodified(child)
            else:
                yield child.get_fullpath()
    else:
        yield dirtree.get_fullpath() + dirtree.sep