Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Python_Engine/Python/src/python_toolkit/bhom/analytics.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,7 @@ def decorator(function: Callable):
@wraps(function)
def wrapper(*args, **kwargs) -> Any:
"""A wrapper around the function that captures usage analytics."""

if disable:
CONSOLE_LOGGER.debug("bhom_analytics is curently disabled.")
return function(*args, **kwargs)
Expand Down
303 changes: 303 additions & 0 deletions Python_Engine/Python/src/python_toolkit/bhom/bhom_object.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,303 @@
from base64 import decode
import copy
from ctypes import ArgumentError
from json import encoder
from pathlib import Path
import uuid
import re
from typing import Any, List, Dict, Union
import json
from json import JSONEncoder, JSONDecoder
from .logging import CONSOLE_LOGGER
from . import BHOM_VERSION
import pandas as pd
import numpy as np

BHOM_SHORT_VERSION = ".".join(BHOM_VERSION.split(".")[0:2])

def convert_pascal_to_camel(s: str):
"""Converts a string to camel_case."""
sections = re.split("(?<=.)(?=[A-Z])", s) #zero-length match before capitals, skipping capital at the 0th index
parts = []
for sec in sections:
parts.append(sec.lower())

return "_".join(parts)

def convert_camel_to_pascal(s: str):
"""Converts a string to PascalCase, ignoring _ if it is the first character."""
sections = re.split(r'(?<=.)_', s) #match all `_` except if it is the first character.

parts = []
for sec in sections: #capitalise each section unless the section is empty (in which case, append an underscore) or the section starts with an underscore (first section can begin with _)
if sec == "":
parts.append("_")
continue
elif sec.startswith("_"):
parts.append(sec)

parts.append(sec.capitalize())

return ''.join(parts)

class BHoMJSONDecoder(JSONDecoder):
def __init__(self, *args, **kwargs):
json.JSONDecoder.__init__(self, object_hook=self.object_hook, *args, **kwargs)

def object_hook(self, d):
if "_t" not in d:
CONSOLE_LOGGER.debug(f"BHoMJSONDecoder could not convert the following dictionary into a BHoMObject due to a missing '_t' property. Falling back to dictionary: {d}")
return self.deserialise_unknown(d)

props = {
"_t": d.pop("_t"),
"_bhom_version": d.pop("_bhomVersion", BHOM_SHORT_VERSION)
}

if (props["_bhom_version"] is not None) and (props["_bhom_version"] != BHOM_SHORT_VERSION):
CONSOLE_LOGGER.warning(f"The bhom version specified in the encoded json ({props['_bhom_version']}) is different from the BHoM version that python_toolkit was installed with ({BHOM_SHORT_VERSION}). There may be versioning issues with this object. Consider deserialising and then serialising again with the BHoM serialiser to get the correct version, or update BHoM to the correct version.")

if d.get("BHoM_Guid", None) is not None:
#deserialise as BHoM Object

#get default BHoMObject properties and replace with defaults if not present
props["name"] = d.pop("Name", "")
props["bhom_guid"] = uuid.UUID(d.pop("BHoM_Guid"))
props["tags"] = d.pop("Tags", [])
props["fragments"] = d.pop("Fragments", [])
props["custom_data"] = d.pop("CustomData", {})

#convert all other properties to camel_case as python users expect
for prop_name in d:
props[convert_pascal_to_camel(prop_name)] = d[prop_name]

return self.deserialise_unknown(BHoMObject(**props))
else:
if props["_t"].startswith("System.Collections.Generic.List"): #this handles when the BHoM serialiser makes generic lists from CustomObject list properties, which get converted to a dictionary. The BHoM serialiser does recognise lists if it can find the object definition via reflection.
return d["_v"]

#deserialise as IObject
for prop_name in d:
props[convert_pascal_to_camel(prop_name)] = d[prop_name]

return self.deserialise_unknown(IObject(**props))

