Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __add__(self, stage_or_pipeline):
assert isinstance(stage_or_pipeline, (StageSpec, PipelineSpec)), "Expected StageSpec or PipelineSpec"
output = self.__class__(self)
if isinstance(stage_or_pipeline, StageSpec):
output.add_stage(stage_or_pipeline)
elif isinstance(stage_or_pipeline, PipelineSpec):
for stage in stage_or_pipeline.stages:
output.add_stage(stage)
return output
def __str__(self):
return json.dumps(self.spec, indent=4)
def __add__(self, other):
return self.pipeline + other
def __dir__(self):
extra_keys = [e["name"][len(self.key):] for e in PDAL_DRIVERS if e["name"].startswith(self.key)] + ["auto"]
return super().__dir__() + [e for e in extra_keys if len(e) > 0]
def execute(self):
return self.pipeline.execute()
readers = StageSpec("readers")
filters = StageSpec("filters")
writers = StageSpec("writers")
class PipelineSpec(object):
stages = []
def __init__(self, other=None):
if other is not None:
self.stages = copy.copy(other.stages)
@property
def spec(self):
"""
Return a `dict` containing the pdal pipeline suitable for dumping to json
"""
return {
return json.dumps(self.spec, indent=4)
def __add__(self, other):
return self.pipeline + other
def __dir__(self):
extra_keys = [e["name"][len(self.key):] for e in PDAL_DRIVERS if e["name"].startswith(self.key)] + ["auto"]
return super().__dir__() + [e for e in extra_keys if len(e) > 0]
def execute(self):
return self.pipeline.execute()
readers = StageSpec("readers")
filters = StageSpec("filters")
writers = StageSpec("writers")
class PipelineSpec(object):
stages = []
def __init__(self, other=None):
if other is not None:
self.stages = copy.copy(other.stages)
@property
def spec(self):
"""
Return a `dict` containing the pdal pipeline suitable for dumping to json
"""
return {
"pipeline": [stage.spec for stage in self.stages]
def __str__(self):
return json.dumps(self.spec, indent=4)
def __add__(self, other):
return self.pipeline + other
def __dir__(self):
extra_keys = [e["name"][len(self.key):] for e in PDAL_DRIVERS if e["name"].startswith(self.key)] + ["auto"]
return super().__dir__() + [e for e in extra_keys if len(e) > 0]
def execute(self):
return self.pipeline.execute()
readers = StageSpec("readers")
filters = StageSpec("filters")
writers = StageSpec("writers")
class PipelineSpec(object):
stages = []
def __init__(self, other=None):
if other is not None:
self.stages = copy.copy(other.stages)
@property
def spec(self):
"""
Return a `dict` containing the pdal pipeline suitable for dumping to json
"""