from owrx.form import Input, CheckboxInput, DropdownInput, DropdownEnum, TextInput from owrx.soapy import SoapySettings from functools import reduce from operator import and_ class GainInput(Input): def __init__(self, id, label, has_agc, gain_stages=None): super().__init__(id, label) self.has_agc = has_agc self.gain_stages = gain_stages def render_input(self, value, errors): # TODO display errors try: display_value = float(value) except (ValueError, TypeError): display_value = "0.0" return """
{stageoption}
""".format( id=self.id, classes=self.input_classes(errors), value=display_value, label=self.label, options=self.render_options(value), stageoption="" if self.gain_stages is None else self.render_stage_option(value, errors), disabled="disabled" if self.disabled else "", ) def render_options(self, value): options = [] if self.has_agc: options.append(("auto", "Enable hardware AGC")) options.append(("manual", "Specify manual gain")), if self.gain_stages: options.append(("stages", "Specify gain stages individually")) mode = self.getMode(value) return "".join( """ """.format( value=v[0], text=v[1], selected="selected" if mode == v[0] else "" ) for v in options ) def getMode(self, value): if value is None: return "auto" if self.has_agc else "manual" if value == "auto": return "auto" try: float(value) return "manual" except (ValueError, TypeError): pass return "stages" def render_stage_option(self, value, errors): try: value_dict = {k: v for item in SoapySettings.parse(value) for k, v in item.items()} except (AttributeError, ValueError): value_dict = {} return """ """.format( inputs="".join( """
""".format( id=self.id, stage=stage, value=value_dict[stage] if stage in value_dict else "", classes=self.input_classes(errors), disabled="disabled" if self.disabled else "", ) for stage in self.gain_stages ) ) def parse(self, data): def getStageValue(stage): input_id = "{id}-{stage}".format(id=self.id, stage=stage) if input_id in data: return data[input_id][0] else: return None select_id = "{id}-select".format(id=self.id) if select_id in data: if self.has_agc and data[select_id][0] == "auto": return {self.id: "auto"} if data[select_id][0] == "manual": input_id = "{id}-manual".format(id=self.id) value = 0.0 if input_id in data: try: value = float(data[input_id][0]) except ValueError: pass return {self.id: value} if self.gain_stages is not None and data[select_id][0] == "stages": settings_dict = [{s: getStageValue(s)} for s in self.gain_stages] # filter out empty ones settings_dict = [s for s in settings_dict if next(iter(s.values()))] return {self.id: SoapySettings.encode(settings_dict)} return {} class BiasTeeInput(CheckboxInput): def __init__(self): super().__init__("bias_tee", "Enable Bias-Tee power supply") class DirectSamplingOptions(DropdownEnum): DIRECT_SAMPLING_OFF = (0, "Off") DIRECT_SAMPLING_I = (1, "Direct Sampling (I branch)") DIRECT_SAMPLING_Q = (2, "Direct Sampling (Q branch)") def __new__(cls, *args, **kwargs): value, description = args obj = object.__new__(cls) obj._value_ = value obj.description = description return obj def __str__(self): return self.description class DirectSamplingInput(DropdownInput): def __init__(self): super().__init__( "direct_sampling", "Direct Sampling", DirectSamplingOptions, ) class RemoteInput(TextInput): def __init__(self): super().__init__( "remote", "Remote IP and Port", infotext="Remote hostname or IP and port to connect to. Format = IP:Port" ) class SchedulerInput(Input): def __init__(self, id, label): super().__init__(id, label) self.profiles = {} def render(self, config, errors): if "profiles" in config: self.profiles = config["profiles"] return super().render(config, errors) def render_profiles_select(self, value, errors, config_key, stage, extra_classes=""): stage_value = "" if value and "schedule" in value and config_key in value["schedule"]: stage_value = value["schedule"][config_key] return """ """.format( id="{}-{}".format(self.id, stage), classes=self.input_classes(errors), extra_classes=extra_classes, disabled="disabled" if self.disabled else "", options="".join( """ """.format( id=p_id, name=p["name"], selected="selected" if stage_value == p_id else "", ) for p_id, p in self.profiles.items() ), ) def render_static_entires(self, value, errors): def render_time_inputs(v): values = ["{}:{}".format(x[0:2], x[2:4]) for x in [v[0:4], v[5:9]]] return '
-
'.join( """ """.format( id="{}-{}-{}".format(self.id, "time", "start" if i == 0 else "end"), classes=self.input_classes(errors), disabled="disabled" if self.disabled else "", value=v, ) for i, v in enumerate(values) ) schedule = {"0000-0000": ""} if value is not None and value and "schedule" in value and "type" in value and value["type"] == "static": schedule = value["schedule"] rows = "".join( """
{time_inputs} {select}
""".format( time_inputs=render_time_inputs(slot), select=self.render_profiles_select(value, errors, slot, "profile"), ) for slot, entry in schedule.items() ) return """ {rows}
""".format( rows=rows, time_inputs=render_time_inputs("0000-0000"), select=self.render_profiles_select("", errors, "0000-0000", "profile"), ) def render_daylight_entries(self, value, errors): return "".join( """
{select}
""".format( name=name, select=self.render_profiles_select(value, errors, stage, stage, extra_classes="col-9"), ) for stage, name in [("day", "Day"), ("night", "Night"), ("greyline", "Greyline")] ) def render_input(self, value, errors): return """
""".format( id=self.id, classes=self.input_classes(errors), disabled="disabled" if self.disabled else "", options=self.render_options(value), entries=self.render_static_entires(value, errors), stages=self.render_daylight_entries(value, errors), ) def _get_mode(self, value): if value is not None and "type" in value: return value["type"] return "" def render_options(self, value): options = [ ("static", "Static scheduler"), ("daylight", "Daylight scheduler"), ] mode = self._get_mode(value) return "".join( """ """.format( value=value, name=name, selected="selected" if mode == value else "" ) for value, name in options ) def parse(self, data): def getStageValue(stage): input_id = "{id}-{stage}".format(id=self.id, stage=stage) if input_id in data: return data[input_id][0] else: return None select_id = "{id}-select".format(id=self.id) if select_id in data: if data[select_id][0] == "static": keys = ["{}-{}".format(self.id, x) for x in ["time-start", "time-end", "profile"]] keys_present = reduce(and_, [key in data for key in keys], True) if not keys_present: return {} lists = [data[key] for key in keys] settings_dict = { "{}{}-{}{}".format(start[0:2], start[3:5], end[0:2], end[3:5]): profile for start, end, profile in zip(*lists) } return {self.id: {"type": "static", "schedule": settings_dict}} elif data[select_id][0] == "daylight": settings_dict = {s: getStageValue(s) for s in ["day", "night", "greyline"]} # filter out empty ones settings_dict = {s: v for s, v in settings_dict.items() if v} return {self.id: {"type": "daylight", "schedule": settings_dict}} return {} class WaterfallLevelsInput(Input): def render_input(self, value, errors): # TODO display errors return """
{inputs}
""".format( id=self.id, inputs="".join( """
dBFS
""".format( id=self.id, name=name, label=label, value=value[name] if value and name in value else "0", classes=self.input_classes(errors), disabled="disabled" if self.disabled else "", ) for name, label in [("min", "Minimum"), ("max", "Maximum")] ) ) def parse(self, data): def getValue(name): key = "{}-{}".format(self.id, name) if key in data: return {name: float(data[key][0])} raise KeyError("waterfall key not found") try: return {self.id: {k: v for name in ["min", "max"] for k, v in getValue(name).items()}} except KeyError: return {}