--- /dev/null
+# -*- 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>
+"""
+
+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
projects / gem5.git / commitdiff