def deserialise_unknown(self, obj: Union['IObject', 'BHoMObject', dict]):
"""Override this method in a subclass to add extra object hook on top of the existing method. If the object is an object serialised by the BHoM serialiser, the output object will be a BHoMObject or IObject, so use isinstance() to select."""
return obj

class BHoMJSONEncoder(JSONEncoder):
def default(self, o):
if isinstance(o, BHoMObject):
#initialise special BHoMObject properties

props = {
"Name": o.name,
"BHoM_Guid": str(o.bhom_guid),
"_t": o._t
}

if len(o.tags) > 0:
props["Tags"] = o.tags

if len(o.fragments) > 0:
props["Fragments"] = o.fragments

if len(o.custom_data) > 0:
props["CustomData"] = o.custom_data

if o._bhom_version is not None:
props["_bhomVersion"] = o._bhom_version

#get property names with reflection and convert all properties to PascalCase as the BHoM serialiser expects
for prop_name, value in vars(o).copy().items():
if prop_name in ["name", "bhom_guid", "tags", "fragments", "custom_data", "_t", "_bhom_version"]:
continue

props[convert_camel_to_pascal(prop_name)] = value

return props
elif isinstance(o, IObject):
props = {
"_t": o._t
}

if o._bhom_version is not None:
props["_bhomVersion"] = o._bhom_version

for prop_name, value in vars(o).copy().items():
if prop_name in ["_t", "_bhom_version"]:
continue

props[convert_camel_to_pascal(prop_name)] = value

return props
#handle common non-serialisable types
elif isinstance(o, uuid.UUID): #UUID object is not json serialisable by default
return str(o)
elif isinstance(o, pd.DatetimeIndex):
return [date.isoformat() for date in o]
elif isinstance(o, np.ndarray):
return o.tolist()
elif isinstance(o, pd.Timestamp):
return o.isoformat()
elif isinstance(o, pd.Series):
return dict(zip(o.index.astype(str), o))
elif isinstance(o, Path):
return { "FileName": str(o.name), "Directory": str(o.parent), "_t": "BH.oM.Adapter.FileSettings", "BHoM_Guid": str(uuid.uuid4()) }

return self.serialise_unknown(o) #if the object is unknown at this point, call serialise_unknown to allow subclasses to add their own serialisation logic.

def serialise_unknown(self, obj: Any) -> Union[Dict[str, str], str, List[str], Any]:
"""
This is called by the BHoMJSONEncoder to serialise unknown objects so that the BHoM can handle them if they exist as BHoM Objects in c# but not in python (i.e. an object from an external library).

Override this method in a sub class with its own logic, and call super().serialise_unknown(obj) for objects that are unknown, or do not wish to serialise.
The following objects won't be passed to this method: BHoMObject, IObject, uuid.UUID, pandas.DatetimeIndex, numpy.ndarray, pandas.Timestamp, pandas.Series, pathlib.Path
"""
return super(type(self), self).default(obj) #fallback to default json decoder (ValueError) if object is not a BHoMObject or common serialisable type.

class IObject:
"""More generic version of BHoMObject, for non-native objects serialised by the BHoM serialiser, but do not inherit from BHoMObject."""
_t: str
_bhom_version: str

def __init__(
self,
_t: str,
_bhom_version: str = None,
**kwargs
) -> 'IObject':
self._t = _t
self._bhom_version = _bhom_version

#set properties with reflection.
for kwarg in kwargs:
setattr(self, kwarg, kwargs[kwarg])

def __repr__(self) -> str:
return f"{type(self).__name__} of type {self._t}, version: '{getattr(self, '_bhom_version', 'Unknown')}'"

def __eq__(self, other) -> bool:
if not isinstance(other, IObject):
return False

if self._t != other._t:
return False

vself = vars(self).copy()
vother = vars(other).copy()

#ignore these properties when comparing by property.
ignore = ["_bhom_version"]
_ = [(vself.pop(p, None), vother.pop(p, None)) for p in ignore]

return vself == vother

