Skip to content

Commit

Permalink
Update array_stream.py (#42)
Browse files Browse the repository at this point in the history
  • Loading branch information
Beakerboy committed Jan 6, 2024
1 parent 3aa0bc5 commit 044cd6b
Show file tree
Hide file tree
Showing 8 changed files with 41 additions and 13 deletions.
27 changes: 23 additions & 4 deletions src/ms_cfb/Models/DataStreams/array_stream.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,11 @@
from ms_cfb.Models.DataStreams.stream_base import StreamBase
from typing import Any, Iterator, TypeVar, overload
from typing import Any, MutableSequence, TypeVar, overload


T = TypeVar('T', bound='ArrayStream')


class ArrayStream(StreamBase):
class ArrayStream(StreamBase, MutableSequence[T]):
"""
An array stream is a stream in which renderable data is
saved in an array.
Expand All @@ -19,12 +19,31 @@ def __init__(self: T, child_sector_size: int) -> None:
self._child_sector_size = child_sector_size

# Dunder Methods
def __iter__(self: T) -> Iterator:
return iter(self._data)

@overload
def __getitem__(self: T, idx: int) -> T: ...

@overload
def __getitem__(self: T, s: slice) -> MutableSequence[T]: ...

def __getitem__(self: T, item):
if isinstance(item, slice):
raise Exception("Subclass disallows slicing")

return self._data[item]

def __len__(self: T) -> int:
return len(self._data)

def __delitem__(self: T, key: Any) -> None:
del self._data[key]

def __setitem__(self: T, key: Any, value: Any) -> None:
self._data[key] = value

def insert(self: T, key: Any, value: Any) -> None:
self._data.insert(key, value)

# Public Methods
def to_file(self: T, path: str) -> None:
f = open(path, "wb")
Expand Down
2 changes: 1 addition & 1 deletion src/ms_cfb/Models/DataStreams/stream_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ def set_additional_sectors(self: T, sectors: list) -> None:
def get_sectors(self: T) -> list:
return self._sectors

def append(self: T, data: bytes) -> None:
def append(self: T, data: Any) -> None:
"""
Extend the data in this stream.
Request additional chain storage if needed
Expand Down
4 changes: 4 additions & 0 deletions src/ms_cfb/Models/Directories/directory.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,10 @@ def __init__(self: T) -> None:
# This object's index in the flattened representation of the tree.
self._flattened_index = 0

self.prev_index: int
self.next_index: int
self.sub_index: int

def __str__(self: T) -> str:
return (self.get_name() +
"\n\tCreated: " + str(self._created) +
Expand Down
3 changes: 2 additions & 1 deletion src/ms_cfb/Models/Directories/directory_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
class DirectoryFactory:

@classmethod
def from_binary(cls: Type[T], data: bytes) -> 'Directory':
def from_binary(cls: Type[T], data: bytes) -> Directory:
format = "<64shbb3I16sIQQIII"
(name, name_size, type, color,
previous_directory_id, next_directory_id,
Expand All @@ -30,6 +30,7 @@ def from_binary(cls: Type[T], data: bytes) -> 'Directory':
created = Filetime.from_msfiletime(created)

guid = uuid.UUID(bytes_le=class_id)
obj: Directory
if type == 1:
obj = StorageDirectory(name)
if file_size != 0:
Expand Down
6 changes: 3 additions & 3 deletions src/ms_cfb/Models/Filesystems/fat_filesystem.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,8 @@ def to_file(self: T, path: str) -> None:
fill = self._sector_size - length % self._sector_size
c.write(b'\xff' * fill)
c.close()
c = open("fat_chain.bin", "rb")
f.write(c.read(self._sector_size))
chain_file = open("fat_chain.bin", "rb")
f.write(chain_file.read(self._sector_size))
f.close()
c.close()
chain_file.close()
os.remove("fat_chain.bin")
4 changes: 2 additions & 2 deletions src/ms_cfb/Models/Filesystems/filesystem_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import string
from ms_cfb.Models.DataStreams.stream_base import StreamBase
from random import choice
from typing import Sequence, TypeVar
from typing import MutableSequence, TypeVar


T = TypeVar('T', bound='FilesystemBase')
Expand All @@ -23,7 +23,7 @@ def __init__(self: T, size: int) -> None:

# Each stream begins at the start of a sector and is padded to fill
# the end of a sector.
self._streams: Sequence = []
self._streams: MutableSequence = []

def __len__(self: T) -> int:
return self._next_free_sector
Expand Down
3 changes: 2 additions & 1 deletion src/ms_cfb/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,8 @@ def update_attributes(dir: 'Directory', conf: dict) -> None:
if "clsid" in conf:
dir.set_clsid(uuid.UUID(conf["clsid"]))
if "flags" in conf:
dir.set_flags(conf["flags"])
...
# dir.set_flags(conf["flags"])


def create_storage(direntry: os.DirEntry,
Expand Down
5 changes: 4 additions & 1 deletion src/ms_cfb/ole_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -440,9 +440,12 @@ def create_from_file(cls: Type[T], path: str) -> T:
directory.right = right
right.parent = directory
if directory.sub_index != 0xFFFFFFFF:
assert isinstance(directory, StorageDirectory)
child = flat_directories[directory.sub_index]
directory.directories.root = child
obj.root_directory = flat_directories[0]
dir_0 = flat_directories[0]
assert isinstance(dir_0, RootDirectory)
obj.root_directory = dir_0

# extract minifat chain
return obj

0 comments on commit 044cd6b

Please sign in to comment.