mirror of
https://github.com/espressif/esp-idf.git
synced 2024-09-21 06:56:11 -04:00
79 lines
3.5 KiB
Python
79 lines
3.5 KiB
Python
# SPDX-FileCopyrightText: 2021-2022 Espressif Systems (Shanghai) CO LTD
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
from typing import List, Optional
|
|
|
|
from .cluster import Cluster
|
|
from .exceptions import NoFreeClusterException
|
|
from .fatfs_state import BootSectorState
|
|
|
|
|
|
class FAT:
|
|
"""
|
|
The FAT represents the FAT region in file system. It is responsible for storing clusters
|
|
and chaining them in case we need to extend file or directory to more clusters.
|
|
"""
|
|
|
|
def allocate_root_dir(self) -> None:
|
|
"""
|
|
The root directory is implicitly created with the FatFS,
|
|
its block is on the index 1 (second index) and is allocated implicitly.
|
|
"""
|
|
self.clusters[Cluster.ROOT_BLOCK_ID].allocate_cluster()
|
|
|
|
def __init__(self, boot_sector_state: BootSectorState, init_: bool) -> None:
|
|
self._first_free_cluster_id = 0
|
|
self.boot_sector_state = boot_sector_state
|
|
self.clusters: List[Cluster] = [Cluster(cluster_id=i,
|
|
boot_sector_state=self.boot_sector_state,
|
|
init_=init_) for i in range(self.boot_sector_state.clusters)]
|
|
if init_:
|
|
self.allocate_root_dir()
|
|
|
|
def get_cluster_value(self, cluster_id_: int) -> int:
|
|
fat_cluster_value_: int = self.clusters[cluster_id_].get_from_fat()
|
|
return fat_cluster_value_
|
|
|
|
def is_cluster_last(self, cluster_id_: int) -> bool:
|
|
value_ = self.get_cluster_value(cluster_id_)
|
|
is_cluster_last_: bool = value_ == (1 << self.boot_sector_state.fatfs_type) - 1
|
|
return is_cluster_last_
|
|
|
|
def get_chained_content(self, cluster_id_: int, size: Optional[int] = None) -> bytearray:
|
|
"""
|
|
The purpose of the method is retrieving the content from chain of clusters when the FAT FS partition
|
|
is analyzed. The file entry provides the reference to the first cluster, this method
|
|
traverses linked list of clusters and append partial results to the content.
|
|
"""
|
|
binary_image: bytearray = self.boot_sector_state.binary_image
|
|
|
|
data_address_ = Cluster.compute_cluster_data_address(self.boot_sector_state, cluster_id_)
|
|
content_ = binary_image[data_address_: data_address_ + self.boot_sector_state.sector_size]
|
|
|
|
while not self.is_cluster_last(cluster_id_):
|
|
cluster_id_ = self.get_cluster_value(cluster_id_)
|
|
data_address_ = Cluster.compute_cluster_data_address(self.boot_sector_state, cluster_id_)
|
|
content_ += binary_image[data_address_: data_address_ + self.boot_sector_state.sector_size]
|
|
# the size is None if the object is directory
|
|
if size is None:
|
|
return content_
|
|
return content_[:size]
|
|
|
|
def find_free_cluster(self) -> Cluster:
|
|
# finds first empty cluster and allocates it
|
|
for cluster_id, cluster in enumerate(self.clusters[self._first_free_cluster_id:],
|
|
start=self._first_free_cluster_id):
|
|
if cluster.is_empty:
|
|
cluster.allocate_cluster()
|
|
self._first_free_cluster_id = cluster_id
|
|
return cluster
|
|
raise NoFreeClusterException('No free cluster available!')
|
|
|
|
def allocate_chain(self, first_cluster: Cluster, size: int) -> None:
|
|
current = first_cluster
|
|
for _ in range(size - 1):
|
|
free_cluster = self.find_free_cluster()
|
|
current.next_cluster = free_cluster
|
|
current.set_in_fat(free_cluster.id)
|
|
current = free_cluster
|