Remove unused or unnecessary code (aws#257)

* remove unused or unnecessary code, also use collectionKeys at one point

* Remove training ended method from trial class

Author: rahul003

tests/core/test_numpy.py

@@ -16,7 +16,7 @@ def rw(path):
             fw.write_tensor(tdata=data, tname=f'foo_{i}')
 
     fr = FileReader(fname=fname)
-    for i,ts in enumerate(fr.read_tensors(read_data=True)):
+    for i,ts in enumerate(fr.read_tensors()):
         """
         read_data returns name, step and data (if read_data==True)
         """
@@ -51,7 +51,7 @@ def test_string():
         fw.write_tensor(tdata=s_written, tname=f'foo_string')
 
     fr = FileReader(fname=fname)
-    read = list(fr.read_tensors(read_data=True))
+    read = list(fr.read_tensors())
     assert len(read)==1
     s_read = np.array(read[0][2])
     assert np.all(s_written == s_read)

tornasole/core/index_reader.py

@@ -97,7 +97,7 @@ def fetch_tensor_value(tensor_location):
             res = [tensor_object]
 
         tr = TensorReader(res[0])  # Access the only element in res
-        tensor_tuple = list(tr.read_tensors(read_data=True))[0]  # Access the only element in the list
+        tensor_tuple = list(tr.read_tensors())[0]  # Access the only element in the list
         tensor_name, step, tensor_data, mode, mode_step = tensor_tuple
         return tensor_data
 

tornasole/core/reader.py

@@ -21,7 +21,7 @@
 
 
 class FileReader():
-    def __init__(self, fname, wtype='tfevent', verbose=True):
+    def __init__(self, fname, wtype='tfevent'):
         """Creates a `FileWriter` and an  file.
         On construction the summary writer creates a new event file in `logdir`.
 
@@ -35,11 +35,9 @@ def __init__(self, fname, wtype='tfevent', verbose=True):
                 How often, in seconds, to flush the pending events and summaries to disk.
             filename_suffix : str
                 Every event file's name is suffixed with `filename_suffix` if provided.
-            verbose : bool
-                Determines whether to print logging messages.
         """
         if wtype == 'tfevent':
-            self._reader = EventFileReader(fname=fname, verbose=verbose)
+            self._reader = EventFileReader(fname=fname)
         else:
             assert False
 
@@ -51,8 +49,8 @@ def __exit__(self, exc_type, exc_value, traceback):
         """Make usable with "with" statement."""
         self._reader.__exit__(exc_type, exc_value, traceback)
 
-    def read_tensors(self, read_data=True, check=False):
-        return self._reader.read_tensors(read_data=read_data, check=check)
+    def read_tensors(self, check=False):
+        return self._reader.read_tensors(check=check)
 
     #def __del__(self):
     #    self._reader.__del__()

tornasole/core/tfevent/event_file_reader.py

@@ -87,7 +87,7 @@ class EventsReader(object):
     """Writes `Event` protocol buffers to an event file. This class is ported from
     EventsReader defined in
     https://github.com/tensorflow/tensorflow/blob/master/tensorflow/core/util/events_writer.cc"""
-    def __init__(self, filename, verbose=True):
+    def __init__(self, filename):
         """
         Events files have a name of the form
         '/file/path/events.out.tfevents.[timestamp].[hostname][file_suffix]'
@@ -116,7 +116,7 @@ class EventFileReader():
     is encoded using the tfrecord format, which is similar to RecordIO.
     """
 
-    def __init__(self, fname, verbose=True):
+    def __init__(self, fname):
         """Creates a `EventFileWriter` and an event file to write to.
         On construction the summary writer creates a new event file in `logdir`.
         This event file will contain `Event` protocol buffers, which are written to
@@ -125,21 +125,18 @@ def __init__(self, fname, verbose=True):
         the event file:
         """
         self._filename = fname
-        self._ev_reader = EventsReader(self._filename, verbose=verbose)
+        self._ev_reader = EventsReader(self._filename)
 
     def __exit__(self,exc_type, exc_value, traceback):
         self._ev_reader.__exit__(exc_type, exc_value, traceback)
 
-    def read_tensors(self, read_data=False, check=False):
+    def read_tensors(self, check=False):
         for step, summ in self.read_summaries(check=check):
             for v in summ.value:
                 assert v.WhichOneof('value') == 'tensor'
                 tensor_name = v.tag
                 # We have found the right tensor at the right step
-                if read_data:
-                    tensor_data = get_tensor_data(v.tensor)
-                else:
-                    tensor_data = None
+                tensor_data = get_tensor_data(v.tensor)
 
                 # default values
                 # todo: validate the logic extensively

tornasole/core/tfevent/event_file_writer.py

@@ -173,7 +173,7 @@ def run(self):
             if isinstance(event_in_queue, EventWithIndex):
                 # checking whether there is an object of IndexArgs,
                 # which is written by write_summary_with_index
-                event = event_in_queue.get_event()
+                event = event_in_queue.event
             else:
                 event = event_in_queue
 
@@ -189,7 +189,7 @@ def run(self):
                     eventfile = self._ev_writer.name()
                     tname = event_in_queue.tensorname
                     mode = event_in_queue.get_mode()
-                    mode_step = event_in_queue.get_mode_step()
+                    mode_step = event_in_queue.mode_step
                     eventfile = get_relative_event_file_path(eventfile)
                     tensorlocation = TensorLocation(
                             tname, mode, mode_step, eventfile,

tornasole/core/tfevent/index_file_writer.py

@@ -64,14 +64,5 @@ def __init__(self, event, tensorname, mode, mode_step):
         self.mode = mode
         self.mode_step = mode_step
 
-    def get_event(self):
-        return self.event
-
-    def get_tensorname(self):
-        return self.tensorname
-
     def get_mode(self):
         return str(self.mode).split('.')[-1]
-
-    def get_mode_step(self):
-        return self.mode_step

tornasole/core/tfrecord/record_reader.py

@@ -62,7 +62,7 @@ def read_record(self, check=True):
         if check:
             computed_len_crc = masked_crc32c(strlen_bytes)
             assert saved_len_crc == computed_len_crc
-        #print( f'Payload_Len={strlen} LENCRC={saved_len_crc},{computed_len_crc}')
+
         payload = self._reader.read(strlen)
         saved_payload_crc = struct.unpack('I', self._reader.read(4))[0]
         if check:
@@ -72,7 +72,6 @@ def read_record(self, check=True):
             computed_payload_crc = masked_crc32c(CHECKSUM_MAGIC_BYTES)
             assert saved_payload_crc == computed_payload_crc
 
-        #print( f'Payload_CRC={saved_payload_crc},{computed_payload_crc}')
         return payload
 
     def flush(self):
@@ -81,7 +80,6 @@ def flush(self):
 
     def close(self):
         """Closes the record reader."""
-        print( "Closing READER", self)
         if self._reader is not None:
             self._reader.close()
             self._reader = None

tornasole/core/tfrecord/tensor_reader.py

@@ -35,16 +35,13 @@ def read_record(self, check=False):
             assert saved_payload_crc == computed_payload_crc
         return payload
 
-    def read_tensors(self, read_data=False, check=False):
+    def read_tensors(self, check=False):
         for (step,summ) in self.read_summaries(check=check):
             for v in summ.value:
                 assert v.WhichOneof('value') == 'tensor'
                 tensor_name = v.tag
                 # We have found the right tensor at the right step
-                if read_data:
-                    tensor_data = get_tensor_data(v.tensor)
-                else:
-                    tensor_data = None
+                tensor_data = get_tensor_data(v.tensor)
 
                 # default values
                 # todo: validate the logic extensively

tornasole/mxnet/mxnet_collection.py

@@ -55,7 +55,7 @@ def get_collection_manager():
     return _collection_manager
 
 def add_to_default_collection(args):
-    add_to_collection('default', args)
+    add_to_collection(CollectionKeys.DEFAULT, args)
 
 def get_collection(collection_name):
     return _collection_manager.get(collection_name, create=True)

tornasole/pytorch/torch_collection.py

@@ -56,7 +56,7 @@ def get_collection_manager():
     return _collection_manager
 
 def add_to_default_collection(args):
-    add_to_collection('default', args)
+    add_to_collection(CollectionKeys.DEFAULT, args)
 
 def get_collection(collection_name):
     return _collection_manager.get(collection_name, create=True)

tornasole/trials/local_trial.py

@@ -5,7 +5,6 @@
 from tornasole.core.collection_manager import CollectionManager, \
     COLLECTIONS_FILE_NAME
 from tornasole.core.reader import FileReader
-from tornasole.core.access_layer.utils import has_training_ended
 
 import time
 import os
@@ -75,15 +74,6 @@ def _load_collections(self):
                     self.logger.debug('Waiting to read collections')
                 continue
 
-    def training_ended(self):
-        return has_training_ended(self.trial_dir)
-
-    def __hash__(self):
-        return hash((self.name, self.trial_dir))
-
-    def __eq__(self, other):
-        return (self.name, self.trial_dir) == (other.name, other.trial_dir)
-
     def get_tensors(self, tname_steps_dict, should_regex_match=False):
         # now we do not need to do anything since we read the full event file
         pass
@@ -101,7 +91,7 @@ def _read_step_dirs(self, step_dirs):
                 dirnames_efts = Parallel(n_jobs=multiprocessing.cpu_count(), verbose=0) \
                     (delayed(self._read_folder) \
                          (EventFileLocation.get_step_dir_path(self.trial_dir, step_dir),
-                          read_data=self.read_data, check=self.check) \
+                          check=self.check) \
                      for step_dir in step_dirs)
                 # sort them as parallel returns in random order
                 # we want to sort them by dirname
@@ -114,7 +104,6 @@ def _read_step_dirs(self, step_dirs):
             for step_dir in step_dirs:
                 step_dir_path = EventFileLocation.get_step_dir_path(self.trial_dir, step_dir)
                 dirnames_efts.extend(self._read_folder(step_dir_path,
-                                                       read_data=self.read_data,
                                                        check=self.check))
 
         for dirname, dir_efts in dirnames_efts:
@@ -133,13 +122,13 @@ def _read_step_dirs(self, step_dirs):
             # temp file for writing event files and do atomic move
 
     @staticmethod
-    def _read_folder(dirname, read_data=True, check=True):
+    def _read_folder(dirname, check=True):
         res = []
         for fname in os.listdir(dirname):
             if fname.endswith(".tfevents"):
                 full_fname = os.path.join(dirname, fname)
                 fr = FileReader(fname=full_fname)
-                summary_values = fr.read_tensors(read_data=read_data, check=check)
+                summary_values = fr.read_tensors(check=check)
                 for sv in summary_values:
                     n, s, d, mode, mode_step = sv
                     eft = EventFileTensor(fname, tensor_name=n, step_num=s, tensor_value=d,

tornasole/trials/s3_trial.py

@@ -2,8 +2,6 @@
 import os
 
 from tornasole.core.access_layer.s3handler import ReadObjectRequest, S3Handler
-from tornasole.core.access_layer.utils import has_training_ended
-from tornasole.core.locations import EventFileLocation
 from tornasole.core.s3_utils import list_s3_objects
 from tornasole.core.locations import EventFileLocation
 from tornasole.core.collection_manager import CollectionManager, \
@@ -33,7 +31,7 @@ def __init__(self, name, bucket_name, prefix_name,
         self.logger.info(f'Loading trial {name} at path s3://{bucket_name}/{prefix_name}')
         self.bucket_name = bucket_name
         self.prefix_name = os.path.join(prefix_name, '')
-        self.path = "s3://"+os.path.join(self.bucket_name, self.prefix_name)
+        self.path = "s3://" + os.path.join(self.bucket_name, self.prefix_name)
         self.s3_handler = S3Handler()
         self._load_collections()
         self.load_tensors()
@@ -44,9 +42,6 @@ def _load_tensors_from_index_tensors(self, index_tensors_dict):
             for step in steps:
                 self.add_tensor(step, index_tensors_dict[tname][step]['tensor_location'])
 
-    def training_ended(self):
-        return has_training_ended("s3://{}/{}".format(self.bucket_name, self.prefix_name))
-
     def _load_collections(self):
         num_times_before_warning = 10
         while True:
@@ -67,13 +62,6 @@ def _load_collections(self):
             self.logger.debug('Loaded collections for trial {}'.format(self.name))
             return
 
-    def __hash__(self):
-        return hash((self.name, self.bucket_name, self.prefix_name))
-
-    def __eq__(self, other):
-        return (self.name, self.bucket_name, self.prefix_name) \
-               == (other.name, other.bucket_name, other.prefix_name)
-
     def get_tensors(self, tname_steps_dict, should_regex_match=False):
         # to be used when getting selective tensors from S3
         # now we do not need to do anything since we read the full event file from S3
@@ -134,5 +122,5 @@ def _get_s3_location(self, obj):
 
     def _read_tensors_from_data(self, data):
         tr = TensorReader(data)
-        res = tr.read_tensors(read_data=self.read_data, check=self.check)
+        res = tr.read_tensors(check=self.check)
         return list(res)

tornasole/trials/trial.py

@@ -3,6 +3,7 @@
 from bisect import bisect_left
 from abc import ABC, abstractmethod
 
+from tornasole.core.access_layer.utils import has_training_ended
 from tornasole.core.tensor import Tensor, StepState
 from tornasole.exceptions import *
 from tornasole.analysis.utils import refresh
@@ -40,13 +41,13 @@ class Trial(ABC):
         _index_tensors_dict
 
     ['name', '_tensors', '_mode_to_global', '_global_to_mode', 'logger', 'parallel',
-    'read_data', 'check', 'range_steps', 'collection_manager', 'loaded_all_steps', 'cache', 'path',
+    'check', 'range_steps', 'collection_manager', 'loaded_all_steps', 'cache', 'path',
     'index_tensors_dict', 'index_mode', 'last_event_token', 'last_index_token', 'index_reader',
     'dynamic_refresh', 'trial_dir']
     """
 
     def __init__(self, name, range_steps=None, parallel=True,
-                 read_data=True, check=False, index_mode=True, cache=False):
+                 check=False, index_mode=True, cache=False):
         self.name = name
         self._tensors = {}
 
@@ -60,7 +61,6 @@ def __init__(self, name, range_steps=None, parallel=True,
 
         self.logger = get_logger()
         self.parallel = parallel
-        self.read_data = read_data
         self.check = check
         self.range_steps = range_steps
         self.collection_manager = None
@@ -97,19 +97,21 @@ def _load_collections(self):
     def _load_tensors_from_index_tensors(self, index_tensors_dict):
         pass
 
-    @abstractmethod
-    def training_ended(self):
-        pass
-
     @abstractmethod
     def _load_tensors_from_event_files(self, start_after_key=None):
         pass
 
+    def __hash__(self):
+        return hash((self.name, self.path))
+
+    def __eq__(self, other):
+        return (self.name, self.path) == (other.name, other.path)
+
     def maybe_refresh(self, name=None):
         if self.loaded_all_steps or not self.dynamic_refresh:
             return
         retry_count = 1
-        training_ended = self.training_ended()
+        training_ended = has_training_ended(self.path)
         if training_ended and self.loaded_all_steps== False:
             retry_count = 2
         while retry_count > 0:

tornasole/xgboost/collection.py

@@ -44,7 +44,7 @@ def get_collection_manager():
 
 def add_to_default_collection(args):
 
-    add_to_collection('default', args)
+    add_to_collection(CollectionKeys.DEFAULT, args)
 
 
 def get_collection(collection_name):