Add function "is" (wip)

http://hl7.org/fhirpath/#is-type-specifier

fhirpathpy/engine/__init__.py

@@ -1,6 +1,7 @@
 import json
 import numbers
 import fhirpathpy.engine.util as util
+from fhirpathpy.engine.nodes import TypeInfo
 from fhirpathpy.engine.evaluators import evaluators
 from fhirpathpy.engine.invocations import invocations
 
@@ -89,6 +90,21 @@ def doInvoke(ctx, fn_name, data, raw_params):
     return util.arraify(res)
 
 
+def type_specifier(ctx, parent_data, node):
+    identifiers = node["text"].replace("`", "").split(".")
+    namespace = None
+    name = None
+
+    if len(identifiers) == 2:
+        namespace, name = identifiers
+    elif len(identifiers) == 1:
+        (name,) = identifiers
+    else:
+        raise Exception(f"Expected TypeSpecifier node, got {node}")
+
+    return TypeInfo(name=name, namespace=namespace)
+
+
 param_check_table = {
     "Integer": check_integer_param,
     "Number": check_number_param,
@@ -98,7 +114,6 @@ def doInvoke(ctx, fn_name, data, raw_params):
 
 
 def make_param(ctx, parentData, node_type, param):
-
     if node_type == "Expr":
 
         def func(data):
@@ -108,7 +123,7 @@ def func(data):
         return func
 
     if node_type == "AnyAtRoot":
-        ctx["$this"] = ctx["$this"] if "$this" in ctx else ctx['dataRoot']
+        ctx["$this"] = ctx["$this"] if "$this" in ctx else ctx["dataRoot"]
         return do_eval(ctx, ctx["dataRoot"], param)
 
     if node_type == "Identifier":
@@ -117,6 +132,9 @@ def func(data):
 
         raise Exception("Expected identifier node, got " + json.dumps(param))
 
+    if node_type == "TypeSpecifier":
+        return type_specifier(ctx, parentData, param)
+
     ctx["$this"] = parentData
     res = do_eval(ctx, parentData, param)
 

fhirpathpy/engine/invocations/__init__.py

@@ -10,6 +10,7 @@
 import fhirpathpy.engine.invocations.equality as equality
 import fhirpathpy.engine.invocations.logic as logic
 import fhirpathpy.engine.invocations.datetime as datetime
+import fhirpathpy.engine.invocations.types as types
 
 invocations = {
     "empty": {"fn": existence.empty_fn},
@@ -33,6 +34,7 @@
     "first": {"fn": filtering.first_fn},
     "last": {"fn": filtering.last_fn},
     "ofType": {"fn": filtering.of_type_fn, "arity": {1: ["Identifier"]}},
+    "is": {"fn": types.is_fn, "arity": {1: ["TypeSpecifier"]}},
     "tail": {"fn": filtering.tail_fn},
     "take": {"fn": filtering.take_fn, "arity": {1: ["Integer"]}},
     "skip": {"fn": filtering.skip_fn, "arity": {1: ["Integer"]}},

fhirpathpy/engine/invocations/types.py

@@ -0,0 +1,20 @@
+from fhirpathpy.engine.nodes import TypeInfo
+
+def type_fn(ctx, coll):
+    return [TypeInfo.from_value(value) for value in coll]
+
+
+def is_fn(ctx, coll, type_info):
+    if not coll:
+        return []
+    if len(coll) > 1:
+        raise ValueError(f"Expected singleton on left side of 'is', got {coll}")
+    return TypeInfo.from_value(coll[0]).is_(type_info)
+
+
+def as_fn(ctx, coll, type_info):
+    if not coll:
+        return []
+    if len(coll) > 1:
+        raise ValueError(f"Expected singleton on left side of 'as', got {coll}")
+    return coll if TypeInfo.from_value(coll[0]).is_(type_info) else []

fhirpathpy/engine/nodes.py

@@ -208,7 +208,7 @@ def conv_unit_to(fromUnit, value, toUnit):
             converted_value = (Decimal(from_lbs_kg_magnitude) * value) / Decimal(
                 to_lbs_kg_magnitude
             )
-            rounded_value = converted_value.quantize(Decimal('1.'), rounding=ROUND_UP)
+            rounded_value = converted_value.quantize(Decimal("1."), rounding=ROUND_UP)
             return FP_Quantity(rounded_value, toUnit)
 
         return None
@@ -520,3 +520,58 @@ def create_node(data, path=None):
         if isinstance(data, ResourceNode):
             return data
         return ResourceNode(data, path)
+
+
+class TypeInfo:
+    model = None
+    System = "System"
+    FHIR = "FHIR"
+
+    def __init__(self, name, namespace):
+        self.name = name
+        self.namespace = namespace
+
+    @staticmethod
+    def is_type(type_name, super_type):
+        while type_name:
+            if type_name == super_type:
+                return True
+            type_name = TypeInfo.model.type2Parent.get(type_name)
+        return False
+
+    def is_(self, other):
+        if isinstance(other, TypeInfo) and (
+            not self.namespace or not other.namespace or self.namespace == other.namespace
+        ):
+            if TypeInfo.model and (not self.namespace or self.namespace == TypeInfo.FHIR):
+                return TypeInfo.is_type(self.name, other.name)
+            else:
+                return self.name == other.name
+        return False
+
+    @staticmethod
+    def create_by_value_in_namespace(namespace, value):
+        name = type(value).__name__
+
+        if isinstance(value, int):
+            name = "integer"
+        elif isinstance(value, float) or isinstance(value, Decimal):
+            name = "decimal"
+        elif isinstance(value, FP_DateTime):
+            name = "dateTime"
+        elif isinstance(value, FP_Time):
+            name = "time"
+        elif isinstance(value, FP_Quantity):
+            name = "Quantity"
+
+        if namespace == TypeInfo.System:
+            name = name.capitalize()
+
+        return TypeInfo(name, namespace)
+
+    @staticmethod
+    def from_value(value):
+        if isinstance(value, ResourceNode):
+            return value.get_type_info()
+        else:
+            return TypeInfo.create_by_value_in_namespace(TypeInfo.System, value)