<commit_before>"""
Driver class for Hagisonic Stargazer, with no ROS dependencies. 
"""
from serial import Serial
from collections import deque
import re
import yaml
import time
import logging
import rospy
import numpy as np
from threading import Thread, Event
from tf import transformations

# STX: char that represents the start of a properly formed message
STX = '~'
# ETX: char that represents the end of a properly formed message
ETX = '`'
# DELIM: char that splits data
DELIM = '|'
# CMD: char that indicates command
CMD = '#'
# CMD: char that indicates command
RESPONSE = '!'
# RESULT: char that indicates that the message contains result data
RESULT = '^'


class StarGazer(object):
    def __init__(self, device, marker_map, callback_global=None, callback_local=None):
        """
        Connect to a Hagisonic StarGazer device and receive poses.

        device:          The device location for the serial connection. 

        marker_map:      dictionary of marker transforms, formatted:
                         {marker_id: (4,4) matrix}

        callback_global: will be called whenever a new pose is received from the
                         Stargazer, will be called with (n,4,4) matrix of poses
                         of the location of the Stargazer in the global frame.
                         These are computed from marker_map. 

        callback_local: will be called whenever a new poses is received from the 
                        Stargazer, with a dict: {marker_id: [xyz, angle]}
        """
        self.device = device
        self.marker_map = marker_map
        self.connection = None

        # chunk_size: how many characters to read from the serial bus in
        # between checking the buffer for the STX/ETX characters
        self._chunk_size = 80

        self._callback_global =  callback_global
        self._callback_local  =  callback_local

        self._stopped = Event()
        self._thread = None

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, type, value, traceback):
        if self.is_connected:
            self.disconnect()

    @property
    def is_connected(self):
        """
        Returns whether the driver is currently connected to a serial port.
        """
        return self.connection is not None

    def connect(self):
        """
        Connect to the StarGazer over the specified RS-232 port.
        """
        assert not self.is_connected
        self.connection = Serial(port=self.device, baudrate=115200, timeout=1.0)

    def disconnect(self):
        """
        Disconnects from the StarGazer and closes the RS-232 port.
        """
        if self.is_streaming:
            self.stop_streaming()

        if self.is_connected:
            self.connection.close()
            self.connection = None

    @property
    def is_streaming(self):
        """
        Returns whether the driver is currently streaming pose data. 
        """
        return self._thread is not None

    def start_streaming(self):
        """
        Begin streaming pose data from the StarGazer.
        """
        assert self.is_connected and not self.is_streaming
        self._send_command('CalcStart')
        self._thread = Thread(target=self._read, args=()).start()

    def stop_streaming(self):
        """
        Stop streaming pose data from the StarGazer.
        """
        assert self.is_connected
        if self.is_streaming:
            self._stopped.set()
            self._thread.join()
            self._send_command('CalcStop')

    def set_parameter(self, name, value):
        """
        Set a StarGazer configuration parameter.

        This function can only be called while the StarGazer is
        connected, but not streaming.

        Arguments
        ---------
        name:  string name of the parameter to set
        value: string value of the parameter to set

        Example
        -------
            set_parameter('MarkType', 'HLD1L')
        """
        assert not self.is_streaming
        self._send_command(name, value)

    def _send_command(self, *args):
        """
        Send a command to the StarGazer.
<commit_msg>Cleaned up a few IO disconnection issues.<commit_after>"""
Driver class for Hagisonic Stargazer, with no ROS dependencies. 
"""
from serial import Serial
from collections import deque
import re
import yaml
import time
import logging
import rospy
import numpy as np
from threading import Thread, Event
from tf import transformations

# STX: char that represents the start of a properly formed message
STX = '~'
# ETX: char that represents the end of a properly formed message
ETX = '`'
# DELIM: char that splits data
DELIM = '|'
# CMD: char that indicates command
CMD = '#'
# CMD: char that indicates command
RESPONSE = '!'
# RESULT: char that indicates that the message contains result data
RESULT = '^'


class StarGazer(object):
    def __init__(self, device, marker_map, callback_global=None, callback_local=None):
        """
        Connect to a Hagisonic StarGazer device and receive poses.

        device:          The device location for the serial connection. 

        marker_map:      dictionary of marker transforms, formatted:
                         {marker_id: (4,4) matrix}

        callback_global: will be called whenever a new pose is received from the
                         Stargazer, will be called with (n,4,4) matrix of poses
                         of the location of the Stargazer in the global frame.
                         These are computed from marker_map. 

        callback_local: will be called whenever a new poses is received from the 
                        Stargazer, with a dict: {marker_id: [xyz, angle]}
        """
        self.device = device
        self.marker_map = marker_map
        self.connection = None

        # chunk_size: how many characters to read from the serial bus in
        # between checking the buffer for the STX/ETX characters
        self._chunk_size = 80

        self._callback_global =  callback_global
        self._callback_local  =  callback_local

        self._stopped = Event()
        self._thread = None

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, type, value, traceback):
        if self.is_connected:
            self.disconnect()

    @property
    def is_connected(self):
        """
        Returns whether the driver is currently connected to a serial port.
        """
        return self.connection is not None

    def connect(self):
        """
        Connect to the StarGazer over the specified RS-232 port.
        """
        assert not self.is_connected
        self.connection = Serial(port=self.device, baudrate=115200, timeout=1.0)

    def disconnect(self):
        """
        Disconnects from the StarGazer and closes the RS-232 port.
        """
        if self.is_connected:
            self.connection.close()
            self.connection = None

    @property
    def is_streaming(self):
        """
        Returns whether the driver is currently streaming pose data. 
        """
        return self._thread is not None

    def start_streaming(self):
        """
        Begin streaming pose data from the StarGazer.
        """
        assert self.is_connected and not self.is_streaming
        self._send_command('CalcStart')
        self._thread = Thread(target=self._read, args=()).start()

    def stop_streaming(self):
        """
        Stop streaming pose data from the StarGazer.
        """
        assert self.is_connected
        if self.is_streaming:
            self._stopped.set()
            self._thread.join()
        self._send_command('CalcStop')

    def set_parameter(self, name, value):
        """
        Set a StarGazer configuration parameter.

        This function can only be called while the StarGazer is
        connected, but not streaming.

        Arguments
        ---------
        name:  string name of the parameter to set
        value: string value of the parameter to set

        Example
        -------
            set_parameter('MarkType', 'HLD1L')
        """
        assert self.is_connected and not self.is_streaming
        self._send_command(name, value)

    def _send_command(self, *args):
        """
        Send a command to the StarGazer.

{'fraction_non_alphanumeric': 0.0750030001200048, 'fraction_numerical': 0.005040201608064322, 'mean_word_length': 2.113186402689578, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3486018', 'n_tokens_mistral': 2265, 'n_tokens_neox': 2115, 'n_words': 795}