| # -*- 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 |
