fwks/tasks.py
def make_training_task(
        noise=None,
        evaluation_metrics=None,
        evaluation_selection=None,
    ):
    """
    Factory of basic model training tasks
    """
    # TODO: add training using noisy instead of clean
    _evaluation_selection = evaluation_selection


    class AbstractModelTraining(Task):
    
        how_much = 9000
        noise_gen = noise
        epochs = 250
        from_path = "datasets/clarin-long/data"
        metrics = evaluation_metrics or []
        evaluation_selection = _evaluation_selection

        def __new__(self, name, bases, dct):
            this = self
            _metrics = self.metrics
            _evaluation_selection = self.evaluation_selection
            
            @classmethod
            def get_dataset(self):
                dset = dataset.Dataset(noise_gen=this.noise_gen)
                dset.loader_adapter = "clarin"
                dset.get_from(self.from_path)
                return dset

            @classmethod
            def validate(self, cache):
                return os.path.exists(os.path.join(cache, "model.zip"))

            @classmethod
            def run(self, cache):
                try:
                    if not os.path.exists(cache):
                        os.mkdir(cache)
                except:
                    pass
                dset = self.get_dataset()
                dset.select_first(self.how_much)
                am = self.get_acoustic_model()
                am.num_epochs = this.epochs
                am.name = name
                am.build(dset)
                am.summary()
                if self._metrics:
                    metric_obj = metricization.TrainedModelMetricization(am, self._metrics)
                    results = metric_obj.on_dataset(dset, partial=self._evaluation_selection)
                    results.summary()
                am.save(os.path.join(cache, "model.zip"), save_full=True)
                print("=" * 60)
                print("Task done!\n")

            @classmethod
            def summary(self, cache, show=False):
                try:
                    print(cache)
                    am = model.AcousticModel.load(os.path.join(cache, "model.zip"))
                    return am.summary(show=show)
                except FileNotFoundError:
                    print("Cannot find the model archive - aborting")

            new_dct = {"run": run, "validate": validate, "summary": summary, 
                       "how_much": this.how_much, "get_dataset": get_dataset,
                       "_metrics": _metrics, "_evaluation_selection": _evaluation_selection}
            new_dct.update(dct)
            return super().__new__(self, name, bases, new_dct)
        
        @classmethod
        def add_metric(self, metric):
            self.metrics.append(metric)

    metaclass = AbstractModelTraining    
    return metaclass

    Factory of basic model training tasks
    
Factory of basic model training tasks
Factory of basic model training tasks
def make_training_task(
        noise=None,
        evaluation_metrics=None,
        evaluation_selection=None,
    ):
    _evaluation_selection = evaluation_selection
    class AbstractModelTraining(Task):
        how_much = 9000
        noise_gen = noise
        epochs = 250
        from_path = "datasets/clarin-long/data"
        metrics = evaluation_metrics or []
        evaluation_selection = _evaluation_selection
        def __new__(self, name, bases, dct):
            this = self
            _metrics = self.metrics
            _evaluation_selection = self.evaluation_selection
            @classmethod
            def get_dataset(self):
                dset = dataset.Dataset(noise_gen=this.noise_gen)
                dset.loader_adapter = "clarin"
                dset.get_from(self.from_path)
                return dset
            @classmethod
            def validate(self, cache):
                return os.path.exists(os.path.join(cache, "model.zip"))
            @classmethod
            def run(self, cache):
                try:
                    if not os.path.exists(cache):
                        os.mkdir(cache)
                except:
                    pass
                dset = self.get_dataset()
                dset.select_first(self.how_much)
                am = self.get_acoustic_model()
                am.num_epochs = this.epochs
                am.name = name
                am.build(dset)
                am.summary()
                if self._metrics:
                    metric_obj = metricization.TrainedModelMetricization(am, self._metrics)
                    results = metric_obj.on_dataset(dset, partial=self._evaluation_selection)
                    results.summary()
                am.save(os.path.join(cache, "model.zip"), save_full=True)
                print("=" * 60)
                print("Task done!\n")
            @classmethod
            def summary(self, cache, show=False):
                try:
                    print(cache)
                    am = model.AcousticModel.load(os.path.join(cache, "model.zip"))
                    return am.summary(show=show)
                except FileNotFoundError:
                    print("Cannot find the model archive - aborting")
            new_dct = {"run": run, "validate": validate, "summary": summary, 
                       "how_much": this.how_much, "get_dataset": get_dataset,
                       "_metrics": _metrics, "_evaluation_selection": _evaluation_selection}
            new_dct.update(dct)
            return super().__new__(self, name, bases, new_dct)
        @classmethod
        def add_metric(self, metric):
            self.metrics.append(metric)
    metaclass = AbstractModelTraining    
    return metaclass
def make_training_task ( noise = None , evaluation_metrics = None , evaluation_selection = None , ) : _evaluation_selection = evaluation_selection class AbstractModelTraining ( Task ) : how_much = 9000 noise_gen = noise epochs = 250 from_path = "datasets/clarin-long/data" metrics = evaluation_metrics or [ ] evaluation_selection = _evaluation_selection def __new__ ( self , name , bases , dct ) : this = self _metrics = self . metrics _evaluation_selection = self . evaluation_selection @ classmethod def get_dataset ( self ) : dset = dataset . Dataset ( noise_gen = this . noise_gen ) dset . loader_adapter = "clarin" dset . get_from ( self . from_path ) return dset @ classmethod def validate ( self , cache ) : return os . path . exists ( os . path . join ( cache , "model.zip" ) ) @ classmethod def run ( self , cache ) : try : if not os . path . exists ( cache ) : os . mkdir ( cache ) except : pass dset = self . get_dataset ( ) dset . select_first ( self . how_much ) am = self . get_acoustic_model ( ) am . num_epochs = this . epochs am . name = name am . build ( dset ) am . summary ( ) if self . _metrics : metric_obj = metricization . TrainedModelMetricization ( am , self . _metrics ) results = metric_obj . on_dataset ( dset , partial = self . _evaluation_selection ) results . summary ( ) am . save ( os . path . join ( cache , "model.zip" ) , save_full = True ) print ( "=" * 60 ) print ( "Task done!\n" ) @ classmethod def summary ( self , cache , show = False ) : try : print ( cache ) am = model . AcousticModel . load ( os . path . join ( cache , "model.zip" ) ) return am . summary ( show = show ) except FileNotFoundError : print ( "Cannot find the model archive - aborting" ) new_dct = { "run" : run , "validate" : validate , "summary" : summary , "how_much" : this . how_much , "get_dataset" : get_dataset , "_metrics" : _metrics , "_evaluation_selection" : _evaluation_selection } new_dct . update ( dct ) return super ( ) . __new__ ( self , name , bases , new_dct ) @ classmethod def add_metric ( self , metric ) : self . metrics . append ( metric ) metaclass = AbstractModelTraining return metaclass
Factory of basic model training tasks
Factory of basic model training tasks
"""
    Factory of basic model training tasks
    """ # TODO: add training using noisy instead of clean
noise
evaluation_metrics
evaluation_selection