X-Git-Url: https://git.ao2.it/SaveMySugar/python3-savemysugar.git/blobdiff_plain/bbe279dcaf1fd1f0fe68dfe2dce0ce566ce09f73..3baafa27ab24bf277e27371a6205884651abb964:/src/savemysugar/CallDistanceTransceiver.py?ds=sidebyside diff --git a/src/savemysugar/CallDistanceTransceiver.py b/src/savemysugar/CallDistanceTransceiver.py index 29c3fb8..f04c533 100755 --- a/src/savemysugar/CallDistanceTransceiver.py +++ b/src/savemysugar/CallDistanceTransceiver.py @@ -32,22 +32,22 @@ import time class SymbolTime(object): """ In theory the symbol distance (the distance to discriminate symbols) can - be arbitrary, and it's only bounded below by the stability of the pulse + be arbitrary, and it's only bounded below by the stability of the period time, but in practice it can be necessary to wait a predefined minimum - amount of time between pulses because of technological limits of the + amount of time between periods because of technological limits of the transmitting devices, we call this time the "minimum inter-symbol distance". """ # pylint: disable=too-few-public-methods - def __init__(self, pulse_min, pulse_max, multiplier, + def __init__(self, period_min, period_max, multiplier, min_inter_symbol_distance=0.0): assert multiplier >= 0 - if (pulse_min == pulse_max) and (min_inter_symbol_distance == 0): - raise ValueError("If (pulse_min == pulse_max) a non-zero", + if (period_min == period_max) and (min_inter_symbol_distance == 0): + raise ValueError("If (period_min == period_max) a non-zero", "inter-symbol distance MUST be specified") - symbol_distance = 2 * (pulse_max - pulse_min) + symbol_distance = 2 * (period_max - period_min) if symbol_distance == 0: symbol_distance = min_inter_symbol_distance @@ -56,14 +56,116 @@ class SymbolTime(object): self.dist = min_inter_symbol_distance + symbol_distance * multiplier # The minimum time which represents the symbol at the receiving end. - self.min = min_inter_symbol_distance + pulse_min + \ + self.min = min_inter_symbol_distance + period_min + \ symbol_distance * multiplier # The maximum time which represents the symbol at the receiving end - self.max = min_inter_symbol_distance + pulse_min + \ + self.max = min_inter_symbol_distance + period_min + \ symbol_distance * (multiplier + 1) +class MorseDistanceModulator(object): + def __init__(self, period_min, period_max, pulse_min, pulse_max, + inter_symbol_distance): + self.set_parameters(period_min, period_max, + pulse_min, pulse_max, + inter_symbol_distance) + + def set_parameters(self, period_min, period_max, pulse_min, pulse_max, + inter_symbol_distance): + self.period_min = period_min + self.period_max = period_max + self.pulse_min = pulse_min + self.pulse_max = pulse_max + self.inter_symbol_distance = inter_symbol_distance + + self.dot_time = SymbolTime(period_min, + period_max, 0, + inter_symbol_distance) + self.dash_time = SymbolTime(period_min, + period_max, 1, + inter_symbol_distance) + self.signalspace_time = SymbolTime(period_min, + period_max, 2, + inter_symbol_distance) + self.wordspace_time = SymbolTime(period_min, + period_max, 3, + inter_symbol_distance) + self.eom_time = SymbolTime(period_min, + period_max, 4, + inter_symbol_distance) + + def symbol_to_distance(self, symbol): + if symbol == ".": + return self.dot_time.dist + elif symbol == "-": + return self.dash_time.dist + elif symbol == " ": + return self.signalspace_time.dist + elif symbol == "/" or symbol == " / ": + return self.wordspace_time.dist + elif symbol == "EOM": + return self.eom_time.dist + + raise ValueError("Unexpected symbol %s" % symbol) + + def is_same_period(self, distance): + return distance > self.pulse_min and distance <= self.pulse_max + + def distance_to_symbol(self, distance): + if distance > self.dot_time.min and \ + distance <= self.dot_time.max: + return "." + + if distance > self.dash_time.min and \ + distance <= self.dash_time.max: + return "-" + + if distance > self.signalspace_time.min and \ + distance <= self.signalspace_time.max: + return " " + + if distance > self.wordspace_time.min and \ + distance <= self.wordspace_time.max: + return "/" + + if distance > self.eom_time.min: + return "EOM" + + raise ValueError("Unexpected distance %.2f" % distance) + + def modulate(self, morse): + signals = morse.split(' ') + distances = [] + for i, signal in enumerate(signals): + for symbol in signal: + distances.append(self.symbol_to_distance(symbol)) + + # Transmit a signal separator only when strictly necessary. + # + # Avoid it in these cases: + # - after the last symbol, because EOM is going to ne transmitted + # anyway and that will mark the end of the last symbol. + # - between words, because the word separator act as a symbol + # separator too. + if i != len(signals) - 1 and signals[i + 1] != "/" and signal != "/": + distances.append(self.symbol_to_distance(" ")) + + distances.append(self.symbol_to_distance("EOM")) + + # Since the Morse signals are encoded in the distance between calls, an + # extra call is needed in order for receiver actually get the EOM and + # see that the transmission has terminated. + distances.append(0) + + return distances + + +def log_symbol(distance, symbol, extra_info=""): + logging.info("distance: %.2f Received \"%s\"%s", distance, symbol, + extra_info) + + class CallDistanceTransceiver(object): """Transmit Morse messages using the distance between calls. @@ -119,14 +221,8 @@ class CallDistanceTransceiver(object): self.modem = modem self.translator = MorseTranslator() - self.destination_number = "" - - self.call_setup_time_min = call_setup_time_min self.call_setup_time_max = call_setup_time_max - self.ring_time_min = ring_time_min - self.ring_time_max = ring_time_max - if add_inter_call_distance: # Analog modems don't like to dial a new call immediately after # they hung up the previous one; an extra delay of about one @@ -135,66 +231,56 @@ class CallDistanceTransceiver(object): else: inter_symbol_distance = 0 - self.dot_time = SymbolTime(call_setup_time_min, - call_setup_time_max, 0, - inter_symbol_distance) - self.dash_time = SymbolTime(call_setup_time_min, - call_setup_time_max, 1, - inter_symbol_distance) - self.signalspace_time = SymbolTime(call_setup_time_min, - call_setup_time_max, 2, - inter_symbol_distance) - self.wordspace_time = SymbolTime(call_setup_time_min, - call_setup_time_max, 3, - inter_symbol_distance) - self.eom_time = SymbolTime(call_setup_time_min, - call_setup_time_max, 4, - inter_symbol_distance) + self.modulator = MorseDistanceModulator(call_setup_time_min, + call_setup_time_max, + ring_time_min, + ring_time_max, + inter_symbol_distance) logging.debug("call setup time between %.2f and %.2f " "--------- dot transmit time: %.2f + %.2f " "receive time: between %.2f and %.2f", - self.call_setup_time_min, - self.call_setup_time_max, - self.call_setup_time_max, - self.dot_time.dist, - self.dot_time.min, - self.dot_time.max) + self.modulator.period_min, + self.modulator.period_max, + self.modulator.period_max, + self.modulator.dot_time.dist, + self.modulator.dot_time.min, + self.modulator.dot_time.max) logging.debug("call setup time between %.2f and %.2f " "-------- dash transmit time: %.2f + %.2f " "receive time: between %.2f and %.2f", - self.call_setup_time_min, - self.call_setup_time_max, - self.call_setup_time_max, - self.dash_time.dist, - self.dash_time.min, - self.dash_time.max) + self.modulator.period_min, + self.modulator.period_max, + self.modulator.period_max, + self.modulator.dash_time.dist, + self.modulator.dash_time.min, + self.modulator.dash_time.max) logging.debug("call setup time between %.2f and %.2f " "- signalspace transmit time: %.2f + %.2f " "receive time: between %.2f and %.2f", - self.call_setup_time_min, - self.call_setup_time_max, - self.call_setup_time_max, - self.signalspace_time.dist, - self.signalspace_time.min, - self.signalspace_time.max) + self.modulator.period_min, + self.modulator.period_max, + self.modulator.period_max, + self.modulator.signalspace_time.dist, + self.modulator.signalspace_time.min, + self.modulator.signalspace_time.max) logging.debug("call setup time between %.2f and %.2f " "--- wordspace transmit time: %.2f + %.2f " "receive time: between %.2f and %.2f", - self.call_setup_time_min, - self.call_setup_time_max, - self.call_setup_time_max, - self.wordspace_time.dist, - self.wordspace_time.min, - self.wordspace_time.max) + self.modulator.period_min, + self.modulator.period_max, + self.modulator.period_max, + self.modulator.wordspace_time.dist, + self.modulator.wordspace_time.min, + self.modulator.wordspace_time.max) logging.debug("call setup time between %.2f and %.2f " "--------- EOM transmit time: %.2f + %.2f " "receive time: between %.2f and inf", - self.call_setup_time_min, - self.call_setup_time_max, - self.call_setup_time_max, - self.eom_time.dist, - self.eom_time.min) + self.modulator.period_min, + self.modulator.period_max, + self.modulator.period_max, + self.modulator.eom_time.dist, + self.modulator.eom_time.min) self.previous_ring_time = -1 self.previous_call_time = -1 @@ -204,17 +290,13 @@ class CallDistanceTransceiver(object): self.end_of_message = False - def log_symbol(self, distance, symbol, extra_info=""): - logging.info("distance: %.2f Received \"%s\"%s", distance, symbol, - extra_info) - - def receive_character(self): + def receive_symbol(self): current_ring_time = time.time() if self.previous_ring_time == -1: self.previous_ring_time = current_ring_time self.previous_call_time = current_ring_time - self.log_symbol(0, "", "(The very first ring)") + log_symbol(0, "", "(The very first ring)") return ring_distance = current_ring_time - self.previous_ring_time @@ -222,8 +304,7 @@ class CallDistanceTransceiver(object): self.previous_ring_time = current_ring_time # Ignore multiple rings in the same call - if ring_distance > self.ring_time_min and \ - ring_distance <= self.ring_time_max: + if self.modulator.is_same_period(ring_distance): logging.debug("multiple rings in the same call, distance: %.2f", ring_distance) return @@ -231,51 +312,36 @@ class CallDistanceTransceiver(object): call_distance = current_ring_time - self.previous_call_time self.previous_call_time = current_ring_time - if call_distance > self.dot_time.min and \ - call_distance <= self.dot_time.max: - self.log_symbol(call_distance, '.') - self.morse_message += "." - return - - if call_distance > self.dash_time.min and \ - call_distance <= self.dash_time.max: - self.log_symbol(call_distance, '-') - self.morse_message += "-" + try: + symbol = self.modulator.distance_to_symbol(call_distance) + except ValueError as err: + logging.error("%s", err) + logging.error("Check the transmitter and receiver parameters") return - if call_distance > self.signalspace_time.min and \ - call_distance <= self.signalspace_time.max: + extra_info = "" + if symbol in [" ", "/", "EOM"]: signal = self.morse_message.strip().split(' ')[-1] character = self.translator.signal_to_character(signal) - self.log_symbol(call_distance, ' ', " got \"%s\"" % character) - self.morse_message += " " - return + extra_info = " got \"%s\"" % character - if call_distance > self.wordspace_time.min and \ - call_distance <= self.wordspace_time.max: - signal = self.morse_message.strip().split(' ')[-1] - character = self.translator.signal_to_character(signal) - self.log_symbol(call_distance, '/', " got \"%s\"" % character) - self.morse_message += " / " - return + log_symbol(call_distance, symbol, extra_info) - if call_distance > self.eom_time.min: - signal = self.morse_message.strip().split(' ')[-1] - character = self.translator.signal_to_character(signal) - self.log_symbol(call_distance, 'EOM', " got \"%s\"" % character) + if symbol != "EOM": + # Add spaces around the wordspace symbol to make it easier to split + # the Morse message in symbols later on + if symbol == "/": + symbol = " / " + self.morse_message += symbol + else: self.end_of_message = True self.previous_ring_time = -1 self.previous_call_time = -1 - return - - # if the code made it up to here, something fishy is going on - logging.error("Unexpected distance: %.2f", call_distance) - logging.error("Check the transmitter and receiver parameters") def receive_loop(self): while not self.end_of_message: self.modem.get_response("RING") - self.receive_character() + self.receive_symbol() logging.debug("Current message: %s", self.morse_message) self.end_of_message = False @@ -285,22 +351,7 @@ class CallDistanceTransceiver(object): def get_text(self): return self.text_message - def transmit_symbol(self, symbol): - if symbol == ".": - sleep_time = self.dot_time.dist - elif symbol == "-": - sleep_time = self.dash_time.dist - elif symbol == " ": - sleep_time = self.signalspace_time.dist - elif symbol == "/": - sleep_time = self.wordspace_time.dist - elif symbol == "EOM": - sleep_time = self.eom_time.dist - elif symbol is None: - # To terminate the transmission just call and hangup, with no extra - # distance - sleep_time = 0 - + def transmit_symbol(self, destination_number, symbol, sleep_time): logging.info("Dial and wait %.2f = %.2f + %.2f seconds " "(transmitting '%s')", self.call_setup_time_max + sleep_time, @@ -311,77 +362,43 @@ class CallDistanceTransceiver(object): # Dial, then wait self.call_setup_time_max to make sure the receiver # gets at least one RING, and then hangup and sleep the time needed to # transmit a symbol. - self.modem.send_command("ATDT" + self.destination_number + ";") + self.modem.send_command("ATDT" + destination_number + ";") time.sleep(self.call_setup_time_max) self.modem.send_command("ATH") self.modem.get_response() time.sleep(sleep_time) - def transmit_signal(self, signal): - logging.debug("Transmitting signal: %s", signal) - for symbol in signal: - self.transmit_symbol(symbol) - def transmit(self, message, destination_number): - self.destination_number = destination_number - morse_message = self.translator.text_to_morse(message) - signals = morse_message.split() + distances = self.modulator.modulate(morse_message) logging.debug("Starting the transmission") - for i, signal in enumerate(signals): - logging.debug("Transmitting '%s' as '%s'", message[i], signal) - self.transmit_signal(signal) - - # Transmit a signal separator only when strictly necessary: - # - after the last symbol, we are going to transmit an EOM - # anyway, and that will mark the end of the last symbol. - # - between words the word separator act as a symbol separator - # too. - if i != len(signals) - 1 and signals[i + 1] != "/" and signal != "/": - self.transmit_symbol(" ") - - self.transmit_symbol("EOM") + for i, distance in enumerate(distances): + # Use 'None' for the last call + if i == len(distances) - 1: + symbol = None + else: + total_sleep_time = self.call_setup_time_max + distance + symbol = self.modulator.distance_to_symbol(total_sleep_time) - # Since the Morse signals are encoded in the distance between calls, an - # extra call is needed in order for receiver actually get the EOM and - # see that the transmission has terminated. - self.transmit_symbol(None) + self.transmit_symbol(destination_number, symbol, distance) def estimate_transmit_duration(self, message): morsemessage = self.translator.text_to_morse(message) - signals = morsemessage.split() + logging.debug(morsemessage) - logging.debug(signals) + distances = self.modulator.modulate(morsemessage) transmitting_time = 0 - for i, signal in enumerate(signals): - logging.debug("signal: %s", signal) - - for symbol in signal: - transmitting_time += self.call_setup_time_max - if symbol == ".": - transmitting_time += self.dot_time.dist - elif symbol == "-": - transmitting_time += self.dash_time.dist - elif symbol == "/": - transmitting_time += self.wordspace_time.dist - - if i != len(signals) - 1 and signals[i + 1] != "/" and symbol != "/": - transmitting_time += self.call_setup_time_max - transmitting_time += self.signalspace_time.dist - - transmitting_time += self.call_setup_time_max - transmitting_time += self.eom_time.dist - - # The final call needed for the receiver to get the EOM - transmitting_time += self.call_setup_time_max + for distance in distances: + transmitting_time += self.call_setup_time_max + transmitting_time += distance logging.debug("Estimated transmitting time: %.2f seconds", transmitting_time) -def test_send_receive(): +def test_transmit_receive(): logging.basicConfig(level=logging.DEBUG) call_setup_time_min = 0 call_setup_time_max = 0.01 @@ -404,24 +421,27 @@ def test_send_receive(): def send_command(self, command): pass - def get_response(self, response): + def get_response(self, response=None): # pylint: disable=unused-argument - setup_time = random.uniform(self.transceiver.call_setup_time_min, - self.transceiver.call_setup_time_max) + setup_time = random.uniform(self.transceiver.modulator.period_min, + self.transceiver.modulator.period_max) if self.ring_count == 0: # dummy ring pass elif self.ring_count == 1: # received a '.' - time.sleep(setup_time + self.transceiver.dot_time.dist) + dot_time = self.transceiver.modulator.dot_time.dist + time.sleep(setup_time + dot_time) elif self.ring_count == 2: # received a '/' - time.sleep(setup_time + self.transceiver.wordspace_time.dist) + wordspace_time = self.transceiver.modulator.wordspace_time.dist + time.sleep(setup_time + wordspace_time) else: # received an 'EOM' - time.sleep(setup_time + self.transceiver.eom_time.dist) + eom_time = self.transceiver.modulator.eom_time.dist + time.sleep(setup_time + eom_time) self.ring_count += 1 self.ring_count %= 4 @@ -434,13 +454,15 @@ def test_send_receive(): modem.transceiver = xcv + xcv.transmit("CODEX PARIS", "0") + while True: - xcv.receive_loop() modem.ring_count = 0 + xcv.receive_loop() print() print("Message received!") print("\"%s\"" % xcv.get_text(), flush=True) if __name__ == "__main__": - test_send_receive() + test_transmit_receive()