| # -*- coding: utf-8 -*- | |
| """ | |
| Device Tree Blob Parser | |
| Copyright 2014 Neil 'superna' Armstrong <superna9999@gmail.com> | |
| Licensed under the Apache License, Version 2.0 (the "License"); | |
| you may not use this file except in compliance with the License. | |
| You may obtain a copy of the License at | |
| http://www.apache.org/licenses/LICENSE-2.0 | |
| Unless required by applicable law or agreed to in writing, software | |
| distributed under the License is distributed on an "AS IS" BASIS, | |
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| See the License for the specific language governing permissions and | |
| limitations under the License. | |
| @author: Neil 'superna' Armstrong <superna9999@gmail.com> | |
| """ | |
| from __future__ import print_function | |
| from __future__ import absolute_import | |
| import string | |
| import os | |
| import json | |
| from copy import deepcopy, copy | |
| from struct import Struct, unpack, pack | |
| FDT_MAGIC = 0xD00DFEED | |
| FDT_BEGIN_NODE = 0x1 | |
| FDT_END_NODE = 0x2 | |
| FDT_PROP = 0x3 | |
| FDT_NOP = 0x4 | |
| FDT_END = 0x9 | |
| INDENT = " " * 4 | |
| FDT_MAX_VERSION = 17 | |
| class FdtProperty(object): | |
| """ Represents an empty property""" | |
| @staticmethod | |
| def __validate_dt_name(name): | |
| """Checks the name validity""" | |
| return not any([True for char in name if char not in string.printable]) | |
| def __init__(self, name): | |
| """Init with name""" | |
| self.name = name | |
| if not FdtProperty.__validate_dt_name(self.name): | |
| raise Exception("Invalid name '%s'" % self.name) | |
| def get_name(self): | |
| """Get property name""" | |
| return self.name | |
| def __str__(self): | |
| """String representation""" | |
| return "Property(%s)" % self.name | |
| def dts_represent(self, depth=0): | |
| """Get dts string representation""" | |
| return INDENT * depth + self.name + ";" | |
| def dtb_represent(self, string_store, pos=0, version=17): | |
| """Get blob representation""" | |
| # print "%x:%s" % (pos, self) | |
| strpos = string_store.find(self.name + "\0") | |
| if strpos < 0: | |
| strpos = len(string_store) | |
| string_store += self.name + "\0" | |
| pos += 12 | |
| return (pack(">III", FDT_PROP, 0, strpos), string_store, pos) | |
| def json_represent(self, depth=0): | |
| """Ouput JSON""" | |
| return "%s: null" % json.dumps(self.name) | |
| def to_raw(self): | |
| """Return RAW value representation""" | |
| return "" | |
| def __getitem__(self, value): | |
| """Returns No Items""" | |
| return None | |
| def __ne__(self, node): | |
| """Check property inequality | |
| """ | |
| return not self.__eq__(node) | |
| def __eq__(self, node): | |
| """Check node equality | |
| check properties are the same (same values) | |
| """ | |
| if not isinstance(node, FdtProperty): | |
| raise Exception("Invalid object type") | |
| if self.name != node.get_name(): | |
| return False | |
| return True | |
| @staticmethod | |
| def __check_prop_strings(value): | |
| """Check property string validity | |
| Python version of util_is_printable_string from dtc | |
| """ | |
| pos = 0 | |
| posi = 0 | |
| end = len(value) | |
| if not len(value): | |
| return None | |
| # Needed for python 3 support: If a bytes object is passed, | |
| # decode it with the ascii codec. If the decoding fails, assume | |
| # it was not a string object. | |
| try: | |
| value = value.decode("ascii") | |
| except ValueError: | |
| return None | |
| # Test both against string 0 and int 0 because of | |
| # python2/3 compatibility | |
| if value[-1] != "\0": | |
| return None | |
| while pos < end: | |
| posi = pos | |
| while ( | |
| pos < end | |
| and value[pos] != "\0" | |
| and value[pos] in string.printable | |
| and value[pos] not in ("\r", "\n") | |
| ): | |
| pos += 1 | |
| if value[pos] != "\0" or pos == posi: | |
| return None | |
| pos += 1 | |
| return True | |
| @staticmethod | |
| def new_raw_property(name, raw_value): | |
| """Instantiate property with raw value type""" | |
| if FdtProperty.__check_prop_strings(raw_value): | |
| return FdtPropertyStrings.init_raw(name, raw_value) | |
| elif len(raw_value) and len(raw_value) % 4 == 0: | |
| return FdtPropertyWords.init_raw(name, raw_value) | |
| elif len(raw_value) and len(raw_value): | |
| return FdtPropertyBytes.init_raw(name, raw_value) | |
| else: | |
| return FdtProperty(name) | |
| class FdtPropertyStrings(FdtProperty): | |
| """Property with strings as value""" | |
| @classmethod | |
| def __extract_prop_strings(cls, value): | |
| """Extract strings from raw_value""" | |
| return [st for st in value.decode("ascii").split("\0") if len(st)] | |
| def __init__(self, name, strings): | |
| """Init with strings""" | |
| FdtProperty.__init__(self, name) | |
| if not strings: | |
| raise Exception("Invalid strings") | |
| for stri in strings: | |
| if len(stri) == 0: | |
| raise Exception("Invalid strings") | |
| if any( | |
| [ | |
| True | |
| for char in stri | |
| if char not in string.printable or char in ("\r", "\n") | |
| ] | |
| ): | |
| raise Exception("Invalid chars in strings") | |
| self.strings = strings | |
| @classmethod | |
| def init_raw(cls, name, raw_value): | |
| """Init from raw""" | |
| return cls(name, cls.__extract_prop_strings(raw_value)) | |
| def dts_represent(self, depth=0): | |
| """Get dts string representation""" | |
| return ( | |
| INDENT * depth | |
| + self.name | |
| + ' = "' | |
| + '", "'.join(self.strings) | |
| + '";' | |
| ) | |
| def dtb_represent(self, string_store, pos=0, version=17): | |
| """Get blob representation""" | |
| # print "%x:%s" % (pos, self) | |
| blob = pack("") | |
| for chars in self.strings: | |
| blob += chars.encode("ascii") + pack("b", 0) | |
| blob_len = len(blob) | |
| if version < 16 and (pos + 12) % 8 != 0: | |
| blob = pack("b", 0) * (8 - ((pos + 12) % 8)) + blob | |
| if blob_len % 4: | |
| blob += pack("b", 0) * (4 - (blob_len % 4)) | |
| strpos = string_store.find(self.name + "\0") | |
| if strpos < 0: | |
| strpos = len(string_store) | |
| string_store += self.name + "\0" | |
| blob = pack(">III", FDT_PROP, blob_len, strpos) + blob | |
| pos += len(blob) | |
| return (blob, string_store, pos) | |
| def json_represent(self, depth=0): | |
| """Ouput JSON""" | |
| result = '%s: ["strings", ' % json.dumps(self.name) | |
| result += ", ".join([json.dumps(stri) for stri in self.strings]) | |
| result += "]" | |
| return result | |
| def to_raw(self): | |
| """Return RAW value representation""" | |
| return "".join([chars + "\0" for chars in self.strings]) | |
| def __str__(self): | |
| """String representation""" | |
| return "Property(%s,Strings:%s)" % (self.name, self.strings) | |
| def __getitem__(self, index): | |
| """Get strings, returns a string""" | |
| return self.strings[index] | |
| def __len__(self): | |
| """Get strings count""" | |
| return len(self.strings) | |
| def __eq__(self, node): | |
| """Check node equality | |
| check properties are the same (same values) | |
| """ | |
| if not FdtProperty.__eq__(self, node): | |
| return False | |
| if self.__len__() != len(node): | |
| return False | |
| for index in range(self.__len__()): | |
| if self.strings[index] != node[index]: | |
| return False | |
| return True | |
| class FdtPropertyWords(FdtProperty): | |
| """Property with words as value""" | |
| def __init__(self, name, words): | |
| """Init with words""" | |
| FdtProperty.__init__(self, name) | |
| for word in words: | |
| if not 0 <= word <= 4294967295: | |
| raise Exception( | |
| ( | |
| "Invalid word value %d, requires " | |
| + "0 <= number <= 4294967295" | |
| ) | |
| % word | |
| ) | |
| if not len(words): | |
| raise Exception("Invalid Words") | |
| self.words = words | |
| @classmethod | |
| def init_raw(cls, name, raw_value): | |
| """Init from raw""" | |
| if len(raw_value) % 4 == 0: | |
| words = [ | |
| unpack(">I", raw_value[i : i + 4])[0] | |
| for i in range(0, len(raw_value), 4) | |
| ] | |
| return cls(name, words) | |
| else: | |
| raise Exception("Invalid raw Words") | |
| def dts_represent(self, depth=0): | |
| """Get dts string representation""" | |
| return ( | |
| INDENT * depth | |
| + self.name | |
| + " = <" | |
| + " ".join(["0x%08x" % word for word in self.words]) | |
| + ">;" | |
| ) | |
| def dtb_represent(self, string_store, pos=0, version=17): | |
| """Get blob representation""" | |
| # # print "%x:%s" % (pos, self) | |
| strpos = string_store.find(self.name + "\0") | |
| if strpos < 0: | |
| strpos = len(string_store) | |
| string_store += self.name + "\0" | |
| blob = pack(">III", FDT_PROP, len(self.words) * 4, strpos) + pack( | |
| "" | |
| ).join([pack(">I", word) for word in self.words]) | |
| pos += len(blob) | |
| return (blob, string_store, pos) | |
| def json_represent(self, depth=0): | |
| """Ouput JSON""" | |
| result = '%s: ["words", "' % json.dumps(self.name) | |
| result += '", "'.join(["0x%08x" % word for word in self.words]) | |
| result += '"]' | |
| return result | |
| def to_raw(self): | |
| """Return RAW value representation""" | |
| return "".join([pack(">I", word) for word in self.words]) | |
| def __str__(self): | |
| """String representation""" | |
| return "Property(%s,Words:%s)" % (self.name, self.words) | |
| def __getitem__(self, index): | |
| """Get words, returns a word integer""" | |
| return self.words[index] | |
| def __len__(self): | |
| """Get words count""" | |
| return len(self.words) | |
| def __eq__(self, node): | |
| """Check node equality | |
| check properties are the same (same values) | |
| """ | |
| if not FdtProperty.__eq__(self, node): | |
| return False | |
| if self.__len__() != len(node): | |
| return False | |
| for index in range(self.__len__()): | |
| if self.words[index] != node[index]: | |
| return False | |
| return True | |
| class FdtPropertyBytes(FdtProperty): | |
| """Property with signed bytes as value""" | |
| def __init__(self, name, bytez): | |
| """Init with bytes""" | |
| FdtProperty.__init__(self, name) | |
| for byte in bytez: | |
| if not -128 <= byte <= 127: | |
| raise Exception( | |
| ( | |
| "Invalid value for byte %d, " | |
| + "requires -128 <= number <= 127" | |
| ) | |
| % byte | |
| ) | |
| if not bytez: | |
| raise Exception("Invalid Bytes") | |
| self.bytes = bytez | |
| @classmethod | |
| def init_raw(cls, name, raw_value): | |
| """Init from raw""" | |
| return cls(name, unpack("b" * len(raw_value), raw_value)) | |
| def dts_represent(self, depth=0): | |
| """Get dts string representation""" | |
| return ( | |
| INDENT * depth | |
| + self.name | |
| + " = [" | |
| + " ".join( | |
| ["%02x" % (byte & int("ffffffff", 16)) for byte in self.bytes] | |
| ) | |
| + "];" | |
| ) | |
| def dtb_represent(self, string_store, pos=0, version=17): | |
| """Get blob representation""" | |
| # print "%x:%s" % (pos, self) | |
| strpos = string_store.find(self.name + "\0") | |
| if strpos < 0: | |
| strpos = len(string_store) | |
| string_store += self.name + "\0" | |
| blob = pack(">III", FDT_PROP, len(self.bytes), strpos) | |
| blob += pack("").join([pack(">b", byte) for byte in self.bytes]) | |
| if len(blob) % 4: | |
| blob += pack("b", 0) * (4 - (len(blob) % 4)) | |
| pos += len(blob) | |
| return (blob, string_store, pos) | |
| def json_represent(self, depth=0): | |
| """Ouput JSON""" | |
| result = '%s: ["bytes", "' % json.dumps(self.name) | |
| result += '", "'.join(["%02x" % byte for byte in self.bytes]) | |
| result += '"]' | |
| return result | |
| def to_raw(self): | |
| """Return RAW value representation""" | |
| return "".join([pack(">b", byte) for byte in self.bytes]) | |
| def __str__(self): | |
| """String representation""" | |
| return "Property(%s,Bytes:%s)" % (self.name, self.bytes) | |
| def __getitem__(self, index): | |
| """Get bytes, returns a byte""" | |
| return self.bytes[index] | |
| def __len__(self): | |
| """Get strings count""" | |
| return len(self.bytes) | |
| def __eq__(self, node): | |
| """Check node equality | |
| check properties are the same (same values) | |
| """ | |
| if not FdtProperty.__eq__(self, node): | |
| return False | |
| if self.__len__() != len(node): | |
| return False | |
| for index in range(self.__len__()): | |
| if self.bytes[index] != node[index]: | |
| return False | |
| return True | |
| class FdtNop(object): # pylint: disable-msg=R0903 | |
| """Nop child representation""" | |
| def __init__(self): | |
| """Init with nothing""" | |
| def get_name(self): # pylint: disable-msg=R0201 | |
| """Return name""" | |
| return None | |
| def __str__(self): | |
| """String representation""" | |
| return "" | |
| def dts_represent(self, depth=0): # pylint: disable-msg=R0201 | |
| """Get dts string representation""" | |
| return INDENT * depth + "// [NOP]" | |
| def dtb_represent(self, string_store, pos=0, version=17): | |
| """Get blob representation""" | |
| # print "%x:%s" % (pos, self) | |
| pos += 4 | |
| return (pack(">I", FDT_NOP), string_store, pos) | |
| class FdtNode(object): | |
| """Node representation""" | |
| @staticmethod | |
| def __validate_dt_name(name): | |
| """Checks the name validity""" | |
| return not any([True for char in name if char not in string.printable]) | |
| def __init__(self, name): | |
| """Init node with name""" | |
| self.name = name | |
| self.subdata = [] | |
| self.parent = None | |
| if not FdtNode.__validate_dt_name(self.name): | |
| raise Exception("Invalid name '%s'" % self.name) | |
| def get_name(self): | |
| """Get property name""" | |
| return self.name | |
| def __check_name_duplicate(self, name): | |
| """Checks if name is not in a subnode""" | |
| for data in self.subdata: | |
| if not isinstance(data, FdtNop) and data.get_name() == name: | |
| return True | |
| return False | |
| def add_subnode(self, node): | |
| """Add child, deprecated use append()""" | |
| self.append(node) | |
| def add_raw_attribute(self, name, raw_value): | |
| """Construct a raw attribute and add to child""" | |
| self.append(FdtProperty.new_raw_property(name, raw_value)) | |
| def set_parent_node(self, node): | |
| """Set parent node, None and FdtNode accepted""" | |
| if node is not None and not isinstance(node, FdtNode): | |
| raise Exception("Invalid object type") | |
| self.parent = node | |
| def get_parent_node(self): | |
| """Get parent node""" | |
| return self.parent | |
| def __str__(self): | |
| """String representation""" | |
| return "Node(%s)" % self.name | |
| def dts_represent(self, depth=0): | |
| """Get dts string representation""" | |
| result = ("\n").join( | |
| [sub.dts_represent(depth + 1) for sub in self.subdata] | |
| ) | |
| if len(result) > 0: | |
| result += "\n" | |
| return ( | |
| INDENT * depth | |
| + self.name | |
| + " {\n" | |
| + result | |
| + INDENT * depth | |
| + "};" | |
| ) | |
| def dtb_represent(self, strings_store, pos=0, version=17): | |
| """Get blob representation | |
| Pass string storage as strings_store, pos for current node start | |
| and version as current dtb version | |
| """ | |
| # print "%x:%s" % (pos, self) | |
| strings = strings_store | |
| if self.get_name() == "/": | |
| blob = pack(">II", FDT_BEGIN_NODE, 0) | |
| else: | |
| blob = pack(">I", FDT_BEGIN_NODE) | |
| blob += self.get_name().encode("ascii") + pack("b", 0) | |
| if len(blob) % 4: | |
| blob += pack("b", 0) * (4 - (len(blob) % 4)) | |
| pos += len(blob) | |
| for sub in self.subdata: | |
| (data, strings, pos) = sub.dtb_represent(strings, pos, version) | |
| blob += data | |
| pos += 4 | |
| blob += pack(">I", FDT_END_NODE) | |
| return (blob, strings, pos) | |
| def json_represent(self, depth=0): | |
| """Get dts string representation""" | |
| result = (",\n" + INDENT * (depth + 1)).join( | |
| [ | |
| sub.json_represent(depth + 1) | |
| for sub in self.subdata | |
| if not isinstance(sub, FdtNop) | |
| ] | |
| ) | |
| if len(result) > 0: | |
| result = INDENT + result + "\n" + INDENT * depth | |
| if self.get_name() == "/": | |
| return "{\n" + INDENT * (depth) + result + "}" | |
| else: | |
| return ( | |
| json.dumps(self.name) | |
| + ": {\n" | |
| + INDENT * (depth) | |
| + result | |
| + "}" | |
| ) | |
| def __getitem__(self, index): | |
| """Get subnodes, returns either a Node, a Property or a Nop""" | |
| return self.subdata[index] | |
| def __setitem__(self, index, subnode): | |
| """Set node at index, replacing previous subnode, | |
| must not be a duplicate name | |
| """ | |
| if self.subdata[ | |
| index | |
| ].get_name() != subnode.get_name() and self.__check_name_duplicate( | |
| subnode.get_name() | |
| ): | |
| raise Exception("%s : %s subnode already exists" % (self, subnode)) | |
| if not isinstance(subnode, (FdtNode, FdtProperty, FdtNop)): | |
| raise Exception("Invalid object type") | |
| self.subdata[index] = subnode | |
| def __len__(self): | |
| """Get strings count""" | |
| return len(self.subdata) | |
| def __ne__(self, node): | |
| """Check node inequality | |
| i.e. is subnodes are the same, in either order | |
| and properties are the same (same values) | |
| The FdtNop is excluded from the check | |
| """ | |
| return not self.__eq__(node) | |
| def __eq__(self, node): | |
| """Check node equality | |
| i.e. is subnodes are the same, in either order | |
| and properties are the same (same values) | |
| The FdtNop is excluded from the check | |
| """ | |
| if not isinstance(node, FdtNode): | |
| raise Exception("Invalid object type") | |
| if self.name != node.get_name(): | |
| return False | |
| curnames = set( | |
| [ | |
| subnode.get_name() | |
| for subnode in self.subdata | |
| if not isinstance(subnode, FdtNop) | |
| ] | |
| ) | |
| cmpnames = set( | |
| [ | |
| subnode.get_name() | |
| for subnode in node | |
| if not isinstance(subnode, FdtNop) | |
| ] | |
| ) | |
| if curnames != cmpnames: | |
| return False | |
| for subnode in [ | |
| subnode | |
| for subnode in self.subdata | |
| if not isinstance(subnode, FdtNop) | |
| ]: | |
| index = node.index(subnode.get_name()) | |
| if subnode != node[index]: | |
| return False | |
| return True | |
| def append(self, subnode): | |
| """Append subnode, same as add_subnode""" | |
| if self.__check_name_duplicate(subnode.get_name()): | |
| raise Exception("%s : %s subnode already exists" % (self, subnode)) | |
| if not isinstance(subnode, (FdtNode, FdtProperty, FdtNop)): | |
| raise Exception("Invalid object type") | |
| self.subdata.append(subnode) | |
| def pop(self, index=-1): | |
| """Remove and returns subnode at index, default the last""" | |
| return self.subdata.pop(index) | |
| def insert(self, index, subnode): | |
| """Insert subnode before index, must not be a duplicate name""" | |
| if self.__check_name_duplicate(subnode.get_name()): | |
| raise Exception("%s : %s subnode already exists" % (self, subnode)) | |
| if not isinstance(subnode, (FdtNode, FdtProperty, FdtNop)): | |
| raise Exception("Invalid object type") | |
| self.subdata.insert(index, subnode) | |
| def _find(self, name): | |
| """Find name in subnodes""" | |
| for i in range(0, len(self.subdata)): | |
| if ( | |
| not isinstance(self.subdata[i], FdtNop) | |
| and name == self.subdata[i].get_name() | |
| ): | |
| return i | |
| return None | |
| def remove(self, name): | |
| """Remove subnode with the name | |
| Raises ValueError is not present | |
| """ | |
| index = self._find(name) | |
| if index is None: | |
| raise ValueError("Not present") | |
| return self.subdata.pop(index) | |
| def index(self, name): | |
| """Returns position of subnode with the name | |
| Raises ValueError is not present | |
| """ | |
| index = self._find(name) | |
| if index is None: | |
| raise ValueError("Not present") | |
| return index | |
| def merge(self, node): | |
| """Merge two nodes and subnodes | |
| Replace current properties with the given properties | |
| """ | |
| if not isinstance(node, FdtNode): | |
| raise Exception("Can only merge with a FdtNode") | |
| for subnode in [ | |
| obj for obj in node if isinstance(obj, (FdtNode, FdtProperty)) | |
| ]: | |
| index = self._find(subnode.get_name()) | |
| if index is None: | |
| dup = deepcopy(subnode) | |
| if isinstance(subnode, FdtNode): | |
| dup.set_parent_node(self) | |
| self.append(dup) | |
| elif isinstance(subnode, FdtNode): | |
| self.subdata[index].merge(subnode) | |
| else: | |
| self.subdata[index] = copy(subnode) | |
| def walk(self): | |
| """Walk into subnodes and yield paths and objects | |
| Returns set with (path string, node object) | |
| """ | |
| node = self | |
| start = 0 | |
| hist = [] | |
| curpath = [] | |
| while True: | |
| for index in range(start, len(node)): | |
| if isinstance(node[index], (FdtNode, FdtProperty)): | |
| yield ( | |
| "/" + "/".join(curpath + [node[index].get_name()]), | |
| node[index], | |
| ) | |
| if isinstance(node[index], FdtNode): | |
| if len(node[index]): | |
| hist.append((node, index + 1)) | |
| curpath.append(node[index].get_name()) | |
| node = node[index] | |
| start = 0 | |
| index = -1 | |
| break | |
| if index >= 0: | |
| if len(hist): | |
| (node, start) = hist.pop() | |
| curpath.pop() | |
| else: | |
| break | |
| class Fdt(object): | |
| """Flattened Device Tree representation""" | |
| def __init__(self, version=17, last_comp_version=16, boot_cpuid_phys=0): | |
| """Init FDT object with version and boot values""" | |
| self.header = { | |
| "magic": FDT_MAGIC, | |
| "totalsize": 0, | |
| "off_dt_struct": 0, | |
| "off_dt_strings": 0, | |
| "off_mem_rsvmap": 0, | |
| "version": version, | |
| "last_comp_version": last_comp_version, | |
| "boot_cpuid_phys": boot_cpuid_phys, | |
| "size_dt_strings": 0, | |
| "size_dt_struct": 0, | |
| } | |
| self.rootnode = None | |
| self.prenops = None | |
| self.postnops = None | |
| self.reserve_entries = None | |
| def add_rootnode(self, rootnode, prenops=None, postnops=None): | |
| """Add root node""" | |
| self.rootnode = rootnode | |
| self.prenops = prenops | |
| self.postnops = postnops | |
| def get_rootnode(self): | |
| """Get root node""" | |
| return self.rootnode | |
| def add_reserve_entries(self, reserve_entries): | |
| """Add reserved entries as list of dict with | |
| 'address' and 'size' keys""" | |
| self.reserve_entries = reserve_entries | |
| def to_dts(self): | |
| """Export to DTS representation in string format""" | |
| result = "/dts-v1/;\n" | |
| result += "// version:\t\t%d\n" % self.header["version"] | |
| result += ( | |
| "// last_comp_version:\t%d\n" % self.header["last_comp_version"] | |
| ) | |
| if self.header["version"] >= 2: | |
| result += ( | |
| "// boot_cpuid_phys:\t0x%x\n" % self.header["boot_cpuid_phys"] | |
| ) | |
| result += "\n" | |
| if self.reserve_entries is not None: | |
| for entry in self.reserve_entries: | |
| result += "/memreserve/ " | |
| if entry["address"]: | |
| result += "%#x " % entry["address"] | |
| else: | |
| result += "0 " | |
| if entry["size"]: | |
| result += "%#x" % entry["size"] | |
| else: | |
| result += "0" | |
| result += ";\n" | |
| if self.prenops: | |
| result += "\n".join([nop.dts_represent() for nop in self.prenops]) | |
| result += "\n" | |
| if self.rootnode is not None: | |
| result += self.rootnode.dts_represent() | |
| if self.postnops: | |
| result += "\n" | |
| result += "\n".join([nop.dts_represent() for nop in self.postnops]) | |
| return result | |
| def to_dtb(self): | |
| """Export to Blob format""" | |
| if self.rootnode is None: | |
| return None | |
| blob_reserve_entries = pack("") | |
| if self.reserve_entries is not None: | |
| for entry in self.reserve_entries: | |
| blob_reserve_entries += pack( | |
| ">QQ", entry["address"], entry["size"] | |
| ) | |
| blob_reserve_entries += pack(">QQ", 0, 0) | |
| header_size = 7 * 4 | |
| if self.header["version"] >= 2: | |
| header_size += 4 | |
| if self.header["version"] >= 3: | |
| header_size += 4 | |
| if self.header["version"] >= 17: | |
| header_size += 4 | |
| header_adjust = pack("") | |
| if header_size % 8 != 0: | |
| header_adjust = pack("b", 0) * (8 - (header_size % 8)) | |
| header_size += len(header_adjust) | |
| dt_start = header_size + len(blob_reserve_entries) | |
| # print "dt_start %d" % dt_start | |
| (blob_dt, blob_strings, dt_pos) = self.rootnode.dtb_represent( | |
| "", dt_start, self.header["version"] | |
| ) | |
| if self.prenops is not None: | |
| blob_dt = ( | |
| pack("").join( | |
| [nop.dtb_represent("")[0] for nop in self.prenops] | |
| ) | |
| + blob_dt | |
| ) | |
| if self.postnops is not None: | |
| blob_dt += pack("").join( | |
| [nop.dtb_represent("")[0] for nop in self.postnops] | |
| ) | |
| blob_dt += pack(">I", FDT_END) | |
| self.header["size_dt_strings"] = len(blob_strings) | |
| self.header["size_dt_struct"] = len(blob_dt) | |
| self.header["off_mem_rsvmap"] = header_size | |
| self.header["off_dt_struct"] = dt_start | |
| self.header["off_dt_strings"] = dt_start + len(blob_dt) | |
| self.header["totalsize"] = dt_start + len(blob_dt) + len(blob_strings) | |
| blob_header = pack( | |
| ">IIIIIII", | |
| self.header["magic"], | |
| self.header["totalsize"], | |
| self.header["off_dt_struct"], | |
| self.header["off_dt_strings"], | |
| self.header["off_mem_rsvmap"], | |
| self.header["version"], | |
| self.header["last_comp_version"], | |
| ) | |
| if self.header["version"] >= 2: | |
| blob_header += pack(">I", self.header["boot_cpuid_phys"]) | |
| if self.header["version"] >= 3: | |
| blob_header += pack(">I", self.header["size_dt_strings"]) | |
| if self.header["version"] >= 17: | |
| blob_header += pack(">I", self.header["size_dt_struct"]) | |
| return ( | |
| blob_header | |
| + header_adjust | |
| + blob_reserve_entries | |
| + blob_dt | |
| + blob_strings.encode("ascii") | |
| ) | |
| def to_json(self): | |
| """Ouput JSON""" | |
| if self.rootnode is None: | |
| return None | |
| return self.rootnode.json_represent() | |
| def resolve_path(self, path): | |
| """Resolve path like /memory/reg and return either a FdtNode, | |
| a FdtProperty or None""" | |
| if self.rootnode is None: | |
| return None | |
| if not path.startswith("/"): | |
| return None | |
| if len(path) > 1 and path.endswith("/"): | |
| path = path[:-1] | |
| if path == "/": | |
| return self.rootnode | |
| curnode = self.rootnode | |
| for subpath in path[1:].split("/"): | |
| found = None | |
| if not isinstance(curnode, FdtNode): | |
| return None | |
| for node in curnode: | |
| if subpath == node.get_name(): | |
| found = node | |
| break | |
| if found is None: | |
| return None | |
| curnode = found | |
| return curnode | |
| def _add_json_to_fdtnode(node, subjson): | |
| """Populate FdtNode with JSON dict items""" | |
| for (key, value) in subjson.items(): | |
| if isinstance(value, dict): | |
| subnode = FdtNode(key) | |
| subnode.set_parent_node(node) | |
| node.append(subnode) | |
| _add_json_to_fdtnode(subnode, value) | |
| elif isinstance(value, list): | |
| if len(value) < 2: | |
| raise Exception("Invalid list for %s" % key) | |
| if value[0] == "words": | |
| words = [int(word, 16) for word in value[1:]] | |
| node.append(FdtPropertyWords(key, words)) | |
| elif value[0] == "bytes": | |
| bytez = [int(byte, 16) for byte in value[1:]] | |
| node.append(FdtPropertyBytes(key, bytez)) | |
| elif value[0] == "strings": | |
| node.append(FdtPropertyStrings(key, [s for s in value[1:]])) | |
| else: | |
| raise Exception("Invalid list for %s" % key) | |
| elif value is None: | |
| node.append(FdtProperty(key)) | |
| else: | |
| raise Exception("Invalid value for %s" % key) | |
| def FdtJsonParse(buf): | |
| """Import FDT from JSON representation, see JSONDeviceTree.md for | |
| structure and encoding | |
| Returns an Fdt object | |
| """ | |
| tree = json.loads(buf) | |
| root = FdtNode("/") | |
| _add_json_to_fdtnode(root, tree) | |
| fdt = Fdt() | |
| fdt.add_rootnode(root) | |
| return fdt | |
| def FdtFsParse(path): | |
| """Parse device tree filesystem and return a Fdt instance | |
| Should be /proc/device-tree on a device, or the fusemount.py | |
| mount point. | |
| """ | |
| root = FdtNode("/") | |
| if path.endswith("/"): | |
| path = path[:-1] | |
| nodes = {path: root} | |
| for subpath, subdirs, files in os.walk(path): | |
| if subpath not in nodes.keys(): | |
| raise Exception("os.walk error") | |
| cur = nodes[subpath] | |
| for f in files: | |
| with open(subpath + "/" + f, "rb") as content_file: | |
| content = content_file.read() | |
| prop = FdtProperty.new_raw_property(f, content) | |
| cur.add_subnode(prop) | |
| for subdir in subdirs: | |
| subnode = FdtNode(subdir) | |
| cur.add_subnode(subnode) | |
| subnode.set_parent_node(cur) | |
| nodes[subpath + "/" + subdir] = subnode | |
| fdt = Fdt() | |
| fdt.add_rootnode(root) | |
| return fdt | |
| class FdtBlobParse(object): # pylint: disable-msg=R0903 | |
| """Parse from file input""" | |
| __fdt_header_format = ">IIIIIII" | |
| __fdt_header_names = ( | |
| "magic", | |
| "totalsize", | |
| "off_dt_struct", | |
| "off_dt_strings", | |
| "off_mem_rsvmap", | |
| "version", | |
| "last_comp_version", | |
| ) | |
| __fdt_reserve_entry_format = ">QQ" | |
| __fdt_reserve_entry_names = ("address", "size") | |
| __fdt_dt_cell_format = ">I" | |
| __fdt_dt_prop_format = ">II" | |
| __fdt_dt_tag_name = { | |
| FDT_BEGIN_NODE: "node_begin", | |
| FDT_END_NODE: "node_end", | |
| FDT_PROP: "prop", | |
| FDT_NOP: "nop", | |
| FDT_END: "end", | |
| } | |
| def __extract_fdt_header(self): | |
| """Extract DTB header""" | |
| header = Struct(self.__fdt_header_format) | |
| header_entry = Struct(">I") | |
| data = self.infile.read(header.size) | |
| result = dict(zip(self.__fdt_header_names, header.unpack_from(data))) | |
| if result["version"] >= 2: | |
| data = self.infile.read(header_entry.size) | |
| result["boot_cpuid_phys"] = header_entry.unpack_from(data)[0] | |
| if result["version"] >= 3: | |
| data = self.infile.read(header_entry.size) | |
| result["size_dt_strings"] = header_entry.unpack_from(data)[0] | |
| if result["version"] >= 17: | |
| data = self.infile.read(header_entry.size) | |
| result["size_dt_struct"] = header_entry.unpack_from(data)[0] | |
| return result | |
| def __extract_fdt_reserve_entries(self): | |
| """Extract reserved memory entries""" | |
| header = Struct(self.__fdt_reserve_entry_format) | |
| entries = [] | |
| self.infile.seek(self.fdt_header["off_mem_rsvmap"]) | |
| while True: | |
| data = self.infile.read(header.size) | |
| result = dict( | |
| zip(self.__fdt_reserve_entry_names, header.unpack_from(data)) | |
| ) | |
| if result["address"] == 0 and result["size"] == 0: | |
| return entries | |
| entries.append(result) | |
| def __extract_fdt_nodename(self): | |
| """Extract node name""" | |
| data = "" | |
| pos = self.infile.tell() | |
| while True: | |
| byte = self.infile.read(1) | |
| if ord(byte) == 0: | |
| break | |
| data += byte.decode("ascii") | |
| align_pos = pos + len(data) + 1 | |
| align_pos = ((align_pos) + ((4) - 1)) & ~((4) - 1) | |
| self.infile.seek(align_pos) | |
| return data | |
| def __extract_fdt_string(self, prop_string_pos): | |
| """Extract string from string pool""" | |
| data = "" | |
| pos = self.infile.tell() | |
| self.infile.seek(self.fdt_header["off_dt_strings"] + prop_string_pos) | |
| while True: | |
| byte = self.infile.read(1) | |
| if ord(byte) == 0: | |
| break | |
| data += byte.decode("ascii") | |
| self.infile.seek(pos) | |
| return data | |
| def __extract_fdt_prop(self): | |
| """Extract property""" | |
| prop = Struct(self.__fdt_dt_prop_format) | |
| pos = self.infile.tell() | |
| data = self.infile.read(prop.size) | |
| (prop_size, prop_string_pos) = prop.unpack_from(data) | |
| prop_start = pos + prop.size | |
| if self.fdt_header["version"] < 16 and prop_size >= 8: | |
| prop_start = ((prop_start) + ((8) - 1)) & ~((8) - 1) | |
| self.infile.seek(prop_start) | |
| value = self.infile.read(prop_size) | |
| align_pos = self.infile.tell() | |
| align_pos = ((align_pos) + ((4) - 1)) & ~((4) - 1) | |
| self.infile.seek(align_pos) | |
| return (self.__extract_fdt_string(prop_string_pos), value) | |
| def __extract_fdt_dt(self): | |
| """Extract tags""" | |
| cell = Struct(self.__fdt_dt_cell_format) | |
| tags = [] | |
| self.infile.seek(self.fdt_header["off_dt_struct"]) | |
| while True: | |
| data = self.infile.read(cell.size) | |
| if len(data) < cell.size: | |
| break | |
| tag, = cell.unpack_from(data) | |
| # print "*** %s" % self.__fdt_dt_tag_name.get(tag, '') | |
| if self.__fdt_dt_tag_name.get(tag, "") in "node_begin": | |
| name = self.__extract_fdt_nodename() | |
| if len(name) == 0: | |
| name = "/" | |
| tags.append((tag, name)) | |
| elif self.__fdt_dt_tag_name.get(tag, "") in ("node_end", "nop"): | |
| tags.append((tag, "")) | |
| elif self.__fdt_dt_tag_name.get(tag, "") in "end": | |
| tags.append((tag, "")) | |
| break | |
| elif self.__fdt_dt_tag_name.get(tag, "") in "prop": | |
| propdata = self.__extract_fdt_prop() | |
| tags.append((tag, propdata)) | |
| else: | |
| print("Unknown Tag %d" % tag) | |
| return tags | |
| def __init__(self, infile): | |
| """Init with file input""" | |
| self.infile = infile | |
| self.fdt_header = self.__extract_fdt_header() | |
| if self.fdt_header["magic"] != FDT_MAGIC: | |
| raise Exception("Invalid Magic") | |
| if self.fdt_header["version"] > FDT_MAX_VERSION: | |
| raise Exception("Invalid Version %d" % self.fdt_header["version"]) | |
| if self.fdt_header["last_comp_version"] > FDT_MAX_VERSION - 1: | |
| raise Exception( | |
| "Invalid last compatible Version %d" | |
| % self.fdt_header["last_comp_version"] | |
| ) | |
| self.fdt_reserve_entries = self.__extract_fdt_reserve_entries() | |
| self.fdt_dt_tags = self.__extract_fdt_dt() | |
| def __to_nodes(self): | |
| """Represent fdt as Node and properties structure | |
| Returns a set with the pre-node Nops, the Root Node, | |
| and the post-node Nops. | |
| """ | |
| prenops = [] | |
| postnops = [] | |
| rootnode = None | |
| curnode = None | |
| for tag in self.fdt_dt_tags: | |
| if self.__fdt_dt_tag_name.get(tag[0], "") in "node_begin": | |
| newnode = FdtNode(tag[1]) | |
| if rootnode is None: | |
| rootnode = newnode | |
| if curnode is not None: | |
| curnode.add_subnode(newnode) | |
| newnode.set_parent_node(curnode) | |
| curnode = newnode | |
| elif self.__fdt_dt_tag_name.get(tag[0], "") in "node_end": | |
| if curnode is not None: | |
| curnode = curnode.get_parent_node() | |
| elif self.__fdt_dt_tag_name.get(tag[0], "") in "nop": | |
| if curnode is not None: | |
| curnode.add_subnode(FdtNop()) | |
| elif rootnode is not None: | |
| postnops.append(FdtNop()) | |
| else: | |
| prenops.append(FdtNop()) | |
| elif self.__fdt_dt_tag_name.get(tag[0], "") in "prop": | |
| if curnode is not None: | |
| curnode.add_raw_attribute(tag[1][0], tag[1][1]) | |
| elif self.__fdt_dt_tag_name.get(tag[0], "") in "end": | |
| continue | |
| return (prenops, rootnode, postnops) | |
| def to_fdt(self): | |
| """Create a fdt object | |
| Returns a Fdt object | |
| """ | |
| if self.fdt_header["version"] >= 2: | |
| boot_cpuid_phys = self.fdt_header["boot_cpuid_phys"] | |
| else: | |
| boot_cpuid_phys = 0 | |
| fdt = Fdt( | |
| version=self.fdt_header["version"], | |
| last_comp_version=self.fdt_header["last_comp_version"], | |
| boot_cpuid_phys=boot_cpuid_phys, | |
| ) | |
| (prenops, rootnode, postnops) = self.__to_nodes() | |
| fdt.add_rootnode(rootnode, prenops=prenops, postnops=postnops) | |
| fdt.add_reserve_entries(self.fdt_reserve_entries) | |
| return fdt |