Initial commit for half implemented IfcFM library to support new FM exchange requirements and related FM tasks. Initially just an IfcCOBie replacement but will grow.

This commit is contained in:
Dion Moult
2021-07-23 18:03:07 +10:00
parent beb5dda6dd
commit 16fd06a4e9
3 changed files with 961 additions and 0 deletions
+8
View File
@@ -0,0 +1,8 @@
# IfcFM
IfcFM is a library that handles the extraction and analysis of IFC data for the
purposes of facility management.
It is currently a prototype and will supersede the IfcCOBie library. It is
planned to support workflows related to the old COBie standard, as well as
upcoming IFC Facility Management related MVDs.
+372
View File
@@ -0,0 +1,372 @@
import datetime
import ifcopenshell
import ifcopenshell.util.fm
import ifcopenshell.util.selector
import ifcopenshell.util.date
import ifcopenshell.util.schema
class Parser:
def __init__(self, logger):
self.logger = logger
self.file = None
self.sheets = [
"contacts",
"facilities",
"floors",
"spaces",
"zones",
"types",
"components",
"systems",
"assemblies",
"connections",
"spares",
"resources",
"jobs",
"impacts",
"documents",
"attributes",
"coordinates",
"issues",
]
for sheet in self.sheets:
setattr(self, sheet, {})
self.picklists = {
"Category-Role": [],
"Category-Facility": [],
"FloorType": [],
"Category-Space": [],
"ZoneType": [],
"Category-Product": [],
"AssetType": [],
"DurationUnit": ["day"], # See note about hardcoded day below
"Category-Element": [],
"SpareType": [],
"ApprovalBy": [],
"StageType": [],
"objType": [],
}
self.default_date = (datetime.datetime(1970, 1, 1) + datetime.timedelta(seconds=-2177452801)).isoformat()
def parse(self, files):
self.files = files
# self.file = ifcopenshell.open(file)
# self.type_assets = self.selector.parse(self.file, type_query)
# self.component_assets = self.selector.parse(self.file, component_query)
self.get_contacts()
self.get_facilities()
self.get_floors()
self.get_spaces()
self.get_zones()
self.get_types()
self.get_components()
self.get_systems()
# self.get_assemblies()
# self.get_connections()
# self.get_spares()
# self.get_resources()
# self.get_jobs()
# self.get_impacts()
self.get_documents()
# self.get_attributes()
# self.get_coordinates()
# self.get_issues()
def get_contacts(self):
for element in self.files["arch"].by_type("IfcOrganization"):
if "IfcApplication" in [e.is_a() for e in self.files["arch"].get_inverse(element)]:
continue
name = element.Name
self.contacts[name] = self.get_organisation(element)
def get_organisation(self, element):
return {
"Name": element.Name,
"Category": self.get_organisation_category(element),
"Email": self.get_organisation_address(element, "ElectronicMailAddresses"),
"Phone": self.get_organisation_address(element, "TelephoneNumbers"),
"Department": self.get_organisation_address(element, "InternalLocation"),
"Street": self.get_organisation_address(element, "AddressLines"),
"PostalBox": self.get_organisation_address(element, "PostalBox"),
"Town": self.get_organisation_address(element, "Town"),
"StateRegion": self.get_organisation_address(element, "Region"),
"PostalCode": self.get_organisation_address(element, "PostalCode"),
"Country": self.get_organisation_address(element, "Country"),
"CompanyURL": self.get_organisation_address(element, "WWWHomePageURL"),
}
def get_organisation_category(self, element):
for role in element.Roles or []:
if role.UserDefinedRole:
return role.UserDefinedRole
def get_organisation_address(self, element, name):
for address in element.Addresses or []:
if hasattr(address, name) and getattr(address, name, None):
result = getattr(address, name)
if isinstance(result, tuple):
return result[0]
return result
def get_facilities(self):
element = self.files["arch"].by_type("IfcBuilding")[0]
self.facilities[element.Name] = {
"Name": element.Name,
"AuthorOrganizationName": element.OwnerHistory.OwningUser.TheOrganization.Name,
"AuthorDate": ifcopenshell.util.date.ifc2datetime(
self.files["arch"].by_type("IfcProject")[0].OwnerHistory.CreationDate
).isoformat(),
"Category": self.get_classification(element),
"ProjectName": element.Decomposes[0].RelatingObject.Decomposes[0].RelatingObject.Name,
"SiteName": element.Decomposes[0].RelatingObject.Name,
"LinearUnits": "millimeters",
"AreaUnits": "square meters",
"AreaMeasurement": "TODO",
"Phase": element.Decomposes[0].RelatingObject.Decomposes[0].RelatingObject.Phase,
"ModelSoftware": element.OwnerHistory.OwningApplication.ApplicationFullName,
"ModelProjectID": self.files["arch"].by_type("IfcProject")[0].GlobalId,
"ModelSiteID": element.Decomposes[0].RelatingObject.GlobalId,
"ModelBuildingID": element.GlobalId,
}
def get_classification(self, element):
rel = [r for r in element.HasAssociations or [] if r.is_a("IfcRelAssociatesClassification")]
if rel:
classification = rel[0].RelatingClassification
if getattr(classification, "Identification", None) and getattr(classification, "Name", None):
return "{}:{}".format(classification.Identification, classification.Name)
elif getattr(classification, "ItemReference", None) and getattr(classification, "Name", None):
return "{}:{}".format(classification.ItemReference, classification.Name)
def get_floors(self):
storeys = self.files["arch"].by_type("IfcBuildingStorey")
self.floors["Site"] = {
"Name": "Site",
"AuthorOrganizationName": storeys[0].OwnerHistory.OwningUser.TheOrganization.Name,
"AuthorDate": datetime.datetime.now().replace(microsecond=0).isoformat(),
"Category": "Site",
"ModelSoftware": "IfcFM",
"ModelObject": "IfcExternalSpatialElement",
"ModelID": ifcopenshell.guid.new(), # TODO
"Elevation": None,
}
for element in storeys:
self.get_floor(element)
def get_floor(self, element):
name = element.Name
elevation = element.ObjectPlacement.RelativePlacement.Location.Coordinates[2]
self.floors[name] = {
"Name": name,
"AuthorOrganizationName": element.OwnerHistory.OwningUser.TheOrganization.Name,
"AuthorDate": ifcopenshell.util.date.ifc2datetime(element.OwnerHistory.CreationDate).isoformat(),
"Category": "Level",
"ModelSoftware": element.OwnerHistory.OwningApplication.ApplicationFullName,
"ModelObject": element.is_a(),
"ModelID": element.GlobalId,
"Elevation": elevation,
}
def get_spaces(self):
primary_keys = []
for element in self.files["arch"].by_type("IfcSpace"):
name = element.Name
primary_keys.append(name)
# TODO: not correct mapping
psets = ifcopenshell.util.element.get_psets(element)
category = None
if "Data" in psets and "COBie.Space.Category" in psets["Data"]:
category = psets["Data"]["COBie.Space.Category"].replace(" : ", ":")
usable_height = None
if "Data" in psets and "COBie.Space.Category" in psets["Data"]:
usable_height = round(psets["Data"]["COBie.Space.UsableHeight"], 2) or None
area_gross = None
if "Data" in psets and "COBie.Space.GrossArea" in psets["Data"]:
area_gross = round(psets["Data"]["COBie.Space.GrossArea"], 2) or None
area_net = None
if "Data" in psets and "COBie.Space.NetArea" in psets["Data"]:
area_net = round(psets["Data"]["COBie.Space.NetArea"], 2) or None
self.spaces[name] = {
"Name": name,
"AuthorOrganizationName": element.OwnerHistory.OwningUser.TheOrganization.Name,
"AuthorDate": ifcopenshell.util.date.ifc2datetime(element.OwnerHistory.CreationDate).isoformat(),
"Category": category,
"LevelName": element.Decomposes[0].RelatingObject.Name,
"Description": element.LongName,
"ModelSoftware": element.OwnerHistory.OwningApplication.ApplicationFullName,
"ModelID": element.GlobalId,
"BuildingRoomNumber": None,
"UsableHeight": usable_height,
"AreaGross": area_gross,
"AreaNet": area_net,
}
def get_zones(self):
for element in self.files["arch"].by_type("IfcZone"):
for rel in element.IsGroupedBy:
for space in rel.RelatedObjects:
if not space.is_a("IfcSpace"):
continue
self.zones[element.Name + space.Name] = {
"Name": element.Name,
"AuthorOrganizationName": "Cox Architecture",
"AuthorDate": ifcopenshell.util.date.ifc2datetime(
element.OwnerHistory.CreationDate
).isoformat(),
"Category": "Occupancy",
"SpaceName": space.Name,
"ModelSoftware": element.OwnerHistory.OwningApplication.ApplicationFullName,
"ModelID": element.GlobalId,
"ParentZoneName": None,
}
def get_systems(self):
for discipline, ifc in self.files.items():
#if discipline == "arch":
# continue
for element in ifc.by_type("IfcSystem"):
name = element.Name
self.systems[name] = {
"Name": name,
"AuthorOrganizationName": "Fredon",
"AuthorDate": ifcopenshell.util.date.ifc2datetime(element.OwnerHistory.CreationDate).isoformat(),
"Category": None,
"ModelSoftware": element.OwnerHistory.OwningApplication.ApplicationFullName,
"ModelID": element.GlobalId,
"ParentSystemName": None,
}
def get_types(self):
for discipline, ifc in self.files.items():
self.get_types_from_file(discipline)
def get_types_from_file(self, ifc_file):
primary_keys = []
for element in ifcopenshell.util.fm.get_fmhem_types(self.files[ifc_file]):
name = element.Name
primary_keys.append(name)
psets = ifcopenshell.util.element.get_psets(element)
category = None
if "Data" in psets and "COBie.Type.Category" in psets["Data"]:
if ":" in psets["Data"]["COBie.Type.Category"]:
category = psets["Data"]["COBie.Type.Category"].replace(" : ", ":")
self.types[name] = {
"Name": name,
"AuthorOrganizationName": element.OwnerHistory.OwningUser.TheOrganization.Name,
"AuthorDate": ifcopenshell.util.date.ifc2datetime(element.OwnerHistory.CreationDate).isoformat(),
"Category": category,
"ProcurementMethod": None,
"Description": element.Description,
"ManufacturerOrganizationName": None,
"SupplierOrganizationName": None,
"ModelNumber": None,
"WarrantyOrganizationName": None,
"WarrantyDuration": None,
"ModelSoftware": element.OwnerHistory.OwningApplication.ApplicationFullName,
"ModelObject": element.is_a(),
"ModelID": element.GlobalId,
"SpecificationSection": None,
"SubmittalID": None,
"ProductURL": None,
}
def get_components(self):
org_map = {"arch": "Cox Architecture", "arch": "Bates Smart", "elec": "Fredon", "fire": "Premier Fire"}
for discipline, ifc in self.files.items():
self.get_components_from_file(discipline)
def get_components_from_file(self, ifc_file):
for element_type in ifcopenshell.util.fm.get_fmhem_types(self.files[ifc_file]):
for element in element_type.ObjectTypeOf[0].RelatedObjects:
name = element.Name
system = None
for rel in element.HasAssignments:
if rel.is_a("IfcRelAssignsToGroup") and rel.RelatingGroup.is_a("IfcSystem"):
system = rel.RelatingGroup.Name
space_name = None
# space = element.ContainedInStructure[0].RelatingStructure
# if space.is_a("IfcSpace"):
# space_name = space.Name
psets = ifcopenshell.util.element.get_psets(element)
if "Data" in psets and "COBie.Component.Space" in psets["Data"]:
space_name = psets["Data"]["COBie.Component.Space"]
if space_name not in self.spaces:
space_name = None
self.components[name] = {
"Name": name,
"AuthorOrganizationName": element.OwnerHistory.OwningUser.TheOrganization.Name,
"AuthorDate": ifcopenshell.util.date.ifc2datetime(element.OwnerHistory.CreationDate).isoformat(),
"TypeName": element_type.Name,
"SpaceName": space_name,
"InstallationDate": None,
"WarrantyStartDate": None,
"ModelSoftware": element.OwnerHistory.OwningApplication.ApplicationFullName,
"ModelObject": element.is_a(),
"ModelID": element.GlobalId,
"InstalledModelNumber": None,
"SerialNumber": None,
"BarCode": None,
"TagNumber": None,
"OwnerAssetID": None,
"SystemName": system,
"FluidHotFeedName": None,
"FluidColdFeedName": None,
"ElectricPanelName": None,
"ElectricCircuitName": None,
"ControlledByName": None,
"InterlockedWithName": None,
"PartOfAssemblyName": None,
}
def get_documents(self):
for rel in self.files["arch"].by_type("IfcRelAssociatesDocument"):
element = rel.RelatingDocument
for related_object in rel.RelatedObjects:
name = element.Name
worksheet_row = related_object.Name
if self.files["arch"].schema == "IFC2X3":
submittal_id = element.ItemReference
referenced_document = element.ReferenceToDocument[0]
author_date = None
if referenced_document.CreationTime:
author_date = ifcopenshell.util.date.ifc2datetime(referenced_document.CreationTime).isoformat()
else:
submittal_id = element.Identification
referenced_document = element.ReferencedDocument
author_date = referenced_document.CreationTime
worksheet_name = None
if related_object.is_a("IfcSpace"):
worksheet_name = "Space"
elif related_object.is_a("IfcTypeObject"):
worksheet_name = "Type"
self.documents[element.Name + worksheet_name + worksheet_row] = {
"Name": name,
"AuthorOrganizationName": referenced_document.DocumentOwner.Name,
"AuthorDate": author_date,
"Category": referenced_document.Purpose,
"WorksheetName": worksheet_name,
"WorksheetRow": worksheet_row,
"Revision": referenced_document.Revision,
"Location": referenced_document.Name,
"Description": referenced_document.Description,
"SpecificationSection": None,
"SubmittalID": submittal_id,
"SourceURL": None,
}
+581
View File
@@ -0,0 +1,581 @@
import csv
try:
from xlsxwriter import Workbook
except:
pass # No XLSX support
try:
from odf.opendocument import OpenDocumentSpreadsheet
from odf.style import Style, TableCellProperties
from odf.table import Table, TableRow, TableCell
from odf.text import P
except:
pass # No ODF support
# https://stackoverflow.com/questions/1143671/how-to-sort-objects-by-multiple-keys-in-python
from operator import itemgetter as i
from functools import cmp_to_key
def cmp(x, y):
"""
Replacement for built-in function cmp that was removed in Python 3
Compare the two objects x and y and return an integer according to
the outcome. The return value is negative if x < y, zero if x == y
and strictly positive if x > y.
https://portingguide.readthedocs.io/en/latest/comparisons.html#the-cmp-function
"""
try:
return (x > y) - (x < y)
except:
return 0
def multikeysort(items, columns):
comparers = [((i(col[1:].strip()), -1) if col.startswith("-") else (i(col.strip()), 1)) for col in columns]
def comparer(left, right):
comparer_iter = (cmp(fn(left), fn(right)) * mult for fn, mult in comparers)
return next((result for result in comparer_iter if result), 0)
return sorted(items, key=cmp_to_key(comparer))
class Writer:
def __init__(self, parser, filename=None):
self.filename = filename
self.parser = parser
self.sheets = []
self.sheet_data = {}
self.colours = {
"r": "fdff8e", # Required
"i": "fdcd94", # Internal reference
"e": "cd95ff", # External reference
"o": "cdffc8", # Optional
"s": "c0c0c0", # Secondary information
"p": "9ccaff", # Project specific
"n": "000000", # Not used
}
self.colours = {
"r": "dc8774", # Required
"i": "eda786", # Internal reference
"e": "96c7d0", # External reference
"o": "ddb873", # Optional or edd889
"s": "dddddd", # Secondary information
"p": "b8dd73", # Project specific
"n": "000000", # Not used
}
def write(self):
self.sheets = [
"Contact",
"Facility",
"Floor",
"Space",
"Zone",
"Type",
"Component",
"System",
# "Assembly",
# "Connection",
# "Spare",
# "Resource",
# "Job",
# "Impact",
"Document",
# "Attribute",
# "Coordinate",
# "Issue",
]
self.write_data(
"Contact",
self.parser.contacts,
[
"Name",
"Category",
"Email",
"Phone",
"Department",
"Street",
"PostalBox",
"Town",
"StateRegion",
"PostalCode",
"Country",
"CompanyURL",
],
"rirrrrrrrrrr",
["Name"],
)
self.write_data(
"Facility",
self.parser.facilities,
[
"Name",
"AuthorOrganizationName",
"AuthorDate",
"Category",
"ProjectName",
"SiteName",
"LinearUnits",
"AreaUnits",
"AreaMeasurement",
"Phase",
"ModelSoftware",
"ModelProjectID",
"ModelSiteID",
"ModelBuildingID",
],
"ririrrrrrreeee",
["Name"],
)
self.write_data(
"Floor",
self.parser.floors,
[
"Name",
"AuthorOrganizationName",
"AuthorDate",
"Category",
"ModelSoftware",
"ModelObject",
"ModelID",
"Elevation",
],
"ririeeer",
["Elevation"],
)
self.write_data(
"Space",
self.parser.spaces,
[
"Name",
"AuthorOrganizationName",
"AuthorDate",
"Category",
"LevelName",
"Description",
"ModelSoftware",
"ModelID",
"BuildingRoomNumber",
"UsableHeight",
"AreaGross",
"AreaNet",
],
"ririireerrrr",
["LevelName", "Name"],
)
self.write_data(
"Zone",
self.parser.zones,
[
"Name",
"AuthorOrganizationName",
"AuthorDate",
"Category",
"SpaceName",
"ModelSoftware",
"ModelID",
"ParentZoneName",
],
"ririieei",
["Name", "SpaceName"],
)
self.write_data(
"Type",
self.parser.types,
[
"Name",
"AuthorOrganizationName",
"AuthorDate",
"Category",
"ProcurementMethod",
"Description",
"ManufacturerOrganizationName",
"SupplierOrganizationName",
"ModelNumber",
"WarrantyOrganizationName",
"WarrantyDuration",
"ModelSoftware",
"ModelObject",
"ModelID",
"SpecificationSection",
"SubmittalID",
"ProductURL",
],
"ririiriirireeerrr",
["ModelObject", "Name"],
)
self.write_data(
"Component",
self.parser.components,
[
"Name",
"AuthorOrganizationName",
"AuthorDate",
"TypeName",
"SpaceName",
"InstallationDate",
"WarrantyStartDate",
"ModelSoftware",
"ModelObject",
"ModelID",
"InstalledModelNumber",
"SerialNumber",
"BarCode",
"TagNumber",
"OwnerAssetID",
"SystemName",
"FluidHotFeedName",
"FluidColdFeedName",
"ElectricPanelName",
"ElectricCircuitName",
"ControlledByName",
"InterlockedWithName",
"PartOfAssemblyName",
],
"ririirreeerrrrriiiiiiii",
["ModelObject", "Name"],
)
self.write_data(
"System",
self.parser.systems,
[
"Name",
"AuthorOrganizationName",
"AuthorDate",
"Category",
"ModelSoftware",
"ModelID",
"ParentSystemName",
],
"ririeei",
["Name"],
)
# self.write_data(
# "Assembly",
# self.parser.assemblies,
# "Name",
# [
# "Name",
# "CreatedBy",
# "CreatedOn",
# "SheetName",
# "ParentName",
# "ChildNames",
# "AssemblyType",
# "ExtSystem",
# "ExtObject",
# "ExtIdentifier",
# "Description",
# ],
# "rirrrrreeeo",
# self.parser.custom_data["assemblies"],
# )
# self.write_data(
# "Connection",
# self.parser.connections,
# "Name",
# [
# "Name",
# "CreatedBy",
# "CreatedOn",
# "ConnectionType",
# "SheetName",
# "RowName1",
# "RowName2",
# "RealizingElement",
# "PortName1",
# "PortName2",
# "ExtSystem",
# "ExtObject",
# "ExtIdentifier",
# "Description",
# ],
# "ririiiiiiieeeo",
# self.parser.custom_data["connections"],
# )
# self.write_data(
# "Spare",
# self.parser.spares,
# "Name",
# [
# "Name",
# "CreatedBy",
# "CreatedOn",
# "Category",
# "TypeName",
# "Suppliers",
# "ExtSystem",
# "ExtObject",
# "ExtIdentifier",
# "Description",
# "SetNumber",
# "PartNumber",
# ],
# "ririiieeeooo",
# self.parser.custom_data["spares"],
# )
# self.write_data(
# "Resource",
# self.parser.resources,
# "Name",
# [
# "Name",
# "CreatedBy",
# "CreatedOn",
# "Category",
# "ExtSystem",
# "ExtObject",
# "ExtIdentifier",
# "Description",
# ],
# "ririeeeo",
# self.parser.custom_data["resources"],
# )
# self.write_data(
# "Job",
# self.parser.jobs,
# "Name",
# [
# "Name",
# "CreatedBy",
# "CreatedOn",
# "Category",
# "Status",
# "TypeName",
# "Description",
# "Duration",
# "DurationUnit",
# "Start",
# "TaskStartUnit",
# "Frequency",
# "FrequencyUnit",
# "ExtSystem",
# "ExtObject",
# "ExtIdentifier",
# "TaskNumber",
# "Priors",
# "ResourceNames",
# ],
# "ririiirriririeeeoii",
# self.parser.custom_data["jobs"],
# )
# self.write_data(
# "Impact",
# self.parser.impacts,
# "Name",
# [
# "Name",
# "CreatedBy",
# "CreatedOn",
# "ImpactType",
# "ImpactStage",
# "SheetName",
# "RowName",
# "Value",
# "Unit",
# "LeadInTime",
# "Duration",
# "LeadOutTime",
# "ExtSystem",
# "ExtObject",
# "ExtIdentifier",
# "Description",
# ],
# "ririiiirioooeeeo",
# self.parser.custom_data["impacts"],
# )
self.write_data(
"Document",
self.parser.documents,
[
"Name",
"AuthorOrganizationName",
"AuthorDate",
"Category",
"WorksheetName",
"WorksheetRow",
"Revision",
"Location",
"Description",
"SpecificationSection",
"SubmittalID",
"SourceURL",
],
"ririiirrrrer",
["Name"],
)
# self.write_data(
# "Attribute",
# self.parser.attributes,
# "Name",
# [
# "Name",
# "CreatedBy",
# "CreatedOn",
# "Category",
# "SheetName",
# "RowName",
# "Value",
# "Unit",
# "ExtSystem",
# "ExtObject",
# "ExtIdentifier",
# "Description",
# "AllowedValues",
# ],
# "ririiirreeeoo",
# )
# self.write_data(
# "Coordinate",
# self.parser.coordinates,
# "Name",
# [
# "Name",
# "CreatedBy",
# "CreatedOn",
# "Category",
# "SheetName",
# "RowName",
# "CoordinateXAxis",
# "CoordinateYAxis",
# "CoordinateZAxis",
# "ExtSystem",
# "ExtObject",
# "ExtIdentifier",
# "ClockwiseRotation",
# "ElevationalRotation",
# "YawRotation",
# ],
# "ririiooooeeeooo",
# )
# self.write_data(
# "Issue",
# self.parser.issues,
# "Name",
# [
# "Name",
# "CreatedBy",
# "CreatedOn",
# "Type",
# "Risk",
# "Chance",
# "Impact",
# "SheetName1",
# "RowName1",
# "SheetName2",
# "RowName2",
# "Description",
# "Owner",
# "Mitigation",
# "ExtSystem",
# "ExtObject",
# "ExtIdentifier",
# ],
# "ririooooooooooeee",
# )
def write_data(self, sheet, data, fieldnames, colours, sort_fields, custom_data={}):
self.sheet_data[sheet] = {"headers": fieldnames + list(custom_data.keys()), "colours": colours, "rows": []}
for row in multikeysort(list(data.values()), sort_fields):
values = []
for fieldname in fieldnames:
values.append(row[fieldname])
for fieldname in custom_data.keys():
values.append(row[fieldname])
self.sheet_data[sheet]["rows"].append(values)
class CsvWriter(Writer):
def write(self):
super().write()
for sheet, data in self.sheet_data.items():
with open(os.path.join(self.filename, "{}.csv".format(sheet)), "w", newline="", encoding="utf-8") as file:
writer = csv.writer(file)
writer.writerow(data["headers"])
for row in data["rows"]:
writer.writerow(row)
class XlsWriter(Writer):
def write(self):
super().write()
self.workbook = Workbook(self.filename + ".xlsx")
self.cell_formats = {}
for key, value in self.colours.items():
self.cell_formats[key] = self.workbook.add_format()
self.cell_formats[key].set_bg_color(value)
for sheet in self.sheets:
self.write_worksheet(sheet)
self.workbook.close()
def write_worksheet(self, name):
worksheet = self.workbook.add_worksheet(name)
r = 0
c = 0
for header in self.sheet_data[name]["headers"]:
cell = worksheet.write(r, c, header, self.cell_formats["s"])
c += 1
c = 0
r += 1
for row in self.sheet_data[name]["rows"]:
c = 0
for col in row:
if c >= len(self.sheet_data[name]["colours"]):
cell_format = "p"
else:
cell_format = self.sheet_data[name]["colours"][c]
cell = worksheet.write(r, c, col, self.cell_formats[cell_format])
c += 1
r += 1
class OdsWriter(Writer):
def write(self):
super().write()
self.doc = OpenDocumentSpreadsheet()
self.cell_formats = {}
for key, value in self.colours.items():
style = Style(name=key, family="table-cell")
style.addElement(TableCellProperties(backgroundcolor="#" + value))
self.doc.automaticstyles.addElement(style)
self.cell_formats[key] = style
for sheet in self.sheets:
self.write_table(sheet)
self.doc.save(self.filename, True)
def write_table(self, name):
table = Table(name=name)
tr = TableRow()
for header in self.sheet_data[name]["headers"]:
tc = TableCell(valuetype="string", stylename="s")
tc.addElement(P(text=header))
tr.addElement(tc)
table.addElement(tr)
for row in self.sheet_data[name]["rows"]:
tr = TableRow()
c = 0
for col in row:
if c >= len(self.sheet_data[name]["colours"]):
cell_format = "p"
else:
cell_format = self.sheet_data[name]["colours"][c]
tc = TableCell(valuetype="string", stylename=cell_format)
if col is None:
col = "NULL"
tc.addElement(P(text=col))
tr.addElement(tc)
c += 1
table.addElement(tr)
self.doc.spreadsheet.addElement(table)