@classmethod
def from_json(cls, j: str, decoder_class: type = BHoMJSONDecoder) -> 'IObject':
obj = json.loads(j, cls=decoder_class)

if not isinstance(obj, cls): #this only tests that the top level object was deserialised correctly, if there are problems with deep properties, change the CONSOLE_LOGGER log level to debug.
raise TypeError("The object provided does not deserialise to a valid BHoM object.")

return obj

def to_json(self, encoder_class: type = BHoMJSONEncoder) -> str:
return json.dumps(self, cls=encoder_class)

@classmethod
def from_dict(cls, d: dict) -> 'IObject':
try:
return cls(**d) #should be valid as long as the dictionary has all necessary entries.
except ArgumentError as ae:
raise ArgumentError("Input dictionary was missing some required arguments, see traceback for more information.") from ae

def to_dict(self, encoder_class: type = BHoMJSONEncoder) -> dict:
"""Convert this IObject to a dictionary via deep copying vars(self)."""
return copy.deepcopy(vars(self))

class BHoMObject(IObject):
name: str
bhom_guid: uuid.UUID
tags: List[str]
fragments: List[Dict[str, object]]
custom_data: Dict[str, object]

def __init__(
self,
_t: str, #don't make default, as subclasses of this class should set this with super().__init__
name: str = "",
bhom_guid: uuid.UUID = uuid.uuid4(),
tags: List[str] = [],
fragments: List[Dict[str, object]] = [],
custom_data: Dict[str, object] = {},
_bhom_version: str = None,
**kwargs
) -> 'BHoMObject':
self._t = _t
self.name = name
self.bhom_guid = bhom_guid
self.fragments = fragments
self.tags = tags
self.custom_data = custom_data
self._bhom_version = _bhom_version

#set non-CustomData properties with reflection.
for kwarg in kwargs:
setattr(self, kwarg, kwargs[kwarg])

def __repr__(self) -> str:
return f"{type(self).__name__} of type {self._t}, name: '{self.name}', version: '{getattr(self, '_bhom_version', 'Unknown')}', id: '{self.bhom_guid}'"

def __eq__(self, other) -> bool:
if not isinstance(other, BHoMObject):
return False

if self._t != other._t:
return False

vself = vars(self).copy()
vother = vars(other).copy()

#ignore these properties when comparing by property.
ignore = ["bhom_guid", "_bhom_version"]
_ = [(vself.pop(p, None), vother.pop(p, None)) for p in ignore]

return vself == vother

@classmethod
def from_json(cls, j: str, decoder_class: type = BHoMJSONDecoder):
obj = json.loads(j, cls=decoder_class)

if isinstance(obj, list):
CONSOLE_LOGGER.warning("The root element of the JSON provided was a list, assuming that the first item is the desired object. If you intended to deserialise a list, please use the `<cls>.from_json_array(j)` method instead.")
obj = obj[0]

if issubclass(cls, BHoMObject) and cls != BHoMObject:
obj = cls._from_bhom_object(obj)

if not isinstance(obj, cls): #this only tests that the top level object was deserialised correctly, if there are problems with deep properties, change the CONSOLE_LOGGER log level to debug.
raise TypeError("The object provided does not deserialise to a valid BHoM object.")

return obj

@classmethod
def from_json_array(cls, j: str, decoder_class: type = BHoMJSONDecoder):
objs = json.loads(j, cls=decoder_class)

if not isinstance(objs, list):
raise TypeError("The root element of the JSON provided was not a JSON array. Perhaps you intended to use `<cls>.from_json(j)` instead?")

out = []
for obj in objs:
if issubclass(cls, BHoMObject) and cls != BHoMObject:
out.append(cls._from_bhom_object(obj))
continue
out.append(obj)

return out

@classmethod
def _from_bhom_object(cls, o: 'BHoMObject'):
return cls(**vars(o).copy()) #assuming that the sub class is correctly set up, then this should work
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
from ._bhom_callable_decorator import bhom_wrapper
Loading